diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs index 1569cfadad..743f25d91a 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 d6e3273611..6d32c22320 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 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 0737e23bf3..e251219bc6 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>, /// `client_id` -> slot index. Rebuilt on decode. index: HashMap, + /// 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, } @@ -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) { + 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 = 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 { + let mut entries: Vec = 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 { - 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 { + 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 { + 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 3f732bb48e..5ad273c35f 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) -> 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, ) -> (Self, Receiver>) { 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, + reply_sender: Option>>, + ) -> 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> VsrConsensus { 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 { diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs index d5647b52f6..0abad9657b 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 b794264a0a..57359759f3 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 0000000000..892211b741 --- /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, + 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, + 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 { + 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 1af433af23..3cbdf3e177 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 where B: MessageBus, { consensus: VsrConsensus, + /// 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>, /// Highest durably persisted offset. pub offset: Arc, @@ -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, @@ -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, + waiter: Option>>, ) { 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) { + pub async fn on_request( + &mut self, + message: Message, + reply: Option>>, + ) { + // 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, header: &RoutedRequestHeader, status: u32, send_fail_label: &'static str, + waiter: Option>>, ) { 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, + header: &RoutedRequestHeader, + reply: Message, + waiter: Option>>, + 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 87887c9ee8..2a2ee6e1e5 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 Plane> for IggyPartitions +impl IggyPartitions 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: as Consensus>::Message, + reply: Option>>, ) { 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>>, + header: &RoutedRequestHeader, + status: u32, + ) { + if let Some(waiter) = waiter { + let _ = waiter.send(build_deny_reply_from_request_header(header, status)); + } + } +} + +impl Plane> for IggyPartitions +where + B: MessageBus, + SB: SuperblockStore, +{ + async fn on_request( + &self, + message: as Consensus>::Message, + ) { + self.on_request_with_reply(message, None).await; } async fn on_replicate(&self, message: as Consensus>::Message) { diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index 5bd29807db..7444b9d76b 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::() + 2 * size_of::() + size_of::(); + /// 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, } 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 { // Size exactly rather than guess; the reservation assert keeps the @@ -309,8 +332,9 @@ impl ConsumerOffsetsWire { let reserved = CONSUMER_OFFSETS_MAGIC.len() + size_of::() + 2 * size_of::() - + 2 * size_of::() + + 3 * size_of::() + (self.consumers.len() + self.groups.len()) * (size_of::() + size_of::()) + + self.dedup.len() * DEDUP_ENTRY_LEN + size_of::(); 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::(), 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, 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 = 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::() + size_of::()) > 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 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 = 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 79bdc46e88..5584b03788 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 0115b5c609..e614e688f5 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 4da515f8e5..b2e0789bfa 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( 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( + shard: &Rc>, + namespace: IggyNamespace, + request: Message, + transport_client_id: u128, + header: &RoutedRequestHeader, +) where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal, 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 aa1db608c3..b03b94bf58 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, - /// 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, + /// 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, /// 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, } -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 73b2c6cb7b..5f1984cd41 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 d73b2005ee..0babb39528 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 ad4471da98..ec39868e5c 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 8b56a415a4..fe4deaea5a 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>>, + 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, }, + /// 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, + reply: Sender>>, + }, /// 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, + /// 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, + /// 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, + ) -> Result { + 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::>>(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> { + 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, + /// 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>>>, } 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>>>, +} + /// 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 { 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, + ::Target: + Journal, 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::() - { - 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::() 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, + reply: &mut Option>>>, ) -> ParkOutcome 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, + reply: Sender>>, + ) where + B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, 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, + reply: Sender>>, + provenance: Option, + ) where + B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, 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>>>, + ) -> 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) where diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 16007ffec1..4fbd8c7fcf 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 { diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index a866a9027c..1f4fa01170 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 fda8e5bcfd..3d2fd3feaa 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> = 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> = 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> = 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> = 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 b4c3a72a23..952e69fb5e 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}..."