You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/09/03 08:36:04 UTC
(iggy) branch master updated: feat(server): dedup partition writes with per-group client table slices (#3959)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new bad22c03c feat(server): dedup partition writes with per-group client table slices (#3959)
bad22c03c is described below
commit bad22c03c64ae0b1b843a0c3f8375f26c6d38228
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Thu Sep 3 10:35:58 2026 +0200
feat(server): dedup partition writes with per-group client table slices (#3959)
Implement deduplication for partition plane operations.
---------
Co-authored-by: Piotr Gankiewicz <pi...@gmail.com>
Co-authored-by: Krishna Vishal <kr...@gmail.com>
Co-authored-by: Hubert Gruszecki <h....@gmail.com>
---
core/configs/src/server_config/defaults.rs | 1 +
core/configs/src/server_config/partition.rs | 52 ++
core/consensus/src/client_table.rs | 710 +++++++++++++++++++-
core/consensus/src/impls.rs | 65 +-
core/consensus/src/lib.rs | 20 +-
core/integration/tests/cluster/mod.rs | 1 +
core/integration/tests/cluster/partition_dedup.rs | 749 ++++++++++++++++++++++
core/partitions/src/iggy_partition.rs | 281 ++++++--
core/partitions/src/iggy_partitions.rs | 63 +-
core/partitions/src/state_transfer.rs | 300 ++++++++-
core/server/config.toml | 11 +
core/server/src/boot/recovery.rs | 3 +
core/server/src/dispatch/partition.rs | 121 +++-
core/server/src/http/session.rs | 33 +-
core/server/src/http/state.rs | 2 +-
core/server/src/http/submit.rs | 15 +-
core/server/src/partition_helpers.rs | 2 +
core/shard/src/lib.rs | 351 +++++++++-
core/shard/src/metrics.rs | 10 +-
core/shard/src/router.rs | 16 +-
core/simulator/src/lib.rs | 264 ++++----
scripts/ci/storage-compat.sh | 47 +-
22 files changed, 2824 insertions(+), 293 deletions(-)
diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs
index 1569cfada..743f25d91 100644
--- a/core/configs/src/server_config/defaults.rs
+++ b/core/configs/src/server_config/defaults.rs
@@ -176,6 +176,7 @@ impl Default for PartitionConfig {
let partition = &SERVER_CONFIG.partition;
PartitionConfig {
prepare_queue_depth: partition.prepare_queue_depth as usize,
+ dedup_clients_max: partition.dedup_clients_max as usize,
evicted_ring_capacity: partition.evicted_ring_capacity as usize,
evicted_ring_bytes_max: partition.evicted_ring_bytes_max.parse().unwrap(),
transfer_served_cache_bytes_max: partition
diff --git a/core/configs/src/server_config/partition.rs b/core/configs/src/server_config/partition.rs
index d6e327361..6d32c2232 100644
--- a/core/configs/src/server_config/partition.rs
+++ b/core/configs/src/server_config/partition.rs
@@ -113,6 +113,15 @@ pub const DEFAULT_EVICTED_RING_BYTES_MAX: u64 = 16 * 1024 * 1024;
/// trips first evicts; this byte ceiling is the second typo guard.
pub const MAX_EVICTED_RING_BYTES: u64 = 256 * 1024 * 1024;
+/// Shipped default for [`PartitionConfig::dedup_clients_max`]; pinned against
+/// the runtime constant by a bootstrap assert.
+pub const PARTITION_DEDUP_CLIENTS_DEFAULT: usize = 4096;
+
+/// Ceiling for [`PartitionConfig::dedup_clients_max`]. A per-group budget, so
+/// the ceiling bounds worst-case memory at roughly `partitions * this * 146
+/// bytes`: a 112-byte slot entry plus its index-map slot.
+pub const PARTITION_DEDUP_CLIENTS_CEILING: usize = 1 << 16;
+
/// Capacity tunables for the per-partition consensus plane.
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct PartitionConfig {
@@ -123,6 +132,18 @@ pub struct PartitionConfig {
/// pinned request-buffer memory by the partition count.
pub prepare_queue_depth: usize,
+ /// Distinct clients each partition group tracks request watermarks for,
+ /// deduplicating retried produces and consumer-offset writes. At capacity
+ /// the entry whose newest commit is oldest is evicted, which costs dedup
+ /// coverage for that client (its next replay re-executes, exactly as it
+ /// would have before dedup existed) and never correctness. Must be > 0 and
+ /// <= [`PARTITION_DEDUP_CLIENTS_CEILING`].
+ ///
+ /// Unlike `[metadata] clients_table_max`, this budget is PER GROUP, so the
+ /// worst case scales with partition count: size it to the producers a
+ /// single partition actually sees, not the node's client total.
+ pub dedup_clients_max: usize,
+
/// Entries the evicted ring retains per multi-replica partition for
/// journal repair after a peer rejoins. Larger widens the window a
/// restarting peer can be served from the ring before falling back to
@@ -177,6 +198,14 @@ impl Validatable<ConfigurationError> for PartitionConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if self.dedup_clients_max == 0 || self.dedup_clients_max > PARTITION_DEDUP_CLIENTS_CEILING {
+ eprintln!(
+ "{COMPONENT} partition.dedup_clients_max ({}) must be > 0 and <= \
+ {PARTITION_DEDUP_CLIENTS_CEILING}",
+ self.dedup_clients_max
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
if self.evicted_ring_capacity == 0 {
eprintln!("{COMPONENT} partition.evicted_ring_capacity must be > 0");
return Err(ConfigurationError::InvalidConfigurationValue);
@@ -253,6 +282,29 @@ mod tests {
);
}
+ #[test]
+ fn shipped_dedup_default_matches_the_runtime_constant() {
+ assert_eq!(
+ PartitionConfig::default().dedup_clients_max,
+ PARTITION_DEDUP_CLIENTS_DEFAULT,
+ "config.toml dedup_clients_max drifted from the runtime default"
+ );
+ }
+
+ #[test]
+ fn rejects_out_of_range_dedup_clients_max() {
+ for value in [0, PARTITION_DEDUP_CLIENTS_CEILING + 1] {
+ let config = PartitionConfig {
+ dedup_clients_max: value,
+ ..PartitionConfig::default()
+ };
+ assert!(
+ config.validate().is_err(),
+ "dedup_clients_max {value} must be rejected"
+ );
+ }
+ }
+
#[test]
fn rejects_zero_prepare_queue_depth() {
let config = PartitionConfig {
diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs
index 0737e23bf..e251219bc 100644
--- a/core/consensus/src/client_table.rs
+++ b/core/consensus/src/client_table.rs
@@ -23,6 +23,7 @@ use server_common::{
MESSAGE_ALIGN, Message,
iobuf::{Frozen, Owned},
};
+use std::cmp::Reverse;
use std::collections::{HashMap, VecDeque};
use std::fmt;
use std::mem::size_of;
@@ -247,6 +248,12 @@ struct ClientEntry {
/// first app op commits. Survives re-register: a resumed session keeps
/// 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
+ /// on the metadata plane, whose reply ring plays this role.
+ committed_window: u128,
/// `request_checksum` of the watermark request; catches a client reusing
/// a request id for a different operation. Zero when unstamped (integrity
/// fields are zeroed on the wire today), which disables the comparison.
@@ -508,6 +515,89 @@ pub enum CommitReply {
AdvancedFence,
}
+/// Which of the table's mechanisms an instance runs.
+///
+/// The metadata plane needs all of them. A partition group's slice needs only
+/// the watermark: it has no register to mint an epoch from, no result section
+/// worth caching, and one table per group rather than per node, so the
+/// preallocated slot array would reserve ~384 KiB per partition before a single
+/// client connects. The predicates below are the only two combinations that
+/// exist, so the mode is an enum rather than three independent flags.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ClientTableMode {
+ /// Metadata plane: replies cached, epoch fenced, slots preallocated.
+ Metadata,
+ /// One partition consensus group's slice: watermark only.
+ PartitionSlice,
+}
+
+impl ClientTableMode {
+ /// Keep committed replies so a duplicate replays the original bytes. Off:
+ /// duplicates answer [`RequestStatus::AlreadyApplied`] and the caller
+ /// synthesizes the reply.
+ #[must_use]
+ pub const fn cache_replies(self) -> bool {
+ matches!(self, Self::Metadata)
+ }
+
+ /// Enforce the register-minted epoch fence. Off: entries carry no epoch,
+ /// `check_request` ignores the presented one, and a committed request may
+ /// create its own entry (there is no register to do it).
+ #[must_use]
+ pub const fn fence_epoch(self) -> bool {
+ matches!(self, Self::Metadata)
+ }
+
+ /// Allocate every slot up front. Off: slots grow to the cap on demand.
+ /// Slot assignment is identical either way -- both hand out the lowest free
+ /// index -- so eviction order and the wire encoding are unchanged.
+ #[must_use]
+ pub const fn preallocate_slots(self) -> bool {
+ matches!(self, Self::Metadata)
+ }
+}
+
+/// One partition slice entry in its wire and install form.
+///
+/// Named fields rather than a tuple because `watermark` and `latest_commit`
+/// are both `u64` and a positional swap would decode cleanly into the wrong
+/// dedup decision.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DedupWatermark {
+ pub client: u128,
+ /// Acting user the watermark belongs to. A different user committing under
+ /// the same client id resets the entry rather than inheriting it: the id is
+ /// client-supplied (or, for HTTP, re-minted after a logout), so it alone is
+ /// not an identity.
+ pub user_id: u32,
+ /// Highest committed request number.
+ pub watermark: u64,
+ /// Commit op of the newest request folded in; the eviction rank.
+ pub latest_commit: u64,
+ /// Bit `i` set: request `watermark - i` committed. See
+ /// [`COMMITTED_WINDOW_BITS`].
+ pub committed_window: u128,
+}
+
+/// Width of the per-entry committed-request window below the watermark.
+///
+/// A client that pipelines writes can see one of them refused transiently and
+/// 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.
+///
+/// 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.
+pub const COMMITTED_WINDOW_BITS: u64 = 128;
+
/// VSR client table: per-session fence epoch + request-watermark dedup.
///
/// Fixed-size slot array (source of truth) + `HashMap` index (O(1) lookup).
@@ -530,11 +620,11 @@ pub enum CommitReply {
///
/// ## Plane
///
-/// Metadata-plane today. The design spans planes (one logical table,
-/// group-resident slices); partition-plane integration arrives once
-/// partition prepares carry real `(session_id, request)` instead of the
-/// transport id (data-plane request numbering, IGGY-137). Until then the
-/// partition plane stays at-least-once with no dedup.
+/// This table is the metadata plane's. The partition plane runs the same
+/// watermark rule in its own per-group slices ([`ClientTableMode::PartitionSlice`],
+/// held by `partitions::IggyPartition::dedup`), which keep no reply ring and no
+/// epoch: partition prepares carry the VSR client id and request number but no
+/// session, so fencing a stale session waits on identity surviving reconnects.
///
/// ## Tracking
///
@@ -560,19 +650,32 @@ pub enum CommitReply {
#[derive(Debug)]
pub struct ClientTable {
/// `None` = free slot. Deterministic iteration for eviction + serialization.
+ ///
+ /// Under [`ClientTableMode::preallocate_slots`] this is sized to
+ /// `clients_max` at construction; otherwise it grows to that cap on demand.
+ /// Every `Some` has exactly one `index` entry, so `index.len()` is the
+ /// occupied count.
slots: Vec<Option<ClientEntry>>,
/// `client_id` -> slot index. Rebuilt on decode.
index: HashMap<u128, usize>,
+ /// Slot ceiling. Tracked explicitly because `slots.len()` is the allocated
+ /// length, which only equals the cap when slots are preallocated.
+ clients_max: usize,
+ mode: ClientTableMode,
/// Fences of clients capacity eviction reclaimed, oldest at the front.
///
/// Bounded by the slot count. A fence is the entry's header fields plus, at
/// most, the watermark request's own reply, so it costs a fraction of the
/// entry it replaces. Trimmed oldest-first.
///
- /// Replica-local best-effort, NOT replicated state: the bound is
- /// `slots.len()`, which `from_snapshot` and `decode` size per node, and a
- /// state transfer replaces the table wholesale. Losing a fence degrades a
- /// resume to the pre-fence behaviour; it never makes one more permissive.
+ /// Replica-local best-effort, NOT replicated state: the bound is the slot
+ /// ceiling, which `from_snapshot` and `decode` size per node, and a state
+ /// transfer replaces the table wholesale. Losing a fence degrades a resume
+ /// to the pre-fence behaviour; it never makes one more permissive.
+ ///
+ /// Only a plane that mints epochs fills this: a fence exists so a later
+ /// register revives the evicted session's watermark, and a plane with no
+ /// register has nothing to revive it with.
evicted_fences: VecDeque<EvictedFence>,
}
@@ -590,11 +693,24 @@ impl ClientTable {
/// `max_clients` caps slots; index pre-sized to avoid rehash storms.
#[must_use]
pub fn new(max_clients: usize) -> Self {
- let mut slots = Vec::with_capacity(max_clients);
- slots.resize_with(max_clients, || None);
+ Self::with_mode(max_clients, ClientTableMode::Metadata)
+ }
+
+ /// `max_clients` caps slots; `mode` selects which mechanisms run.
+ #[must_use]
+ pub fn with_mode(max_clients: usize, mode: ClientTableMode) -> Self {
+ let (slots, index) = if mode.preallocate_slots() {
+ let mut slots = Vec::with_capacity(max_clients);
+ slots.resize_with(max_clients, || None);
+ (slots, HashMap::with_capacity(max_clients))
+ } else {
+ (Vec::new(), HashMap::new())
+ };
Self {
slots,
- index: HashMap::with_capacity(max_clients),
+ index,
+ clients_max: max_clients,
+ mode,
evicted_fences: VecDeque::new(),
}
}
@@ -611,7 +727,7 @@ impl ClientTable {
self.index.is_empty(),
"set_capacity must run before any client registers"
);
- *self = Self::new(max_clients);
+ *self = Self::with_mode(max_clients, self.mode);
}
/// Snapshot the table for the metadata checkpoint: every occupied slot with its
@@ -738,14 +854,18 @@ impl ClientTable {
user_id: entry.user_id,
watermark: entry.watermark,
watermark_checksum: entry.watermark_checksum,
+ committed_window: 0,
ring,
client_id: entry.client_id,
latest_commit,
});
}
+ let clients_max = slots.len();
let mut table = Self {
slots,
index,
+ clients_max,
+ mode: ClientTableMode::Metadata,
evicted_fences: VecDeque::with_capacity(snapshot.fences.len()),
};
for (position, fence) in snapshot.fences.into_iter().enumerate() {
@@ -793,7 +913,10 @@ impl ClientTable {
) -> RequestStatus {
assert!(client_id != 0, "client_id 0 is reserved for internal use");
// Header validation guarantees both > 0 at wire layer.
- debug_assert!(epoch > 0, "check_request: epoch must be > 0");
+ debug_assert!(
+ epoch > 0 || !self.mode.fence_epoch(),
+ "check_request: epoch must be > 0 when fencing"
+ );
debug_assert!(request > 0, "check_request: request must be > 0");
// Epoch check before request: a fenced zombie must be rejected even
@@ -803,17 +926,21 @@ impl ClientTable {
};
let entry = self.slots[slot_idx].as_ref().expect("index/slot mismatch");
- if epoch < entry.epoch {
- return RequestStatus::Fenced {
- current: entry.epoch,
- received: epoch,
- };
- }
- if epoch > entry.epoch {
- return RequestStatus::EpochAhead {
- current: entry.epoch,
- received: epoch,
- };
+ // A plane with no register mints no epoch, so there is nothing to
+ // fence against and the presented value is ignored.
+ if self.mode.fence_epoch() {
+ if epoch < entry.epoch {
+ return RequestStatus::Fenced {
+ current: entry.epoch,
+ received: epoch,
+ };
+ }
+ if epoch > entry.epoch {
+ return RequestStatus::EpochAhead {
+ current: entry.epoch,
+ received: epoch,
+ };
+ }
}
if request > entry.watermark {
@@ -917,7 +1044,7 @@ impl ClientTable {
// must never hand one user another user's dedup history, nor its
// cached reply bytes, merely because the key was reused.
let fence = self.take_fence(client_id, user_id);
- let freed = if self.index.len() >= self.slots.len() {
+ let freed = if self.index.len() >= self.clients_max {
self.evict_oldest()
} else {
None
@@ -946,6 +1073,7 @@ impl ClientTable {
.as_ref()
.map_or(REGISTER_REQUEST_ID, |fence| fence.watermark),
watermark_checksum: fence.as_ref().map_or(0, |fence| fence.watermark_checksum),
+ committed_window: 0,
ring,
});
self.index.insert(client_id, slot_idx);
@@ -1059,6 +1187,176 @@ impl ClientTable {
CommitReply::Cached
}
+ /// 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.
+ ///
+ /// # 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 {
+ 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;
+ };
+ let entry = self.slots[slot_idx].as_ref().expect("index/slot mismatch");
+ entry.user_id == user_id && request <= entry.watermark && entry.window_has(request)
+ }
+
+ /// Record a committed request without a reply to cache.
+ ///
+ /// 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.
+ ///
+ /// 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
+ /// what stops the next holder of a re-minted id from having its first
+ /// writes absorbed by the previous holder's watermark.
+ ///
+ /// # Panics
+ /// If called on a table whose mode caches replies -- that plane must go
+ /// through [`Self::commit_reply`] so the ring stays populated.
+ pub fn commit_request(&mut self, client_id: u128, user_id: u32, request: u64, commit_op: u64) {
+ debug_assert!(
+ !self.mode.cache_replies(),
+ "commit_request: a reply-caching table must use commit_reply"
+ );
+ // Zero is the reserved client id, refused at every ingress (wire
+ // validation, the HTTP minter, the auto-commit guard at the call
+ // sites). Kept as a return rather than an assert so that an artifact
+ // slipping past the decoder degrades to no dedup for that entry instead
+ // of taking the replica down.
+ if client_id == 0 {
+ return;
+ }
+
+ if let Some(&slot_idx) = self.index.get(&client_id) {
+ let entry = self.slots[slot_idx].as_mut().expect("index/slot mismatch");
+ if entry.user_id != user_id {
+ entry.user_id = user_id;
+ entry.watermark = request;
+ entry.committed_window = 1;
+ entry.latest_commit = commit_op;
+ } else if request > entry.watermark {
+ let gap = request - entry.watermark;
+ entry.committed_window = if gap >= COMMITTED_WINDOW_BITS {
+ 1
+ } else {
+ (entry.committed_window << gap) | 1
+ };
+ entry.watermark = request;
+ entry.latest_commit = commit_op;
+ } else {
+ let below = entry.watermark - request;
+ if below < COMMITTED_WINDOW_BITS && entry.committed_window & (1 << below) == 0 {
+ entry.committed_window |= 1 << below;
+ // Commits walk in op order, so a newly folded reordered id
+ // is the newest commit unless an install re-walk replays an
+ // older one.
+ entry.latest_commit = entry.latest_commit.max(commit_op);
+ }
+ }
+ return;
+ }
+
+ let freed = if self.index.len() >= self.clients_max {
+ self.evict_oldest()
+ } else {
+ None
+ };
+ let Some(slot_idx) = freed.or_else(|| self.first_free_slot()) else {
+ // Only reachable at a zero cap, which config validation rejects.
+ return;
+ };
+ self.index.insert(client_id, slot_idx);
+ self.slots[slot_idx] = Some(ClientEntry::watermark_only(
+ client_id, user_id, request, 1, commit_op,
+ ));
+ }
+
+ /// Replace every entry, as a state-transfer install does. An empty iterator
+ /// is the clear: there is no separate `clear`, and the one caller that
+ /// needs one (a failed install converging to empty) comes through here.
+ ///
+ /// The peer's cap may exceed this node's, so when the input is longer than
+ /// `clients_max` the entries with the newest commits survive, which is what
+ /// the oldest-commit eviction would have converged on had the surplus been
+ /// folded in one by one. Zero client ids are dropped, as
+ /// [`Self::commit_request`] drops them.
+ ///
+ /// # Panics
+ /// If called on a table whose mode caches replies (those install through
+ /// the snapshot / wire codecs, which carry the rings).
+ pub fn install_watermarks(&mut self, entries: impl IntoIterator<Item = DedupWatermark>) {
+ debug_assert!(
+ !self.mode.cache_replies(),
+ "install_watermarks: a reply-caching table installs via decode"
+ );
+ self.slots.clear();
+ self.index.clear();
+ let mut entries: Vec<DedupWatermark> = entries
+ .into_iter()
+ .filter(|entry| entry.client != 0)
+ .collect();
+ entries.sort_unstable_by_key(|entry| Reverse(entry.latest_commit));
+ entries.truncate(self.clients_max);
+ for entry in entries {
+ // The wire form is strictly ascending by client, so this only
+ // guards a caller-built iterator; the first (newest) copy wins.
+ if self.index.contains_key(&entry.client) {
+ continue;
+ }
+ self.index.insert(entry.client, self.slots.len());
+ self.slots.push(Some(ClientEntry::watermark_only(
+ entry.client,
+ entry.user_id,
+ entry.watermark,
+ entry.committed_window,
+ entry.latest_commit,
+ )));
+ }
+ }
+
+ /// Every entry ascending by client: the deterministic form a wire encoding
+ /// needs.
+ #[must_use]
+ pub fn watermarks_sorted(&self) -> Vec<DedupWatermark> {
+ let mut entries: Vec<DedupWatermark> = self
+ .slots
+ .iter()
+ .flatten()
+ .map(|entry| DedupWatermark {
+ client: entry.client_id,
+ user_id: entry.user_id,
+ watermark: entry.watermark,
+ latest_commit: entry.latest_commit,
+ committed_window: entry.committed_window,
+ })
+ .collect();
+ entries.sort_unstable_by_key(|entry| entry.client);
+ entries
+ }
+
/// Remove a client session and cached replies.
///
/// **LOCAL ONLY -- does NOT replicate.** Two correct call sites:
@@ -1152,7 +1450,7 @@ impl ClientTable {
/// latency shows up in the logs rather than as a silent re-execution.
#[must_use]
pub const fn fence_retention(&self) -> usize {
- self.slots.len()
+ self.clients_max
}
/// Evict the client whose latest cached reply has the oldest commit.
@@ -1212,6 +1510,12 @@ impl ClientTable {
/// Record an evicted entry's dedup fence, trimming oldest-first.
fn remember_fence(&mut self, entry: &ClientEntry) {
+ // Only a register revives a fence, and a plane that mints no epoch has
+ // none, so storing one there would cost a slot's worth of memory per
+ // group for something nothing can read back.
+ if !self.mode.fence_epoch() {
+ return;
+ }
// Nothing committed under this session, so there is nothing to dedup.
// Worth skipping rather than storing: `evict_oldest` ranks on the oldest
// `latest_commit`, and a session idle since its register carries its own
@@ -1310,8 +1614,25 @@ impl ClientTable {
self.evicted_fences.remove(position)
}
- fn first_free_slot(&self) -> Option<usize> {
- self.slots.iter().position(Option::is_none)
+ /// Lowest free slot, growing the array when slots are allocated lazily.
+ /// Assignment is identical to the preallocated case: both hand out the
+ /// lowest free index, so eviction order and the wire encoding do not
+ /// depend on the mode.
+ ///
+ /// The hole scan runs only when a hole exists (`index.len()` is the
+ /// occupied count): a lazily grown table below its cap has none, and
+ /// scanning it before every push would make the fill quadratic on the
+ /// commit path.
+ fn first_free_slot(&mut self) -> Option<usize> {
+ if self.index.len() < self.slots.len()
+ && let Some(index) = self.slots.iter().position(Option::is_none)
+ {
+ return Some(index);
+ }
+ (self.slots.len() < self.clients_max).then(|| {
+ self.slots.push(None);
+ self.slots.len() - 1
+ })
}
/// Latest cached reply for a client.
@@ -1649,6 +1970,7 @@ impl ClientTable {
user_id,
watermark,
watermark_checksum,
+ committed_window: 0,
ring,
client_id,
latest_commit,
@@ -1724,11 +2046,42 @@ impl ClientTable {
/// can exceed `[metadata] clients_table_max`.
#[must_use]
pub const fn capacity(&self) -> usize {
- self.slots.len()
+ self.clients_max
}
}
impl ClientEntry {
+ /// Entry for a plane that mints no epoch and caches no reply: the fields a
+ /// [`ClientTableMode::PartitionSlice`] table never reads stay at their
+ /// zero values. Bit 0 of the window is forced on: the watermark itself is
+ /// committed by definition.
+ const fn watermark_only(
+ client_id: u128,
+ user_id: u32,
+ watermark: u64,
+ committed_window: u128,
+ commit_op: u64,
+ ) -> Self {
+ Self {
+ epoch: 0,
+ user_id,
+ watermark,
+ watermark_checksum: 0,
+ committed_window: committed_window | 1,
+ ring: VecDeque::new(),
+ client_id,
+ latest_commit: commit_op,
+ }
+ }
+
+ /// 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
@@ -3002,6 +3355,303 @@ mod tests {
// Capacity resize (boot-only)
+ // --- ClientTableMode::PartitionSlice: watermark-only dedup ---
+ //
+ // One consensus group's slice. No register mints entries here, no reply is
+ // cached, and slots grow on demand, so these pin the behaviour the
+ // partition plane actually relies on.
+
+ const SLICE_USER: u32 = 3;
+ const OTHER_USER: u32 = 4;
+
+ fn slice(clients_max: usize) -> ClientTable {
+ ClientTable::with_mode(clients_max, ClientTableMode::PartitionSlice)
+ }
+
+ fn watermark(client: u128, watermark: u64, latest_commit: u64) -> DedupWatermark {
+ DedupWatermark {
+ client,
+ user_id: SLICE_USER,
+ watermark,
+ latest_commit,
+ committed_window: 1,
+ }
+ }
+
+ fn clients_of(table: &ClientTable) -> Vec<u128> {
+ table
+ .watermarks_sorted()
+ .into_iter()
+ .map(|entry| entry.client)
+ .collect()
+ }
+
+ #[test]
+ fn given_partition_slice_when_empty_should_admit_and_not_preallocate() {
+ // The reason this plane cannot use the metadata mode: one table per
+ // group, so preallocating the cap would reserve hundreds of KiB per
+ // partition before a single client connects.
+ 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));
+ }
+
+ #[test]
+ fn given_partition_slice_when_request_replayed_should_report_duplicate() {
+ // Only what committed is a duplicate: an id below the watermark that
+ // never committed is a reordered arrival and still executes.
+ 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));
+ }
+
+ #[test]
+ fn given_partition_slice_when_request_id_gaps_should_accept_the_jump() {
+ // One client counter feeds several groups, so a slice legitimately sees
+ // only a subset of the ids that client mints; the skipped ids stay
+ // admissible in case they were routed here late rather than elsewhere.
+ let mut table = slice(4);
+ 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));
+ }
+
+ #[test]
+ fn given_partition_slice_when_commit_replayed_should_be_idempotent() {
+ let mut table = slice(4);
+ table.commit_request(7, SLICE_USER, 5, 100);
+ table.commit_request(7, SLICE_USER, 5, 100);
+ table.commit_request(7, SLICE_USER, 5, 100);
+
+ assert_eq!(table.watermarks_sorted(), vec![watermark(7, 5, 100)]);
+ }
+
+ #[test]
+ fn given_partition_slice_when_lower_id_commits_late_should_admit_then_absorb() {
+ // A pipelining client had request 2 refused transiently and replays it
+ // after 3 committed: the replay is a new write, not a duplicate, and
+ // only once it commits does it read as one.
+ let mut table = slice(4);
+ table.commit_request(7, SLICE_USER, 1, 100);
+ table.commit_request(7, SLICE_USER, 3, 101);
+
+ assert!(!table.is_duplicate(7, SLICE_USER, 2));
+ 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_eq!(
+ table.watermarks_sorted(),
+ vec![DedupWatermark {
+ client: 7,
+ user_id: SLICE_USER,
+ watermark: 3,
+ latest_commit: 102,
+ committed_window: 0b111,
+ }]
+ );
+ }
+
+ #[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.
+ 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));
+ }
+
+ #[test]
+ fn given_partition_slice_when_watermark_jumps_should_shift_the_window() {
+ let mut table = slice(4);
+ table.commit_request(7, SLICE_USER, 1, 100);
+ table.commit_request(7, SLICE_USER, 2, 101);
+ table.commit_request(7, SLICE_USER, 5, 102);
+
+ // 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));
+ }
+
+ #[test]
+ fn given_partition_slice_when_other_user_commits_under_same_id_should_reset() {
+ // The id is client-supplied (or re-minted after an HTTP logout), so the
+ // previous holder's watermark must not absorb the next holder's writes.
+ let mut table = slice(4);
+ table.commit_request(7, SLICE_USER, u64::MAX, 100);
+
+ assert!(!table.is_duplicate(7, OTHER_USER, 1));
+ 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, SLICE_USER, 5),
+ "the previous holder's history is gone with the reset"
+ );
+ assert_eq!(table.count(), 1, "a reset reuses the slot");
+ }
+
+ #[test]
+ fn given_partition_slice_when_full_should_evict_the_oldest_commit() {
+ let mut table = slice(2);
+ table.commit_request(1, SLICE_USER, 1, 10);
+ table.commit_request(2, SLICE_USER, 1, 20);
+ table.commit_request(3, SLICE_USER, 1, 30);
+
+ assert_eq!(table.count(), 2);
+ assert_eq!(
+ clients_of(&table),
+ vec![2, 3],
+ "oldest commit is the victim"
+ );
+ }
+
+ #[test]
+ fn given_partition_slice_when_entry_evicted_should_admit_its_replay_again() {
+ // Losing an entry costs dedup coverage, never correctness: the replay
+ // re-executes exactly as it would have before the slice existed.
+ let mut table = slice(1);
+ table.commit_request(1, SLICE_USER, 5, 10);
+ table.commit_request(2, SLICE_USER, 1, 20);
+
+ assert!(!table.is_duplicate(1, SLICE_USER, 5));
+ }
+
+ #[test]
+ fn given_partition_slice_when_entry_touched_should_spare_it_from_eviction() {
+ let mut table = slice(2);
+ table.commit_request(1, SLICE_USER, 1, 10);
+ table.commit_request(2, SLICE_USER, 1, 20);
+ // Client 1 commits again, so client 2 now holds the oldest commit.
+ table.commit_request(1, SLICE_USER, 2, 30);
+ table.commit_request(3, SLICE_USER, 1, 40);
+
+ assert_eq!(clients_of(&table), vec![1, 3]);
+ }
+
+ #[test]
+ fn given_partition_slice_when_filled_to_cap_should_grow_without_holes() {
+ // The lazily grown array must hand out every index once and never
+ // rescan for a hole that cannot exist below the cap.
+ let mut table = slice(64);
+ for client in 1..=64u128 {
+ table.commit_request(client, SLICE_USER, 1, client as u64);
+ }
+
+ assert_eq!(table.count(), 64);
+ assert_eq!(table.slots.len(), 64);
+ assert!(table.slots.iter().all(Option::is_some));
+ }
+
+ #[test]
+ fn given_partition_slice_when_watermarks_installed_should_replace_not_merge() {
+ let mut table = slice(4);
+ table.commit_request(9, SLICE_USER, 3, 1);
+ table.install_watermarks([watermark(1, 4, 50), watermark(2, 7, 60)]);
+
+ assert_eq!(table.count(), 2);
+ assert!(
+ !table.is_duplicate(9, SLICE_USER, 3),
+ "install replaces rather than merges"
+ );
+ assert!(table.is_duplicate(1, SLICE_USER, 4));
+ assert!(!table.is_duplicate(2, SLICE_USER, 8));
+ }
+
+ #[test]
+ fn given_partition_slice_when_install_exceeds_cap_should_keep_newest_commits() {
+ // A peer with a larger cap ships more entries than fit; the survivors
+ // are the ones eviction would have converged on, not wire order.
+ let mut table = slice(2);
+ table.install_watermarks([
+ watermark(1, 1, 300),
+ watermark(2, 1, 100),
+ watermark(3, 1, 200),
+ ]);
+
+ assert_eq!(table.count(), 2);
+ assert_eq!(clients_of(&table), vec![1, 3]);
+ }
+
+ #[test]
+ fn given_partition_slice_when_install_carries_user_should_keep_it() {
+ let mut table = slice(4);
+ table.install_watermarks([DedupWatermark {
+ client: 1,
+ user_id: OTHER_USER,
+ watermark: 4,
+ latest_commit: 50,
+ committed_window: 1,
+ }]);
+
+ assert!(table.is_duplicate(1, OTHER_USER, 4));
+ assert!(!table.is_duplicate(1, SLICE_USER, 4));
+ }
+
+ #[test]
+ fn given_partition_slice_when_exported_should_sort_ascending_by_client() {
+ let mut table = slice(8);
+ for (commit_op, client) in [30u128, 10, 20].into_iter().enumerate() {
+ table.commit_request(client, SLICE_USER, 1, commit_op as u64);
+ }
+
+ assert_eq!(clients_of(&table), vec![10, 20, 30]);
+ }
+
+ #[test]
+ fn given_partition_slice_when_cleared_should_admit_everything() {
+ let mut table = slice(4);
+ table.commit_request(7, SLICE_USER, 5, 100);
+ table.install_watermarks(std::iter::empty());
+
+ assert_eq!(table.count(), 0);
+ assert!(!table.is_duplicate(7, SLICE_USER, 5));
+ }
+
+ #[test]
+ fn given_partition_slice_when_client_is_reserved_zero_should_record_nothing() {
+ // Zero is reserved cluster-wide and refused at every ingress; a commit
+ // or install that still carries it degrades to no entry, not a panic.
+ let mut table = slice(4);
+ table.commit_request(0, SLICE_USER, 5, 100);
+ table.install_watermarks([watermark(0, 5, 100), watermark(1, 1, 101)]);
+
+ assert_eq!(clients_of(&table), vec![1]);
+ }
+
+ #[cfg(debug_assertions)]
+ #[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);
+ }
+
+ #[cfg(debug_assertions)]
+ #[test]
+ #[should_panic(expected = "a reply-caching table must use commit_reply")]
+ fn given_metadata_table_when_commit_request_called_should_panic() {
+ ClientTable::new(4).commit_request(7, SLICE_USER, 1, 1);
+ }
+
// Resizing an empty table swaps its slot count in: a smaller cap then
// evicts once the new bound is reached.
#[test]
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 3f732bb48..5ad273c35 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -176,6 +176,14 @@ pub const PROBE_ATTEMPTS_MAX: u32 = 5;
/// When exceeded, the client with the oldest committed request is evicted.
pub const CLIENTS_TABLE_MAX: usize = 8192;
+/// Default live dedup entries per PARTITION consensus group.
+///
+/// Far below [`CLIENTS_TABLE_MAX`] because this budget is per group rather than
+/// per node: the worst case scales with partition count, so it is sized to the
+/// producers one partition sees. Pinned against the
+/// `[partition] dedup_clients_max` default by a bootstrap assert.
+pub const PARTITION_DEDUP_CLIENTS_MAX: usize = 4096;
+
#[derive(Debug)]
pub struct PipelineEntry {
pub header: PrepareHeader,
@@ -286,13 +294,10 @@ pub struct RequestEntry {
}
impl RequestEntry {
+ /// Queued request on the network reply path: no in-process subscriber.
#[must_use]
pub const fn new(message: Message<RoutedRequestHeader>) -> Self {
- Self {
- message,
- received_at: 0,
- reply_sender: None,
- }
+ Self::with_sender(message, None)
}
/// Queued request paired with a fresh receiver that resolves when the
@@ -305,12 +310,22 @@ impl RequestEntry {
message: Message<RoutedRequestHeader>,
) -> (Self, Receiver<Message<ReplyHeader>>) {
let (sender, receiver) = oneshot::channel();
- let entry = Self {
+ (Self::with_sender(message, Some(sender)), receiver)
+ }
+
+ /// Queued request carrying a sender the caller already owns, for a submit
+ /// that parked before reaching a prepare slot. `None` is the network reply
+ /// path; the other two constructors are this one with a fixed sender.
+ #[must_use]
+ pub const fn with_sender(
+ message: Message<RoutedRequestHeader>,
+ reply_sender: Option<Sender<Message<ReplyHeader>>>,
+ ) -> Self {
+ Self {
message,
received_at: 0,
- reply_sender: Some(sender),
- };
- (entry, receiver)
+ reply_sender,
+ }
}
/// Take the reply sender for hand-off to the promoted pipeline entry.
@@ -656,6 +671,24 @@ impl LocalPipeline {
.any(|r| r.message.header().client == client)
}
+ /// True if either queue already holds this exact `(client, request)`.
+ ///
+ /// The partition-plane in-flight check. Narrower than
+ /// [`Self::has_message_from_client`] on purpose: the partition pipeline is
+ /// depth-`prepare_queue_depth` by design, so blocking every concurrent
+ /// request from one client would serialize it to one in-flight write per
+ /// group. Only an exact replay needs absorbing.
+ #[must_use]
+ pub fn has_message_from_client_request(&self, client: u128, request: u64) -> bool {
+ self.prepare_queue
+ .iter()
+ .any(|p| p.header.client == client && p.header.request == request)
+ || self.request_queue.iter().any(|r| {
+ let header = r.message.header();
+ header.client == client && header.request == request
+ })
+ }
+
/// Verify pipeline invariants.
///
/// # Panics
@@ -769,6 +802,10 @@ impl Pipeline for LocalPipeline {
Self::has_message_from_client(self, client_id)
}
+ fn has_message_from_client_request(&self, client_id: u128, request: u64) -> bool {
+ Self::has_message_from_client_request(self, client_id, request)
+ }
+
fn cancel_all_subscribers(&mut self) {
Self::cancel_all_subscribers(self);
}
@@ -1763,6 +1800,16 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
self.pipeline.borrow().has_message_from_client(client_id)
}
+ /// True iff this exact `(client, request)` is already in flight. The
+ /// partition plane's in-flight dedup: absorbs a replay without serializing
+ /// a client's pipeline depth.
+ #[must_use]
+ pub fn pipeline_has_message_from_client_request(&self, client_id: u128, request: u64) -> bool {
+ self.pipeline
+ .borrow()
+ .has_message_from_client_request(client_id, request)
+ }
+
/// Header of the oldest in-flight prepare.
#[must_use]
pub fn pipeline_head_header(&self) -> Option<PrepareHeader> {
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index d5647b52f..0abad9657 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -70,13 +70,22 @@ pub trait Pipeline {
fn verify(&self);
/// True iff either queue carries `client_id`. Used by metadata-plane
- /// preflight for in-flight dedup. Partition plane is at-least-once
- /// and skips. Default `false`; falls through to slot dedup in
+ /// preflight for in-flight dedup; the partition plane uses the narrower
+ /// [`Self::has_message_from_client_request`] instead, to keep a client's
+ /// pipeline depth. Default `false`; falls through to slot dedup in
/// `check_request`.
fn has_message_from_client(&self, _client_id: u128) -> bool {
false
}
+ /// True iff either queue carries this exact `(client, request)`. The
+ /// partition-plane in-flight dedup check: narrow on purpose, so a client
+ /// keeps its pipeline depth and only an exact replay is absorbed.
+ /// Default `false`.
+ fn has_message_from_client_request(&self, _client_id: u128, _request: u64) -> bool {
+ false
+ }
+
/// Drop reply senders on every entry; receivers wake `Canceled`.
/// View-change reset uses this to unblock awaiters while preserving
/// pipeline for DVC reconciliation.
@@ -161,8 +170,9 @@ where
pub mod client_table;
pub mod le_cursor;
pub use client_table::{
- CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableSnapshot,
- ClientTableWireError, CommitReply, DISCONNECT_LOGOUT_REQUEST_ID, FenceSnapshot, SessionEnd,
+ CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableMode,
+ ClientTableSnapshot, ClientTableWireError, CommitReply, DISCONNECT_LOGOUT_REQUEST_ID,
+ DedupWatermark, FenceSnapshot, SessionEnd,
};
pub mod state_manifest;
pub use state_manifest::{
@@ -176,7 +186,7 @@ pub use state_transfer::{
};
// One-shot per `PipelineEntry` for in-process commit awaiters.
pub(crate) mod oneshot;
-pub use oneshot::{Canceled, Receiver};
+pub use oneshot::{Canceled, Receiver, Sender, channel as oneshot_channel};
mod fatal;
pub use fatal::{FatalReason, fatal};
diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs
index b794264a0..57359759f 100644
--- a/core/integration/tests/cluster/mod.rs
+++ b/core/integration/tests/cluster/mod.rs
@@ -26,6 +26,7 @@ mod metadata_checkpoint_restart;
mod metadata_state_transfer;
mod multi_shard_partition_convergence;
mod parked_frame_redispatch;
+mod partition_dedup;
mod partition_primary_routing;
mod partition_state_transfer;
mod register_forwarding;
diff --git a/core/integration/tests/cluster/partition_dedup.rs b/core/integration/tests/cluster/partition_dedup.rs
new file mode 100644
index 000000000..892211b74
--- /dev/null
+++ b/core/integration/tests/cluster/partition_dedup.rs
@@ -0,0 +1,749 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Spec tests for partition-plane request dedup (IGGY-274).
+//!
+//! Each partition consensus group keeps a slice of the VSR client table:
+//! per-client request watermarks folded in at commit. A replay of an
+//! already-committed `(client, request)` is answered with the empty success its
+//! original earned instead of committing a second copy.
+//!
+//! The frames are hand-crafted on a raw TCP socket for the same reason
+//! `client_table_restart` does it: the Rust SDK mints a fresh `client_id` and
+//! request id per attempt, so it cannot express "the same request, twice" --
+//! which is precisely the input under test. The SDK is still used for setup and
+//! for reading the log back, where it is the more honest observer.
+
+use bytes::{Bytes, BytesMut};
+use futures::future::join_all;
+use iggy::prelude::*;
+use iggy_binary_protocol::codec::WireEncode;
+use iggy_binary_protocol::consensus::{
+ Command, Operation, ReplyHeader, RequestHeader, read_size_field,
+};
+use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::send_messages::{RawMessage, SendMessagesEncoder};
+use iggy_binary_protocol::requests::users::LoginRegisterRequest;
+use iggy_binary_protocol::{
+ AckLevel, ClientVersionInfo, HEADER_SIZE, IGGY_PROTOCOL_VERSION, WireConsumer, WireIdentifier,
+ WireName, WirePartitioning,
+};
+use integration::harness::TestHarness;
+use integration::iggy_harness;
+use secrecy::SecretString;
+use std::mem::offset_of;
+use std::net::SocketAddr;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::time::{Instant, sleep, timeout};
+
+const STREAM_NAME: &str = "partition-dedup-stream";
+const TOPIC_NAME: &str = "partition-dedup-topic";
+const PARTITION_ID: u32 = 0;
+
+/// Fixed wire identity, so the replay frame is byte-identical to the original.
+/// The SDK would randomize this.
+const CLIENT_ID: u128 = 0x0DED_1234_5678;
+
+/// Second identity for liveness probes. The dedup watermark is a per-client
+/// max, so a probe under [`CLIENT_ID`] would raise that client's watermark and
+/// mask a missing transfer; the probe must not touch the identity under test.
+const PROBE_CLIENT_ID: u128 = 0x0DED_9999_0001;
+
+const REPLY_WAIT: Duration = Duration::from_secs(10);
+const COMMIT_BUDGET: Duration = Duration::from_secs(20);
+const RETRY_PAUSE: Duration = Duration::from_millis(100);
+
+#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_committed_send_when_replayed_should_absorb_without_a_second_copy(
+ harness: &mut TestHarness,
+) {
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client");
+ seed_topic(&client).await;
+
+ let addr = harness.node(0).tcp_addr().expect("node tcp address");
+ let (mut stream, session) = register(addr).await;
+
+ let body = send_messages_body(b"only-once");
+ let header = request_header(Operation::SendMessages, session, 1, body.len());
+
+ let original = exchange_until_committed(&mut stream, &header, &body).await;
+ assert_eq!(original, 0, "the original send must commit");
+
+ // Byte-identical replay: what a retry after a lost reply looks like.
+ let replayed = exchange_until_committed(&mut stream, &header, &body).await;
+ assert_eq!(
+ replayed, 0,
+ "an absorbed duplicate is a success, not an error"
+ );
+
+ let polled = poll_all(&client).await;
+ assert_eq!(
+ polled, 1,
+ "the replayed send must not append a second copy (got {polled} messages)"
+ );
+}
+
+#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_committed_send_when_next_request_id_arrives_should_admit_it(
+ harness: &mut TestHarness,
+) {
+ // The watermark must not wedge the client: the id above it still commits.
+ // Without this, "dedup works" and "the plane is broken" look identical.
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client");
+ seed_topic(&client).await;
+
+ let addr = harness.node(0).tcp_addr().expect("node tcp address");
+ let (mut stream, session) = register(addr).await;
+
+ for request in 1..=3u64 {
+ let body = send_messages_body(format!("message-{request}").as_bytes());
+ let header = request_header(Operation::SendMessages, session, request, body.len());
+ let status = exchange_until_committed(&mut stream, &header, &body).await;
+ assert_eq!(status, 0, "request {request} must commit");
+ }
+
+ let polled = poll_all(&client).await;
+ assert_eq!(polled, 3, "each distinct request id must append once");
+}
+
+#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_gapped_request_id_when_sent_should_commit(harness: &mut TestHarness) {
+ // One client counter feeds every group it writes to, so a slice only ever
+ // sees a subset of the ids minted. Gaps must be legal, not a wedge.
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client");
+ seed_topic(&client).await;
+
+ let addr = harness.node(0).tcp_addr().expect("node tcp address");
+ let (mut stream, session) = register(addr).await;
+
+ for request in [1u64, 9, 40] {
+ let body = send_messages_body(format!("gap-{request}").as_bytes());
+ let header = request_header(Operation::SendMessages, session, request, body.len());
+ let status = exchange_until_committed(&mut stream, &header, &body).await;
+ assert_eq!(status, 0, "gapped request {request} must commit");
+ }
+
+ let polled = poll_all(&client).await;
+ assert_eq!(polled, 3, "a gapped id is new, not a duplicate");
+}
+
+#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_committed_consumer_offset_when_replayed_should_absorb(harness: &mut TestHarness) {
+ // Dedup covers every replicated partition write, not just produces. A
+ // replayed offset store must answer success rather than committing twice.
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client");
+ seed_topic(&client).await;
+
+ let addr = harness.node(0).tcp_addr().expect("node tcp address");
+ let (mut stream, session) = register(addr).await;
+
+ // Seed a message so offset 0 is in range for the store.
+ let produce = send_messages_body(b"seed");
+ let produce_header = request_header(Operation::SendMessages, session, 1, produce.len());
+ assert_eq!(
+ exchange_until_committed(&mut stream, &produce_header, &produce).await,
+ 0,
+ "the seed produce must commit"
+ );
+
+ let body = store_offset_body(0);
+ let header = request_header(Operation::StoreConsumerOffset, session, 2, body.len());
+
+ let original = exchange_until_committed(&mut stream, &header, &body).await;
+ assert_eq!(original, 0, "the original offset store must commit");
+
+ let replayed = exchange_until_committed(&mut stream, &header, &body).await;
+ assert_eq!(
+ replayed, 0,
+ "a replayed offset store is absorbed as a success"
+ );
+
+ // The next id still gets through: the watermark must not wedge the client.
+ let next = store_offset_body(0);
+ let next_header = request_header(Operation::StoreConsumerOffset, session, 3, next.len());
+ assert_eq!(
+ exchange_until_committed(&mut stream, &next_header, &next).await,
+ 0,
+ "the id above the watermark must still commit"
+ );
+}
+
+/// More connections than the prepare queue holds, each with one write in
+/// flight at the same instant, so the surplus parks in the request queue and is
+/// promoted into a prepare slot at a later commit. Every one of them must be
+/// answered: a write promoted without the reply sender it parked with commits
+/// but leaves its connection waiting out a timeout, and the count is the
+/// second discriminator (each writer's single id must commit exactly once).
+const CONCURRENT_WRITERS: u64 = 40;
+
+/// Distinct identity per writer, so each connection's watermark is its own and
+/// the request ids can all be 1.
+const WRITER_CLIENT_BASE: u128 = 0x0DED_C0DE_0000;
+
+#[iggy_harness(
+ cluster_nodes = 3,
+ server(
+ system.sharding.cpu_allocation = "0..1",
+ partition.prepare_queue_depth = "4"
+ )
+)]
+async fn given_more_writers_than_prepare_slots_when_all_send_at_once_should_answer_every_one(
+ harness: &mut TestHarness,
+) {
+ // Three nodes so a prepare needs a replication round trip to commit and the
+ // pipeline actually fills; a solo primary self-acks per frame and never
+ // exposes the request queue. The queue depth is pinned low so forty writers
+ // overflow it deterministically rather than by timing luck.
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client");
+ seed_topic(&client).await;
+
+ let addr = harness.node(0).tcp_addr().expect("node tcp address");
+ // Register every connection first so the writes race each other, not the
+ // logins.
+ let mut connections = Vec::with_capacity(CONCURRENT_WRITERS as usize);
+ for writer in 0..CONCURRENT_WRITERS {
+ let client_id = WRITER_CLIENT_BASE + u128::from(writer);
+ let (stream, session) = register_client_with_budget(addr, client_id, COMMIT_BUDGET).await;
+ connections.push((client_id, stream, session));
+ }
+
+ let sends = connections
+ .iter_mut()
+ .map(|(client_id, stream, session)| async move {
+ let body = send_messages_body(format!("writer-{client_id:x}").as_bytes());
+ let header =
+ request_header_for(*client_id, Operation::SendMessages, *session, 1, body.len());
+ exchange_with_budget(stream, &header, &body, COMMIT_BUDGET).await
+ });
+ let statuses = join_all(sends).await;
+ for (writer, status) in statuses.into_iter().enumerate() {
+ assert_eq!(
+ status, 0,
+ "writer {writer} must be answered with its commit (got status {status})"
+ );
+ }
+
+ let polled = poll_all(&client).await;
+ assert_eq!(
+ u64::from(polled),
+ CONCURRENT_WRITERS,
+ "every writer's single request must commit exactly once"
+ );
+}
+
+/// `StoreConsumerOffset` body for the raw connection's own consumer id.
+fn store_offset_body(offset: u64) -> Bytes {
+ StoreConsumerOffsetRequest {
+ consumer: WireConsumer::consumer(WireIdentifier::numeric(1)),
+ stream_id: WireIdentifier::named(STREAM_NAME).expect("stream identifier"),
+ topic_id: WireIdentifier::named(TOPIC_NAME).expect("topic identifier"),
+ partition_id: Some(PARTITION_ID),
+ offset,
+ ack: AckLevel::Quorum,
+ }
+ .to_bytes()
+}
+
+/// State-transfer choreography end to end: rejoin, view changes, and the
+/// final commits ride slow CI runners.
+const TRANSFER_BUDGET: Duration = Duration::from_secs(60);
+const FINAL_COMMIT_BUDGET: Duration = Duration::from_secs(120);
+const MARKER_POLL: Duration = Duration::from_millis(200);
+const INSTALL_MARKER: &str = "partition state transfer installed";
+
+/// Pre-stop produces fold into every replica's slice live; the rest commit
+/// while node 2 is down. Total must push the evicted ring (capacity 64) past
+/// the rejoiner's durable end, or repair closes the gap and no transfer runs.
+const PRE_STOP_SENDS: u64 = 40;
+/// The identity under test stops sending here; everything after comes from the
+/// filler client. The rejoiner's tail repair re-applies the ring window (the
+/// LAST ~64 commits) through the ordinary commit path, and the watermark is a
+/// per-client max -- so if the tested client appeared anywhere in that window,
+/// repair alone would cover every lower id and the artifact would be
+/// redundant. The filler pushes the tested client's last send out of the ring,
+/// leaving the transferred artifact as node 2's ONLY source for it.
+const TESTED_CLIENT_SENDS: u64 = 140;
+const FILLER_SENDS: u64 = 100;
+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;
+
+/// Filler identity whose sends evict the tested client from the repair ring.
+const FILLER_CLIENT_ID: u128 = 0x0DED_F111_E400;
+
+/// Sentinel status for an Eviction frame: the connection's session is gone and
+/// the caller must reconnect and re-register before retrying.
+const EVICTED: u32 = u32::MAX;
+
+/// Sentinel status for a socket the server closed mid-exchange (a node stopped
+/// under the connection). Same contract as [`EVICTED`]: reconnect, re-register,
+/// retry the identical frame.
+const DISCONNECTED: u32 = u32::MAX - 1;
+
+#[iggy_harness(
+ cluster_nodes = 3,
+ server(
+ system.sharding.cpu_allocation = "0..1",
+ partition.evicted_ring_capacity = "64"
+ )
+)]
+async fn given_transferred_dedup_slice_when_old_request_replays_should_absorb(
+ harness: &mut TestHarness,
+) {
+ // Phase 1: node 0 is every group's view-0 primary. Produce the pre-stop
+ // window with node 2 live, then the rest with it stopped, so the second
+ // window exists on node 2 only via state transfer.
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client");
+ seed_topic(&client).await;
+ let addr = harness.node(0).tcp_addr().expect("node 0 tcp address");
+ let (mut stream, session) = register(addr).await;
+ raw_produce(&mut stream, session, 1..=PRE_STOP_SENDS, COMMIT_BUDGET).await;
+ sleep(Duration::from_secs(1)).await;
+ harness.stop_node(2).expect("stop node 2");
+ raw_produce(
+ &mut stream,
+ session,
+ (PRE_STOP_SENDS + 1)..=TESTED_CLIENT_SENDS,
+ COMMIT_BUDGET,
+ )
+ .await;
+ drop(stream);
+ let (mut filler, filler_session) =
+ register_client_with_budget(addr, FILLER_CLIENT_ID, COMMIT_BUDGET).await;
+ raw_produce_for(
+ &mut filler,
+ FILLER_CLIENT_ID,
+ filler_session,
+ 1..=FILLER_SENDS,
+ COMMIT_BUDGET,
+ )
+ .await;
+ drop(filler);
+ drop(client);
+
+ // Phase 2: the rejoin cannot repair past the survivors' evicted ring, so
+ // it converts to state transfer; the install carries the dedup section.
+ harness.restart_node(2).expect("restart node 2");
+ await_marker(harness, 2, INSTALL_MARKER).await;
+
+ // Phase 3: walk the primaries off node 0 and node 1 so the REPLAY is
+ // admitted by the transferred node. Stopping node 0 elects node 1
+ // (view 1); after node 0 rejoins, stopping node 1 elects node 2 (view 2)
+ // with quorum {0, 2}.
+ harness.stop_node(0).expect("stop node 0");
+ sleep(Duration::from_secs(2)).await;
+ harness.restart_node(0).expect("restart node 0");
+ sleep(Duration::from_secs(2)).await;
+ harness.stop_node(1).expect("stop node 1");
+
+ // Phase 4, on node 2. The probe send goes FIRST and under a DIFFERENT
+ // client: its commit proves the view settled on node 2 and the rejoined
+ // node 0 is acking, and it pins the expected count -- while leaving
+ // CLIENT_ID's watermark exactly what the transfer installed (the watermark
+ // is a per-client max, so a same-client probe would mask a missing
+ // transfer). Sends reconnect + re-register on eviction; dedup keys on the
+ // client id and must hold across a re-register.
+ let addr = harness.node(2).tcp_addr().expect("node 2 tcp address");
+ let fresh = send_reconnecting(addr, PROBE_CLIENT_ID, 1, FINAL_COMMIT_BUDGET).await;
+ assert_eq!(
+ fresh, 0,
+ "the probe client's send must commit on the new primary"
+ );
+
+ let replayed = send_reconnecting(addr, CLIENT_ID, REPLAYED_REQUEST, FINAL_COMMIT_BUDGET).await;
+ assert_eq!(
+ replayed, 0,
+ "a replay of a transferred watermark is absorbed as a success"
+ );
+
+ // 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.
+ let client = harness
+ .root_client_for_node(2)
+ .await
+ .expect("connect a root client to node 2");
+ let polled = poll_up_to(&client, (TOTAL_SENDS + 16) as u32).await;
+ assert_eq!(
+ u64::from(polled),
+ TOTAL_SENDS + 1,
+ "the transferred slice must absorb the replay instead of re-executing it"
+ );
+}
+
+/// One send under `request`, surviving evictions: reconnect, re-register, and
+/// retry the identical frame until it answers or the budget runs out.
+async fn send_reconnecting(addr: SocketAddr, client: u128, request: u64, budget: Duration) -> u32 {
+ let deadline = Instant::now() + budget;
+ let body = send_messages_body(format!("send-{request}").as_bytes());
+ loop {
+ let remaining = deadline.saturating_duration_since(Instant::now());
+ assert!(
+ remaining > Duration::ZERO,
+ "request {request} did not resolve within {budget:?}"
+ );
+ let (mut stream, session) = register_client_with_budget(addr, client, remaining).await;
+ let header = request_header_for(
+ client,
+ Operation::SendMessages,
+ session,
+ request,
+ body.len(),
+ );
+ let status = exchange_with_budget(&mut stream, &header, &body, remaining).await;
+ if status != EVICTED && status != DISCONNECTED {
+ return status;
+ }
+ sleep(RETRY_PAUSE).await;
+ }
+}
+
+/// Produce one single-message batch per request id over the lockstep raw
+/// connection, waiting out each commit.
+async fn raw_produce(
+ stream: &mut TcpStream,
+ session: u64,
+ requests: std::ops::RangeInclusive<u64>,
+ budget: Duration,
+) {
+ raw_produce_for(stream, CLIENT_ID, session, requests, budget).await;
+}
+
+async fn raw_produce_for(
+ stream: &mut TcpStream,
+ client: u128,
+ session: u64,
+ requests: std::ops::RangeInclusive<u64>,
+ budget: Duration,
+) {
+ for request in requests {
+ let body = send_messages_body(format!("send-{request}").as_bytes());
+ let header = request_header_for(
+ client,
+ Operation::SendMessages,
+ session,
+ request,
+ body.len(),
+ );
+ let status = exchange_with_budget(stream, &header, &body, budget).await;
+ assert_ne!(
+ status, DISCONNECTED,
+ "server closed the lockstep connection under request {request}"
+ );
+ assert_eq!(status, 0, "request {request} must commit");
+ }
+}
+
+async fn await_marker(harness: &TestHarness, node: usize, marker: &str) {
+ let deadline = Instant::now() + TRANSFER_BUDGET;
+ while !harness.node(node).stdout_contains(marker) {
+ assert!(
+ Instant::now() < deadline,
+ "node {node} never logged {marker:?} within {TRANSFER_BUDGET:?}"
+ );
+ sleep(MARKER_POLL).await;
+ }
+}
+
+async fn seed_topic(client: &IggyClient) {
+ client
+ .create_stream(STREAM_NAME)
+ .await
+ .expect("create stream");
+ let stream_id = Identifier::named(STREAM_NAME).expect("stream identifier");
+ client
+ .create_topic(
+ &stream_id,
+ TOPIC_NAME,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ // Every commit flushes and ring-evicts, which is what marches
+ // the repair floor past a rejoiner and forces the transfer the
+ // transferred-slice spec depends on.
+ messages_required_to_save: Some(1),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+}
+
+async fn poll_all(client: &IggyClient) -> u32 {
+ poll_up_to(client, 100).await
+}
+
+async fn poll_up_to(client: &IggyClient, max: u32) -> u32 {
+ let stream_id = Identifier::named(STREAM_NAME).expect("stream identifier");
+ let topic_id = Identifier::named(TOPIC_NAME).expect("topic identifier");
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(PARTITION_ID),
+ &Consumer::new(Identifier::numeric(1).expect("consumer identifier")),
+ &PollingStrategy::offset(0),
+ max,
+ false,
+ )
+ .await
+ .expect("poll messages")
+ .messages
+ .len() as u32
+}
+
+/// Full `SendMessages` body: metadata prefix, batch header, one message.
+fn send_messages_body(payload: &[u8]) -> Bytes {
+ let stream_id = WireIdentifier::named(STREAM_NAME).expect("stream identifier");
+ let topic_id = WireIdentifier::named(TOPIC_NAME).expect("topic identifier");
+ let partitioning = WirePartitioning::PartitionId(PARTITION_ID);
+ let messages = [RawMessage {
+ // A fixed id keeps the replay byte-identical; a zero would be
+ // server-stamped and the two frames would diverge.
+ id: 0x5EED,
+ origin_timestamp: 0,
+ headers: None,
+ payload,
+ }];
+ let size = SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages);
+ let mut buf = BytesMut::with_capacity(size);
+ SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages)
+ .expect("encode send_messages body");
+ buf.freeze()
+}
+
+fn request_header(
+ operation: Operation,
+ session: u64,
+ request: u64,
+ body_len: usize,
+) -> RequestHeader {
+ request_header_for(CLIENT_ID, operation, session, request, body_len)
+}
+
+fn request_header_for(
+ client: u128,
+ operation: Operation,
+ session: u64,
+ request: u64,
+ body_len: usize,
+) -> RequestHeader {
+ RequestHeader {
+ command: Command::Request,
+ operation,
+ size: u32::try_from(HEADER_SIZE + body_len).unwrap(),
+ client,
+ session,
+ request,
+ ..Default::default()
+ }
+}
+
+/// Exchange until the server stops answering transiently, returning the reply
+/// status. A transient means the request was never admitted, so replaying it
+/// keeps the same id -- exactly what the SDK's own retry loop does.
+async fn exchange_until_committed(
+ stream: &mut TcpStream,
+ header: &RequestHeader,
+ body: &Bytes,
+) -> u32 {
+ let status = exchange_with_budget(stream, header, body, COMMIT_BUDGET).await;
+ assert_ne!(
+ status, DISCONNECTED,
+ "server closed the lockstep connection under request {}",
+ header.request
+ );
+ status
+}
+
+async fn exchange_with_budget(
+ stream: &mut TcpStream,
+ header: &RequestHeader,
+ body: &Bytes,
+ budget: Duration,
+) -> u32 {
+ let deadline = Instant::now() + budget;
+ loop {
+ let status = exchange(stream, header, body).await;
+ if !is_transient(status) {
+ return status;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "request {} stayed transient for {budget:?}",
+ header.request
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+}
+
+/// Write one frame, read one frame, return the reply status. The connection is
+/// lockstep, so the reply that comes back is this request's. A socket the
+/// server closed (a node stopping under the connection) answers
+/// [`DISCONNECTED`] rather than panicking, so the reconnecting callers can
+/// treat it like an eviction; a reply that never comes is still a failure.
+async fn exchange(stream: &mut TcpStream, header: &RequestHeader, body: &Bytes) -> u32 {
+ if stream.write_all(bytemuck::bytes_of(header)).await.is_err() {
+ return DISCONNECTED;
+ }
+ if !body.is_empty() && stream.write_all(body).await.is_err() {
+ return DISCONNECTED;
+ }
+
+ let mut reply_header = [0u8; HEADER_SIZE];
+ match timeout(REPLY_WAIT, stream.read_exact(&mut reply_header)).await {
+ Ok(Ok(_)) => {}
+ Ok(Err(_)) => return DISCONNECTED,
+ Err(_) => panic!("reply header timed out"),
+ }
+
+ let command_offset = offset_of!(RequestHeader, command);
+ if reply_header[command_offset] == Command::Eviction as u8 {
+ // The session died (view change, epoch fence): the contract is
+ // reconnect + re-register, and dedup must still hold because it keys
+ // on the client id, not the session.
+ return EVICTED;
+ }
+ assert_eq!(
+ reply_header[command_offset],
+ Command::Reply as u8,
+ "expected a Reply frame"
+ );
+
+ let status_offset = offset_of!(ReplyHeader, status);
+ let status = u32::from_le_bytes(
+ reply_header[status_offset..status_offset + 4]
+ .try_into()
+ .unwrap(),
+ );
+ let total_size = read_size_field(&reply_header).expect("reply size field") as usize;
+ if total_size > HEADER_SIZE {
+ let mut discard = vec![0u8; total_size - HEADER_SIZE];
+ match timeout(REPLY_WAIT, stream.read_exact(&mut discard)).await {
+ Ok(Ok(_)) => {}
+ Ok(Err(_)) => return DISCONNECTED,
+ Err(_) => panic!("reply body timed out"),
+ }
+ }
+ status
+}
+
+/// Register `CLIENT_ID` as root, returning the connection and its bound
+/// session. The session binds to THIS socket server-side, so every frame in a
+/// test must reuse the returned stream.
+async fn register(addr: SocketAddr) -> (TcpStream, u64) {
+ register_with_budget(addr, COMMIT_BUDGET).await
+}
+
+/// Fresh socket per attempt: a login refused mid-election may come back as an
+/// eviction that poisons the connection.
+async fn register_with_budget(addr: SocketAddr, budget: Duration) -> (TcpStream, u64) {
+ register_client_with_budget(addr, CLIENT_ID, budget).await
+}
+
+async fn register_client_with_budget(
+ addr: SocketAddr,
+ client: u128,
+ budget: Duration,
+) -> (TcpStream, u64) {
+ let deadline = Instant::now() + budget;
+ loop {
+ let mut stream = TcpStream::connect(addr).await.unwrap();
+ if let Some(session) = login_on(&mut stream, client).await {
+ return (stream, session);
+ }
+ assert!(
+ Instant::now() < deadline,
+ "register did not commit within {budget:?}"
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+}
+
+async fn login_on(stream: &mut TcpStream, client: u128) -> Option<u64> {
+ let body = LoginRegisterRequest {
+ version_info: ClientVersionInfo {
+ protocol_version: IGGY_PROTOCOL_VERSION,
+ sdk_name: WireName::new("iggy274-raw").unwrap(),
+ sdk_version: WireName::new("0.0.1").unwrap(),
+ },
+ username: WireName::new(DEFAULT_ROOT_USERNAME).unwrap(),
+ password: SecretString::from(DEFAULT_ROOT_PASSWORD),
+ client_context: None,
+ }
+ .to_bytes();
+ let header = request_header_for(client, Operation::Register, 0, 0, body.len());
+
+ stream.write_all(bytemuck::bytes_of(&header)).await.unwrap();
+ stream.write_all(&body).await.unwrap();
+
+ let mut reply_header = [0u8; HEADER_SIZE];
+ let Ok(Ok(_)) = timeout(REPLY_WAIT, stream.read_exact(&mut reply_header)).await else {
+ return None;
+ };
+ let command_offset = offset_of!(RequestHeader, command);
+ if reply_header[command_offset] != Command::Reply as u8 {
+ return None;
+ }
+
+ let status_offset = offset_of!(ReplyHeader, status);
+ let status = u32::from_le_bytes(
+ reply_header[status_offset..status_offset + 4]
+ .try_into()
+ .unwrap(),
+ );
+ let total_size = read_size_field(&reply_header).expect("login reply size") as usize;
+ let mut reply_body = vec![0u8; total_size - HEADER_SIZE];
+ let Ok(Ok(_)) = timeout(REPLY_WAIT, stream.read_exact(&mut reply_body)).await else {
+ return None;
+ };
+ if status != 0 {
+ return None;
+ }
+ let session_offset = offset_of!(ReplyHeader, commit);
+ Some(u64::from_le_bytes(
+ reply_header[session_offset..session_offset + 8]
+ .try_into()
+ .unwrap(),
+ ))
+}
+
+fn is_transient(code: u32) -> bool {
+ code == IggyError::TransientNotCommitted.as_code()
+ || code == IggyError::TransientNotAccepted.as_code()
+}
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 1af433af2..3cbdf3e17 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -36,9 +36,9 @@ use crate::{
PollingConsumer,
};
use consensus::{
- CommitLogEvent, Consensus, PartitionDiagEvent, PipelineEntry, PlaneKind, Project,
- ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight,
- ack_quorum_reached, build_deny_reply_from_request, build_reply_from_request,
+ ClientTable, ClientTableMode, CommitLogEvent, Consensus, PartitionDiagEvent, PipelineEntry,
+ PlaneKind, Project, ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus,
+ ack_preflight, ack_quorum_reached, build_deny_reply_from_request, build_reply_from_request,
build_reply_message, drain_committable_prefix, emit_namespace_progress_event,
emit_partition_diag, emit_sim_event, fence_old_prepare_by_commit, repaired_frontier_update,
replicate_frozen_to_next_in_chain, replicate_preflight, restamp_prepare_view,
@@ -53,7 +53,7 @@ use iggy_binary_protocol::responses::messages::{
use iggy_binary_protocol::{
AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireEncode, WireIdentifier,
};
-use iggy_binary_protocol::{PrepareOkHeader, RoutedRequestHeader};
+use iggy_binary_protocol::{PrepareOkHeader, ReplyHeader, RoutedRequestHeader};
use iggy_common::{
ConsumerGroupId, ConsumerGroupOffsets, ConsumerKind, ConsumerOffset, ConsumerOffsets,
IggyByteSize, IggyError, IggyExpiry, IggyTimestamp, PartitionStats, PollingKind,
@@ -86,16 +86,24 @@ use tokio::sync::Mutex as TokioMutex;
use tracing::{debug, error, warn};
// This struct aliases in terms of the code contained the `LocalPartition from `core/server/src/streaming/partitions/local_partition.rs`.
-//
-// Note: there is no per-client write dedup at the partition plane.
-// `SendMessages` retries are at-least-once and may commit multiple times.
-// Duplicate suppression is a consensus-layer concern: the VSR client table
-// dedups by request id (at-most-once), so the data plane needs no message-id set.
pub struct IggyPartition<B = IggyMessageBus, SB = PingPongSuperblock>
where
B: MessageBus,
{
consensus: VsrConsensus<B>,
+ /// This group's slice of the VSR client table, run in
+ /// [`ClientTableMode::PartitionSlice`]: per-client request watermarks
+ /// folded in at commit. Replica-local and memory-only: boot lifts the commit
+ /// frontier without re-applying the log, so a restarted replica comes back
+ /// with an empty slice while its peers keep theirs, and only commits folded
+ /// in after boot, or a state-transfer install, rebuild it. The mode turns
+ /// off what this plane cannot use -- no reply ring (`SendMessages` has no
+ /// result section, so a duplicate is answered by synthesizing the empty
+ /// success its original earned), no epoch fence (a partition group never
+ /// observes a `Register`), and no preallocated slot array (one table per
+ /// group, where preallocating the cap would reserve hundreds of KiB per
+ /// partition before a client connects).
+ dedup: ClientTable,
pub log: SegmentedLog<PartitionJournal<PartitionJournalMemStorage>>,
/// Highest durably persisted offset.
pub offset: Arc<AtomicU64>,
@@ -460,6 +468,10 @@ where
let single_replica = consensus.replica_count() == 1;
let partition = Self {
consensus,
+ dedup: ClientTable::with_mode(
+ consensus::PARTITION_DEDUP_CLIENTS_MAX,
+ ClientTableMode::PartitionSlice,
+ ),
log: SegmentedLog::default(),
offset: Arc::new(AtomicU64::new(0)),
dirty_offset: AtomicU64::new(0),
@@ -555,6 +567,26 @@ where
&self.consensus
}
+ /// This group's dedup slice. Read at admission to classify a request,
+ /// written only from the commit path.
+ #[must_use]
+ pub(crate) const fn dedup(&self) -> &ClientTable {
+ &self.dedup
+ }
+
+ /// Mutable slice, for the commit path and state-transfer install.
+ pub(crate) const fn dedup_mut(&mut self) -> &mut ClientTable {
+ &mut self.dedup
+ }
+
+ /// Size the dedup slice to `[partition] dedup_clients_max`. Boot-only:
+ /// `set_capacity` replaces the table rather than evicting into the new
+ /// bound, and panics if the slice already holds an entry. Config
+ /// validation rejects a zero cap before it can reach here.
+ pub fn set_dedup_clients_max(&mut self, clients_max: usize) {
+ self.dedup.set_capacity(clients_max);
+ }
+
#[must_use]
pub fn with_in_memory_storage(
stats: Arc<PartitionStats>,
@@ -1435,8 +1467,9 @@ where
}
/// `AckLevel::NoAck` fast path: persist, apply, send reply, no
- /// replication. Single-replica durability. No reply cache: partition
- /// plane is at-least-once; session lifecycle lives on metadata.
+ /// replication. Single-replica durability. Never recorded in the dedup
+ /// slice: it does not replicate, so folding it in would fork the slice
+ /// across replicas. Session lifecycle lives on metadata.
#[allow(clippy::future_not_send)]
async fn apply_consumer_offset_no_ack(
&self,
@@ -1444,6 +1477,7 @@ where
kind: ConsumerKind,
consumer_id: u32,
offset: Option<u64>,
+ waiter: Option<consensus::Sender<Message<ReplyHeader>>>,
) {
let pending = offset.map_or_else(
|| PendingConsumerOffsetCommit::delete(kind, consumer_id),
@@ -1474,6 +1508,12 @@ where
&request_header,
committed_reply_body(request_header.operation),
);
+ // Same rule as the committed path: a submit's waiter takes the reply,
+ // because `header.client` is then the VSR consensus id.
+ if let Some(waiter) = waiter {
+ let _ = waiter.send(reply);
+ return;
+ }
let reply_buffers = reply.into_generic().into_frozen();
if let Err(error) = self
.consensus
@@ -1951,16 +1991,28 @@ where
/// Project a client request into a prepare.
///
- /// At-least-once: no per-client dedup. `SendMessages` retry -> fresh
- /// prepare, may re-commit at new offset. Consumers handle dedup
- /// (message key / content / producer-id+seq). Session lifecycle +
- /// eviction live on metadata plane.
+ /// A replay of a committed `(client, request)` is absorbed by this group's
+ /// dedup slice; anything above the watermark projects into a prepare.
+ /// Session lifecycle + eviction live on the metadata plane.
+ ///
+ /// `reply` is the in-process channel a `PartitionSubmit` carried in. When
+ /// present the committed reply fires on it instead of going to the bus:
+ /// the connection-owning shard writes it to the socket it holds, because
+ /// `header.client` is the VSR consensus id and carries no home-shard
+ /// routing. `None` keeps the bus path (auto-commit ops, tests).
///
/// # Panics
/// Panics if called when this partition's consensus instance is not the
/// primary, is not in normal status, or is currently syncing.
#[allow(clippy::future_not_send, clippy::too_many_lines)]
- pub async fn on_request(&mut self, message: Message<RoutedRequestHeader>) {
+ pub async fn on_request(
+ &mut self,
+ message: Message<RoutedRequestHeader>,
+ reply: Option<consensus::Sender<Message<ReplyHeader>>>,
+ ) {
+ // Taken by whichever arm answers: the deny paths, the NoAck fast path,
+ // or the pipeline entry that fires it at commit. Exactly one runs.
+ let mut reply = reply;
self.clear_pending_consumer_offset_commits_if_view_changed();
let namespace = IggyNamespace::from_raw(message.header().group);
let client_id = message.header().client;
@@ -2023,6 +2075,7 @@ where
message.header(),
IggyError::TransientNotAccepted.as_code(),
"non-primary transient reply send failed",
+ reply.take(),
)
.await;
return;
@@ -2054,6 +2107,63 @@ where
_ => None,
};
+ // Dedup BEFORE the admission checks below: a replay of an
+ // already-committed delete must answer the success its original
+ // earned, not the typed 404 the existence check would raise now
+ // that the offset is gone.
+ //
+ // A replay racing its own in-flight original is absorbed here: the
+ // slice only knows committed ops, so it cannot yet see the copy
+ // still in the pipeline. Keyed on the exact `(client, request)`
+ // for the transports that keep several writes in flight per
+ // client (HTTP handlers on one session, the pipelining SDKs):
+ // matching any request from the client would serialize them to
+ // one in-flight write per group. A lockstep TCP connection never
+ // has a second request here to begin with.
+ //
+ // Those same transports can deliver a client's ids out of order:
+ // a write refused transiently here is replayed after its
+ // successors committed. The slice therefore keeps a committed-id
+ // window under the watermark (`consensus::COMMITTED_WINDOW_BITS`)
+ // and admits an unmarked id inside it instead of absorbing it.
+ if !is_auto_commit_client(client_id) {
+ if consensus.pipeline_has_message_from_client_request(client_id, request) {
+ Self::send_partition_deny_or_log(
+ consensus,
+ message.header(),
+ IggyError::TransientNotCommitted.as_code(),
+ "in-flight dedup transient reply send failed",
+ reply.take(),
+ )
+ .await;
+ return;
+ }
+ // An absorbed duplicate answers the operation's empty success.
+ // For `SendMessages` that is LESS than the original reply
+ // 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
+ .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;
+ }
+ }
+
if matches!(message.header().operation, Operation::DeleteConsumerOffset)
&& let Some((kind, consumer_id, _, _)) = consumer_offset
&& let Err(error) = self.ensure_consumer_offset_exists(kind, consumer_id)
@@ -2078,6 +2188,7 @@ where
message.header(),
error.as_code(),
"delete_consumer_offset deny reply send failed",
+ reply.take(),
)
.await;
return;
@@ -2112,6 +2223,7 @@ where
message.header(),
IggyError::InvalidOffset(requested_offset).as_code(),
"store_consumer_offset deny reply send failed",
+ reply.take(),
)
.await;
return;
@@ -2136,9 +2248,10 @@ where
// request room -> buffer; both full -> drop+warn (client retries
// via read-timeout).
if consensus.pipeline_is_full() {
- let push_result =
- consensus.push_queued_request(consensus::RequestEntry::new(message));
- if push_result.is_err() {
+ let push_result = consensus.push_queued_request(
+ consensus::RequestEntry::with_sender(message, reply.take()),
+ );
+ if let Err(mut refused) = push_result {
emit_partition_diag(
tracing::Level::WARN,
&PartitionDiagEvent::new(
@@ -2146,13 +2259,32 @@ where
"on_request: prepare and request queues both full, dropping",
),
);
+ // The request provably never entered either queue, so a
+ // waiter can be told so instead of waiting out its
+ // timeout.
+ let waiter = refused.take_reply_sender();
+ Self::send_partition_deny_or_log(
+ consensus,
+ refused.message.header(),
+ IggyError::TransientNotAccepted.as_code(),
+ "queues-full transient reply send failed",
+ waiter,
+ )
+ .await;
}
return;
}
let prepare = message.project(consensus);
consensus.verify_pipeline();
- consensus.pipeline_message(PlaneKind::Partitions, &prepare);
+ match reply.take() {
+ Some(sender) => consensus.pipeline_message_with_sender(
+ PlaneKind::Partitions,
+ &prepare,
+ sender,
+ ),
+ None => consensus.pipeline_message(PlaneKind::Partitions, &prepare),
+ }
Disposition::Replicate(prepare)
}
};
@@ -2165,17 +2297,23 @@ where
consumer_id,
offset,
} => {
- self.apply_consumer_offset_no_ack(request_header, kind, consumer_id, offset)
- .await;
+ self.apply_consumer_offset_no_ack(
+ request_header,
+ kind,
+ consumer_id,
+ offset,
+ reply.take(),
+ )
+ .await;
}
}
}
/// Promote up to `slots_freed` buffered requests into prepares post-commit.
///
- /// No preflight: partition plane is at-least-once with no `ClientTable`
- /// dedup. Buffered `SendMessages` retry commits at fresh offset; consumers
- /// dedup by message key / content / producer-id+seq.
+ /// Promotion runs no preflight: the request was classified at admission
+ /// and the slice cannot have gained a higher watermark for it since (only
+ /// a commit moves it, and this entry has not committed).
///
/// Per-iteration `is_primary && is_normal && !is_transferring` asserts inlined
/// (closure form's `&consensus` borrow conflicts with `&mut self`). Guards
@@ -2192,7 +2330,7 @@ where
pub async fn drain_request_queue_into_prepares(&mut self, slots_freed: usize) {
for _ in 0..slots_freed {
let req = self.consensus().pop_queued_request();
- let Some(req) = req else { break };
+ let Some(mut req) = req else { break };
let prepare = {
let consensus = self.consensus();
@@ -2208,9 +2346,19 @@ where
!consensus.is_transferring(),
"drain_request_queue_into_prepares: must not be transferring state"
);
+ // The waiter parked with the request; it must travel into the
+ // prepare slot or the commit has nobody to answer.
+ let reply_sender = req.take_reply_sender();
let prepare = req.message.project(consensus);
consensus.verify_pipeline();
- consensus.pipeline_message(PlaneKind::Partitions, &prepare);
+ match reply_sender {
+ Some(sender) => consensus.pipeline_message_with_sender(
+ PlaneKind::Partitions,
+ &prepare,
+ sender,
+ ),
+ None => consensus.pipeline_message(PlaneKind::Partitions, &prepare),
+ }
prepare
};
self.on_replicate(prepare).await;
@@ -3262,7 +3410,7 @@ where
let committed_batch_stats = self.resolve_committed_visible_offsets(&drained);
let mut messages_committed = false;
- for (entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) {
+ for (mut entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) {
let prepare_header = entry.header;
if !self
.commit_partition_entry(
@@ -3312,6 +3460,20 @@ where
self.consensus.advance_commit_min(prepare_header.op);
+ // Fold the committed request into this group's dedup slice. Runs on
+ // EVERY replica, not just the one that replies, so a promoted
+ // primary can absorb a replay of what its predecessor committed.
+ // Auto-commit ops carry the reserved sentinel client and no client
+ // ever replays them.
+ if !is_auto_commit_client(prepare_header.client) {
+ self.dedup.commit_request(
+ prepare_header.client,
+ prepare_header.user_id,
+ prepare_header.request,
+ prepare_header.op,
+ );
+ }
+
let pipeline_depth = self.consensus.pipeline_len();
let event = CommitLogEvent {
replica: ReplicaLogContext::from_consensus(&self.consensus, PlaneKind::Partitions),
@@ -3329,9 +3491,11 @@ where
pipeline_depth,
);
- // No reply cache: at-least-once means retries re-commit at new
- // offsets. Only primary delivers replies; backups just advance
- // commit. Session lifecycle is metadata-only.
+ // No reply cache: an absorbed duplicate is answered by
+ // synthesizing the same empty success at admission, so no committed
+ // bytes need keeping. Only the primary delivers replies; backups
+ // just advance commit and fold the slice. Session lifecycle is
+ // metadata-only.
//
// A server-generated auto-commit op (a poll's `auto_commit`,
// replicated for failover) carries the reserved
@@ -3346,23 +3510,36 @@ where
operation => committed_reply_body(operation),
};
let reply = build_reply_message(&prepare_header, &body);
- let reply_buffers = reply.into_generic().into_frozen();
emit_sim_event(SimEventKind::ClientReplyEmitted, &event);
- if let Err(error) = self
+ // An in-process waiter takes the reply instead of the bus: it
+ // arrived as a `PartitionSubmit`, so `header.client` is the VSR
+ // consensus id and carries no home-shard routing. The awaiting
+ // shard owns the socket. A dropped receiver is ignored -- the
+ // client recovers on its own read-timeout.
+ //
+ // Without a waiter the bus is tried, and for a TCP client it
+ // cannot route the VSR id. That is the expected shape of every
+ // op re-committed after a view change (the rebuilt pipeline
+ // entries carry no sender), so it logs at debug: the original
+ // waiter was cancelled by the view change and the client is
+ // already on its read-timeout.
+ if let Some(sender) = entry.take_reply_sender() {
+ let _ = sender.send(reply);
+ } else if let Err(error) = self
.consensus
.message_bus()
- .send_to_client(prepare_header.client, reply_buffers)
+ .send_to_client(prepare_header.client, reply.into_generic().into_frozen())
.await
{
- tracing::error!(
+ tracing::debug!(
target: "iggy.partitions.diag",
plane = "partitions",
client = prepare_header.client,
op = prepare_header.op,
namespace_raw,
%error,
- "client reply forward failed, no retransmit path; client will time out",
+ "client reply not routable by the bus; client will time out",
);
}
}
@@ -3570,14 +3747,34 @@ where
/// Send `header`'s deny reply with `status` on `ReplyHeader.status` (empty
/// body, op=0), logging a WARN under `send_fail_label` if the reply send
/// fails. Callers deny on the primary, before the op enters the pipeline,
- /// so nothing replicates.
+ /// so nothing replicates. `waiter` is the submit's in-process channel,
+ /// taken by the caller.
async fn send_partition_deny_or_log(
consensus: &VsrConsensus<B>,
header: &RoutedRequestHeader,
status: u32,
send_fail_label: &'static str,
+ waiter: Option<consensus::Sender<Message<ReplyHeader>>>,
) {
let reply = build_deny_reply_from_request(consensus, header, status);
+ Self::deliver_reply_or_log(consensus, header, reply, waiter, send_fail_label).await;
+ }
+
+ /// Deliver an admission-time reply (a deny, or an absorbed duplicate's
+ /// success). When `waiter` is present the reply goes there: `header.client`
+ /// is then the VSR consensus id, which the bus cannot route. Otherwise the
+ /// bus carries it, and a failed send logs a WARN under `send_fail_label`.
+ async fn deliver_reply_or_log(
+ consensus: &VsrConsensus<B>,
+ header: &RoutedRequestHeader,
+ reply: Message<ReplyHeader>,
+ waiter: Option<consensus::Sender<Message<ReplyHeader>>>,
+ send_fail_label: &'static str,
+ ) {
+ if let Some(waiter) = waiter {
+ let _ = waiter.send(reply);
+ return;
+ }
if let Err(send_error) = consensus
.message_bus()
.send_to_client(header.client, reply.into_generic().into_frozen())
@@ -5664,7 +5861,7 @@ mod tests {
let consumer_id: u32 = 5;
partition
- .on_request(delete_offset_request(client_id, 7, consumer_id))
+ .on_request(delete_offset_request(client_id, 7, consumer_id), None)
.await;
{
@@ -5702,7 +5899,7 @@ mod tests {
ConsumerOffset::new(ConsumerKind::Consumer, consumer_id, 3, String::new()),
);
partition
- .on_request(delete_offset_request(client_id, 8, consumer_id))
+ .on_request(delete_offset_request(client_id, 8, consumer_id), None)
.await;
assert_eq!(
partition.consensus().pipeline_len(),
@@ -5717,7 +5914,7 @@ mod tests {
let client_id = 42;
partition
- .on_request(delete_offset_request(client_id, 7, 5))
+ .on_request(delete_offset_request(client_id, 7, 5), None)
.await;
let sent = sent_to_clients.borrow();
@@ -7233,6 +7430,7 @@ mod tests {
next_offset: 50,
consumers: Vec::new(),
groups: Vec::new(),
+ dedup: Vec::new(),
};
let refused = partition
.install_state_transfer(&repair_config(), 12, Vec::new(), &behind.encode(), 0)
@@ -7260,6 +7458,7 @@ mod tests {
next_offset: 0,
consumers: Vec::new(),
groups: Vec::new(),
+ dedup: Vec::new(),
};
let accepted = partition
.install_state_transfer(&repair_config(), 12, Vec::new(), &purged.encode(), 0)
@@ -7299,6 +7498,7 @@ mod tests {
next_offset: 50,
consumers: Vec::new(),
groups: Vec::new(),
+ dedup: Vec::new(),
};
let refused = partition
.install_state_transfer(&repair_config(), 12, Vec::new(), &offer.encode(), 1)
@@ -7348,6 +7548,7 @@ mod tests {
next_offset: 0,
consumers: Vec::new(),
groups: Vec::new(),
+ dedup: Vec::new(),
};
let installed = partition
.install_state_transfer(&repair_config(), 12, Vec::new(), &reset.encode(), 1)
diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs
index 87887c9ee..2a2ee6e1e 100644
--- a/core/partitions/src/iggy_partitions.rs
+++ b/core/partitions/src/iggy_partitions.rs
@@ -21,12 +21,17 @@ use crate::poll_plan::PollPlan;
use crate::types::PartitionsConfig;
use crate::{IggyPartition, Partition, PollingArgs, PollingConsumer};
use ahash::AHashSet;
-use consensus::{Consensus, Plane, PlaneIdentity, VsrConsensus};
+use consensus::{
+ Consensus, Plane, PlaneIdentity, VsrConsensus, build_deny_reply_from_request_header,
+};
use iggy_binary_protocol::{
- Command, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, RoutedRequestHeader,
+ Command, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, ReplyHeader,
+ RoutedRequestHeader,
};
+use iggy_common::IggyError;
use journal::superblock::{PingPongSuperblock, SuperblockStore};
use message_bus::MessageBus;
+use server_common::Message;
use server_common::send_messages::{ChecksumMode, convert_request_message, encrypt_batch_request};
use server_common::sharding::{IggyNamespace, LocalIdx, ShardId};
#[cfg(debug_assertions)]
@@ -528,22 +533,35 @@ where
}
}
-impl<B, SB> Plane<VsrConsensus<B>> for IggyPartitions<B, SB>
+impl<B, SB> IggyPartitions<B, SB>
where
B: MessageBus,
SB: SuperblockStore,
{
- async fn on_request(
+ /// [`Plane::on_request`] carrying the in-process reply channel a
+ /// `PartitionSubmit` arrived with; `None` keeps the bus-reply path.
+ pub async fn on_request_with_reply(
&self,
message: <VsrConsensus<B> as Consensus>::Message<RoutedRequestHeader>,
+ reply: Option<consensus::Sender<Message<ReplyHeader>>>,
) {
let namespace = IggyNamespace::from_raw(message.header().group);
+ // Every exit below that drops the request answers its waiter first: a
+ // submit's reply cannot be routed by `header.client` (the VSR id), so
+ // an unanswered channel costs the client a full read-timeout for an
+ // outcome that was decided here and now.
+ let mut reply = reply;
if self.is_tombstoned(&namespace) {
warn!(
target: "iggy.partitions.diag",
namespace_raw = namespace.inner(),
"dropping request: namespace tombstoned"
);
+ Self::answer_waiter(
+ reply.take(),
+ message.header(),
+ IggyError::TransientNotAccepted.as_code(),
+ );
return;
}
// At-rest encryption happens HERE, once, before the op enters
@@ -559,6 +577,9 @@ where
// `encrypt_batch_request`'s decode before re-encryption, and the
// re-encrypted batch (checksum kept by `encrypt_batch_request`) then
// re-enters `convert` as the canonical-vs-legacy discriminator.
+ // The header outlives the message consumed by the conversion, so a
+ // failure can still be answered with the frame's own identity.
+ let header = *message.header();
let canonical = convert_request_message(namespace, message, ChecksumMode::Compute)
.and_then(|message| encrypt_batch_request(message, encryptor));
match canonical {
@@ -570,6 +591,7 @@ where
%error,
"dropping send_messages: failed to encrypt batch at ingestion"
);
+ Self::answer_waiter(reply.take(), &header, error.as_code());
return;
}
}
@@ -584,9 +606,40 @@ where
operation = ?message.header().operation,
"partition not initialized for namespace"
);
+ Self::answer_waiter(
+ reply.take(),
+ message.header(),
+ IggyError::TransientNotAccepted.as_code(),
+ );
return;
};
- partition.on_request(message).await;
+ partition.on_request(message, reply).await;
+ }
+
+ /// Deny a request that never reached its partition on the submit channel
+ /// it arrived with, if any. The bus path has no waiter to answer and keeps
+ /// its drop-and-warn behaviour.
+ fn answer_waiter(
+ waiter: Option<consensus::Sender<Message<ReplyHeader>>>,
+ header: &RoutedRequestHeader,
+ status: u32,
+ ) {
+ if let Some(waiter) = waiter {
+ let _ = waiter.send(build_deny_reply_from_request_header(header, status));
+ }
+ }
+}
+
+impl<B, SB> Plane<VsrConsensus<B>> for IggyPartitions<B, SB>
+where
+ B: MessageBus,
+ SB: SuperblockStore,
+{
+ async fn on_request(
+ &self,
+ message: <VsrConsensus<B> as Consensus>::Message<RoutedRequestHeader>,
+ ) {
+ self.on_request_with_reply(message, None).await;
}
async fn on_replicate(&self, message: <VsrConsensus<B> as Consensus>::Message<PrepareHeader>) {
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 5bd29807d..7444b9d76 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -37,7 +37,9 @@ use crate::{IggyIndexWriter, IggyPartition};
use compio::io::{AsyncReadAtExt, AsyncWriteAtExt};
use consensus::le_cursor::{LeCursor, Truncated, split_verified_trailer};
use consensus::state_manifest::artifact_kind;
-use consensus::{ArtifactProgress, Sequencer as _, StateArtifactHasher, state_artifact_checksum};
+use consensus::{
+ ArtifactProgress, DedupWatermark, Sequencer as _, StateArtifactHasher, state_artifact_checksum,
+};
use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyByteSize};
use journal::superblock::SuperblockStore;
use message_bus::MessageBus;
@@ -50,15 +52,26 @@ use std::path::{Path, PathBuf};
use std::rc::Rc;
use std::sync::atomic::Ordering;
-/// Framing marker for the consumer-offsets wire artifact, "ICO1".
-pub(crate) const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO1";
+/// Framing marker for the consumer-offsets wire artifact, "ICO2". Bumped with
+/// the version when the dedup section was appended, so the magic alone tells
+/// the two layouts apart.
+pub(crate) const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO2";
/// Version byte following the magic.
///
/// Any layout change bumps this, INCLUDING appended fields: the decoder
/// deliberately fails closed on unknown versions and on trailing bytes,
/// because a v2 field can change the meaning of fields v1 already read.
-pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 1;
+pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 2;
+
+/// The previous framing, "ICO1" at version 1: the same layout without the
+/// dedup section. Still decoded so a rolling upgrade works in both orders --
+/// an upgraded replica rejoining behind the repair floor of an un-upgraded
+/// primary installs its artifact with an empty slice (dedup for that window
+/// degrades to at-least-once, exactly the pre-dedup behaviour) instead of
+/// refusing it and re-pulling forever.
+pub(crate) const CONSUMER_OFFSETS_MAGIC_V1: [u8; 4] = *b"ICO1";
+const CONSUMER_OFFSETS_VERSION_V1: u8 = 1;
/// Per-section entry ceiling for the consumer-offsets artifact.
///
@@ -67,6 +80,10 @@ pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 1;
/// entry ceiling.
pub(crate) const CONSUMER_OFFSETS_ENTRIES_MAX: u32 = 1 << 20;
+/// Wire stride of one dedup entry: client u128 + watermark u64 + commit u64 +
+/// user u32 + committed window u128.
+const DEDUP_ENTRY_LEN: usize = 2 * size_of::<u128>() + 2 * size_of::<u64>() + size_of::<u32>();
+
/// One in-flight partition state transfer on the receiving replica.
///
/// Mirrors the metadata plane's session, plus `staged`: completed
@@ -295,13 +312,19 @@ pub(crate) struct ConsumerOffsetsWire {
pub consumers: Vec<(u32, u64)>,
/// `(consumer group id, offset)`, ascending by id.
pub groups: Vec<(u32, u64)>,
+ /// This group's dedup slice, ascending by client. Carried so a replica
+ /// rejoining behind the repair floor can absorb a replay of what the group
+ /// already committed instead of re-executing it.
+ pub dedup: Vec<DedupWatermark>,
}
impl ConsumerOffsetsWire {
/// Encode: `magic | version u8 | purge_generation u64 | next_offset u64 |
- /// consumer_count u32 | group_count u32 | {id u32, offset u64}xN |
- /// {id u32, offset u64}xM | XxHash3_64 trailer`. Little-endian
- /// throughout.
+ /// consumer_count u32 | group_count u32 | dedup_count u32 |
+ /// {id u32, offset u64}xN | {id u32, offset u64}xM |
+ /// {client u128, watermark u64, latest_commit u64, user_id u32,
+ /// committed_window u128}xD |
+ /// XxHash3_64 trailer`. Little-endian throughout.
#[must_use]
pub fn encode(&self) -> Vec<u8> {
// Size exactly rather than guess; the reservation assert keeps the
@@ -309,8 +332,9 @@ impl ConsumerOffsetsWire {
let reserved = CONSUMER_OFFSETS_MAGIC.len()
+ size_of::<u8>()
+ 2 * size_of::<u64>()
- + 2 * size_of::<u32>()
+ + 3 * size_of::<u32>()
+ (self.consumers.len() + self.groups.len()) * (size_of::<u32>() + size_of::<u64>())
+ + self.dedup.len() * DEDUP_ENTRY_LEN
+ size_of::<u64>();
let mut out = Vec::with_capacity(reserved);
out.extend_from_slice(&CONSUMER_OFFSETS_MAGIC);
@@ -321,10 +345,19 @@ impl ConsumerOffsetsWire {
out.extend_from_slice(&(self.consumers.len() as u32).to_le_bytes());
#[allow(clippy::cast_possible_truncation)]
out.extend_from_slice(&(self.groups.len() as u32).to_le_bytes());
+ #[allow(clippy::cast_possible_truncation)]
+ out.extend_from_slice(&(self.dedup.len() as u32).to_le_bytes());
for (id, offset) in self.consumers.iter().chain(self.groups.iter()) {
out.extend_from_slice(&id.to_le_bytes());
out.extend_from_slice(&offset.to_le_bytes());
}
+ for entry in &self.dedup {
+ out.extend_from_slice(&entry.client.to_le_bytes());
+ out.extend_from_slice(&entry.watermark.to_le_bytes());
+ out.extend_from_slice(&entry.latest_commit.to_le_bytes());
+ out.extend_from_slice(&entry.user_id.to_le_bytes());
+ out.extend_from_slice(&entry.committed_window.to_le_bytes());
+ }
debug_assert_eq!(out.len() + size_of::<u64>(), reserved, "encode reservation");
let trailer = state_artifact_checksum(&out);
out.extend_from_slice(&trailer.to_le_bytes());
@@ -351,19 +384,32 @@ impl ConsumerOffsetsWire {
})?;
let mut cursor = LeCursor::new(content);
let magic = cursor.take(CONSUMER_OFFSETS_MAGIC.len())?;
- if magic != CONSUMER_OFFSETS_MAGIC {
- return Err(ConsumerOffsetsWireError::BadMagic);
- }
let version = cursor.u8()?;
- if version != CONSUMER_OFFSETS_VERSION {
- return Err(ConsumerOffsetsWireError::UnsupportedVersion { version });
- }
+ let carries_dedup = if magic == CONSUMER_OFFSETS_MAGIC {
+ if version != CONSUMER_OFFSETS_VERSION {
+ return Err(ConsumerOffsetsWireError::UnsupportedVersion { version });
+ }
+ true
+ } else if magic == CONSUMER_OFFSETS_MAGIC_V1 {
+ if version != CONSUMER_OFFSETS_VERSION_V1 {
+ return Err(ConsumerOffsetsWireError::UnsupportedVersion { version });
+ }
+ false
+ } else {
+ return Err(ConsumerOffsetsWireError::BadMagic);
+ };
let purge_generation = cursor.u64()?;
let next_offset = cursor.u64()?;
let consumer_count = cursor.u32()?;
let group_count = cursor.u32()?;
+ let dedup_count = if carries_dedup { cursor.u32()? } else { 0 };
let consumers = Self::decode_section(&mut cursor, "consumers", consumer_count)?;
let groups = Self::decode_section(&mut cursor, "groups", group_count)?;
+ let dedup = if carries_dedup {
+ Self::decode_dedup_section(&mut cursor, dedup_count)?
+ } else {
+ Vec::new()
+ };
if !cursor.remaining().is_empty() {
// Distinct from `Truncated`: extra bytes point at a NEWER
// encoder, and telling the operator the artifact is short would
@@ -377,9 +423,56 @@ impl ConsumerOffsetsWire {
next_offset,
consumers,
groups,
+ dedup,
})
}
+ /// Same guards as [`Self::decode_section`] at the dedup stride: peer count
+ /// against the ceiling, then against the bytes actually present, then
+ /// ascending-strict client order so the encoding stays canonical. Client
+ /// zero is the reserved id no ingress admits, so an artifact carrying it is
+ /// a peer bug and fails closed rather than being silently dropped at
+ /// install.
+ fn decode_dedup_section(
+ cursor: &mut LeCursor<'_>,
+ count: u32,
+ ) -> Result<Vec<DedupWatermark>, ConsumerOffsetsWireError> {
+ if count > CONSUMER_OFFSETS_ENTRIES_MAX {
+ return Err(ConsumerOffsetsWireError::TooManyEntries {
+ section: "dedup",
+ count,
+ max: CONSUMER_OFFSETS_ENTRIES_MAX,
+ });
+ }
+ if count as usize * DEDUP_ENTRY_LEN > cursor.remaining().len() {
+ return Err(ConsumerOffsetsWireError::Truncated);
+ }
+ let mut entries = Vec::with_capacity(count as usize);
+ let mut previous: Option<u128> = None;
+ for _ in 0..count {
+ let client = cursor.u128()?;
+ let watermark = cursor.u64()?;
+ let latest_commit = cursor.u64()?;
+ let user_id = cursor.u32()?;
+ let committed_window = cursor.u128()?;
+ if client == 0 {
+ return Err(ConsumerOffsetsWireError::ReservedClient);
+ }
+ if previous.is_some_and(|previous| client <= previous) {
+ return Err(ConsumerOffsetsWireError::NonAscendingClient { client });
+ }
+ previous = Some(client);
+ entries.push(DedupWatermark {
+ client,
+ user_id,
+ watermark,
+ latest_commit,
+ committed_window,
+ });
+ }
+ Ok(entries)
+ }
+
fn decode_section(
cursor: &mut LeCursor<'_>,
section: &'static str,
@@ -397,9 +490,12 @@ impl ConsumerOffsetsWire {
// The count is peer input and the reservation is 12 bytes per element
// after alignment, so it is checked against the bytes actually present
// before allocating: a ~30 byte artifact could otherwise ask for tens of
- // megabytes across the two sections. 12 is the wire stride below -- the
- // groups call sees exactly `12 * count` bytes remaining, so a wider
- // guard would reject every non-empty artifact.
+ // megabytes across the two sections. 12 is the wire stride below, and
+ // the guard is a lower bound on purpose: what trails a section varies
+ // (groups trails consumers, the dedup section trails groups and is
+ // absent from a v1 artifact), so only "at least this many bytes
+ // present" holds for both calls. Demanding that `12 * count` be all
+ // that remains would reject valid input.
if count as usize * (size_of::<u32>() + size_of::<u64>()) > cursor.remaining().len() {
return Err(ConsumerOffsetsWireError::Truncated);
}
@@ -450,6 +546,14 @@ pub enum ConsumerOffsetsWireError {
section: &'static str,
id: u32,
},
+ /// Dedup clients are not strictly ascending. Same encoder bug as
+ /// [`Self::NonAscendingId`], on the u128-keyed section.
+ NonAscendingClient {
+ client: u128,
+ },
+ /// A dedup entry carries client id zero, which is reserved and refused at
+ /// every ingress: a peer encoder bug.
+ ReservedClient,
}
impl From<Truncated> for ConsumerOffsetsWireError {
@@ -490,6 +594,17 @@ impl fmt::Display for ConsumerOffsetsWireError {
"consumer-offsets artifact {section} id {id} does not ascend \
(duplicate, or out of order)"
),
+ Self::NonAscendingClient { client } => write!(
+ f,
+ "consumer-offsets artifact dedup client {client} does not ascend \
+ (duplicate, or out of order)"
+ ),
+ Self::ReservedClient => {
+ write!(
+ f,
+ "consumer-offsets artifact dedup entry carries reserved client 0"
+ )
+ }
}
}
}
@@ -506,6 +621,20 @@ mod tests {
next_offset: 43,
consumers: vec![(1, 10), (7, 42)],
groups: vec![(2, 5)],
+ dedup: vec![
+ dedup_entry(11, 4, 90),
+ dedup_entry(usize::MAX as u128 + 5, 9, 91),
+ ],
+ }
+ }
+
+ fn dedup_entry(client: u128, watermark: u64, latest_commit: u64) -> DedupWatermark {
+ DedupWatermark {
+ client,
+ user_id: 1,
+ watermark,
+ latest_commit,
+ committed_window: 0b1011,
}
}
@@ -525,6 +654,7 @@ mod tests {
next_offset: 0,
consumers: Vec::new(),
groups: Vec::new(),
+ dedup: Vec::new(),
};
let encoded = empty.encode();
assert_eq!(
@@ -584,6 +714,126 @@ mod tests {
);
}
+ #[test]
+ fn given_unordered_dedup_clients_when_decoded_should_reject() {
+ let unordered = ConsumerOffsetsWire {
+ purge_generation: 0,
+ next_offset: 0,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ dedup: vec![dedup_entry(9, 1, 1), dedup_entry(4, 2, 2)],
+ };
+ assert_eq!(
+ ConsumerOffsetsWire::decode(&unordered.encode()),
+ Err(ConsumerOffsetsWireError::NonAscendingClient { client: 4 })
+ );
+ }
+
+ #[test]
+ fn given_reserved_client_in_dedup_when_decoded_should_reject() {
+ let reserved = ConsumerOffsetsWire {
+ purge_generation: 0,
+ next_offset: 0,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ dedup: vec![dedup_entry(0, 1, 1), dedup_entry(4, 2, 2)],
+ };
+ assert_eq!(
+ ConsumerOffsetsWire::decode(&reserved.encode()),
+ Err(ConsumerOffsetsWireError::ReservedClient)
+ );
+ }
+
+ #[test]
+ fn given_v1_artifact_when_decoded_should_install_empty_dedup() {
+ // An un-upgraded primary still ships "ICO1": same fields minus the
+ // dedup count and section. It must decode, with nothing to absorb.
+ let mut bytes = Vec::new();
+ bytes.extend_from_slice(&CONSUMER_OFFSETS_MAGIC_V1);
+ bytes.push(CONSUMER_OFFSETS_VERSION_V1);
+ bytes.extend_from_slice(&3u64.to_le_bytes());
+ bytes.extend_from_slice(&43u64.to_le_bytes());
+ bytes.extend_from_slice(&1u32.to_le_bytes());
+ bytes.extend_from_slice(&1u32.to_le_bytes());
+ for (id, offset) in [(7u32, 42u64), (2, 5)] {
+ bytes.extend_from_slice(&id.to_le_bytes());
+ bytes.extend_from_slice(&offset.to_le_bytes());
+ }
+ let trailer = state_artifact_checksum(&bytes);
+ bytes.extend_from_slice(&trailer.to_le_bytes());
+ assert_eq!(
+ ConsumerOffsetsWire::decode(&bytes),
+ Ok(ConsumerOffsetsWire {
+ purge_generation: 3,
+ next_offset: 43,
+ consumers: vec![(7, 42)],
+ groups: vec![(2, 5)],
+ dedup: Vec::new(),
+ })
+ );
+ }
+
+ #[test]
+ fn given_v1_magic_with_wrong_version_when_decoded_should_reject() {
+ let mut bytes = Vec::new();
+ bytes.extend_from_slice(&CONSUMER_OFFSETS_MAGIC_V1);
+ bytes.push(CONSUMER_OFFSETS_VERSION);
+ bytes.extend_from_slice(&0u64.to_le_bytes());
+ bytes.extend_from_slice(&0u64.to_le_bytes());
+ bytes.extend_from_slice(&0u32.to_le_bytes());
+ bytes.extend_from_slice(&0u32.to_le_bytes());
+ let trailer = state_artifact_checksum(&bytes);
+ bytes.extend_from_slice(&trailer.to_le_bytes());
+ assert_eq!(
+ ConsumerOffsetsWire::decode(&bytes),
+ Err(ConsumerOffsetsWireError::UnsupportedVersion {
+ version: CONSUMER_OFFSETS_VERSION
+ })
+ );
+ }
+
+ #[test]
+ fn given_dedup_count_past_ceiling_when_decoded_should_reject_before_allocating() {
+ let mut bytes = Vec::new();
+ bytes.extend_from_slice(&CONSUMER_OFFSETS_MAGIC);
+ bytes.push(CONSUMER_OFFSETS_VERSION);
+ bytes.extend_from_slice(&0u64.to_le_bytes());
+ bytes.extend_from_slice(&0u64.to_le_bytes());
+ bytes.extend_from_slice(&0u32.to_le_bytes());
+ bytes.extend_from_slice(&0u32.to_le_bytes());
+ bytes.extend_from_slice(&(CONSUMER_OFFSETS_ENTRIES_MAX + 1).to_le_bytes());
+ let trailer = state_artifact_checksum(&bytes);
+ bytes.extend_from_slice(&trailer.to_le_bytes());
+ assert_eq!(
+ ConsumerOffsetsWire::decode(&bytes),
+ Err(ConsumerOffsetsWireError::TooManyEntries {
+ section: "dedup",
+ count: CONSUMER_OFFSETS_ENTRIES_MAX + 1,
+ max: CONSUMER_OFFSETS_ENTRIES_MAX,
+ })
+ );
+ }
+
+ #[test]
+ fn given_dedup_count_exceeding_bytes_when_decoded_should_reject_as_truncated() {
+ // Under the ceiling but past the bytes present: the stride guard is
+ // what stops a ~30 byte artifact reserving megabytes.
+ let mut bytes = Vec::new();
+ bytes.extend_from_slice(&CONSUMER_OFFSETS_MAGIC);
+ bytes.push(CONSUMER_OFFSETS_VERSION);
+ bytes.extend_from_slice(&0u64.to_le_bytes());
+ bytes.extend_from_slice(&0u64.to_le_bytes());
+ bytes.extend_from_slice(&0u32.to_le_bytes());
+ bytes.extend_from_slice(&0u32.to_le_bytes());
+ bytes.extend_from_slice(&1_000u32.to_le_bytes());
+ let trailer = state_artifact_checksum(&bytes);
+ bytes.extend_from_slice(&trailer.to_le_bytes());
+ assert_eq!(
+ ConsumerOffsetsWire::decode(&bytes),
+ Err(ConsumerOffsetsWireError::Truncated)
+ );
+ }
+
#[test]
fn given_count_past_ceiling_when_decoded_should_reject_before_allocating() {
let mut bytes = Vec::new();
@@ -593,6 +843,7 @@ mod tests {
bytes.extend_from_slice(&0u64.to_le_bytes());
bytes.extend_from_slice(&(CONSUMER_OFFSETS_ENTRIES_MAX + 1).to_le_bytes());
bytes.extend_from_slice(&0u32.to_le_bytes());
+ bytes.extend_from_slice(&0u32.to_le_bytes());
let trailer = state_artifact_checksum(&bytes);
bytes.extend_from_slice(&trailer.to_le_bytes());
assert_eq!(
@@ -612,6 +863,7 @@ mod tests {
next_offset: 0,
consumers: vec![(5, 1), (5, 2)],
groups: Vec::new(),
+ dedup: Vec::new(),
};
assert_eq!(
ConsumerOffsetsWire::decode(&duplicate.encode()),
@@ -625,6 +877,7 @@ mod tests {
next_offset: 0,
consumers: Vec::new(),
groups: vec![(9, 1), (4, 2)],
+ dedup: Vec::new(),
};
assert_eq!(
ConsumerOffsetsWire::decode(&unordered.encode()),
@@ -1704,11 +1957,13 @@ where
// sealed segment while the counter stands at N, and the receiver
// must resume minting at N either way.
let next_offset = self.offset_frontier();
+ let dedup = self.dedup().watermarks_sorted();
ConsumerOffsetsWire {
purge_generation: self.applied_purge_generation,
next_offset,
consumers,
groups,
+ dedup,
}
}
@@ -2460,6 +2715,13 @@ where
// file put a rejoin carrying thousands of consumers on the pump for
// thousands of sequential open + write + optional fsync round trips;
// the tick's superblock pre-pass sets the precedent for the width.
+ // The dedup slice is memory-only, so it installs here with the maps
+ // rather than being written anywhere. No frontier fence is needed: the
+ // install lifts `commit_min` to the offer's `commit_op`, so the commit
+ // walk that follows starts strictly above everything this artifact
+ // covers, and `record_commit` is idempotent besides.
+ self.dedup_mut()
+ .install_watermarks(offsets_wire.dedup.iter().copied());
let mut planned: Vec<PlannedOffsetWrite> =
Vec::with_capacity(offsets_wire.consumers.len() + offsets_wire.groups.len());
if let Some(dir) = self.consumer_offsets_path.clone() {
@@ -2682,6 +2944,10 @@ where
// promise rested on the caller clearing it first.
self.segment_checksum_cache.borrow_mut().clear();
self.reuse_scan_memo.borrow_mut().take();
+ // Degrade to at-least-once rather than keep watermarks that may now
+ // describe data this partition no longer holds: a stale entry would
+ // absorb a replay whose original was just unlinked.
+ self.dedup_mut().install_watermarks(std::iter::empty());
// Sweep EVERY segment file, not the in-memory count's worth: after
// a late failure the renamed-in new chain is on disk while the
diff --git a/core/server/config.toml b/core/server/config.toml
index 79bdc46e8..5584b0378 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -985,6 +985,17 @@ clients_table_max = 8192
# u128 bitset, and this depth bounds that suffix.
prepare_queue_depth = 32
+# Distinct clients each partition group tracks request watermarks for, so a
+# retried produce or consumer-offset write is answered instead of committing a
+# second time. At capacity the client whose newest commit is oldest is evicted;
+# that client's next replay re-executes, exactly as it would have before dedup
+# existed, so under-sizing degrades rather than breaks. Must be > 0 and <= 65536.
+#
+# Unlike [metadata] clients_table_max, this budget is PER GROUP, so worst-case
+# memory scales with partition count. Size it to the producers one partition
+# actually sees, not the node's client total.
+dedup_clients_max = 4096
+
# Entries the evicted ring retains per multi-replica partition for journal
# repair after a peer rejoins. Larger widens the window a restarting peer can be
# served from the ring before falling back to bulk sync, at the cost of pinned
diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs
index 0115b5c60..e614e688f 100644
--- a/core/server/src/boot/recovery.rs
+++ b/core/server/src/boot/recovery.rs
@@ -291,6 +291,9 @@ const _: () = assert!(
configs::partition::DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH
== consensus::PIPELINE_PREPARE_QUEUE_MAX
);
+const _: () = assert!(
+ configs::partition::PARTITION_DEDUP_CLIENTS_DEFAULT == consensus::PARTITION_DEDUP_CLIENTS_MAX
+);
const _: () =
assert!(configs::metadata::DEFAULT_METADATA_CLIENTS_TABLE_MAX == consensus::CLIENTS_TABLE_MAX);
const _: () =
diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs
index 4da515f8e..b2e0789bf 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -319,9 +319,9 @@ fn build_auto_commit_request(
operation: Operation::StoreConsumerOffset,
size,
client: AUTO_COMMIT_CLIENT_ID,
- // The partition plane is sessionless (no `ClientTable` dedup); a
- // nonzero session + request just satisfy the wire header
- // validation.
+ // The reserved sentinel client is never deduped and never
+ // replied to; a nonzero session + request just satisfy the wire
+ // header validation.
session: 1,
request: 1,
group: namespace.inner(),
@@ -334,10 +334,18 @@ fn build_auto_commit_request(
/// Route a partition data-plane op (`SendMessages` / consumer-offset writes)
/// through the shard mesh by namespace: the op belongs to the partition's
/// own consensus group, not the metadata group. The owning shard's
-/// partitions plane runs at-least-once consensus and replies directly via
-/// `send_to_client`. `header.client` therefore stays the TRANSPORT id
-/// (home-shard routing bits), not the VSR session id -- partition ops are
-/// sessionless ("session lifecycle is metadata-only").
+/// partitions plane dedups the request against its group's slice, so
+/// `header.client` carries the VSR consensus id (the dedup key) rather than
+/// the transport id.
+///
+/// How the committed reply gets back depends on whether the bus can route that
+/// id. HTTP registers each session under its own shard-0 transport id, so the
+/// two are equal and the plane's `send_to_client` fires the session's
+/// in-process reply slot directly; the request is dispatched and forgotten
+/// here (`?ack=none` relies on exactly that: nothing listening, reply shed at
+/// the bus). Every other transport registers under a client-chosen id the bus
+/// cannot route, so the request is submitted with an in-process channel and
+/// the reply is relayed to the socket this shard holds.
///
/// Callers must have authenticated the transport already: `vsr_client_id` /
/// `bound_session` come from its bound VSR session. Every failure before
@@ -475,17 +483,98 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>(
let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
new_header.group = namespace;
- new_header.client = transport_client_id;
- // Header validation requires `session > 0 && request > 0` for
- // non-register ops. The partition plane itself is sessionless
- // (at-least-once, no `ClientTable` dedup), so the bound VSR
- // session merely satisfies validation. Current SDKs do number
- // partition ops, but older and internal callers may still send
- // zero, so a zero id is normalized to the compatibility value 1.
+ // The VSR consensus id, exactly as metadata ops are stamped: it is the
+ // dedup key every replica keys its slice by, and unlike the transport
+ // id it stays valid across nodes. Replies therefore cannot be routed
+ // by this field -- they ride the submit's channel back to this shard,
+ // which owns the socket.
+ new_header.client = vsr_client_id;
+ // Header validation requires `session > 0` for non-register ops. The
+ // slices mint no epoch of their own, so the bound VSR session only
+ // satisfies validation here.
new_header.session = bound_session;
- new_header.request = new_header.request.max(1);
+ // The session's owner as this shard resolved it, never the
+ // client-supplied value: the dedup slice keys on it to tell a re-minted
+ // id's next holder apart from its previous one, so it has to be
+ // trustworthy on every replica. Same rule as the metadata plane. The
+ // gate above failed closed on `None`, so `0` (unattributed) is only a
+ // type-level fallback here.
+ new_header.user_id = acting_user_id.unwrap_or(0);
+ });
+ if vsr_client_id == transport_client_id {
+ shard.dispatch(request.into_generic());
+ return;
+ }
+ relay_partition_reply(
+ shard,
+ IggyNamespace::from_raw(namespace),
+ request,
+ transport_client_id,
+ &header,
+ )
+ .await;
+}
+
+/// Submit a partition write whose reply the bus cannot route (the routed
+/// header's `client` is the VSR consensus id, whose bits encode no home shard)
+/// and relay the committed reply to the socket this shard holds.
+///
+/// Only the submit runs on the caller's drain loop: it is what fixes the order
+/// two writes from one connection reach the owning shard in. The wait for the
+/// commit is spawned, so a connection keeps draining its queued polls and
+/// metadata ops instead of holding them behind one replication round trip.
+#[allow(clippy::future_not_send)]
+async fn relay_partition_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ namespace: IggyNamespace,
+ request: Message<RoutedRequestHeader>,
+ transport_client_id: u128,
+ header: &RoutedRequestHeader,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Ok(ticket) = shard.partition_submit(namespace, request) else {
+ // `PartitionSubmitRefused`: the frame never reached the owning shard,
+ // so this is a known outcome and the client can be told now rather
+ // than after its read-timeout. Same transient the plane itself answers
+ // for a request it could not admit.
+ send_deny_reply(
+ shard,
+ transport_client_id,
+ header,
+ IggyError::TransientNotAccepted.as_code(),
+ )
+ .await;
+ return;
+ };
+ let operation = header.operation;
+ let shard = Rc::clone(shard);
+ // Through the bus, not the runtime directly: the simulator supplies its
+ // own executor and virtual clock.
+ shard.bus.clone().spawn(async move {
+ let Some(reply) = shard.await_partition_submit(ticket).await else {
+ // Abandoned or expired. Deliberately silent: the outcome is
+ // unknown, and a synthesized failure could contradict a write that
+ // commits moments later. The client's read-timeout is the recovery.
+ return;
+ };
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id, reply.into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ operation = ?operation,
+ error = %error,
+ "failed to forward committed partition reply to its socket"
+ );
+ }
});
- shard.dispatch(request.into_generic());
}
/// Serve `poll_messages`: resolve the partition namespace, run the read on
diff --git a/core/server/src/http/session.rs b/core/server/src/http/session.rs
index aa1db608c..b03b94bf5 100644
--- a/core/server/src/http/session.rs
+++ b/core/server/src/http/session.rs
@@ -104,12 +104,18 @@ pub(in crate::http) struct HttpSession {
/// a larger one would arrive at or below the watermark and be refused as a
/// duplicate.
pub(in crate::http) gate: Mutex<u64>,
- /// Next data-plane request id. A separate, gate-free counter: partition ops
- /// are at-least-once with no consensus dedup, so the id only correlates the
- /// in-process reply slot and concurrent produces on one session are legal.
- /// A plain `Cell` suffices on single-threaded shard 0; ids are minted
- /// monotonically and never reused, which the slot-guard contract requires.
- pub(in crate::http) data_request: Cell<u64>,
+ /// Serializes this session's data-plane writes the way `gate` does its
+ /// metadata writes: the guarded value is the NEXT request id, and the write
+ /// path holds the lock from the mint until the request has been handed to
+ /// the owning shard's inbox. The partition slice dedups on a per-client
+ /// watermark, so two handlers that minted in one order but reached the
+ /// shard in the other (one slept in the routable wait, say) would have the
+ /// lower id absorbed as a duplicate with a success status. Concurrent
+ /// awaits stay legal: the lock covers admission, not the commit round trip.
+ /// Shared with the `?ack=none` path so a shed reply's id never collides
+ /// with a live awaited slot on this session. Ids are minted monotonically
+ /// and never reused, which the slot-guard contract requires.
+ pub(in crate::http) data_gate: Mutex<u64>,
/// Registry token of this session's lazily-installed in-process reply
/// target (`None` until the first awaited partition write). Stored so
/// session eviction can tear the registry entry down fenced by the same
@@ -121,17 +127,6 @@ pub(in crate::http) struct HttpSession {
pub(in crate::http) in_flight_writes: Cell<u32>,
}
-impl HttpSession {
- /// Mint the next data-plane request id. Also consumed by the `?ack=none`
- /// path, which installs no slot: sharing one counter keeps a shed reply's
- /// id from ever colliding with a live awaited slot on this session.
- pub(in crate::http) fn next_data_request_id(&self) -> u64 {
- let id = self.data_request.get();
- self.data_request.set(id + 1);
- id
- }
-}
-
/// Serializes first-use VSR registration per credential key so a herd of
/// concurrent first-requests for one token runs exactly one `Register` instead
/// of N that each mint a client id and orphan N-1 slots last-writer-wins.
@@ -271,7 +266,7 @@ mod tests {
user_id: DEFAULT_ROOT_USER_ID,
expiry: u64::MAX,
gate: Mutex::new(FIRST_REQUEST_ID),
- data_request: Cell::new(FIRST_REQUEST_ID),
+ data_gate: Mutex::new(FIRST_REQUEST_ID),
registry_token: Cell::new(None),
in_flight_writes: Cell::new(0),
});
@@ -303,7 +298,7 @@ mod tests {
user_id: DEFAULT_ROOT_USER_ID,
expiry,
gate: Mutex::new(FIRST_REQUEST_ID),
- data_request: Cell::new(FIRST_REQUEST_ID),
+ data_gate: Mutex::new(FIRST_REQUEST_ID),
registry_token: Cell::new(None),
in_flight_writes: Cell::new(0),
})
diff --git a/core/server/src/http/state.rs b/core/server/src/http/state.rs
index 73b2c6cb7..5f1984cd4 100644
--- a/core/server/src/http/state.rs
+++ b/core/server/src/http/state.rs
@@ -367,7 +367,7 @@ impl HttpInner {
user_id,
expiry,
gate: Mutex::new(FIRST_REQUEST_ID),
- data_request: Cell::new(FIRST_REQUEST_ID),
+ data_gate: Mutex::new(FIRST_REQUEST_ID),
registry_token: Cell::new(None),
in_flight_writes: Cell::new(0),
}))
diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs
index d73b2005e..0babb3952 100644
--- a/core/server/src/http/submit.rs
+++ b/core/server/src/http/submit.rs
@@ -377,7 +377,13 @@ pub(in crate::http) async fn partition_write_replicated(
// timeout. Held across every exit below; released by `Drop`.
let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?;
ensure_in_process_reply_target(state, session);
- let request_id = session.next_data_request_id();
+ // Held from the mint until `dispatch_partition_request` returns, which is
+ // past the owning shard's inbox: the ids of this session's writes must
+ // reach the partition in mint order or the watermark absorbs the overtaken
+ // one. Released before the commit wait so writes still overlap there.
+ let mut next_data_request_id = session.data_gate.lock().await;
+ let request_id = *next_data_request_id;
+ *next_data_request_id += 1;
let message = build_request_message(
operation,
session.client_id,
@@ -407,6 +413,7 @@ pub(in crate::http) async fn partition_write_replicated(
Some(session.user_id),
)
.await;
+ drop(next_data_request_id);
let outcome = compio::time::timeout(PARTITION_WRITE_REPLY_TIMEOUT, receiver).await;
// Removes the slot unless the reply already fired, so a late commit
// reply after a timeout sheds at the bus instead of leaking a waiter.
@@ -442,7 +449,10 @@ pub(in crate::http) async fn produce_unacked(
body: &[u8],
) -> Result<(), PartitionWriteError> {
let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?;
- let request_id = session.next_data_request_id();
+ // Same gate as the acked path, for the same ordering reason.
+ let mut next_data_request_id = session.data_gate.lock().await;
+ let request_id = *next_data_request_id;
+ *next_data_request_id += 1;
let message = build_request_message(
Operation::SendMessages,
session.client_id,
@@ -459,6 +469,7 @@ pub(in crate::http) async fn produce_unacked(
Some(session.user_id),
)
.await;
+ drop(next_data_request_id);
Ok(())
}
diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs
index ad4471da9..ec39868e5 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -797,6 +797,7 @@ async fn load_partition(
config.partition.evicted_ring_capacity,
config.partition.evicted_ring_bytes_max.as_bytes_u64(),
);
+ partition.set_dedup_clients_max(config.partition.dedup_clients_max);
partition.set_partition_dir(partition_dir.clone());
// Before the hydrate: the durable record is keyed by incarnation, so a
// `purge.gen` left behind by a previous life of this namespace reads 0.
@@ -1210,6 +1211,7 @@ pub async fn build_partition_fresh(
config.partition.evicted_ring_capacity,
config.partition.evicted_ring_bytes_max.as_bytes_u64(),
);
+ partition.set_dedup_clients_max(config.partition.dedup_clients_max);
partition.set_partition_dir(partition_dir);
// Fresh dirs read generation 0; a dir surviving from a crashed process
// (this "fresh" build races repair re-materialization) reads the last
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 8b56a415a..fe4deaea5 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -402,6 +402,27 @@ pub type PartitionReadHandler =
/// deadline.
const PARTITION_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+/// Budget for a partition write's wait on its committed reply. Longer than a
+/// read: the wait spans replication quorum plus any park-and-promote the
+/// request rides through, and a view change mid-flight re-proposes under the
+/// new primary. Expiry leaves the client to its own read-timeout.
+const PARTITION_SUBMIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+
+/// A partition write admitted onto its owning shard's inbox, awaiting the
+/// committed reply. Redeem with [`IggyShard::await_partition_submit`].
+pub struct PartitionSubmitTicket {
+ receiver: Receiver<Option<Message<GenericHeader>>>,
+ target: u16,
+}
+
+/// The write never reached the owning shard.
+///
+/// No sender existed for the target, or its inbox refused the frame. Either
+/// way the outcome is known, unlike a reply that fails to arrive, so the caller
+/// may deny the client outright.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PartitionSubmitRefused;
+
/// Race `future` against a bus timer.
///
/// `Some` if it finishes within `budget`, `None` if the timer fires first.
@@ -703,6 +724,17 @@ pub enum LifecycleFrame {
read: PartitionRead,
reply: Sender<PartitionReadReply>,
},
+ /// Admit a partition write (`SendMessages` / consumer-offset write) on
+ /// the shard owning its namespace, carrying the channel its committed
+ /// reply travels back on. The partition plane cannot route a reply by
+ /// `header.client` -- that field is the VSR consensus id, whose bits
+ /// carry no home-shard routing -- so the reply returns to the
+ /// connection-owning shard, which writes it to the socket it holds.
+ /// See [`IggyShard::partition_submit`].
+ PartitionSubmit {
+ request: Message<RoutedRequestHeader>,
+ reply: Sender<Option<Message<GenericHeader>>>,
+ },
/// Shard 0 broadcasts after a partition-shaped metadata commit; wakes
/// the per-shard reconciler. No payload: reconciler re-reads target
/// state. Drops covered by the periodic safety tick.
@@ -1429,6 +1461,11 @@ where
/// from. Cleared when the park map empties, so one episode warns once.
shard_park_shedding: Cell<bool>,
+ /// Set once a partition submit has waited out its budget and warned;
+ /// cleared by the next reply that arrives. Gates the timeout warning to
+ /// one line per stall episode (see [`Self::await_partition_submit`]).
+ partition_submit_stalled: Cell<bool>,
+
/// Live ceiling on prepares served per `RequestPrepares` round. Defaults
/// to [`REPAIR_CHUNK_MAX`]; the server overrides it from
/// `[cluster] repair_chunk_max` at bootstrap.
@@ -1607,6 +1644,7 @@ where
parked_partition_bytes: Cell::new(0),
redispatch_queue: RefCell::new(VecDeque::new()),
shard_park_shedding: Cell::new(false),
+ partition_submit_stalled: Cell::new(false),
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
@@ -1872,6 +1910,121 @@ where
}
}
+ /// Admit a partition write on the shard owning `namespace`. Routes through
+ /// the shards table exactly like [`Self::partition_read`], self-sends
+ /// included, so a locally-owned partition takes the same path.
+ ///
+ /// Synchronous up to the inbox `try_send`, so two writes a caller admits
+ /// back to back reach the owning shard in that order; the committed reply
+ /// is awaited separately through [`Self::await_partition_submit`], which a
+ /// connection's drain loop spawns rather than blocks on.
+ ///
+ /// # Errors
+ /// [`PartitionSubmitRefused`] when the frame provably never reached the
+ /// owning shard (no sender for the target, or a full inbox), so the caller
+ /// can deny the client outright instead of leaving it to a read-timeout
+ /// for an outcome that is already known.
+ pub fn partition_submit(
+ &self,
+ namespace: IggyNamespace,
+ request: Message<RoutedRequestHeader>,
+ ) -> Result<PartitionSubmitTicket, PartitionSubmitRefused> {
+ let target = self.shards_table.shard_for(namespace).unwrap_or_else(|| {
+ // Same fallback as `route_typed`: a miss means "not seeded yet",
+ // not "unroutable", and the owning shard parks what arrives early.
+ crate::shards_table::calculate_shard_from_consensus_ns(
+ namespace.inner(),
+ self.shard_count,
+ )
+ });
+ let (reply_tx, reply_rx) = channel::<Option<Message<GenericHeader>>>(1);
+ let frame = ShardFrame::lifecycle(LifecycleFrame::PartitionSubmit {
+ request,
+ reply: reply_tx,
+ });
+ let Some(sender) = self.senders.get(target as usize) else {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::UNROUTABLE,
+ );
+ return Err(PartitionSubmitRefused);
+ };
+ if let Err(error) = sender.try_send(frame) {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::coordinator::classify_try_send_err(&error),
+ );
+ tracing::warn!(
+ shard = self.id,
+ target,
+ "partition_submit: inbox rejected PartitionSubmit frame: {error:?}"
+ );
+ return Err(PartitionSubmitRefused);
+ }
+ Ok(PartitionSubmitTicket {
+ receiver: reply_rx,
+ target,
+ })
+ }
+
+ /// Wait out a submitted write's committed reply.
+ ///
+ /// `None` = reply channel dropped before a reply (view-change reset, park
+ /// eviction, shutdown) or budget expiry. The caller stays silent on `None`:
+ /// the outcome is unknown, so a synthesized failure could contradict a
+ /// write that commits moments later, and the client's own read-timeout is
+ /// the recovery. Both exits count under
+ /// `frame_drops_total{variant=partition}` with their own reasons. The
+ /// timeout warning fires once per stall episode, reset by the next reply
+ /// that does arrive: one wedged group would otherwise log a line per
+ /// request, and the counter carries the volume.
+ #[allow(clippy::future_not_send)]
+ pub async fn await_partition_submit(
+ &self,
+ ticket: PartitionSubmitTicket,
+ ) -> Option<Message<GenericHeader>> {
+ let PartitionSubmitTicket { receiver, target } = ticket;
+ match bus_timeout(&self.bus, PARTITION_SUBMIT_TIMEOUT, receiver.recv()).await {
+ Some(Ok(Some(reply))) => {
+ self.partition_submit_stalled.set(false);
+ Some(reply)
+ }
+ Some(Ok(None) | Err(_)) => {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::SUBMIT_ABANDONED,
+ );
+ tracing::debug!(
+ shard = self.id,
+ target,
+ "partition_submit: reply channel dropped before commit"
+ );
+ None
+ }
+ None => {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::SUBMIT_TIMEOUT,
+ );
+ if self.partition_submit_stalled.replace(true) {
+ tracing::debug!(
+ shard = self.id,
+ target,
+ "partition_submit: owning shard did not reply within budget"
+ );
+ } else {
+ tracing::warn!(
+ shard = self.id,
+ target,
+ "partition_submit: owning shard did not reply within budget; \
+ further expiries log at debug until a reply arrives"
+ );
+ }
+ None
+ }
+ }
+ }
+
/// Return a clone of the shard-0 coordinator handle, if attached.
/// Bootstrap uses this to wire the listener accept callbacks
/// (replica + client) to coordinator-driven fd-delegation instead
@@ -1937,6 +2090,7 @@ where
parked_partition_bytes: Cell::new(0),
redispatch_queue: RefCell::new(VecDeque::new()),
shard_park_shedding: Cell::new(false),
+ partition_submit_stalled: Cell::new(false),
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
@@ -2413,6 +2567,12 @@ struct ParkedFrame {
/// the outcome from a reply rather than a timeout.
passes: u32,
message: Message<GenericHeader>,
+ /// Channel the committed reply travels back on, for a frame that arrived
+ /// as a [`LifecycleFrame::PartitionSubmit`]. `None` for replicated
+ /// prepares and for writes admitted without a waiter. Dropping the frame
+ /// (expiry, teardown, shutdown) drops this, which wakes the awaiting
+ /// dispatch with a receive error it maps to silence.
+ reply: Option<Sender<Option<Message<GenericHeader>>>>,
}
impl ParkedFrame {
@@ -2426,6 +2586,17 @@ impl ParkedFrame {
}
}
+/// One staged frame classified for re-delivery.
+///
+/// `reply` is the submit channel the frame parked with, if any: it decides which
+/// admission path the pump re-enters, since a submit's committed reply cannot be
+/// routed by `header.client` (that field is the VSR consensus id).
+struct RedispatchedFrame {
+ message: MessageBag,
+ provenance: ParkProvenance,
+ reply: Option<Sender<Option<Message<GenericHeader>>>>,
+}
+
/// What a frame keeps if it parks again after the pump re-delivers it.
///
/// Production prevents that race by ranking redispatch above inbox work and by
@@ -2573,12 +2744,13 @@ where
/// arm. The queue borrow ends before dispatch awaits, so simulator
/// materialisation can append off-pump without colliding with a suspended
/// `RefCell` guard.
- fn pop_redispatched_frame(&self) -> Option<(MessageBag, ParkProvenance)> {
+ fn pop_redispatched_frame(&self) -> Option<RedispatchedFrame> {
loop {
let ParkedFrame {
epoch,
passes,
message,
+ reply,
} = self.redispatch_queue.borrow_mut().pop_front()?;
let provenance = ParkProvenance { epoch, passes };
// Parked frames are stored generic (the buffer holds every variant
@@ -2586,7 +2758,13 @@ where
// the rare path - a post-`CreateTopic` convergence window, not the
// per-message steady state the bag handoff exists for.
match MessageBag::try_from(message) {
- Ok(bag) => return Some((bag, provenance)),
+ Ok(message) => {
+ return Some(RedispatchedFrame {
+ message,
+ provenance,
+ reply,
+ });
+ }
Err(error) => {
// The frame classified once already, on the way in, so this
// is unreachable short of memory corruption. The consumed
@@ -2607,6 +2785,42 @@ where
}
}
+ /// Deliver one staged frame, through the admission path it arrived on.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn dispatch_redispatched_frame(&self, frame: RedispatchedFrame)
+ where
+ B: MessageBus + 'static,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ M: RestorableMetadataStm,
+ T: ShardsTable,
+ {
+ let RedispatchedFrame {
+ message,
+ provenance,
+ reply,
+ } = frame;
+ match (reply, message) {
+ (Some(reply), MessageBag::Request(request)) => {
+ self.dispatch_partition_submit(request, reply, Some(provenance))
+ .await;
+ }
+ (Some(_), _) => {
+ // Only a client request parks with a waiter attached, so this is
+ // unreachable short of a classify that disagrees with the one
+ // the frame passed on the way in. Dropping the sender wakes the
+ // awaiting shard, which maps the receive error to silence.
+ tracing::error!(
+ shard = self.id,
+ "staged partition frame carries a reply channel but is not a client request; \
+ dropping it"
+ );
+ }
+ (None, message) => self.dispatch_message(message, Some(provenance)).await,
+ }
+ }
+
/// Test-only delivery of one staged frame. Production obtains frames through
/// the router's ranked select arm, which also processes loopback after each
/// one. This hook exists for the reconciler's defence-in-depth interleaving.
@@ -2621,10 +2835,10 @@ where
M: RestorableMetadataStm,
T: ShardsTable,
{
- let Some((message, provenance)) = self.pop_redispatched_frame() else {
+ let Some(frame) = self.pop_redispatched_frame() else {
return false;
};
- self.dispatch_message(message, Some(provenance)).await;
+ self.dispatch_redispatched_frame(frame).await;
true
}
@@ -2665,7 +2879,9 @@ where
let header = request.header();
(header.operation, header.group)
};
- match self.park_if_unmaterialised(request, routing.0, routing.1, provenance) {
+ match self
+ .park_if_unmaterialised(request, routing.0, routing.1, provenance, &mut None)
+ {
// The incarnation fence runs only here, on client traffic.
// A backup denying what the primary admitted would diverge
// the replicas, so replicated frames are never fenced.
@@ -2696,7 +2912,9 @@ where
// A tombstoned prepare still flows to the plane: replicated
// traffic has no client awaiting a reply on this node, and
// the plane's own tombstone guard drops it.
- match self.park_if_unmaterialised(prepare, routing.0, routing.1, provenance) {
+ match self
+ .park_if_unmaterialised(prepare, routing.0, routing.1, provenance, &mut None)
+ {
ParkOutcome::Deliver(prepare) | ParkOutcome::Tombstoned(prepare) => {
self.on_replicate(prepare).await;
// A follower learns the cluster commit point from the
@@ -3129,12 +3347,21 @@ where
/// repair must fill. The `false` return is what makes callers bump
/// `frame_drops_total{variant=partition,reason=park_dropped}`.
fn deny_parked_client_request(&self, frame: ParkedFrame) -> bool {
- if frame.message.header().command == Command::Request
- && let Ok(request) = frame.message.try_into_typed::<RoutedRequestHeader>()
- {
- return self.stage_transient_deny(request.header());
+ let ParkedFrame { message, reply, .. } = frame;
+ if message.header().command != Command::Request {
+ return false;
}
- false
+ let Ok(request) = message.try_into_typed::<RoutedRequestHeader>() else {
+ return false;
+ };
+ // A submit cannot be answered through `stage_transient_deny`: it routes
+ // by `header.client`, which on a partition request is the VSR consensus
+ // id and addresses no connection. Its own channel reaches the shard
+ // holding the socket.
+ if reply.is_some() {
+ return Self::answer_partition_submit_transient(request.header(), reply);
+ }
+ self.stage_transient_deny(request.header())
}
/// A parked frame addressed an incarnation this shard no longer holds.
@@ -3204,6 +3431,7 @@ where
operation: Operation,
namespace_raw: u64,
provenance: Option<ParkProvenance>,
+ reply: &mut Option<Sender<Option<Message<GenericHeader>>>>,
) -> ParkOutcome<H>
where
H: iggy_binary_protocol::ConsensusHeader,
@@ -3331,6 +3559,7 @@ where
epoch,
passes,
message: message.into_generic(),
+ reply: reply.take(),
});
drop(pending);
self.parked_partition_bytes
@@ -3413,6 +3642,106 @@ where
true
}
+ /// Admit a `PartitionSubmit`: same gates as the [`MessageBag::Request`]
+ /// arm, but every refusal answers on `reply` instead of the bus, and the
+ /// admitted request carries an in-process reply channel down to the
+ /// pipeline entry so its committed reply comes back here rather than
+ /// being routed by `header.client`.
+ #[allow(clippy::future_not_send)]
+ pub async fn on_partition_submit(
+ &self,
+ request: Message<RoutedRequestHeader>,
+ reply: Sender<Option<Message<GenericHeader>>>,
+ ) where
+ B: MessageBus + 'static,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ M: RestorableMetadataStm,
+ T: ShardsTable,
+ {
+ self.dispatch_partition_submit(request, reply, None).await;
+ }
+
+ /// [`Self::on_partition_submit`] carrying the park provenance of a submit
+ /// the pump is re-delivering, for the same reason
+ /// [`Self::dispatch_message`] carries it.
+ #[allow(clippy::future_not_send)]
+ async fn dispatch_partition_submit(
+ &self,
+ request: Message<RoutedRequestHeader>,
+ reply: Sender<Option<Message<GenericHeader>>>,
+ provenance: Option<ParkProvenance>,
+ ) where
+ B: MessageBus + 'static,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ M: RestorableMetadataStm,
+ T: ShardsTable,
+ {
+ let routing = {
+ let header = request.header();
+ (header.operation, header.group)
+ };
+ // The frame takes a clone of the sender only when it parks; every other
+ // outcome answers on the original below, so no arm can lose the waiter
+ // to a `None` it would have to guard against. The clone is an `Rc`
+ // bump, and whichever half is not used drops with this scope.
+ match self.park_if_unmaterialised(
+ request,
+ routing.0,
+ routing.1,
+ provenance,
+ &mut Some(reply.clone()),
+ ) {
+ ParkOutcome::Deliver(request)
+ if !self.serves_committed_incarnation(routing.0, routing.1) =>
+ {
+ Self::answer_partition_submit_transient(request.header(), Some(reply));
+ }
+ ParkOutcome::Deliver(request) => {
+ let (sender, receiver) = consensus::oneshot_channel();
+ self.plane
+ .partitions()
+ .on_request_with_reply(request, Some(sender))
+ .await;
+ // Await OFF the pump: the commit that fires this receiver needs
+ // the pump to keep draining acks, so blocking here would
+ // deadlock the very reply being waited on. The task holds only
+ // owned channel halves, never a partitions borrow.
+ //
+ // Through the bus, not the runtime directly: the simulator
+ // supplies its own executor and virtual clock.
+ self.bus.spawn(async move {
+ let committed = receiver.await.ok().map(Message::into_generic);
+ let _ = reply.try_send(committed);
+ });
+ }
+ ParkOutcome::Tombstoned(request) | ParkOutcome::Overflow(request) => {
+ Self::answer_partition_submit_transient(request.header(), Some(reply));
+ }
+ // The clone travelled with the parked frame; it answers on drain
+ // or wakes the awaiter with a receive error when the frame expires.
+ ParkOutcome::Parked => {}
+ }
+ }
+
+ /// Answer a refused `PartitionSubmit` with the same transient deny the bus
+ /// path sends, over the submit's own channel. `false` = nobody was
+ /// answered, so the frame still counts as dropped.
+ fn answer_partition_submit_transient(
+ request_header: &RoutedRequestHeader,
+ reply: Option<Sender<Option<Message<GenericHeader>>>>,
+ ) -> bool {
+ let Some(reply) = reply else { return false };
+ let deny = build_deny_reply_from_request_header(
+ request_header,
+ IggyError::TransientNotAccepted.as_code(),
+ );
+ reply.try_send(Some(deny.into_generic())).is_ok()
+ }
+
#[allow(clippy::future_not_send)]
pub async fn on_request(&self, request: Message<RoutedRequestHeader>)
where
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index 16007ffec..4fbd8c7fc 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -132,6 +132,12 @@ pub mod frame_drop_reason {
pub const MISROUTED: &str = "misrouted";
pub const PARK_OVERFLOW: &str = "park_overflow";
pub const PARK_DROPPED: &str = "park_dropped";
+ /// A partition write's reply channel was dropped before a reply arrived
+ /// (view-change pipeline reset, park teardown, shutdown): the outcome is
+ /// unknown and the client is left to its read-timeout.
+ pub const SUBMIT_ABANDONED: &str = "submit_abandoned";
+ /// A partition write's reply did not arrive within the submit budget.
+ pub const SUBMIT_TIMEOUT: &str = "submit_timeout";
}
// The tables only index the lazy fast-path cache below; a `{variant, reason}`
@@ -139,7 +145,7 @@ pub mod frame_drop_reason {
// site actually produces it, so the unreachable corners of the 7 x 9 cross
// product never appear as permanent zero-valued series.
const VARIANT_COUNT: usize = 7;
-const REASON_COUNT: usize = 9;
+const REASON_COUNT: usize = 11;
const VARIANTS: [&str; VARIANT_COUNT] = [
frame_drop_variant::CONSENSUS,
@@ -161,6 +167,8 @@ const REASONS: [&str; REASON_COUNT] = [
frame_drop_reason::MISROUTED,
frame_drop_reason::PARK_OVERFLOW,
frame_drop_reason::PARK_DROPPED,
+ frame_drop_reason::SUBMIT_ABANDONED,
+ frame_drop_reason::SUBMIT_TIMEOUT,
];
fn variant_index(s: &str) -> Option<usize> {
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index a866a9027..1f4fa0117 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -351,13 +351,13 @@ where
self.apply_reconcile_ops();
consensus_tick.set(rearm_tick());
}
- (message, provenance) = poll_fn(|_| {
+ frame = poll_fn(|_| {
self.pop_redispatched_frame().map_or(Poll::Pending, Poll::Ready)
}).fuse() => {
// One frame per select iteration. Ranking this arm above
// the inbox preserves park order without making a full
// queue stall ticks and commit broadcasts for other groups.
- self.dispatch_message(message, Some(provenance)).await;
+ self.dispatch_redispatched_frame(frame).await;
// A request handled by a solo primary self-acks here. If
// loopback waited for another inbox frame, the request
// would remain uncommitted indefinitely on a quiet shard.
@@ -480,8 +480,8 @@ where
M: RestorableMetadataStm,
{
loop {
- while let Some((message, provenance)) = self.pop_redispatched_frame() {
- self.dispatch_message(message, Some(provenance)).await;
+ while let Some(frame) = self.pop_redispatched_frame() {
+ self.dispatch_redispatched_frame(frame).await;
self.process_loopback(loopback_buf, namespace_scratch).await;
self.apply_reconcile_ops();
if let Some(fault) = self.first_partition_commit_fault() {
@@ -718,6 +718,14 @@ where
// times out.
(self.on_partition_read)(namespace, read, reply);
}
+ LifecycleFrame::PartitionSubmit { request, reply } => {
+ // Addressed to the shard owning the request's namespace (the
+ // sender resolved it via the shards table, same fallback as
+ // `route_typed`). Every refusal answers on `reply`, so the
+ // awaiting shard never waits out its budget on a decision
+ // already made.
+ self.on_partition_submit(request, reply).await;
+ }
LifecycleFrame::MetadataCommitTick => {
// Reconciler may not yet be wired (e.g. mid-bootstrap, or
// single-shard tests that never enable the reconciler loop).
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index fda8e5bcf..3d2fd3fea 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -2002,122 +2002,6 @@ mod tests {
);
}
- /// At-least-once failover: a `SendMessages` retry on a new primary re-executes.
- /// The retry reply carries a HIGHER `commit` op, proof of re-execution rather
- /// than dedup, and the duplicate payload lives at two offsets. Consumers dedup
- /// if they want at-most-once-per-payload.
- #[test]
- fn failover_retry_re_executes_under_at_least_once() {
- server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
- enabled: false,
- size: iggy_common::IggyByteSize::from(0u64),
- bucket_capacity: 1,
- });
-
- let replica_count: u8 = 5;
- let client_id: u128 = 1;
- let network_opts = packet::PacketSimulatorOptions {
- node_count: replica_count,
- client_count: 1,
- ..packet::PacketSimulatorOptions::default()
- };
-
- let mut sim = Simulator::new(
- replica_count as usize,
- std::iter::once(client_id),
- network_opts,
- );
- let client = SimClient::new(client_id);
- let ns = IggyNamespace::new(1, 1, 0);
- sim.init_partition(ns);
- sim.register_client_with_primary(&client);
-
- // Same `(client, session, request)` for replay; mirrors SDK's
- // connection-loss retry.
- let original_req = client.send_messages(ns, &[Bytes::from_static(b"failover-test")]);
- let replay_req = original_req.deep_copy();
- let original_request_id = original_req.header().request;
-
- sim.submit_request(client_id, 0, original_req.into_generic());
-
- let mut original_reply: Option<Message<ReplyHeader>> = None;
- for _ in 0..200 {
- let replies = sim.step();
- if !replies.is_empty() {
- original_reply = Some(replies[0].deep_copy());
- break;
- }
- }
- let original_reply = original_reply.expect("commit reply must arrive before primary crash");
- let original_commit_op = original_reply.header().commit;
- assert_eq!(
- original_reply.header().request,
- original_request_id,
- "sanity: original reply must echo the request id"
- );
-
- // Crash primary. Real-world: TCP buffer might have lost reply
- // before ack; same retry path.
- sim.replica_crash(0);
-
- // Steps for view change across 4 survivors.
- for _ in 0..800 {
- sim.step();
- }
-
- // Find new primary via any live replica.
- let live = &sim.replicas[1].shards[0];
- let live_consensus = live
- .plane
- .partitions()
- .get_by_ns(&ns)
- .expect("partition must exist on a live replica")
- .consensus();
- assert!(
- live_consensus.view() > 0,
- "view must have advanced past the crashed primary"
- );
- let new_primary_idx = live_consensus.primary_index(live_consensus.view());
- assert_ne!(
- new_primary_idx, 0,
- "new primary must not be the crashed replica"
- );
-
- // Replay the SAME request to the new primary. No dedup, so re-execution.
- sim.submit_request(client_id, new_primary_idx, replay_req.into_generic());
-
- let mut retry_reply: Option<Message<ReplyHeader>> = None;
- for _ in 0..200 {
- let replies = sim.step();
- if !replies.is_empty() {
- retry_reply = Some(replies[0].deep_copy());
- break;
- }
- }
- let retry_reply = retry_reply.expect(
- "reply must arrive after retry; new primary re-commits as \
- fresh prepare (at-least-once)",
- );
-
- // At-least-once: same request id (correlation), HIGHER commit op
- // (re-execution). No dedup absorbs the retry.
- assert_eq!(
- retry_reply.header().request,
- original_request_id,
- "retry's reply must correlate to the request id"
- );
- assert!(
- retry_reply.header().commit > original_commit_op,
- "retry must re-execute (commit op > original={original_commit_op}, got {})",
- retry_reply.header().commit
- );
- assert_eq!(
- retry_reply.header().client,
- client_id,
- "retry must echo original client_id"
- );
- }
-
/// Determinism: fresh simulator + workload from the same seed (network
/// and workload) produces an identical reply-header sequence.
#[test]
@@ -2617,6 +2501,11 @@ mod tests {
let client = SimClient::new(CLIENT_ID);
sim.shell_login(&client);
+ // Both sends in flight at once. The simulated link delays each packet
+ // independently, so the lower request id can reach the primary after
+ // the higher one committed; the dedup slice's committed-id window is
+ // what keeps that reordered arrival a new write rather than an absorbed
+ // duplicate, and this loop is the check that both payloads commit.
for payload in [
Bytes::from_static(b"parked-redispatch-0"),
Bytes::from_static(b"parked-redispatch-1"),
@@ -3228,6 +3117,149 @@ mod tests {
);
}
+ /// Failover retry absorbed by the partition dedup slice: a `SendMessages`
+ /// replay of an already-committed `(client, request)` on a NEW primary is
+ /// answered without re-executing. The slice is folded in on every replica
+ /// at commit, so the promoted primary knows the watermark its predecessor
+ /// established -- that inheritance is what this test proves.
+ #[test]
+ #[allow(clippy::too_many_lines)]
+ fn failover_retry_absorbed_by_partition_dedup() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 5;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+
+ let mut sim = Simulator::new(
+ replica_count as usize,
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+ let ns = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns);
+ sim.register_client_with_primary(&client);
+
+ // Same `(client, session, request)` for replay; mirrors SDK's
+ // connection-loss retry.
+ let original_req = client.send_messages(ns, &[Bytes::from_static(b"failover-test")]);
+ let replay_req = original_req.deep_copy();
+ let original_request_id = original_req.header().request;
+
+ sim.submit_request(client_id, 0, original_req.into_generic());
+
+ let mut original_reply: Option<Message<ReplyHeader>> = None;
+ for _ in 0..200 {
+ let replies = sim.step();
+ if !replies.is_empty() {
+ original_reply = Some(replies[0].deep_copy());
+ break;
+ }
+ }
+ let original_reply = original_reply.expect("commit reply must arrive before primary crash");
+ let original_commit_op = original_reply.header().commit;
+ // Offset after exactly one committed batch: the duplicate must not
+ // move it.
+ let offset_after_original = sim.replicas[1].shards[0]
+ .plane
+ .partitions()
+ .get_by_ns(&ns)
+ .expect("partition must exist on a live replica")
+ .stats
+ .current_offset();
+ assert_eq!(
+ original_reply.header().request,
+ original_request_id,
+ "sanity: original reply must echo the request id"
+ );
+
+ // Crash primary. Real-world: TCP buffer might have lost reply
+ // before ack; same retry path.
+ sim.replica_crash(0);
+
+ // Steps for view change across 4 survivors.
+ for _ in 0..800 {
+ sim.step();
+ }
+
+ // Find new primary via any live replica.
+ let live = &sim.replicas[1].shards[0];
+ let live_consensus = live
+ .plane
+ .partitions()
+ .get_by_ns(&ns)
+ .expect("partition must exist on a live replica")
+ .consensus();
+ assert!(
+ live_consensus.view() > 0,
+ "view must have advanced past the crashed primary"
+ );
+ let new_primary_idx = live_consensus.primary_index(live_consensus.view());
+ assert_ne!(
+ new_primary_idx, 0,
+ "new primary must not be the crashed replica"
+ );
+
+ // Replay the SAME request to the new primary: the dedup slice it
+ // inherited at commit must absorb it.
+ sim.submit_request(client_id, new_primary_idx, replay_req.into_generic());
+
+ let mut retry_reply: Option<Message<ReplyHeader>> = None;
+ for _ in 0..200 {
+ let replies = sim.step();
+ if !replies.is_empty() {
+ retry_reply = Some(replies[0].deep_copy());
+ break;
+ }
+ }
+ let retry_reply = retry_reply
+ .expect("reply must arrive after retry; the new primary absorbs it as a duplicate");
+
+ assert_eq!(
+ retry_reply.header().request,
+ original_request_id,
+ "retry's reply must correlate to the request id"
+ );
+ assert_eq!(
+ retry_reply.header().client,
+ client_id,
+ "retry must echo original client_id"
+ );
+ assert_eq!(
+ retry_reply.header().status,
+ 0,
+ "an absorbed duplicate is a success, not an error"
+ );
+ // The absorbed answer is synthesized at admission, so it never earns a
+ // new op. Re-execution would have committed past the original.
+ assert!(
+ retry_reply.header().op <= original_commit_op,
+ "retry must NOT re-execute (original commit={original_commit_op}, reply op={})",
+ retry_reply.header().op
+ );
+ // The payload committed exactly once.
+ let committed = sim.replicas[usize::from(new_primary_idx)].shards[0]
+ .plane
+ .partitions()
+ .get_by_ns(&ns)
+ .expect("partition must exist on the new primary")
+ .stats
+ .current_offset();
+ assert_eq!(
+ committed, offset_after_original,
+ "duplicate must not append a second copy"
+ );
+ }
+
/// With a positive crash probability
/// the driver crashes followers (never the primary) but never below the
/// survivor floor, while the per-tick invariants stay green and the
diff --git a/scripts/ci/storage-compat.sh b/scripts/ci/storage-compat.sh
index b4c3a72a2..952e69fb5 100755
--- a/scripts/ci/storage-compat.sh
+++ b/scripts/ci/storage-compat.sh
@@ -86,7 +86,8 @@ REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
# its own CWD, and the baseline below builds from a worktree elsewhere on
# disk, so a relative value would scatter the two builds and the lookup of
# either binary across three directories. Exported so every cargo call here,
-# nextest included, lands in the same place.
+# nextest included, lands in the same place -- except the baseline build, which
+# must have a target directory of its own (see the worktree build below).
TARGET_DIR="${CARGO_TARGET_DIR:-${REPO_ROOT}/target}"
mkdir -p "${TARGET_DIR}"
TARGET_DIR="$(cd "${TARGET_DIR}" && pwd)"
@@ -201,39 +202,51 @@ else
WORKTREE_DIR="$(mktemp -d "${TMPDIR:-/tmp}/iggy-storage-compat.XXXXXX")"
git worktree add --detach "${WORKTREE_DIR}" "${BASELINE_SHA}"
- # Symmetric to the guard before the HEAD build below: a HEAD binary left by an
- # earlier run makes cargo skip the uplift, and the cp further down would then
- # capture that HEAD binary as the baseline, comparing HEAD against itself.
- rm -f "${HEAD_SERVER}"
-
echo "Building baseline iggy-server from ${BASELINE_SHA}..."
# Built from the worktree as CWD so the baseline's own rust-toolchain.toml
- # applies, into the shared CARGO_TARGET_DIR so the registry graph compiles
- # once.
+ # applies, and into a target directory of the worktree's own.
+ #
+ # The two trees MUST NOT share one. Cargo keys a workspace member's unit hash
+ # on its manifest path RELATIVE to the workspace root, and records that
+ # member's sources in the dep-info relative too, so `core/configs` in the
+ # worktree and `core/configs` here hash identically and both resolve against
+ # whichever root cargo is invoked from. Sharing a target directory therefore
+ # makes the second build read the first build's rlibs as fresh: HEAD's
+ # `server` would compile against MASTER's `configs`, `consensus`,
+ # `partitions` and `shard`. Any PR touching a crate below `server` fails to
+ # build here with errors that do not reproduce anywhere else. The duplicated
+ # dependency compile is the price of the two halves being what they claim.
+ #
+ # Inside the worktree so the cleanup trap reclaims it with the worktree; only
+ # the copied binary below outlives the run.
#
# Debug profile on both sides, and no --all-features: release would compile
# debug_assert! out of the baseline while HEAD still panics on it, and
# --all-features turns on the server's `disable-mimalloc`, so the two halves
# would differ in ways the storage format never changed.
+ BASELINE_TARGET_DIR="${WORKTREE_DIR}/target"
(
cd "${WORKTREE_DIR}"
- cargo build --locked -p server --bin iggy-server
+ CARGO_TARGET_DIR="${BASELINE_TARGET_DIR}" cargo build --locked -p server --bin iggy-server
)
- if [ ! -x "${HEAD_SERVER}" ]; then
- echo "Baseline build did not produce ${HEAD_SERVER}"
+ BASELINE_BUILT="${BASELINE_TARGET_DIR}/debug/iggy-server"
+ if [ ! -x "${BASELINE_BUILT}" ]; then
+ echo "Baseline build did not produce ${BASELINE_BUILT}"
exit 1
fi
- # Copy before HEAD builds: both trees uplift to the same target/debug path.
mkdir -p "$(dirname "${BASELINE_SERVER}")"
- cp "${HEAD_SERVER}" "${BASELINE_SERVER}"
+ cp "${BASELINE_BUILT}" "${BASELINE_SERVER}"
+ # Now, not at cleanup: a second full debug dependency graph is several GB, and
+ # the HEAD build plus the integration test still have to fit on the runner.
+ rm -rf "${BASELINE_TARGET_DIR}"
fi
-# Delete the uplifted binary before building HEAD. Cargo skips re-uplifting when
-# the destination already looks current, and the baseline build just refreshed
-# it, so on a second run the BASELINE binary could survive at
-# target/debug/iggy-server and the test would compare master against master.
+# The baseline never writes here any more, but a binary left by an earlier run
+# of this script (or by any other build in this tree) would satisfy the
+# existence check below without cargo having produced it now. Delete it so that
+# check means what it says.
rm -f "${HEAD_SERVER}"
echo "Building HEAD iggy-server from ${HEAD_SHA}..."