This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch fix/partition-recovery-and-old-requests in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 935a6d5f3e6100c824dd230366a147700c394ce5 Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Sep 30 14:01:42 2026 +0200 fix(consensus): preserve recovery floors and reject aged-out requests Recovery below the creation view could install an empty primary behind acknowledged writes. Aged-out IDs could receive false success. Use the creation view as a floor without certifying the log. Preserve certified WAL history and reject StartView below committed prefixes. Return RequestTooOld for aged-out IDs and stop producer retries: the outcome is unknown. Keep applicable SDK error mappings aligned. --- core/common/src/error/iggy_error.rs | 19 + core/consensus/src/client_table.rs | 152 ++++--- core/consensus/src/impls.rs | 186 ++++---- core/integration/tests/cluster/partition_dedup.rs | 8 + core/partitions/src/iggy_partition.rs | 164 ++++--- core/sdk/src/clients/producer.rs | 141 ++++-- core/sdk/src/vsr.rs | 28 +- core/server/README.md | 30 ++ core/server/src/partition_helpers.rs | 494 ++++++++++++++++++--- core/server/src/server_error.rs | 19 + core/shard/src/lib.rs | 9 +- core/simulator/src/lib.rs | 189 +++++++- foreign/go/errors/errors.yaml | 4 + foreign/go/errors/errors_gen.go | 17 + foreign/go/errors/errors_test.go | 6 + .../org/apache/iggy/exception/IggyErrorCode.java | 1 + .../apache/iggy/exception/IggyErrorCodeTest.java | 1 + foreign/node/src/wire/error.code.test.ts | 8 + foreign/node/src/wire/error.code.ts | 1 + .../swift/Sources/Iggy/Errors/IggyErrorCode.swift | 2 + foreign/swift/Tests/IggyTests/Fixtures/golden.json | 1 + 21 files changed, 1131 insertions(+), 349 deletions(-) diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs index 6a31bcb48..a3f48d57d 100644 --- a/core/common/src/error/iggy_error.rs +++ b/core/common/src/error/iggy_error.rs @@ -173,6 +173,10 @@ pub enum IggyError { InvalidBooleanValue = 83, #[error("Invalid number value")] InvalidNumberValue = 84, + #[error( + "Request is outside the deduplication window; outcome unknown, do not send a new request" + )] + RequestTooOld = 85, #[error("Client with ID: {0} was not found.")] ClientNotFound(u32) = 100, #[error("Invalid client ID")] @@ -637,4 +641,19 @@ mod tests { assert_eq!(IggyError::from_code(3024), error); assert_eq!(IggyError::from_code_as_string(3024), error.as_string()); } + + #[test] + fn aged_out_request_code_does_not_reuse_retired_encryption_error() { + const REQUEST_TOO_OLD_CODE: u32 = 85; + const RETIRED_ENCRYPTION_KEY_CODE: u32 = 60; + assert_eq!(IggyError::RequestTooOld.as_code(), REQUEST_TOO_OLD_CODE); + assert_eq!( + IggyError::from_code(REQUEST_TOO_OLD_CODE), + IggyError::RequestTooOld, + ); + assert_eq!( + IggyError::from_code(RETIRED_ENCRYPTION_KEY_CODE), + IggyError::Error, + ); + } } diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index c72ef8e34..bf5c28d03 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -18,6 +18,7 @@ use crate::le_cursor::{LeCursor, Truncated, split_verified_trailer}; use iggy_binary_protocol::consensus::ConsensusError; use iggy_binary_protocol::{GenericHeader, ReplyHeader}; +use iggy_common::IggyError; use serde::{Deserialize, Serialize}; use server_common::{ MESSAGE_ALIGN, Message, @@ -251,9 +252,9 @@ struct ClientEntry { /// its dedup history. watermark: u64, /// Partition-slice only: bit `i` set means request `watermark - i` has - /// committed, bit 0 being the watermark itself. A request below the - /// watermark with its bit clear is a reordered arrival still to execute, - /// not a duplicate; below the window everything reads as committed. Zero + /// committed, bit 0 being the watermark itself. A request in the retained + /// window with its bit clear is a reordered arrival still to execute, + /// not a duplicate; below the window the outcome is unknown and refused. Zero /// on the metadata plane, whose reply ring plays this role. committed_window: u128, /// `request_checksum` of the watermark request; catches a client reusing @@ -603,17 +604,17 @@ pub struct DedupWatermark { /// replay it after later ids have committed, so "at or below the watermark" /// alone would absorb that replay as a duplicate and lose the write. The window /// records which ids under the watermark actually committed; an unmarked one -/// inside it executes, while one that has aged out below it reads as committed -/// and is absorbed with the operation's empty success. +/// inside it executes, while one that has aged out below it returns +/// [`IggyError::RequestTooOld`] because its outcome is unknown. /// /// The width is in the CLIENT's request-id space, not in this group's writes: /// `ConsensusSession` mints from one counter across every partition, stream and /// metadata op, so a slice only ever sees the subset of those ids routed to it. /// Coverage in a client's own writes to one group is this width divided by the /// number of groups it interleaves, so 128 ids is around 16 writes per group -/// across 8 partitions, and a replay held back longer than that is absorbed and -/// lost. A wider bitmap divides by the same fanout: closing the gap needs -/// per-group request numbering, which waits on the clients-table follow-up. +/// across 8 partitions, and a replay held back longer than that can be refused +/// with an unknown outcome. A wider bitmap divides by the same fanout: closing +/// the gap needs per-group request numbering, which waits on the clients-table follow-up. pub const COMMITTED_WINDOW_BITS: u64 = 128; /// VSR client table: per-session fence epoch + request-watermark dedup. @@ -1210,30 +1211,39 @@ impl ClientTable { /// Watermark-plus-window dedup check for a plane that mints no epoch. /// - /// `true` means `user_id` already committed this request under `client_id`, - /// so it must be answered rather than executed again: it is the watermark, - /// a marked id inside the [`COMMITTED_WINDOW_BITS`] window below it, or - /// anything older than the window. An unmarked id inside the window is a - /// reordered arrival (a transiently refused write replayed after its - /// successors committed) and reads as new. An entry another user left under - /// the same id is not evidence about this caller: the id alone is not an - /// identity (see [`DedupWatermark::user_id`]), so the request reads as new - /// and its commit resets the entry. + /// Only marked IDs in the retained window prove a committed request. + /// Unmarked IDs remain admissible within that window. Another user's + /// entry is not evidence about this caller and reads as new. + /// + /// # Errors + /// Returns [`IggyError::RequestTooOld`] outside the retained window: + /// the outcome is unknown, so neither success nor execution is safe. /// /// # Panics - /// If called on a table that fences epochs -- that plane must go through - /// [`Self::check_request`], which enforces the fence. - #[must_use] - pub fn is_duplicate(&self, client_id: u128, user_id: u32, request: u64) -> bool { + /// If called on a table that fences epochs, which must use + /// [`Self::check_request`]. + pub fn is_duplicate( + &self, + client_id: u128, + user_id: u32, + request: u64, + ) -> Result<bool, IggyError> { debug_assert!( !self.mode.fence_epoch(), "is_duplicate: an epoch-fencing table must use check_request" ); let Some(&slot_idx) = self.index.get(&client_id) else { - return false; + return Ok(false); }; let entry = self.slots[slot_idx].as_ref().expect("index/slot mismatch"); - entry.user_id == user_id && request <= entry.watermark && entry.window_has(request) + if entry.user_id != user_id || request > entry.watermark { + return Ok(false); + } + let below = entry.watermark - request; + if below >= COMMITTED_WINDOW_BITS { + return Err(IggyError::RequestTooOld); + } + Ok(entry.committed_window & (1 << below) != 0) } /// Record a committed request without a reply to cache. @@ -1241,16 +1251,16 @@ impl ClientTable { /// The entry point for a plane that runs /// [`ClientTableMode::PartitionSlice`]: there is no register to create the /// entry, so the first committed request creates it, and there is no result - /// section worth retaining, so a later duplicate answers - /// [`RequestStatus::AlreadyApplied`] and the caller synthesizes the reply. + /// section worth retaining. [`Self::is_duplicate`] proves retained + /// duplicates and the caller synthesizes their reply. /// /// Idempotent and order-insensitive for one user: the watermark only rises /// and the window only gains bits, so replaying an already-folded op is a /// no-op and a state-transfer install followed by a re-walk of the same /// commits converges. A commit above the watermark shifts the window up by - /// the gap (ids that age out read as committed from then on); one below it - /// sets its bit. A commit by a DIFFERENT user under the same client id - /// replaces the entry outright: nothing observes a logout here, so this is + /// the gap (ids that age out are refused with [`IggyError::RequestTooOld`]); + /// one below it sets its bit. A commit by a DIFFERENT user under the same + /// client id replaces the entry outright: nothing observes a logout here, so this is /// what stops the next holder of a re-minted id from having its first /// writes absorbed by the previous holder's watermark. /// @@ -2118,14 +2128,6 @@ impl ClientEntry { } } - /// Whether `request` (at or below the watermark) is inside the window and - /// marked committed, or below the window entirely. Callers check - /// `request <= watermark` first. - const fn window_has(&self, request: u64) -> bool { - let below = self.watermark - request; - below >= COMMITTED_WINDOW_BITS || self.committed_window & (1 << below) != 0 - } - /// Latest committed reply (register or app op). /// /// # Panics @@ -3508,7 +3510,7 @@ mod tests { let table = slice(4096); assert_eq!(table.count(), 0); assert_eq!(table.slots.len(), 0, "slots must grow on demand"); - assert!(!table.is_duplicate(7, SLICE_USER, 1)); + assert!(!table.is_duplicate(7, SLICE_USER, 1).unwrap()); } #[test] @@ -3518,9 +3520,9 @@ mod tests { let mut table = slice(4); table.commit_request(7, SLICE_USER, 5, 100); - assert!(table.is_duplicate(7, SLICE_USER, 5)); - assert!(!table.is_duplicate(7, SLICE_USER, 4)); - assert!(!table.is_duplicate(7, SLICE_USER, 6)); + assert!(table.is_duplicate(7, SLICE_USER, 5).unwrap()); + assert!(!table.is_duplicate(7, SLICE_USER, 4).unwrap()); + assert!(!table.is_duplicate(7, SLICE_USER, 6).unwrap()); } #[test] @@ -3532,10 +3534,10 @@ mod tests { table.commit_request(7, SLICE_USER, 5, 100); table.commit_request(7, SLICE_USER, 9, 101); - assert!(table.is_duplicate(7, SLICE_USER, 5)); - assert!(table.is_duplicate(7, SLICE_USER, 9)); - assert!(!table.is_duplicate(7, SLICE_USER, 7)); - assert!(!table.is_duplicate(7, SLICE_USER, 10)); + assert!(table.is_duplicate(7, SLICE_USER, 5).unwrap()); + assert!(table.is_duplicate(7, SLICE_USER, 9).unwrap()); + assert!(!table.is_duplicate(7, SLICE_USER, 7).unwrap()); + assert!(!table.is_duplicate(7, SLICE_USER, 10).unwrap()); } #[test] @@ -3557,13 +3559,13 @@ mod tests { table.commit_request(7, SLICE_USER, 1, 100); table.commit_request(7, SLICE_USER, 3, 101); - assert!(!table.is_duplicate(7, SLICE_USER, 2)); + assert!(!table.is_duplicate(7, SLICE_USER, 2).unwrap()); table.commit_request(7, SLICE_USER, 2, 102); - assert!(table.is_duplicate(7, SLICE_USER, 2)); - assert!(table.is_duplicate(7, SLICE_USER, 1)); - assert!(table.is_duplicate(7, SLICE_USER, 3)); - assert!(!table.is_duplicate(7, SLICE_USER, 4)); + assert!(table.is_duplicate(7, SLICE_USER, 2).unwrap()); + assert!(table.is_duplicate(7, SLICE_USER, 1).unwrap()); + assert!(table.is_duplicate(7, SLICE_USER, 3).unwrap()); + assert!(!table.is_duplicate(7, SLICE_USER, 4).unwrap()); assert_eq!( table.watermarks_sorted(), vec![DedupWatermark { @@ -3577,18 +3579,24 @@ mod tests { } #[test] - fn given_partition_slice_when_id_ages_out_of_window_should_read_as_committed() { - // Below the window nothing is tracked, so the pre-window rule applies: - // absorbed. Inside it, an unmarked id stays admissible however the - // watermark moved. + fn given_partition_slice_when_id_ages_out_of_window_should_report_unknown_outcome() { let mut table = slice(4); table.commit_request(7, SLICE_USER, 1, 100); table.commit_request(7, SLICE_USER, 1 + COMMITTED_WINDOW_BITS + 10, 101); - assert!(table.is_duplicate(7, SLICE_USER, 1)); - assert!(!table.is_duplicate(7, SLICE_USER, 1 + COMMITTED_WINDOW_BITS)); - assert!(!table.is_duplicate(7, SLICE_USER, 12)); - assert!(table.is_duplicate(7, SLICE_USER, 11)); + for request in [1, 2, 11] { + assert_eq!( + table.is_duplicate(7, SLICE_USER, request), + Err(IggyError::RequestTooOld), + "aged-out request {request} must have an unknown outcome" + ); + } + assert!(!table.is_duplicate(7, SLICE_USER, 12).unwrap()); + assert!( + !table + .is_duplicate(7, SLICE_USER, 1 + COMMITTED_WINDOW_BITS) + .unwrap() + ); } #[test] @@ -3600,9 +3608,9 @@ mod tests { // 5 (bit 0), 2 (bit 3), 1 (bit 4) committed; 3 and 4 did not. assert_eq!(table.watermarks_sorted()[0].committed_window, 0b11001); - assert!(!table.is_duplicate(7, SLICE_USER, 3)); - assert!(!table.is_duplicate(7, SLICE_USER, 4)); - assert!(table.is_duplicate(7, SLICE_USER, 2)); + assert!(!table.is_duplicate(7, SLICE_USER, 3).unwrap()); + assert!(!table.is_duplicate(7, SLICE_USER, 4).unwrap()); + assert!(table.is_duplicate(7, SLICE_USER, 2).unwrap()); } #[test] @@ -3612,13 +3620,13 @@ mod tests { let mut table = slice(4); table.commit_request(7, SLICE_USER, u64::MAX, 100); - assert!(!table.is_duplicate(7, OTHER_USER, 1)); + assert!(!table.is_duplicate(7, OTHER_USER, 1).unwrap()); table.commit_request(7, OTHER_USER, 1, 101); - assert!(table.is_duplicate(7, OTHER_USER, 1)); - assert!(!table.is_duplicate(7, OTHER_USER, 2)); + assert!(table.is_duplicate(7, OTHER_USER, 1).unwrap()); + assert!(!table.is_duplicate(7, OTHER_USER, 2).unwrap()); assert!( - !table.is_duplicate(7, SLICE_USER, 5), + !table.is_duplicate(7, SLICE_USER, 5).unwrap(), "the previous holder's history is gone with the reset" ); assert_eq!(table.count(), 1, "a reset reuses the slot"); @@ -3647,7 +3655,7 @@ mod tests { table.commit_request(1, SLICE_USER, 5, 10); table.commit_request(2, SLICE_USER, 1, 20); - assert!(!table.is_duplicate(1, SLICE_USER, 5)); + assert!(!table.is_duplicate(1, SLICE_USER, 5).unwrap()); } #[test] @@ -3684,11 +3692,11 @@ mod tests { assert_eq!(table.count(), 2); assert!( - !table.is_duplicate(9, SLICE_USER, 3), + !table.is_duplicate(9, SLICE_USER, 3).unwrap(), "install replaces rather than merges" ); - assert!(table.is_duplicate(1, SLICE_USER, 4)); - assert!(!table.is_duplicate(2, SLICE_USER, 8)); + assert!(table.is_duplicate(1, SLICE_USER, 4).unwrap()); + assert!(!table.is_duplicate(2, SLICE_USER, 8).unwrap()); } #[test] @@ -3717,8 +3725,8 @@ mod tests { committed_window: 1, }]); - assert!(table.is_duplicate(1, OTHER_USER, 4)); - assert!(!table.is_duplicate(1, SLICE_USER, 4)); + assert!(table.is_duplicate(1, OTHER_USER, 4).unwrap()); + assert!(!table.is_duplicate(1, SLICE_USER, 4).unwrap()); } #[test] @@ -3738,7 +3746,7 @@ mod tests { table.install_watermarks(std::iter::empty()); assert_eq!(table.count(), 0); - assert!(!table.is_duplicate(7, SLICE_USER, 5)); + assert!(!table.is_duplicate(7, SLICE_USER, 5).unwrap()); } #[test] @@ -3756,7 +3764,7 @@ mod tests { #[test] #[should_panic(expected = "an epoch-fencing table must use check_request")] fn given_metadata_table_when_is_duplicate_called_should_panic() { - let _ = ClientTable::new(4).is_duplicate(7, SLICE_USER, 1); + let _ = ClientTable::new(4).is_duplicate(7, SLICE_USER, 1).unwrap(); } #[cfg(debug_assertions)] diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index e082a26a8..47f288549 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -1262,8 +1262,8 @@ pub struct VsrRestore<'a> { pub timers: &'a ConsensusTimers, /// `(view, log_view)` read back from the group's durable superblock. pub durable_view: Option<(u32, u32)>, - /// View inferred from the last journaled prepare, consulted only when no - /// durable record exists; `log_view` cannot be inferred and stays 0. + /// Known view floor from journaled prepares or committed creation metadata, + /// used without a durable record. It does not certify `log_view`. pub view_fallback: Option<u32>, /// Starting `(view, log_view)` for a group with NO history of its own, /// consulted only when neither of the two above applies: the view the @@ -1295,6 +1295,8 @@ pub struct VsrRestore<'a> { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct FreshGroupStart { pub join: JoinMode, + /// Recovery floor without log certification; see [`VsrRestore::view_fallback`]. + pub view_fallback: Option<u32>, /// Starting view for a group with no durable record; see /// [`VsrRestore::seed_view`]. Always `None` when a durable record exists, /// so a caller can apply the two in either order. @@ -1329,15 +1331,16 @@ pub const fn fresh_group_start( } else { JoinMode::Init }; - // A durable record outranks the seed, and a probing backup takes neither: - // it must sit at or below the group's real view for the primary's - // `StartView` to move it forward, and seeded above that the reply reads as - // stale and is dropped. - let seed_view = match (durable_view, join) { - (None, JoinMode::Init) => Some(created_view), - _ => None, + let (view_fallback, seed_view) = match (durable_view, join) { + (None, JoinMode::Init) => (None, Some(created_view)), + (None, JoinMode::ProbeAsBackup { .. }) => (Some(created_view), None), + (Some(_), _) => (None, None), }; - FreshGroupStart { join, seed_view } + FreshGroupStart { + join, + view_fallback, + seed_view, + } } impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { @@ -3400,29 +3403,20 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { return Vec::new(); } - // Skip an equal-view StartView whose op is below our COMMITTED floor. Such a - // message can only be stale: a live primary's head covers every op it ever - // told us was committed. Already in this view, so re-running - // reset_view_change_state for one would cancel subscribers (waking register - // awaiters Canceled) and clear the pipeline for nothing. log_view (not - // self.view) tracks the last-normal view. - // - // The bound is deliberately the commit floor and NOT the sequencer head. - // Adopting a StartView drops the head (below) without truncating the WAL, so - // a replica that adopted at head H and then restarted re-derives a LONGER - // head from its own journal while log_view stays at that same view. Skipping - // on the head would drop the primary's StartView there -- including the reply - // echoing this replica's own probe -- leaving it to time the probe out and - // elect instead, with a DoViewChange of (log_view, discarded_head) that - // outranks the real primary's and pushes ops that view already discarded back - // over committed bodies. - // - // Re-adoption drops the head again while the WAL still holds the discarded - // suffix. Consensus is sans-io and cannot truncate it; the plane sweeps it - // on every adoption (`reconcile_{metadata,partition}_view_divergence`, - // above-head branch) before the primary's next prepare can collide with a - // relic in `append`'s slot-collision check. - if msg_view == self.log_view.get() && msg_op < self.commit_min() { + // A new view may discard an uncommitted WAL suffix, never a committed prefix. + if msg_op < self.commit_max() { + if msg_view > self.view.get() { + tracing::warn!( + group = self.group, + replica = self.replica, + current_view = self.view.get(), + view = msg_view, + op = msg_op, + commit_min = self.commit_min(), + commit_max = self.commit_max(), + "ignoring StartView that omits known committed entries" + ); + } return Vec::new(); } @@ -4261,6 +4255,7 @@ mod fresh_group_start_tests { let start = fresh_group_start(false, None, CREATED_VIEW); assert_eq!(start.join, JoinMode::Init); assert_eq!(start.seed_view, Some(CREATED_VIEW)); + assert_eq!(start.view_fallback, None); } /// A durable record outranks the seed: it is what this replica actually @@ -4269,16 +4264,18 @@ mod fresh_group_start_tests { fn given_a_durable_record_when_seeding_should_prefer_the_record() { let start = fresh_group_start(false, Some((7, 7)), CREATED_VIEW); assert_eq!(start.seed_view, None); + assert_eq!(start.view_fallback, None); } /// A probing backup must sit at or below the group's real view for the /// primary's `StartView` to move it forward. Seeded above it, the reply /// reads as stale and the replica never rejoins. #[test] - fn given_a_prior_life_when_seeding_should_probe_without_a_seed() { + fn given_a_prior_life_when_seeding_should_probe_at_the_creation_floor() { let start = fresh_group_start(true, None, CREATED_VIEW); assert!(matches!(start.join, JoinMode::ProbeAsBackup { .. })); assert_eq!(start.seed_view, None); + assert_eq!(start.view_fallback, Some(CREATED_VIEW)); } } @@ -4881,70 +4878,63 @@ mod timestamp_clamp_tests { } #[test] - fn given_restored_log_view_when_start_view_head_behind_wal_should_adopt() { - // A replica that adopted a StartView at head 105 in view 7, then crashed, - // recovers log_view = 7 from the superblock but re-derives head 120 from its - // own WAL: adoption drops the head without truncating the journal. The - // primary's StartView for view 7 then carries an op BEHIND that head, and - // skipping it on the head comparison would leave this replica probing until it - // elected instead, carrying a DoViewChange of (7, 120) that outranks the real - // primary's (7, 105) and resurrects ops view 7 already discarded. - // - // Replica 0 of 3 at view 7, whose primary is replica 1 (7 % 3). Incarnation - // left at 0 so the recovering-replica guard stays inert and this exercises the - // equal-view path alone. - let mut consensus = VsrConsensus::with_clock( - 1, - 0, - 3, - METADATA_GROUP, - NoopBus, - LocalPipeline::new(), - ConsensusClock::system(), - ); - consensus.set_view(7); - consensus.set_log_view(7); - consensus.restore_commit_state(105, 105); - consensus.sequencer().set_sequence(120); - - // Below the committed floor: stale by construction, since a live primary's - // head covers every op it told us was committed. - assert!( - consensus - .handle_start_view( - PlaneKind::Metadata, - make_start_view(7, 104, 104, 1, 0).header(), - &[] - ) - .is_empty(), - "an equal-view StartView below the commit floor must be skipped" - ); - assert_eq!( - consensus.sequencer().current_sequence(), - 120, - "a skipped StartView must not move the head" - ); - - // At the committed floor but behind our WAL head: the primary's real head. - // Adopt it and drop the discarded suffix. - assert!( - !consensus - .handle_start_view( - PlaneKind::Metadata, - make_start_view(7, 105, 105, 1, 0).header(), - &[] - ) - .is_empty(), - "an equal-view StartView at or above the commit floor must be adopted, \ - even when its head is behind a WAL suffix the view already discarded" - ); - assert_eq!( - consensus.sequencer().current_sequence(), - 105, - "adoption must drop the head to the primary's, so a later DoViewChange \ - cannot outrank it with a discarded suffix" - ); - assert_eq!(consensus.status(), Status::Normal); + fn given_committed_prefix_when_start_view_head_behind_wal_should_preserve_commits() { + const RESTORED_VIEW: u32 = 7; + const CHECKPOINT: u64 = 100; + const COMMIT: u64 = 105; + const WAL_HEAD: u64 = 120; + for checkpoint in [COMMIT, CHECKPOINT] { + for view in [RESTORED_VIEW, RESTORED_VIEW + 1] { + let mut consensus = VsrConsensus::with_clock( + 1, + 0, + 3, + METADATA_GROUP, + NoopBus, + LocalPipeline::new(), + ConsensusClock::system(), + ); + consensus.set_view(RESTORED_VIEW); + consensus.set_log_view(RESTORED_VIEW); + consensus.restore_commit_state(checkpoint, COMMIT); + consensus.sequencer().set_sequence(WAL_HEAD); + consensus.init_as_backup(); + consensus.begin_view_probe(); + let primary = u8::try_from(view % 3).unwrap(); + + assert!( + consensus + .handle_start_view( + PlaneKind::Metadata, + make_start_view(view, COMMIT - 1, COMMIT - 1, primary, 0).header(), + &[], + ) + .is_empty(), + "StartView in view {view} must preserve commit {COMMIT} above checkpoint {checkpoint}", + ); + assert_eq!(consensus.view(), RESTORED_VIEW); + assert_eq!(consensus.sequencer().current_sequence(), WAL_HEAD); + assert_eq!(consensus.commit_min(), checkpoint); + assert_eq!(consensus.commit_max(), COMMIT); + assert_eq!(consensus.status(), Status::Recovering); + + assert!( + !consensus + .handle_start_view( + PlaneKind::Metadata, + make_start_view(view, COMMIT, COMMIT, primary, 0).header(), + &[], + ) + .is_empty(), + "StartView in view {view} may discard an uncommitted WAL suffix", + ); + assert_eq!(consensus.view(), view); + assert_eq!(consensus.sequencer().current_sequence(), COMMIT); + assert_eq!(consensus.commit_min(), checkpoint); + assert_eq!(consensus.commit_max(), COMMIT); + assert_eq!(consensus.status(), Status::Normal); + } + } } /// The split-brain gate's predicate: `view` and `log_view` each independently diff --git a/core/integration/tests/cluster/partition_dedup.rs b/core/integration/tests/cluster/partition_dedup.rs index dc2c66c4f..b66ece607 100644 --- a/core/integration/tests/cluster/partition_dedup.rs +++ b/core/integration/tests/cluster/partition_dedup.rs @@ -299,6 +299,7 @@ const TOTAL_SENDS: u64 = TESTED_CLIENT_SENDS + FILLER_SENDS; /// Replayed id: the tested client's watermark itself. Absorbing it requires an /// entry for that client, which only the transferred artifact can supply. const REPLAYED_REQUEST: u64 = TESTED_CLIENT_SENDS; +const AGED_OUT_REQUEST: u64 = 1; /// Filler identity whose sends evict the tested client from the repair ring. const FILLER_CLIENT_ID: u128 = 0x0DED_F111_E400; @@ -391,6 +392,13 @@ async fn given_transferred_dedup_slice_when_old_request_replays_should_absorb( "a replay of a transferred watermark is absorbed as a success" ); + let aged_out = send_reconnecting(addr, CLIENT_ID, AGED_OUT_REQUEST, FINAL_COMMIT_BUDGET).await; + assert_eq!( + aged_out, + IggyError::RequestTooOld.as_code(), + "the transferred window must reject an aged-out request with an unknown outcome" + ); + // The count is the discriminator: an absorbed replay leaves it at // TOTAL_SENDS + 1; a re-execution (empty transferred slice) appends a // second copy of the replayed payload. diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index b5d21ba32..2856d8882 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -940,9 +940,11 @@ where &mut self, persistence: &Rc<PartitionPersistence>, ) -> Result<(), IggyError> { + // New empty WALs carry view zero; positive certificates belong to recovered history. if !self.materialization_missing && self.recovered_log_view.is_none() && persistence.head() == 0 + && matches!(persistence.certified_log_view(), None | Some(0)) { persistence.certify_log_view(self.consensus.log_view(), 0, 0); if persistence.start() { @@ -4409,24 +4411,38 @@ where // carried: the offset confirmations are not retained (no reply // ring in this mode), so a retried produce learns it committed // but not where. - if self + match self .dedup .is_duplicate(client_id, message.header().user_id, request) { - let committed = build_reply_from_request( - &self.consensus, - message.header(), - committed_reply_body(message.header().operation), - ); - Self::deliver_reply_or_log( - &self.consensus, - message.header(), - committed, - reply.take(), - "duplicate reply send failed", - ) - .await; - return; + Ok(false) => {} + Ok(true) => { + let committed = build_reply_from_request( + &self.consensus, + message.header(), + committed_reply_body(message.header().operation), + ); + Self::deliver_reply_or_log( + &self.consensus, + message.header(), + committed, + reply.take(), + "duplicate reply send failed", + ) + .await; + return; + } + Err(error) => { + Self::send_partition_deny_or_log( + consensus, + message.header(), + error.as_code(), + "aged-out request rejection failed", + reply.take(), + ) + .await; + return; + } } } @@ -9700,6 +9716,35 @@ mod tests { assert_eq!(partition.consensus().last_prepare_checksum(), checksum); } + #[compio::test] + async fn certified_empty_wal_without_superblock_preserves_its_log_view() { + const CREATED_VIEW: u32 = 2; + const LOG_VIEW: u32 = 5; + let directory = tempfile::tempdir().unwrap(); + let mut partition = partition_at_view(CREATED_VIEW, 0); + partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); + partition.runtime_options.durability = iggy_common::Durability::Persisted; + let wal = directory.path().join("prepares-0"); + let mut journal = + journal::PartitionPrepareJournal::open(&wal, partition.consensus().group(), 0) + .await + .unwrap(); + journal.certify_log_view(LOG_VIEW, 0, 0).await.unwrap(); + drop(journal); + + partition.open_persistence().await.unwrap(); + + assert_eq!(partition.consensus().view(), LOG_VIEW); + assert_eq!(partition.consensus().log_view(), LOG_VIEW); + assert!(!partition.requires_state_transfer()); + drop(partition); + let journal = + journal::PartitionPrepareJournal::open(&wal, IggyNamespace::new(1, 1, 0).inner(), 0) + .await + .unwrap(); + assert_eq!(journal.certified_log_view(), Some(LOG_VIEW)); + } + #[compio::test] async fn uncertified_log_view_requires_transfer_before_voting() { let directory = tempfile::tempdir().unwrap(); @@ -10239,30 +10284,28 @@ mod tests { #[compio::test] async fn given_advanced_view_when_persist_gate_runs_should_write_vsr_state_once() { - let mut partition = partition_at_view(3, 2); - let store = Rc::new(RecordingSuperblock::default()); - partition.set_superblock(store.clone(), None); - - assert!(partition.persist_superblock_if_needed().await); - - let state = consensus::VsrState::try_from(store.writes.borrow()[0].as_slice()) - .expect("recorded payload decodes as a VsrState"); - assert_eq!(state.cluster, TEST_CLUSTER); - assert_eq!(state.view, 3); - assert_eq!(state.log_view, 2); - assert_eq!( - (state.checkpoint_op, state.checkpoint_checksum), - (0, 0), - "no partition checkpoint exists yet, so the pairing fields stay zero" - ); - assert!(!partition.consensus().needs_superblock_persist()); - - assert!(partition.persist_superblock_if_needed().await); - assert_eq!( - store.attempts.get(), - 1, - "an unchanged view must take the lock-free fast path, not rewrite" - ); + for (view, log_view) in [(3, 2), (2, 0)] { + let mut partition = partition_at_view(view, log_view); + let store = Rc::new(RecordingSuperblock::default()); + partition.set_superblock(store.clone(), None); + + assert!(partition.persist_superblock_if_needed().await); + + let state = consensus::VsrState::try_from(store.writes.borrow()[0].as_slice()) + .expect("recorded payload decodes as a VsrState"); + assert_eq!(state.cluster, TEST_CLUSTER); + assert_eq!(state.view, view); + assert_eq!(state.log_view, log_view); + assert_eq!((state.checkpoint_op, state.checkpoint_checksum), (0, 0)); + assert!(!partition.consensus().needs_superblock_persist()); + + assert!(partition.persist_superblock_if_needed().await); + assert_eq!( + store.attempts.get(), + 1, + "an unchanged view must not rewrite" + ); + } } /// The `offset_frontier` of the most recent recorded write. @@ -11218,30 +11261,27 @@ mod tests { #[compio::test] async fn given_failing_superblock_when_persist_gate_runs_should_withhold_and_back_off() { - let mut partition = partition_at_view(1, 1); - let store = Rc::new(RecordingSuperblock::default()); - store.fail_writes.set(true); - partition.set_superblock(store.clone(), None); + for (view, log_view) in [(1, 1), (2, 0)] { + let mut partition = partition_at_view(view, log_view); + let store = Rc::new(RecordingSuperblock::default()); + store.fail_writes.set(true); + partition.set_superblock(store.clone(), None); - assert!( - !partition.persist_superblock_if_needed().await, - "a failed write must withhold the send" - ); - assert_eq!(store.attempts.get(), 1); - - assert!( - !partition.persist_superblock_if_needed().await, - "the backoff window must withhold without retrying the write" - ); - assert_eq!( - store.attempts.get(), - 1, - "a call inside the backoff window must not touch the store" - ); - assert!( - partition.consensus().needs_superblock_persist(), - "the view stays undurable until a write lands" - ); + assert!( + !partition.persist_superblock_if_needed().await, + "a failed write must withhold the send" + ); + assert_eq!(store.attempts.get(), 1); + assert!( + !partition.persist_superblock_if_needed().await, + "the backoff window must withhold without retrying the write" + ); + assert_eq!(store.attempts.get(), 1); + assert!( + partition.consensus().needs_superblock_persist(), + "the view stays undurable until a write lands" + ); + } } /// Client-facing bus that records every `send_to_client` frame so tests diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs index dd21f29f6..91fb882bc 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs @@ -61,22 +61,13 @@ pub(crate) fn no_confirmations() -> SendMessagesResponse { } } -/// True when `error` can only have been raised after the server committed the -/// batch. Resending then turns one committed write into as many copies as the -/// retry budget allows, on a plane that keeps no reply cache to collapse them. -/// -/// Both kinds are raised while decoding the HTTP reply body, which is reached -/// only once the status check has accepted a 2xx, so the batch landed and just -/// its confirmation is unreadable. The binary path degrades an unreadable body -/// to an empty confirmation list and never raises either kind. -/// -/// Membership is scoped to the send path and must stay conservative. An error -/// meaning the request never arrived, or that the server rejected it before -/// committing, has to keep retrying. -fn implies_committed_send(error: &IggyError) -> bool { +/// Stop for HTTP confirmation decode failures after a successful status, or +/// an explicit refusal of an aged-out dedup ID whose outcome is unknown. +/// Other failures retain the producer's at-least-once retry policy. +const fn stops_send_retries(error: &IggyError) -> bool { matches!( error, - IggyError::InvalidBytesResponse | IggyError::InvalidJsonResponse + IggyError::InvalidBytesResponse | IggyError::InvalidJsonResponse | IggyError::RequestTooOld ) } @@ -294,10 +285,10 @@ impl ProducerCore { // failed attempts have none to report. Ok(confirmation) => return Ok(confirmation), Err(error) => { - if implies_committed_send(&error) { + if stops_send_retries(&error) { error!( - "Not retrying a send to topic: {topic}, stream: {stream}: the batch \ - committed and only its confirmation could not be read. {error}." + "Not retrying a send to topic: {topic}, stream: {stream}: \ + another request could duplicate the batch. {error}." ); return Err(error); } @@ -671,8 +662,8 @@ unsafe impl Sync for IggyProducer {} /// /// # Retrying and what a failure means /// -/// A retry is the same request sent again unchanged, meaning a producer never rewrites, splits, or -/// reorders a batch to retry a send. [`send_retries()`] sets how many times it may try and how +/// A producer retry sends the same batch under a new request ID without rewriting, splitting, or +/// reordering it. [`send_retries()`] sets how many times it may try and how /// later retries are paced. The default allows three retries and configures a one-second interval. /// The first retry is immediate. Later retries wait for the next tick of an interval timer, so they /// are at most one interval apart, and an attempt that outlasts the interval is followed by the next @@ -682,15 +673,18 @@ unsafe impl Sync for IggyProducer {} /// A request can fail after the server has already appended it, so the first request of an /// unconfirmed tail may already be in the partition. Sending the tail again can therefore write the /// same messages twice, leaving the batch in the partition at multiple offsets. A consumer that -/// cannot accept duplicates must recognize them itself. +/// cannot accept duplicates must recognize them itself. This at-least-once policy also retries +/// ambiguous failures such as [`IggyError::Disconnected`] and [`IggyError::TransientNotCommitted`]. +/// The explicit aged-out refusal and HTTP confirmation decode failures below stop retries. /// /// What is retried, and what is not: /// /// | Situation | Retried | What the caller ends up with | /// | --- | --- | --- | /// | the client is not signed in, so nothing may be sent yet | yes | the send goes ahead as soon as the client is signed in, or fails with [`IggyError::CannotSendMessagesDueToClientDisconnection`] once the retry budget is spent | -/// | the request failed without an indication that it committed | yes, the identical request is sent again | the confirmation of the attempt that finally succeeds, or the last error once the retry budget is spent | +/// | the request failed with another error | yes, the messages are sent as a new request | the confirmation of the attempt that finally succeeds, or the last error once the retry budget is spent | /// | the batch committed, but its confirmation could not be read ([`IggyError::InvalidBytesResponse`] or [`IggyError::InvalidJsonResponse`], raised on the HTTP transport only) | no | the write did happen and retrying would duplicate it on purpose | +/// | the request aged out of the dedup window ([`IggyError::RequestTooOld`]) | no | its outcome is unknown and another request could duplicate it | /// | encrypting or partitioning the batch failed | no | that cause, with the whole batch returned as unconfirmed because nothing was sent, although earlier messages may already have been encrypted in place | /// | [`send_retries()`] passed `None` or `0` | no | the outcome of the single attempt, which skips the sign-in check above and fails with the transport error when the client is disconnected | /// @@ -1057,19 +1051,34 @@ impl IggyProducer { #[cfg(test)] mod tests { - use super::implies_committed_send; - use iggy_common::IggyError; + use super::stops_send_retries; + use crate::client_wrappers::client_wrapper::ClientWrapper; + use crate::clients::producer_builder::IggyProducerBuilder; + use crate::clients::producer_config::DirectConfig; + use crate::tcp::tcp_client::TcpClient; + use bytes::Bytes; + use iggy_binary_protocol::{Command, HEADER_SIZE, Operation, ReplyHeader, RequestHeader}; + use iggy_common::locking::{IggyRwLock, IggyRwLockFn}; + use iggy_common::{ + BinaryTransport, Client, ClientState, Identifier, IggyError, IggyMessage, Partitioning, + TcpClientConfig, VsrSessionControl, + }; + use std::sync::Arc; + use std::time::Duration; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; #[test] fn test_unreadable_confirmation_of_a_committed_batch_stops_retrying() { - assert!(implies_committed_send(&IggyError::InvalidBytesResponse)); - assert!(implies_committed_send(&IggyError::InvalidJsonResponse)); + assert!(stops_send_retries(&IggyError::InvalidBytesResponse)); + assert!(stops_send_retries(&IggyError::InvalidJsonResponse)); } #[test] - fn test_errors_reachable_before_a_commit_keep_retrying() { + fn test_other_failures_keep_the_at_least_once_retry_policy() { for error in [ IggyError::Disconnected, + IggyError::TransientNotCommitted, IggyError::EmptyResponse, IggyError::Unauthenticated, IggyError::Unauthorized, @@ -1078,9 +1087,85 @@ mod tests { IggyError::ResourceNotFound(String::new()), ] { assert!( - !implies_committed_send(&error), - "{error} does not prove a commit, so it must stay retryable" + !stops_send_retries(&error), + "{error} remains retryable under the producer's at-least-once policy" ); } } + + #[tokio::test] + async fn aged_out_request_stops_producer_retries_without_minting_another_request() { + const RETRIES: u32 = 2; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap().to_string(); + let peer = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut requests = Vec::new(); + let mut header_bytes = [0u8; HEADER_SIZE]; + while stream.read_exact(&mut header_bytes).await.is_ok() { + let request = + bytemuck::checked::try_pod_read_unaligned::<RequestHeader>(&header_bytes) + .unwrap(); + assert_eq!(request.operation, Operation::SendMessages); + let mut body = vec![0u8; usize::try_from(request.size).unwrap() - HEADER_SIZE]; + stream.read_exact(&mut body).await.unwrap(); + requests.push(request.request); + let reply = ReplyHeader { + command: Command::Reply, + operation: request.operation, + client: request.client, + request: request.request, + size: u32::try_from(HEADER_SIZE).unwrap(), + status: IggyError::RequestTooOld.as_code(), + ..Default::default() + }; + stream.write_all(bytemuck::bytes_of(&reply)).await.unwrap(); + } + requests + }); + let client = TcpClient::create(Arc::new(TcpClientConfig { + server_address: address, + ..Default::default() + })) + .unwrap(); + Client::connect(&client).await.unwrap(); + client.bind_vsr_session(1).await.unwrap(); + client.set_state(ClientState::Authenticated).await; + let producer = IggyProducerBuilder::new( + IggyRwLock::new(ClientWrapper::Tcp(client)), + Identifier::numeric(1).unwrap(), + "stream".to_owned(), + Identifier::numeric(1).unwrap(), + "topic".to_owned(), + None, + None, + ) + .partitioning(Partitioning::partition_id(0)) + .direct(DirectConfig::builder().build()) + .send_retries(Some(RETRIES), None) + .build(); + let message = IggyMessage::builder() + .payload(Bytes::from_static(b"unknown-outcome")) + .build() + .unwrap(); + let result = tokio::time::timeout(Duration::from_secs(1), producer.send(vec![message])) + .await + .unwrap(); + let Err(IggyError::ProducerSendFailed { cause, failed, .. }) = result else { + panic!("the producer must return an unconfirmed batch"); + }; + assert!( + matches!(*cause, IggyError::RequestTooOld), + "unexpected producer error: {cause:?}" + ); + assert_eq!(failed.len(), 1); + Client::shutdown(&*producer.core.client.read().await) + .await + .unwrap(); + assert_eq!( + peer.await.unwrap().len(), + 1, + "unknown outcome must not mint a fresh request" + ); + } } diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs index ca5fa7eb9..60cdb519c 100644 --- a/core/sdk/src/vsr.rs +++ b/core/sdk/src/vsr.rs @@ -411,19 +411,21 @@ mod tests { #[test] fn reply_with_nonzero_status_surfaces_as_typed_error() { - // A dispatch-time authorization denial rides `ReplyHeader.status`; the - // decode funnel surfaces it as the typed error before any body decode, - // even though the deny body is empty. - let header = ReplyHeader { - command: Command::Reply, - size: HEADER_SIZE as u32, - status: IggyError::Unauthorized.as_code(), - ..Default::default() - }; - let mut buf = [0u8; HEADER_SIZE]; - buf.copy_from_slice(bytemuck::bytes_of(&header)); - let result = decode_response_split(&buf, Bytes::new()); - assert!(matches!(result, Err(IggyError::Unauthorized))); + for error in [IggyError::Unauthorized, IggyError::RequestTooOld] { + let header = ReplyHeader { + command: Command::Reply, + operation: Operation::SendMessages, + size: HEADER_SIZE as u32, + status: error.as_code(), + ..Default::default() + }; + let mut buffer = [0u8; HEADER_SIZE]; + buffer.copy_from_slice(bytemuck::bytes_of(&header)); + assert!( + matches!(decode_response_split(&buffer, Bytes::new()), Err(result) if result == error), + "an empty denial body must preserve {error}" + ); + } } #[test] diff --git a/core/server/README.md b/core/server/README.md index 0135ce276..f2cac2c03 100644 --- a/core/server/README.md +++ b/core/server/README.md @@ -49,6 +49,36 @@ Older SDKs can lose membership when a backup refuses an offset commit, and new SDKs require servers supporting those commands. HTTP polling is forwarded by the server and keeps its existing client API. +## Upgrade recovery + +Partition recovery refuses a superblock whose `view`, or nonzero `log_view`, +is below the partition's committed `created_view`. It also refuses a WAL +certificate with a nonzero `log_view` below that floor. The log names the +superblock directory or the `prepares-{revision}` WAL directory and the +recorded views. The affected partition is tombstoned locally and is not served; +its durable files are preserved. + +Such state can have been written by older servers, including `server-0.9.0`, +after restarting before the partition's first superblock write. The creation +view was not restored on that path. Raising the recorded views during an +upgrade would certify history that the replica may not hold, so the server +does not repair those records automatically. + +If recovery reports `PartitionViewBelowCreation` or +`PartitionWalViewBelowCreation`: + +1. Pause writes to the affected partition and stop the affected cluster before + changing any recovery files. Keep the refusal logs. +2. Preserve a consistent copy of every replica's metadata, partition + superblocks, WAL directories and segment files together with its configuration. +3. Restore only from a verified consistent backup, or use a separately validated + recovery procedure based on retained replica history. Confirm that it preserves + acknowledged writes and agrees with the committed partition creation metadata. + +There is no automatic in-place migration for below-floor records. Do not edit +view numbers or delete superblocks or WAL directories to bypass the refusal: +an empty history could then replace committed data during a view change. + ## Systemd integration Build with the `systemd` feature to enable readiness and watchdog notifications: diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index c31ecbab1..d488ee951 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -76,10 +76,12 @@ use tracing::error; /// /// [`ServerError::PartitionSuperblockIo`] when the directory or a slot /// cannot be read; the `VersionUnknown` / `Unverifiable` / `Undecodable` / -/// `IdentityMismatch` variants when a record exists but cannot be trusted. +/// `IdentityMismatch` / `PartitionViewBelowCreation` variants when a record +/// exists but cannot be trusted. pub async fn open_partition_superblock( partition_dir: &str, identity: ReplicaIdentity, + created_view: u32, ) -> Result<(Rc<PingPongSuperblock>, Option<VsrState>), ServerError> { let io_error = |source| ServerError::PartitionSuperblockIo { dir: PathBuf::from(partition_dir), @@ -142,6 +144,14 @@ pub async fn open_partition_superblock( state.replica_count.into(), ); } + if state.view < created_view || (state.log_view != 0 && state.log_view < created_view) { + return Err(ServerError::PartitionViewBelowCreation { + dir: PathBuf::from(partition_dir), + view: state.view, + log_view: state.log_view, + created_view, + }); + } } Ok((Rc::new(superblock), recovered_state)) } @@ -353,7 +363,9 @@ pub async fn load_partition_or_fence( | ServerError::PartitionSuperblockVersionUnknown { .. } | ServerError::PartitionSuperblockUnverifiable { .. } | ServerError::PartitionSuperblockUndecodable { .. } - | ServerError::PartitionSuperblockIdentityMismatch { .. }), + | ServerError::PartitionSuperblockIdentityMismatch { .. } + | ServerError::PartitionViewBelowCreation { .. } + | ServerError::PartitionWalViewBelowCreation { .. }), ) => { error!( stream_id, @@ -398,6 +410,7 @@ async fn load_partition( replica_id: self_replica_id, replica_count, }, + partition_metadata.created_view, ) .await?; @@ -434,7 +447,7 @@ async fn load_partition( durable_view: recovered_state .as_ref() .map(|state| (state.view, state.log_view)), - view_fallback: None, + view_fallback: Some(partition_metadata.created_view), seed_view: None, incarnation: None, join, @@ -455,34 +468,46 @@ async fn load_partition( { let directory = Path::new(&partition_dir) .join(format!("prepares-{}", partition_metadata.created_revision)); - Some( - PartitionPersistence::open_with_capacity( - &directory, - namespace.inner(), - partition_metadata.created_revision, - DiskStorage, - config.partition.wal_bytes_max.as_bytes_u64(), - runtime_options - .preallocate_segments - .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS), - ) - .await - .map_err(|source| match source.kind() { - std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof => { - ServerError::PartitionRecovery(PartitionRecoveryError::Refused { - dir: PathBuf::from(&partition_dir), - stream_id: namespace.stream_id(), - topic_id: namespace.topic_id(), - partition_id: namespace.partition_id(), - reason: PartitionRecoveryRefusal::PrepareWal { directory, source }, - }) - } - _ => ServerError::PartitionPrepareWalIo { - dir: directory, - source, - }, - })?, + let (persistence, prepares) = PartitionPersistence::open_with_capacity( + &directory, + namespace.inner(), + partition_metadata.created_revision, + DiskStorage, + config.partition.wal_bytes_max.as_bytes_u64(), + runtime_options + .preallocate_segments + .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS), ) + .await + .map_err(|source| match source.kind() { + std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof => { + ServerError::PartitionRecovery(PartitionRecoveryError::Refused { + dir: PathBuf::from(&partition_dir), + stream_id: namespace.stream_id(), + topic_id: namespace.topic_id(), + partition_id: namespace.partition_id(), + reason: PartitionRecoveryRefusal::PrepareWal { + directory: directory.clone(), + source, + }, + }) + } + _ => ServerError::PartitionPrepareWalIo { + dir: directory.clone(), + source, + }, + })?; + if let Some(log_view) = persistence.certified_log_view() + && log_view != 0 + && log_view < partition_metadata.created_view + { + return Err(ServerError::PartitionWalViewBelowCreation { + dir: directory, + log_view, + created_view: partition_metadata.created_view, + }); + } + Some((persistence, prepares)) } else { None }; @@ -705,7 +730,8 @@ async fn recover_partition_segments( /// in: the metadata plane's view when it committed the create, recorded on /// the committed partition so every replica seeds the same value. See the /// `seed_view` comment below for why a group left at view 0 is unreachable. A -/// restart materialization ignores it and probes for the live view instead. +/// restart keeps it as a view floor while probing for the live view, without +/// assigning it to the recovered log. /// /// The returned partition's `offset` / `dirty_offset` are `0` and its /// `OffsetSpace` is unused, mirroring a clean append starting at the empty @@ -737,8 +763,8 @@ pub async fn build_partition_fresh( // is the marker of a prior life (the .log inside may legitimately be // empty -- committed-but-unflushed data dies with the journal), while a // genuinely fresh create finds nothing. - let restarted = replica_count > 1 - && std::fs::metadata(config.get_partition_path(stream_id, topic_id, partition_id)).is_ok(); + let partition_dir = config.get_partition_path(stream_id, topic_id, partition_id); + let restarted = replica_count > 1 && std::fs::metadata(&partition_dir).is_ok(); create_partition_file_hierarchy(stream_id, topic_id, partition_id, partitions_config) .await .map_err(|source| { @@ -755,18 +781,13 @@ pub async fn build_partition_fresh( if runtime_options.durability.is_persisted() || runtime_options.consumer_offset_durability.is_persisted() { - persist_partition_hierarchy( - &config.get_partition_path(stream_id, topic_id, partition_id), - &config.get_system_path(), - ) - .await?; + persist_partition_hierarchy(&partition_dir, &config.get_system_path()).await?; } // The hierarchy create above guarantees the directory exists; recover this // group's durable (view, log_view) before choosing how to join, so a // restart materialization resumes from the view it last recorded instead // of re-entering an older one. - let partition_dir = config.get_partition_path(stream_id, topic_id, partition_id); let (superblock, recovered_state) = open_partition_superblock( &partition_dir, ReplicaIdentity { @@ -774,6 +795,7 @@ pub async fn build_partition_fresh( replica_id: self_replica_id, replica_count, }, + created_view, ) .await?; @@ -794,8 +816,11 @@ pub async fn build_partition_fresh( .map(|state| (state.view, state.log_view)); // Shared with the simulator's `init_partition`, which cannot call this // builder; see `fresh_group_start`. - let FreshGroupStart { join, seed_view } = - fresh_group_start(restarted, durable_view, created_view); + let FreshGroupStart { + view_fallback, + seed_view, + join, + } = fresh_group_start(restarted, durable_view, created_view); // Request queue holds 2x the prepare depth (buffered requests drain as // prepares commit); depth is the per-partition `[partition]` knob. let prepare_queue_depth = config.partition.prepare_queue_depth; @@ -810,7 +835,7 @@ pub async fn build_partition_fresh( VsrRestore { timers: &timers, durable_view, - view_fallback: None, + view_fallback, // Both planes pick their primary as `view % replica_count` from // their OWN view counter. A group left at view 0 while the // metadata plane sits elsewhere therefore names a different node @@ -988,6 +1013,7 @@ mod tests { use super::*; use bytes::Bytes; use configs::server::ServerConfig; + use consensus::Sequencer; use iggy_binary_protocol::batch::BATCH_HEADER_SIZE; use iggy_binary_protocol::{Command, Operation, PrepareHeader}; use journal::DurableAppend; @@ -1003,6 +1029,327 @@ mod tests { const REPLICA: u8 = 1; const REPLICAS: u8 = 3; + #[compio::test] + async fn given_unrecorded_creation_view_when_loading_or_rebuilding_should_keep_creation_floor() + { + const CREATED_VIEW: u32 = 2; + let namespace = IggyNamespace::new(1, 1, 0); + let mut recovered_views = Vec::new(); + for use_loader in [true, false] { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let partitions_config = solo_partitions_config(&config); + let fresh = build_partition_fresh( + &config, + &partitions_config, + namespace, + Arc::new(PartitionStats::default()), + 0, + TopicRuntimeOptions::default(), + CLUSTER, + REPLICA, + REPLICAS, + CREATED_VIEW, + Rc::new(IggyMessageBus::new(0)), + ) + .await + .unwrap(); + assert_eq!(fresh.consensus().view(), CREATED_VIEW); + assert!(fresh.consensus().needs_superblock_persist()); + drop(fresh); + + let recovered = if use_loader { + load_partition( + &config, + &partitions_config, + namespace, + Arc::new(PartitionStats::default()), + &Partition::new(0, namespace.inner(), IggyTimestamp::now(), 0, CREATED_VIEW), + TopicRuntimeOptions::default(), + CLUSTER, + REPLICA, + REPLICAS, + Rc::new(IggyMessageBus::new(0)), + ) + .await + } else { + build_partition_fresh( + &config, + &partitions_config, + namespace, + Arc::new(PartitionStats::default()), + 0, + TopicRuntimeOptions::default(), + CLUSTER, + REPLICA, + REPLICAS, + CREATED_VIEW, + Rc::new(IggyMessageBus::new(0)), + ) + .await + } + .unwrap(); + assert_eq!( + recovered.consensus().status(), + consensus::Status::Recovering + ); + assert_eq!( + recovered.consensus().log_view(), + 0, + "creation metadata must not certify the recovered log" + ); + assert!(recovered.consensus().needs_superblock_persist()); + recovered_views.push(recovered.consensus().view()); + } + assert_eq!( + recovered_views, [CREATED_VIEW; 2], + "the boot loader and runtime rebuild must not recover below the committed creation view" + ); + } + + #[compio::test] + async fn given_durable_state_below_creation_view_when_loading_should_refuse_unchanged() { + const CREATED_VIEW: u32 = 2; + let namespace = IggyNamespace::new(1, 1, 0); + for state in [recorded_state(1, 1), recorded_state(3, 1)] { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let directory = config.get_partition_path(1, 1, 0); + let (store, _) = open_partition_superblock(&directory, test_identity(), 0) + .await + .unwrap(); + store.write(&state.to_bytes()).await.unwrap(); + drop(store); + let partitions = solo_partitions(&config); + let loaded = load_partition_or_fence( + &config, + namespace, + Arc::new(PartitionStats::default()), + &Partition::new(0, namespace.inner(), IggyTimestamp::now(), 0, CREATED_VIEW), + TopicRuntimeOptions::default(), + CLUSTER, + REPLICA, + REPLICAS, + Rc::new(IggyMessageBus::new(0)), + &partitions, + ) + .await + .unwrap(); + assert!(loaded.is_none()); + assert!(partitions.is_tombstoned(&namespace)); + let (_, recovered) = open_partition_superblock(&directory, test_identity(), 0) + .await + .unwrap(); + assert_eq!( + recovered, + Some(state), + "refusal must preserve contradictory durable state" + ); + assert!(!Path::new(&format!("{directory}.fenced.0")).exists()); + assert!(matches!( + build_partition_fresh( + &config, + partitions.config(), + namespace, + Arc::new(PartitionStats::default()), + 0, + TopicRuntimeOptions::default(), + CLUSTER, + REPLICA, + REPLICAS, + CREATED_VIEW, + Rc::new(IggyMessageBus::new(0)), + ) + .await, + Err(ServerError::PartitionViewBelowCreation { .. }), + )); + } + } + + #[compio::test] + async fn given_certified_wal_below_creation_view_when_loading_should_refuse_unchanged() { + const CREATED_VIEW: u32 = 2; + const LOG_VIEW: u32 = 1; + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let namespace = IggyNamespace::new(1, 1, 0); + let directory = config.get_partition_path(1, 1, 0); + let wal = Path::new(&directory).join("prepares-0"); + create_dir_all(&directory).await.unwrap(); + let mut journal = journal::PartitionPrepareJournal::open(&wal, namespace.inner(), 0) + .await + .unwrap(); + journal.certify_log_view(LOG_VIEW, 0, 0).await.unwrap(); + drop(journal); + let partitions = solo_partitions(&config); + let metadata = Partition::new(0, namespace.inner(), IggyTimestamp::now(), 0, CREATED_VIEW); + let runtime = TopicRuntimeOptions { + durability: iggy_common::Durability::Persisted, + preallocate_segments: Some(false), + ..Default::default() + }; + let result = load_partition( + &config, + partitions.config(), + namespace, + Arc::new(PartitionStats::default()), + &metadata, + runtime, + CLUSTER, + REPLICA, + REPLICAS, + Rc::new(IggyMessageBus::new(0)), + ) + .await; + assert!(matches!( + result, + Err(ServerError::PartitionWalViewBelowCreation { + dir, + log_view: LOG_VIEW, + created_view: CREATED_VIEW, + }) if dir == wal + )); + let loaded = load_partition_or_fence( + &config, + namespace, + Arc::new(PartitionStats::default()), + &metadata, + runtime, + CLUSTER, + REPLICA, + REPLICAS, + Rc::new(IggyMessageBus::new(0)), + &partitions, + ) + .await + .unwrap(); + assert!(loaded.is_none()); + assert!(partitions.is_tombstoned(&namespace)); + assert!(!Path::new(&format!("{directory}.fenced.0")).exists()); + let journal = journal::PartitionPrepareJournal::open(&wal, namespace.inner(), 0) + .await + .unwrap(); + assert_eq!(journal.certified_log_view(), Some(LOG_VIEW)); + } + + #[compio::test] + #[allow(clippy::too_many_lines)] + async fn given_creation_floor_when_replaying_certified_wal_should_preserve_history_and_view() { + const CREATED_VIEW: u32 = 2; + const LOG_VIEW: u32 = 5; + const PAYLOAD: &[u8] = b"certified-history"; + let namespace = IggyNamespace::new(1, 1, 0); + let runtime = TopicRuntimeOptions { + durability: iggy_common::Durability::Persisted, + preallocate_segments: Some(false), + ..Default::default() + }; + for durable_view in [None, Some(LOG_VIEW + 1)] { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let partitions_config = solo_partitions_config(&config); + let fresh = build_partition_fresh( + &config, + &partitions_config, + namespace, + Arc::new(PartitionStats::default()), + 0, + runtime, + CLUSTER, + REPLICA, + REPLICAS, + CREATED_VIEW, + Rc::new(IggyMessageBus::new(0)), + ) + .await + .unwrap(); + assert!(!fresh.requires_state_transfer()); + assert_eq!(fresh.consensus().log_view(), CREATED_VIEW); + drop(fresh); + let directory = config.get_partition_path(1, 1, 0); + let mut messages = IggyMessages::with_capacity(1); + messages.push(IggyMessage { + header: IggyMessageHeader::default(), + payload: Bytes::from_static(PAYLOAD), + user_headers: None, + }); + let batch = SendMessagesOwned::from_messages(namespace, &messages).unwrap(); + let total = size_of::<PrepareHeader>() + batch.header.total_size(); + let mut prepare = Message::<PrepareHeader>::new(total); + let body = &mut prepare.as_mut_slice()[size_of::<PrepareHeader>()..]; + batch.header.encode_into(body); + body[BATCH_HEADER_SIZE..].copy_from_slice(&batch.blob); + let prepare = prepare.transmute_header(|_, header: &mut PrepareHeader| { + header.command = Command::Prepare; + header.operation = Operation::SendMessages; + header.cluster = CLUSTER; + header.group = namespace.inner(); + header.view = LOG_VIEW; + header.op = 1; + header.commit = 1; + header.size = u32::try_from(total).unwrap(); + header.checksum = header.identity_checksum(); + }); + let checksum = prepare.header().checksum; + let wal = Path::new(&directory).join("prepares-0"); + let mut journal = journal::PartitionPrepareJournal::open(&wal, namespace.inner(), 0) + .await + .unwrap(); + journal.append(prepare.into_frozen()).await.unwrap(); + journal + .certify_log_view(LOG_VIEW, 1, checksum) + .await + .unwrap(); + drop(journal); + if let Some(view) = durable_view { + let (store, _) = + open_partition_superblock(&directory, test_identity(), CREATED_VIEW) + .await + .unwrap(); + let state = VsrState { + commit_max: 1, + ..recorded_state(view, LOG_VIEW) + }; + store.write(&state.to_bytes()).await.unwrap(); + } + let recovered = load_partition( + &config, + &partitions_config, + namespace, + Arc::new(PartitionStats::default()), + &Partition::new(0, namespace.inner(), IggyTimestamp::now(), 0, CREATED_VIEW), + runtime, + CLUSTER, + REPLICA, + REPLICAS, + Rc::new(IggyMessageBus::new(0)), + ) + .await + .unwrap(); + assert_eq!( + recovered.consensus().view(), + durable_view.unwrap_or(LOG_VIEW) + ); + assert_eq!(recovered.consensus().log_view(), LOG_VIEW); + assert_eq!(recovered.consensus().sequencer().current_sequence(), 1); + assert_eq!(recovered.consensus().commit_max(), 1); + assert!(!recovered.requires_state_transfer()); + let stored = recovered.log.journal().inner.repair_entry(1).unwrap(); + assert!( + stored + .as_slice() + .windows(PAYLOAD.len()) + .any(|bytes| bytes == PAYLOAD) + ); + drop(recovered); + let journal = journal::PartitionPrepareJournal::open(&wal, namespace.inner(), 0) + .await + .unwrap(); + assert_eq!(journal.certified_log_view(), Some(LOG_VIEW)); + assert_eq!(journal.head(), 1); + } + } + #[compio::test] async fn loading_after_retention_preserves_the_wal_owned_tail_when_the_logical_chain_is_empty() { @@ -1134,7 +1481,7 @@ mod tests { .unwrap(), ); let directory = config.get_partition_path(1, 1, 0); - let (store, _) = open_partition_superblock(&directory, test_identity()) + let (store, _) = open_partition_superblock(&directory, test_identity(), 0) .await .unwrap(); let state = recorded_state(3, 2); @@ -1170,7 +1517,7 @@ mod tests { .unwrap(); assert!(partition.requires_state_transfer()); assert!(partition.consensus().view() >= state.view); - let (_, recovered) = open_partition_superblock(&directory, test_identity()) + let (_, recovered) = open_partition_superblock(&directory, test_identity(), 0) .await .unwrap(); assert_eq!(recovered, Some(state)); @@ -1280,7 +1627,7 @@ mod tests { /// The reservation the partition left on disk, which is the only copy a /// restart or a first send can read. async fn recorded_reservation(dir: &str) -> u64 { - let (_store, recorded) = open_partition_superblock(dir, solo_identity()) + let (_store, recorded) = open_partition_superblock(dir, solo_identity(), 0) .await .expect("reopen the partition superblock"); recorded @@ -1323,7 +1670,7 @@ mod tests { let config = solo_config(&root); let dir = config.get_partition_path(1, 1, 0); - let (store, recovered) = open_partition_superblock(&dir, solo_identity()) + let (store, recovered) = open_partition_superblock(&dir, solo_identity(), 0) .await .expect("open a fresh partition superblock"); assert!(recovered.is_none()); @@ -1443,9 +1790,10 @@ mod tests { // A not-yet-materialized directory must open as fresh, not error: the // helper creates it, since a follower can reach load before its first // segment write. - let (_store, recovered) = open_partition_superblock(&partition_dir(&root), test_identity()) - .await - .expect("open a fresh partition superblock"); + let (_store, recovered) = + open_partition_superblock(&partition_dir(&root), test_identity(), 0) + .await + .expect("open a fresh partition superblock"); assert!( recovered.is_none(), "an empty superblock is a fresh group, never an error" @@ -1454,35 +1802,37 @@ mod tests { #[compio::test] async fn given_recorded_view_when_superblock_reopened_should_recover_state() { - let root = tempfile::tempdir().expect("tempdir"); - let dir = partition_dir(&root); - let (store, recovered) = open_partition_superblock(&dir, test_identity()) - .await - .expect("first open"); - assert!(recovered.is_none()); - let state = recorded_state(3, 2); - store - .write(&state.to_bytes()) - .await - .expect("record the advanced view"); - drop(store); - - let (_store, recovered) = open_partition_superblock(&dir, test_identity()) - .await - .expect("reopen after a restart"); + for (view, log_view, created_view) in [(3, 2, 2), (2, 0, 2), (0, 0, 0)] { + let root = tempfile::tempdir().expect("tempdir"); + let dir = partition_dir(&root); + let (store, recovered) = open_partition_superblock(&dir, test_identity(), created_view) + .await + .expect("first open"); + assert!(recovered.is_none()); + let state = recorded_state(view, log_view); + store + .write(&state.to_bytes()) + .await + .expect("record the advanced view"); + drop(store); - assert_eq!( - recovered, - Some(state), - "a restarted partition must recover exactly the state it recorded" - ); + let (_store, recovered) = + open_partition_superblock(&dir, test_identity(), created_view) + .await + .expect("reopen after a restart"); + assert_eq!( + recovered, + Some(state), + "a restarted partition must recover exactly its recorded state" + ); + } } #[compio::test] async fn given_foreign_cluster_record_when_superblock_opened_should_refuse_boot() { let root = tempfile::tempdir().expect("tempdir"); let dir = partition_dir(&root); - let (store, _) = open_partition_superblock(&dir, test_identity()) + let (store, _) = open_partition_superblock(&dir, test_identity(), 0) .await .expect("first open"); let foreign = VsrState { @@ -1495,7 +1845,7 @@ mod tests { .expect("record a foreign identity"); drop(store); - let refused = open_partition_superblock(&dir, test_identity()).await; + let refused = open_partition_superblock(&dir, test_identity(), 0).await; match refused { Err(ServerError::PartitionSuperblockIdentityMismatch { field, .. }) => { diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 67925924c..7d4c3c4a4 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -207,6 +207,25 @@ pub enum ServerError { expected: u128, found: u128, }, + #[error( + "partition superblock at {dir} falls below committed creation view {created_view}: \ + view {view}, log_view {log_view}" + )] + PartitionViewBelowCreation { + dir: PathBuf, + view: u32, + log_view: u32, + created_view: u32, + }, + #[error( + "partition WAL certificate at {dir} falls below committed creation view {created_view}: \ + log_view {log_view}" + )] + PartitionWalViewBelowCreation { + dir: PathBuf, + log_view: u32, + created_view: u32, + }, // Only the `Refused` shape is per-partition: the loader's fence-or-tombstone // arm catches it. Everything else the partition readers raise fails the // boot, and the reconciler logs it and retries the partition with backoff. diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index b46bd2812..533d90e65 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -4327,8 +4327,11 @@ where .as_ref() .map(|state| (state.view, state.log_view)); let restarted = retained.is_some() && self.partition_consensus.replica_count > 1; - let consensus::FreshGroupStart { join, seed_view } = - consensus::fresh_group_start(restarted, durable_view, created_view); + let consensus::FreshGroupStart { + join, + view_fallback, + seed_view, + } = consensus::fresh_group_start(restarted, durable_view, created_view); // Recorded view first, exactly as the two boot paths order it: restoring // after `init` would advertise a view older than the recorded one. @@ -4336,6 +4339,8 @@ where consensus.set_view(view); consensus.set_log_view(log_view); consensus.mark_superblock_durable(view, log_view); + } else if let Some(view) = view_fallback { + consensus.set_view(view); } else if let Some(view) = seed_view { consensus.set_view(view); consensus.set_log_view(view); diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 5acb01651..89963976d 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1690,10 +1690,10 @@ mod tests { use crate::client::SimClient; use crate::workload::apply_sim_commands; use bytes::Bytes; - use consensus::Status; + use consensus::{Status, client_table::COMMITTED_WINDOW_BITS}; use futures::FutureExt; use iggy_binary_protocol::{AckLevel, RoutedRequestHeader, WireIdentifier}; - use iggy_common::ConsumerKind; + use iggy_common::{ConsumerKind, IggyError}; use server_common::sharding::{IggyNamespace, LIST_CLIENTS_GATHER_TIMEOUT}; const DISCONNECT_STREAM: &str = "sim-stream-0"; @@ -3230,6 +3230,191 @@ mod tests { ); } + #[test] + fn given_aged_out_request_when_replayed_should_report_unknown_outcome() { + const CLIENT_ID: u128 = 1; + for initially_committed in [false, true] { + let mut sim = partition_safety_cluster(&[CLIENT_ID]); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + let client = SimClient::new(CLIENT_ID); + sim.register_client_with_primary(&client); + + let delayed = client.send_messages(namespace, &[Bytes::from_static(b"aged-out")]); + let destination = u8::from(!initially_committed); + let first = + submit_and_wait_for_reply(&mut sim, CLIENT_ID, destination, delayed.deep_copy()); + assert_eq!( + first.header().status, + if initially_committed { + 0 + } else { + IggyError::TransientNotAccepted.as_code() + }, + "the original attempt must establish the intended admission outcome", + ); + for _ in 0..COMMITTED_WINDOW_BITS { + let later = + client.send_messages(namespace, &[Bytes::from_static(b"committed-later")]); + let committed = submit_and_wait_for_reply(&mut sim, CLIENT_ID, 0, later); + assert_eq!(committed.header().status, 0); + } + + let expected = sim + .offsets(0, namespace) + .expect("committed partition offsets"); + let replay = submit_and_wait_for_reply(&mut sim, CLIENT_ID, 0, delayed); + assert_eq!( + replay.header().status, + IggyError::RequestTooOld.as_code(), + "aging out cannot prove whether request 1 committed", + ); + assert_eq!( + sim.offsets(0, namespace), + Some(expected), + "rejecting an aged-out request must not append another payload", + ); + } + } + + #[test] + fn given_reordered_requests_within_dedup_window_when_retried_should_commit_once() { + const CLIENT_ID: u128 = 1; + let mut sim = partition_safety_cluster(&[CLIENT_ID]); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + let client = SimClient::new(CLIENT_ID); + sim.register_client_with_primary(&client); + let delayed = client.send_messages(namespace, &[Bytes::from_static(b"delayed")]); + let later = client.send_messages(namespace, &[Bytes::from_static(b"later")]); + let committed = submit_and_wait_for_reply(&mut sim, CLIENT_ID, 0, later.deep_copy()); + assert_eq!(committed.header().status, 0); + let committed = submit_and_wait_for_reply(&mut sim, CLIENT_ID, 0, delayed.deep_copy()); + assert_eq!(committed.header().status, 0); + let expected = sim.offsets(0, namespace).unwrap(); + assert_eq!( + expected.commit_offset, 1, + "both reordered requests must append" + ); + for replay in [delayed, later] { + let duplicate = submit_and_wait_for_reply(&mut sim, CLIENT_ID, 0, replay); + assert_eq!(duplicate.header().status, 0); + assert_eq!( + sim.offsets(0, namespace), + Some(expected), + "a retained duplicate must not append again" + ); + } + } + + #[test] + #[allow(clippy::too_many_lines)] + fn given_unpersisted_creation_view_when_replicas_restart_should_preserve_acknowledged_writes() { + const CLIENT_ID: u128 = 1; + const PROGRESS_STEPS: usize = 2_000; + const PAYLOAD: &[u8] = b"acknowledged-before-late-materialization"; + for created_view in [1, 2] { + let mut sim = partition_safety_cluster(&[CLIENT_ID]); + let client = SimClient::new(CLIENT_ID); + sim.register_client_with_primary(&client); + let namespace = IggyNamespace::new(1, 1, 0); + sim.partition_created_views.insert(namespace, created_view); + sim.seed_stream_topic_partition(namespace); + for replica in &sim.replicas[..2] { + materialise_partition( + replica, + namespace, + false, + created_view, + sim.consumer_offsets_max, + ); + assert!( + replica.partition_superblocks.borrow()[&namespace] + .read_latest_sync() + .is_none(), + "the crash must precede the first durable view write" + ); + } + for replica in 0..2 { + sim.replica_crash(replica); + } + for replica in 0..2 { + sim.replica_restart(replica); + } + let primary = (0..PROGRESS_STEPS) + .find_map(|_| { + sim.step(); + (0..2).find(|replica| { + sim.partition_consensus_state(usize::from(*replica), namespace) + .is_some_and(|state| state.status == Status::Normal && state.is_primary) + }) + }) + .expect("the restarted quorum must elect a primary"); + let request = client.send_messages(namespace, &[Bytes::from_static(PAYLOAD)]); + let committed = submit_and_wait_for_reply(&mut sim, CLIENT_ID, primary, request); + assert_eq!(committed.header().status, 0); + let committed_op = committed.header().commit; + let expected = sim + .offsets(usize::from(primary), namespace) + .expect("acknowledged partition offsets"); + + materialise_partition( + &sim.replicas[2], + namespace, + false, + created_view, + sim.consumer_offsets_max, + ); + assert!( + (0..PROGRESS_STEPS).any(|_| { + sim.step(); + (0..3).all(|replica| { + sim.partition_consensus_state(replica, namespace) + .is_some_and(|state| { + state.status == Status::Normal && state.commit_min >= committed_op + }) + && sim.offsets(replica, namespace) == Some(expected) + }) + }), + "late materialization must preserve the acknowledged prefix: states={:?}, offsets={:?}", + (0..3) + .map(|replica| sim.partition_consensus_state(replica, namespace)) + .collect::<Vec<_>>(), + (0..3) + .map(|replica| sim.offsets(replica, namespace)) + .collect::<Vec<_>>() + ); + for replica in 0..3 { + let prepare = retained_prepare(&sim, replica, namespace, committed_op); + assert!( + prepare + .body() + .windows(PAYLOAD.len()) + .any(|bytes| bytes == PAYLOAD), + "replica {replica} lost the acknowledged payload at op {committed_op}" + ); + } + } + } + + fn partition_safety_cluster(client_ids: &[u128]) -> Simulator { + const REPLICA_COUNT: u8 = 3; + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + Simulator::new( + usize::from(REPLICA_COUNT), + client_ids.iter().copied(), + packet::PacketSimulatorOptions { + node_count: REPLICA_COUNT, + client_count: u8::try_from(client_ids.len()).expect("test clients fit u8"), + ..packet::PacketSimulatorOptions::default() + }, + ) + } + fn successful_send_reply_count(replies: &[Message<ReplyHeader>]) -> usize { replies .iter() diff --git a/foreign/go/errors/errors.yaml b/foreign/go/errors/errors.yaml index 145cd569f..9ad6a6ec0 100644 --- a/foreign/go/errors/errors.yaml +++ b/foreign/go/errors/errors.yaml @@ -335,6 +335,10 @@ code: 84 format: "invalid number value" fields: [] +- name: RequestTooOld + code: 85 + format: "request is outside the deduplication window; outcome unknown, do not send a new request" + fields: [] - name: ClientNotFound code: 100 format: "client with id: %d was not found." diff --git a/foreign/go/errors/errors_gen.go b/foreign/go/errors/errors_gen.go index 2f80a8a78..f1e3f94f0 100644 --- a/foreign/go/errors/errors_gen.go +++ b/foreign/go/errors/errors_gen.go @@ -722,6 +722,17 @@ func (e InvalidNumberValue) Is(target error) bool { return ok } +type RequestTooOld struct{} + +func (e RequestTooOld) Error() string { + return "request is outside the deduplication window; outcome unknown, do not send a new request" +} +func (e RequestTooOld) Code() Code { return 85 } +func (e RequestTooOld) Is(target error) bool { + _, ok := target.(RequestTooOld) + return ok +} + type ClientNotFound struct { ID uint32 } @@ -2723,6 +2734,7 @@ var ( ErrInvalidNumberEncoding = InvalidNumberEncoding{} ErrInvalidBooleanValue = InvalidBooleanValue{} ErrInvalidNumberValue = InvalidNumberValue{} + ErrRequestTooOld = RequestTooOld{} ErrClientNotFound = ClientNotFound{} ErrInvalidClientId = InvalidClientId{} ErrConnectionClosed = ConnectionClosed{} @@ -2967,6 +2979,7 @@ const ( InvalidNumberEncodingCode Code = 82 InvalidBooleanValueCode Code = 83 InvalidNumberValueCode Code = 84 + RequestTooOldCode Code = 85 ClientNotFoundCode Code = 100 InvalidClientIdCode Code = 101 ConnectionClosedCode Code = 206 @@ -3278,6 +3291,8 @@ func (c Code) String() string { return "InvalidBooleanValue" case InvalidNumberValueCode: return "InvalidNumberValue" + case RequestTooOldCode: + return "RequestTooOld" case ClientNotFoundCode: return "ClientNotFound" case InvalidClientIdCode: @@ -3763,6 +3778,8 @@ func FromCode(code Code) IggyError { return ErrInvalidBooleanValue case InvalidNumberValueCode: return ErrInvalidNumberValue + case RequestTooOldCode: + return ErrRequestTooOld case ClientNotFoundCode: return ErrClientNotFound case InvalidClientIdCode: diff --git a/foreign/go/errors/errors_test.go b/foreign/go/errors/errors_test.go index cb9719983..64045fd5d 100644 --- a/foreign/go/errors/errors_test.go +++ b/foreign/go/errors/errors_test.go @@ -64,6 +64,12 @@ func TestIggyError_ConsensusErrors(t *testing.T) { code: TransientNotAcceptedCode, message: "request transiently not accepted; retry, on any replica", }, + { + err: RequestTooOld{}, + sentinel: ErrRequestTooOld, + code: RequestTooOldCode, + message: "request is outside the deduplication window; outcome unknown, do not send a new request", + }, { err: ConsumerGroupPartitionNotOwned{ClientId: 4, PartitionId: 9}, sentinel: ErrConsumerGroupPartitionNotOwned, diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java index 5d776f827..aa4399829 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java @@ -57,6 +57,7 @@ public enum IggyErrorCode { PERSONAL_ACCESS_TOKEN_EXPIRED(54), TRANSIENT_NOT_COMMITTED(57), TRANSIENT_NOT_ACCEPTED(58), + REQUEST_TOO_OLD(85), ACCESS_TOKEN_MISSING(77), INVALID_ACCESS_TOKEN(78), diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/exception/IggyErrorCodeTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/exception/IggyErrorCodeTest.java index 45feef08b..07b4c422a 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/exception/IggyErrorCodeTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/exception/IggyErrorCodeTest.java @@ -179,6 +179,7 @@ class IggyErrorCodeTest { "54, PERSONAL_ACCESS_TOKEN_EXPIRED", "57, TRANSIENT_NOT_COMMITTED", "58, TRANSIENT_NOT_ACCEPTED", + "85, REQUEST_TOO_OLD", "77, ACCESS_TOKEN_MISSING", "78, INVALID_ACCESS_TOKEN", diff --git a/foreign/node/src/wire/error.code.test.ts b/foreign/node/src/wire/error.code.test.ts index dc768e2d2..9f278b898 100644 --- a/foreign/node/src/wire/error.code.test.ts +++ b/foreign/node/src/wire/error.code.test.ts @@ -19,6 +19,14 @@ import assert from 'node:assert/strict'; import { it } from 'node:test'; import { translateErrorCode } from './error.code.js'; +it('reports an unknown outcome for an aged-out request', () => { + assert.equal( + translateErrorCode(85), + 'Request is outside the deduplication window; outcome unknown, do not send a new request' + ); + assert.equal(translateErrorCode(60), 'error'); +}); + it('translates the consumer-offset capacity error', () => { assert.equal(translateErrorCode(3024), 'Consumer offset limit reached for partition, raise [partition] consumer_offsets_max'); }); diff --git a/foreign/node/src/wire/error.code.ts b/foreign/node/src/wire/error.code.ts index 58773d957..4e9298cda 100644 --- a/foreign/node/src/wire/error.code.ts +++ b/foreign/node/src/wire/error.code.ts @@ -85,6 +85,7 @@ export const translateErrorCode = (code: number): string => { case '82': return "Invalid number encoding"; case '83': return "Invalid boolean value"; case '84': return "Invalid number value"; + case '85': return "Request is outside the deduplication window; outcome unknown, do not send a new request"; case '100': return "Client with ID: {0} was not found."; case '101': return "Invalid client ID"; diff --git a/foreign/swift/Sources/Iggy/Errors/IggyErrorCode.swift b/foreign/swift/Sources/Iggy/Errors/IggyErrorCode.swift index ffbac5d24..8939dae33 100644 --- a/foreign/swift/Sources/Iggy/Errors/IggyErrorCode.swift +++ b/foreign/swift/Sources/Iggy/Errors/IggyErrorCode.swift @@ -92,6 +92,7 @@ public enum IggyErrorCode: UInt32, Sendable, Hashable, CaseIterable, Codable { case invalidNumberEncoding = 82 case invalidBooleanValue = 83 case invalidNumberValue = 84 + case requestTooOld = 85 case clientNotFound = 100 case invalidClientId = 101 case connectionClosed = 206 @@ -338,6 +339,7 @@ extension IggyErrorCode { case .invalidNumberEncoding: "invalid_number_encoding" case .invalidBooleanValue: "invalid_boolean_value" case .invalidNumberValue: "invalid_number_value" + case .requestTooOld: "request_too_old" case .clientNotFound: "client_not_found" case .invalidClientId: "invalid_client_id" case .connectionClosed: "connection_closed" diff --git a/foreign/swift/Tests/IggyTests/Fixtures/golden.json b/foreign/swift/Tests/IggyTests/Fixtures/golden.json index 0cb0e5413..5d81c5d16 100644 --- a/foreign/swift/Tests/IggyTests/Fixtures/golden.json +++ b/foreign/swift/Tests/IggyTests/Fixtures/golden.json @@ -712,6 +712,7 @@ "82": "invalid_number_encoding", "83": "invalid_boolean_value", "84": "invalid_number_value", + "85": "request_too_old", "100": "client_not_found", "101": "invalid_client_id", "206": "connection_closed",
