You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/30 13:22:01 UTC
(iggy) branch fix/partition-recovery-and-old-requests created (now 935a6d5f3)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch fix/partition-recovery-and-old-requests
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 935a6d5f3 fix(consensus): preserve recovery floors and reject aged-out requests
This branch includes the following new commits:
new 935a6d5f3 fix(consensus): preserve recovery floors and reject aged-out requests
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: fix(consensus): preserve recovery floors and reject aged-out requests
Posted by hg...@apache.org.
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 <h....@gmail.com>
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",