diff --git a/safekeeper/src/safekeeper.rs b/safekeeper/src/safekeeper.rs index 0814d9ba67..9d103887ae 100644 --- a/safekeeper/src/safekeeper.rs +++ b/safekeeper/src/safekeeper.rs @@ -875,6 +875,29 @@ where return Ok(Some(AcceptorProposerMessage::AppendResponse(resp))); } + // Disallow any non-sequential writes, which can result in gaps or + // overwrites. If we need to move the pointer, ProposerElected message + // should have truncated WAL first accordingly. Note that the first + // condition (WAL rewrite) is quite expected in real world; it happens + // when walproposer reconnects to safekeeper and writes some more data + // while first connection still gets some packets later. It might be + // better to not log this as error! above. + let write_lsn = self.wal_store.write_lsn(); + if write_lsn > msg.h.begin_lsn { + bail!( + "append request rewrites WAL written before, write_lsn={}, msg lsn={}", + write_lsn, + msg.h.begin_lsn + ); + } + if write_lsn < msg.h.begin_lsn && write_lsn != Lsn(0) { + bail!( + "append request creates gap in written WAL, write_lsn={}, msg lsn={}", + write_lsn, + msg.h.begin_lsn, + ); + } + // Now we know that we are in the same term as the proposer, // processing the message. @@ -1005,6 +1028,10 @@ mod tests { #[async_trait::async_trait] impl wal_storage::Storage for DummyWalStore { + fn write_lsn(&self) -> Lsn { + self.lsn + } + fn flush_lsn(&self) -> Lsn { self.lsn } diff --git a/safekeeper/src/wal_storage.rs b/safekeeper/src/wal_storage.rs index ded8571a3e..5136bdb9d8 100644 --- a/safekeeper/src/wal_storage.rs +++ b/safekeeper/src/wal_storage.rs @@ -37,6 +37,8 @@ use utils::{id::TenantTimelineId, lsn::Lsn}; #[async_trait::async_trait] pub trait Storage { + // Last written LSN. + fn write_lsn(&self) -> Lsn; /// LSN of last durably stored WAL record. fn flush_lsn(&self) -> Lsn; @@ -327,6 +329,10 @@ impl PhysicalStorage { #[async_trait::async_trait] impl Storage for PhysicalStorage { + // Last written LSN. + fn write_lsn(&self) -> Lsn { + self.write_lsn + } /// flush_lsn returns LSN of last durably stored WAL record. fn flush_lsn(&self) -> Lsn { self.flush_record_lsn diff --git a/safekeeper/tests/walproposer_sim/safekeeper_disk.rs b/safekeeper/tests/walproposer_sim/safekeeper_disk.rs index c2db9de78a..be56e86562 100644 --- a/safekeeper/tests/walproposer_sim/safekeeper_disk.rs +++ b/safekeeper/tests/walproposer_sim/safekeeper_disk.rs @@ -177,6 +177,10 @@ impl DiskWALStorage { #[async_trait::async_trait] impl wal_storage::Storage for DiskWALStorage { + // Last written LSN. + fn write_lsn(&self) -> Lsn { + self.write_lsn + } /// LSN of last durably stored WAL record. fn flush_lsn(&self) -> Lsn { self.flush_record_lsn