From 7f71c3f7bba52f4bfe5975b74323c66ceb8acdf3 Mon Sep 17 00:00:00 2001 From: RustFS <155562731+loverustfs@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:17:12 +0800 Subject: [PATCH] Add bounded shard clients and atomic membership fencing --- crates/transport/src/client.rs | 67 ++ crates/transport/src/directory.rs | 175 +++++ crates/transport/src/lib.rs | 8 +- crates/transport/src/plane.rs | 103 ++- crates/transport/src/registered.rs | 35 + crates/transport/src/routing.rs | 1026 +++++++++++++++++++++++++ crates/transport/tests/sharded_tcp.rs | 524 +++++++++++++ docs/adr/0011-bounded-shard-client.md | 156 ++++ 8 files changed, 2090 insertions(+), 4 deletions(-) create mode 100644 crates/transport/src/routing.rs create mode 100644 crates/transport/tests/sharded_tcp.rs create mode 100644 docs/adr/0011-bounded-shard-client.md diff --git a/crates/transport/src/client.rs b/crates/transport/src/client.rs index 1b3f2be..6a2c87d 100644 --- a/crates/transport/src/client.rs +++ b/crates/transport/src/client.rs @@ -17,6 +17,17 @@ use crate::clock::Clock; use crate::frame::{read_envelope, write_envelope}; use crate::pool::Pool; +/// Counts complete application envelopes, including the hello exchange. +/// A framed envelope adds 12 bytes. TCP/IP headers, retransmission, and partial +/// failed writes/reads are excluded; these counters are not NIC wire telemetry. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ClientIoStats { + pub rpc_body_bytes_sent: u64, + pub rpc_body_bytes_received: u64, + pub framed_bytes_sent: u64, + pub framed_bytes_received: u64, +} + pub struct TcpClient { stream: TcpStream, auth: AuthContext, @@ -25,6 +36,7 @@ pub struct TcpClient { epoch: u64, clock: Clock, usable: bool, + io_stats: ClientIoStats, } impl TcpClient { @@ -73,9 +85,27 @@ impl TcpClient { epoch, clock: Clock::new(envelope.server_now), usable: true, + io_stats: ClientIoStats { + rpc_body_bytes_sent: hello.len() as u64, + rpc_body_bytes_received: envelope.body().len() as u64, + framed_bytes_sent: hello.len() as u64 + 12, + framed_bytes_received: envelope.body().len() as u64 + 12, + }, }) } + pub fn io_stats(&self) -> ClientIoStats { + self.io_stats + } + + pub(crate) fn is_usable(&self) -> bool { + self.usable + } + + pub(crate) fn server_now(&self) -> u64 { + self.clock.server_now() + } + pub fn epoch(&self) -> u64 { self.epoch } @@ -158,7 +188,23 @@ impl TcpClient { // incomplete exchange, a late reply must never answer a later call. self.usable = false; write_envelope(&mut self.stream, 0, &bytes, self.max_frame)?; + self.io_stats.rpc_body_bytes_sent = self + .io_stats + .rpc_body_bytes_sent + .saturating_add(bytes.len() as u64); + self.io_stats.framed_bytes_sent = self + .io_stats + .framed_bytes_sent + .saturating_add(bytes.len() as u64 + 12); let envelope = read_envelope(&mut self.stream, self.pool.checkout()?, self.max_frame)?; + self.io_stats.rpc_body_bytes_received = self + .io_stats + .rpc_body_bytes_received + .saturating_add(envelope.body().len() as u64); + self.io_stats.framed_bytes_received = self + .io_stats + .framed_bytes_received + .saturating_add(envelope.body().len() as u64 + 12); match decode_reply(envelope.body())? { ReplyBody::Failed { op, @@ -497,4 +543,25 @@ mod tests { }) )); } + + #[test] + fn io_counters_separate_complete_rpc_body_and_framed_bytes() { + let (mut client, server) = fake_server(vec![encode_reply(&lookup_reply(1, &[1])).unwrap()]); + let before = client.io_stats(); + assert_eq!(before.framed_bytes_sent - before.rpc_body_bytes_sent, 12); + assert_eq!( + before.framed_bytes_received - before.rpc_body_bytes_received, + 12 + ); + client.lookup(vec![lookup_item(1)]).unwrap(); + let after = client.io_stats(); + assert_eq!(after.framed_bytes_sent - after.rpc_body_bytes_sent, 24); + assert_eq!( + after.framed_bytes_received - after.rpc_body_bytes_received, + 24 + ); + assert!(after.rpc_body_bytes_sent > before.rpc_body_bytes_sent); + assert!(after.rpc_body_bytes_received > before.rpc_body_bytes_received); + server.join().unwrap(); + } } diff --git a/crates/transport/src/directory.rs b/crates/transport/src/directory.rs index a3384d6..2c882b6 100644 --- a/crates/transport/src/directory.rs +++ b/crates/transport/src/directory.rs @@ -97,6 +97,17 @@ struct Flight { waiters: BTreeSet<[u8; 16]>, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DirectoryStats { + pub flights: usize, + pub request_entries: usize, + pub notices: usize, + pub tenants: usize, + pub members: usize, + pub fence_count: u64, + pub fenced_fills: u64, +} + pub struct ShardDirectory { shard_id: u64, epoch: u64, @@ -115,6 +126,8 @@ pub struct ShardDirectory { notices: BTreeMap<(Vec, [u8; 16]), FillNotice>, notice_order: VecDeque<(Vec, [u8; 16])>, max_notices: usize, + fence_count: u64, + fenced_fills: u64, } impl ShardDirectory { @@ -145,6 +158,8 @@ impl ShardDirectory { next_flight: 1, notices: BTreeMap::new(), notice_order: VecDeque::new(), + fence_count: 0, + fenced_fills: 0, max_notices: params .max_flights .saturating_mul(params.max_waiters.saturating_add(1)) @@ -152,6 +167,54 @@ impl ShardDirectory { } } + pub fn stats(&self) -> DirectoryStats { + DirectoryStats { + flights: self.flights.len(), + request_entries: self.by_request.len(), + notices: self.notices.len(), + tenants: self.tenant_used.len(), + members: self.members.len(), + fence_count: self.fence_count, + fenced_fills: self.fenced_fills, + } + } + + /// Apply one bounded membership snapshot atomically. Validation and epoch + /// capacity checks finish before membership, credits, or notices change. + /// There is no update queue: callers retry rejected snapshots explicitly. + pub fn apply_membership(&mut self, members: &[MemberView]) -> Result { + if members.is_empty() || members.len() > self.max_members { + return Err(CacheError::Malformed { + reason: "membership", + }); + } + let mut seen = BTreeSet::new(); + for member in members { + if member.incarnation == 0 || !seen.insert(member.node_id) { + return Err(CacheError::Malformed { + reason: "membership", + }); + } + if self + .members + .iter() + .any(|old| old.node_id == member.node_id && old.incarnation > member.incarnation) + { + return Err(CacheError::FencingRejected { reason: "member" }); + } + } + if members.len() == self.members.len() + && members.iter().all(|member| self.members.contains(member)) + { + return Ok(self.epoch); + } + self.ensure_epoch_can_advance()?; + let epoch = self.fence_inflight()?; + self.members = members.to_vec(); + self.members.sort_by_key(|member| member.node_id); + Ok(epoch) + } + pub fn shard_id(&self) -> u64 { self.shard_id } @@ -386,6 +449,8 @@ impl ShardDirectory { kind: QuotaKind::InFlightRequests, })?; self.epoch = next; + self.fence_count = self.fence_count.saturating_add(1); + self.fenced_fills = self.fenced_fills.saturating_add(self.flights.len() as u64); self.by_request.clear(); self.by_id.clear(); let flights = std::mem::take(&mut self.flights); @@ -873,4 +938,114 @@ mod tests { FillNotice::LeaderFinished ); } + + #[test] + fn membership_batch_rejects_every_invalid_snapshot_before_fencing() { + let mut dir = ShardDirectory::single_node(params()); + dir.note_member(2, 2).unwrap(); + dir.begin_or_join(&fill(b"a", b"tenant-a", 10, 1, dir.epoch())) + .unwrap(); + dir.begin_or_join(&fill(b"a", b"tenant-a", 10, 2, dir.epoch())) + .unwrap(); + let before = dir.stats(); + let members = dir.members().to_vec(); + for candidate in [ + vec![], + vec![MemberView { + node_id: 2, + incarnation: 1, + }], + vec![members[0], members[0]], + vec![MemberView { + node_id: 3, + incarnation: 0, + }], + (1..=5) + .map(|node_id| MemberView { + node_id, + incarnation: 2, + }) + .collect(), + ] { + assert!(dir.apply_membership(&candidate).is_err()); + assert_eq!(dir.stats(), before); + assert_eq!(dir.members(), members); + assert_eq!(dir.credits_used(), 10); + assert_eq!( + dir.fill_notice(b"tenant-a", 2u128.to_le_bytes()).unwrap(), + FillNotice::Pending + ); + } + dir.set_epoch_for_test(u64::MAX); + assert!(dir.apply_membership(&[members[0]]).is_err()); + assert_eq!(dir.stats(), before); + assert_eq!(dir.members(), members); + } + + #[test] + fn batch_membership_is_order_independent_and_fences_once() { + let mut dir = ShardDirectory::single_node(params()); + dir.note_member(2, 1).unwrap(); + dir.begin_or_join(&fill(b"a", b"tenant-a", 10, 1, dir.epoch())) + .unwrap(); + let mut same = dir.members().to_vec(); + same.reverse(); + assert_eq!(dir.apply_membership(&same).unwrap(), 2); + assert_eq!(dir.inflight(), 1); + let snapshot = [ + MemberView { + node_id: 2, + incarnation: 2, + }, + MemberView { + node_id: 3, + incarnation: 1, + }, + ]; + assert_eq!(dir.apply_membership(&snapshot).unwrap(), 3); + assert_eq!(dir.inflight(), 0); + assert_eq!(dir.credits_used(), 0); + assert_eq!(dir.stats().fenced_fills, 1); + assert_eq!(dir.stats().fence_count, 2); + assert_eq!( + dir.fill_notice(b"tenant-a", 1u128.to_le_bytes()).unwrap(), + FillNotice::Fenced + ); + } + + #[test] + fn cancel_restart_and_membership_storm_keep_every_table_bounded() { + let mut dir = ShardDirectory::single_node(params()); + for n in 1..=2000u128 { + let epoch = dir.epoch(); + let Admit::Leader { flight_id, .. } = dir + .begin_or_join(&fill(b"a", b"tenant-a", 10, n * 2, epoch)) + .unwrap() + else { + panic!("leader") + }; + dir.begin_or_join(&fill(b"a", b"tenant-a", 10, n * 2 + 1, epoch)) + .unwrap(); + if n % 2 == 0 { + dir.fence_epoch().unwrap(); + assert!(dir + .complete_flight(flight_id, epoch, (n * 2).to_le_bytes()) + .is_err()); + } else { + dir.fail_flight(flight_id, epoch, (n * 2).to_le_bytes()) + .unwrap(); + assert_eq!( + dir.complete_flight(flight_id, epoch, (n * 2).to_le_bytes()), + Err(CacheError::UnknownGeneration) + ); + } + let stats = dir.stats(); + assert_eq!(stats.flights, 0); + assert_eq!(stats.request_entries, 0); + assert_eq!(stats.tenants, 0); + assert!(stats.notices <= dir.max_notices); + assert!(dir.notice_order.len() <= dir.max_notices); + assert_eq!(dir.credits_used(), 0); + } + } } diff --git a/crates/transport/src/lib.rs b/crates/transport/src/lib.rs index 8178411..30d5cb7 100644 --- a/crates/transport/src/lib.rs +++ b/crates/transport/src/lib.rs @@ -20,13 +20,16 @@ mod plane; mod pool; mod rate; mod registered; +mod routing; mod server; mod shard_demo; mod stage; -pub use client::TcpClient; +pub use client::{ClientIoStats, TcpClient}; pub use demo::{run_share_demo, ShareReport}; -pub use directory::{Admit, FillNotice, FillRequest, MemberView, ShardDirectory, ShardParams}; +pub use directory::{ + Admit, DirectoryStats, FillNotice, FillRequest, MemberView, ShardDirectory, ShardParams, +}; pub use fault_demo::{run_fault_demo, FaultReport}; pub use iface::{ rdma_feature_selected, rdma_probe, submit_preferring_rdma, FallbackReason, HardwarePath, @@ -35,6 +38,7 @@ pub use iface::{ pub use plane::{ControlPlane, RebalancePlan, UpgradeStamp}; pub use rate::RateGate; pub use registered::{CheckedBatch, PoolStats, RegisteredPool, SimCompletion, SlotId}; +pub use routing::{ClientLimits, ClientStats, ShardEndpoint, ShardSession, ShardedClient}; pub use server::{serve, ServerConfig, ServerHandle}; pub use shard_demo::{run_shard_demo, ShardReport}; pub use stage::stage_through_tcp; diff --git a/crates/transport/src/plane.rs b/crates/transport/src/plane.rs index 15e561f..925595e 100644 --- a/crates/transport/src/plane.rs +++ b/crates/transport/src/plane.rs @@ -10,7 +10,7 @@ use std::collections::BTreeMap; use gpucache_core_types::{CacheError, QuotaKind}; -use crate::directory::{Admit, FillNotice, FillRequest, ShardDirectory, ShardParams}; +use crate::directory::{Admit, FillNotice, FillRequest, MemberView, ShardDirectory, ShardParams}; #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum Phase { @@ -107,6 +107,19 @@ impl ControlPlane { Ok(()) } + /// Apply one complete membership view through the directory's atomic fence. + /// Failed validation or epoch exhaustion keeps fault-domain assignments and + /// pending fills intact. Draining upgrades do not accept membership changes. + pub fn apply_membership(&mut self, members: &[MemberView]) -> Result { + if self.phase != Phase::Serving { + return Err(CacheError::Cancelled); + } + let epoch = self.dir.apply_membership(members)?; + self.domains + .retain(|node, _| members.iter().any(|member| member.node_id == *node)); + Ok(epoch) + } + /// Remove every member in `domain`. The last remaining member stays. /// In-flight fills are fenced and become misses. No payload is copied. /// If the epoch cannot advance for the entire batch, nothing is changed. @@ -175,8 +188,14 @@ impl ControlPlane { if self.phase != Phase::Draining { return Err(CacheError::Malformed { reason: "upgrade" }); } + let generation = + self.software_generation + .checked_add(1) + .ok_or(CacheError::FencingRejected { + reason: "software generation exhausted", + })?; let epoch = self.dir.fence_epoch()?; - self.software_generation = self.software_generation.saturating_add(1); + self.software_generation = generation; self.upgraded = true; self.phase = Phase::Serving; Ok(UpgradeStamp { @@ -319,6 +338,34 @@ mod tests { assert_eq!(plane.credits_used(), 0); } + #[test] + fn exhausted_software_generation_does_not_fence_or_resume_a_draining_shard() { + let mut plane = ControlPlane::single_node(params()); + let (flight_id, leader, waiter) = begin_leader_and_waiter(&mut plane); + plane.software_generation = u64::MAX; + plane.begin_upgrade().unwrap(); + assert_eq!( + plane.finish_upgrade().err(), + Some(CacheError::FencingRejected { + reason: "software generation exhausted" + }) + ); + assert!(!plane.serving()); + assert_eq!(plane.epoch(), leader.epoch); + assert_eq!(plane.inflight(), 1); + assert_eq!(plane.credits_used(), leader.bytes); + for request in [&leader, &waiter] { + assert_eq!( + plane.fill_notice(&request.tenant, request.request).unwrap(), + FillNotice::Pending + ); + } + plane.rollback().unwrap(); + plane + .complete_flight(flight_id, leader.epoch, leader.request) + .unwrap(); + } + #[test] fn isolate_domain_refuses_to_remove_the_last_member() { let mut plane = ControlPlane::single_node(params()); @@ -344,6 +391,58 @@ mod tests { assert!(plane.dir.members().iter().all(|member| member.node_id != 9)); } + #[test] + fn public_membership_admission_enables_fault_domain_isolation_and_refuses_during_upgrade() { + let mut plane = ControlPlane::single_node(params()); + let members = [ + MemberView { + node_id: 1, + incarnation: 1, + }, + MemberView { + node_id: 2, + incarnation: 1, + }, + ]; + plane.apply_membership(&members).unwrap(); + plane.assign_domain(1, 7).unwrap(); + plane.assign_domain(2, 9).unwrap(); + let (flight_id, leader, _) = begin_leader_and_waiter(&mut plane); + plane.begin_upgrade().unwrap(); + assert_eq!( + plane.apply_membership(&members[..1]), + Err(CacheError::Cancelled) + ); + assert_eq!(plane.inflight(), 1); + plane.rollback().unwrap(); + let epoch = plane.isolate_domain(9).unwrap(); + assert!(epoch > leader.epoch); + assert_eq!(plane.domains, BTreeMap::from([(1, 7)])); + assert_eq!(plane.credits_used(), 0); + assert!(matches!( + plane.complete_flight(flight_id, leader.epoch, leader.request), + Err(CacheError::EpochMismatch { .. }) + )); + } + + #[test] + fn failed_membership_batch_retains_fault_domains_and_pending_fills() { + let mut plane = three_member_plane(u64::MAX); + let (_, leader, waiter) = begin_leader_and_waiter(&mut plane); + let domains = plane.domains.clone(); + let members = plane.dir.members().to_vec(); + assert!(plane.apply_membership(&members[..1]).is_err()); + assert_eq!(plane.dir.members(), members); + assert_eq!(plane.domains, domains); + assert_eq!(plane.credits_used(), leader.bytes); + for request in [&leader, &waiter] { + assert_eq!( + plane.fill_notice(&request.tenant, request.request).unwrap(), + FillNotice::Pending + ); + } + } + #[test] fn isolate_domain_epoch_shortage_keeps_the_entire_batch_and_fills_unchanged() { let mut plane = three_member_plane(u64::MAX - 1); diff --git a/crates/transport/src/registered.rs b/crates/transport/src/registered.rs index 0267c1f..4ff147d 100644 --- a/crates/transport/src/registered.rs +++ b/crates/transport/src/registered.rs @@ -53,6 +53,9 @@ pub struct PoolStats { pub backing_reused: u64, pub epoch: u64, pub held_bytes: u64, + pub fence_count: u64, + pub drained_slots: u64, + pub reclaimed_slots: u64, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -80,6 +83,9 @@ pub struct RegisteredPool { reused_registrations: u64, backing_reused: u64, epoch: u64, + fence_count: u64, + drained_slots: u64, + reclaimed_slots: u64, } impl RegisteredPool { @@ -113,6 +119,9 @@ impl RegisteredPool { reused_registrations: 0, backing_reused: 0, epoch: 1, + fence_count: 0, + drained_slots: 0, + reclaimed_slots: 0, }) } @@ -139,6 +148,9 @@ impl RegisteredPool { backing_reused: self.backing_reused, epoch: self.epoch, held_bytes: (self.regions.len() * self.slot_bytes) as u64, + fence_count: self.fence_count, + drained_slots: self.drained_slots, + reclaimed_slots: self.reclaimed_slots, } } @@ -305,6 +317,7 @@ impl RegisteredPool { } region.id = None; region.state = RegionState::Unbound; + self.drained_slots = self.drained_slots.saturating_add(1); Ok(()) } @@ -315,6 +328,7 @@ impl RegisteredPool { kind: QuotaKind::Registrations, })?; self.epoch = next; + self.fence_count = self.fence_count.saturating_add(1); for region in &mut self.regions { if matches!( region.state, @@ -336,6 +350,7 @@ impl RegisteredPool { freed += 1; } } + self.reclaimed_slots = self.reclaimed_slots.saturating_add(freed as u64); Ok(freed) } @@ -515,4 +530,24 @@ mod tests { .unwrap(); assert_eq!(pool.stats().idle_registered, 1); } + + #[test] + fn cancellation_storm_requires_drain_and_reports_fence_reclaim_counts() { + let mut pool = RegisteredPool::new(2, 8).unwrap(); + for _ in 0..1000 { + let batch = pool.checkout_batch(2).unwrap(); + pool.complete_batch(&batch, SimCompletion::Unknown).unwrap(); + assert!(pool.checkout_batch(1).is_err()); + pool.on_lease_expiry(batch.slots()[0]).unwrap(); + assert_eq!(pool.stats().quarantined, 2); + pool.attest_drain(batch.slots()[0]).unwrap(); + pool.fence_epoch().unwrap(); + assert_eq!(pool.reclaim_fenced().unwrap(), 1); + assert_eq!(pool.stats().held_bytes, 16); + } + assert_eq!(pool.stats().fence_count, 1000); + assert_eq!(pool.stats().drained_slots, 1000); + assert_eq!(pool.stats().reclaimed_slots, 1000); + assert_eq!(pool.stats().quarantined, 0); + } } diff --git a/crates/transport/src/routing.rs b/crates/transport/src/routing.rs new file mode 100644 index 0000000..6c76954 --- /dev/null +++ b/crates/transport/src/routing.rs @@ -0,0 +1,1026 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Bounded CPU/TCP shard routing. A session keeps the original six RPC methods. +//! Generations remain shard-local: retain the session's shard alongside a generation. +//! No directory, payload, or retry queue is shared across tenants. + +use std::collections::{BTreeMap, BTreeSet}; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Duration, Instant}; + +use gpucache_core_types::{ + content_checksum, encode_physical_key, CacheError, CacheIdentity, IdentityLimits, QuotaKind, + ResourceLimits, +}; +use gpucache_protocol::{ + encode_request, AuthContext, BatchReply, CommitItem, CommitOutcome, GetHit, GetItem, + InvalidateItem, InvalidateOutcome, ItemHeader, LookupItem, LookupOutcome, ReleaseCancelItem, + ReleaseOutcome, RequestBody, ReserveOutcome, ReservePutItem, +}; + +use crate::TcpClient; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ShardEndpoint { + pub shard_id: u64, + pub address: SocketAddr, + /// Expected store epoch, obtained from the deployment's fenced membership. + pub epoch: u64, +} + +#[derive(Clone, Debug)] +pub struct ClientLimits { + pub max_shards: usize, + pub max_connections: usize, + pub max_connections_per_shard: usize, + pub max_retry_entries: usize, + /// Total attempts for one exact request, including its first attempt. + pub max_attempts: u8, + pub max_frame_bytes: usize, + pub io_timeout: Duration, + pub max_request_ttl_ms: u64, + /// Explicit deployment contract, not negotiated over GCK1. None disables + /// mutation retries. Must not exceed any server's configured replay retention. + pub mutation_retry_retention_ms: Option, +} + +impl Default for ClientLimits { + fn default() -> Self { + Self { + max_shards: 64, + max_connections: 64, + max_connections_per_shard: 4, + max_retry_entries: 4096, + max_attempts: 2, + max_frame_bytes: 1024 * 1024, + io_timeout: Duration::from_secs(5), + max_request_ttl_ms: 30_000, + mutation_retry_retention_ms: None, + } + } +} + +impl ClientLimits { + fn validate(&self) -> Result<(), CacheError> { + if self.max_shards == 0 + || self.max_shards > 4096 + || self.max_connections == 0 + || self.max_connections > 4096 + || self.max_connections_per_shard == 0 + || self.max_connections_per_shard > self.max_connections + || self.max_retry_entries == 0 + || self.max_retry_entries > 1_048_576 + || self.max_attempts == 0 + || self.max_attempts > 4 + || self.max_frame_bytes < 64 + || self.max_frame_bytes > 32 * 1024 * 1024 + || self.io_timeout.is_zero() + || self.max_request_ttl_ms == 0 + || self.max_request_ttl_ms > 60_000 + || self + .mutation_retry_retention_ms + .is_some_and(|ttl| ttl <= self.max_request_ttl_ms) + { + return Err(CacheError::Malformed { + reason: "shard client config", + }); + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ClientStats { + pub revision: u64, + pub shards: usize, + pub connections: usize, + pub idle_connections: usize, + pub retry_entries: usize, + pub connection_reuses: u64, + pub retries: u64, + pub fenced_replies: u64, +} + +#[derive(Clone)] +pub struct ShardedClient { + inner: Arc, +} + +struct Inner { + auth: AuthContext, + limits: ClientLimits, + state: Mutex, +} + +struct State { + revision: u64, + endpoints: Vec, + idle: BTreeMap>, + connections: usize, + by_shard: BTreeMap, + retries: BTreeMap<(u64, u64, [u8; 16]), RetryEntry>, + connection_reuses: u64, + retry_attempts: u64, + fenced_replies: u64, +} + +struct RetryEntry { + fingerprint: [u8; 32], + attempts: u8, + active: bool, + expires: Instant, +} + +/// Exclusive connection lease. Drop returns a synchronized connection to the +/// pool; poisoned or retired connections are closed. No blocking checkout queue. +pub struct ShardSession { + owner: Arc, + endpoint: ShardEndpoint, + revision: u64, + client: Option, +} + +impl ShardedClient { + pub fn new( + auth: AuthContext, + endpoints: Vec, + limits: ClientLimits, + ) -> Result { + auth.validate()?; + limits.validate()?; + let endpoints = validate_endpoints(endpoints, limits.max_shards)?; + Ok(Self { + inner: Arc::new(Inner { + auth, + limits, + state: Mutex::new(State { + revision: 1, + endpoints, + idle: BTreeMap::new(), + connections: 0, + by_shard: BTreeMap::new(), + retries: BTreeMap::new(), + connection_reuses: 0, + retry_attempts: 0, + fenced_replies: 0, + }), + }), + }) + } + + pub fn stats(&self) -> ClientStats { + let mut state = self.inner.lock(); + state.expire(); + ClientStats { + revision: state.revision, + shards: state.endpoints.len(), + connections: state.connections, + idle_connections: state.idle.values().map(Vec::len).sum(), + retry_entries: state.retries.len(), + connection_reuses: state.connection_reuses, + retries: state.retry_attempts, + fenced_replies: state.fenced_replies, + } + } + + pub fn shard_for(&self, identity: &CacheIdentity) -> Result { + let hash = self.inner.identity_hash(identity)?; + Ok(select_endpoint(&self.inner.lock().endpoints, &hash).shard_id) + } + + /// Route an identity and obtain an exclusive bounded connection lease. + pub fn route(&self, identity: &CacheIdentity) -> Result { + let hash = self.inner.identity_hash(identity)?; + let state = self.inner.lock(); + let endpoint = select_endpoint(&state.endpoints, &hash); + self.checkout(state, endpoint) + } + + /// Reopen the original shard for a stored generation. Never hash generation + /// numbers: two independent stores can both allocate generation 1. + pub fn shard(&self, shard_id: u64) -> Result { + let state = self.inner.lock(); + let endpoint = state + .endpoints + .iter() + .find(|endpoint| endpoint.shard_id == shard_id) + .copied() + .ok_or(CacheError::UnknownGeneration)?; + self.checkout(state, endpoint) + } + + fn checkout( + &self, + mut state: MutexGuard<'_, State>, + endpoint: ShardEndpoint, + ) -> Result { + let revision = state.revision; + if let Some(client) = state.idle.get_mut(&endpoint.shard_id).and_then(Vec::pop) { + state.connection_reuses = state.connection_reuses.saturating_add(1); + return Ok(ShardSession { + owner: Arc::clone(&self.inner), + endpoint, + revision, + client: Some(client), + }); + } + if state.by_shard.get(&endpoint.shard_id).copied().unwrap_or(0) + >= self.inner.limits.max_connections_per_shard + { + return Err(quota(QuotaKind::InFlightRequests)); + } + // An idle socket on another shard must not monopolize the global + // ceiling. Retire one idle socket before refusing unrelated work. + if state.connections >= self.inner.limits.max_connections { + let victim = state.idle.iter().find_map(|(shard, clients)| { + (*shard != endpoint.shard_id && !clients.is_empty()).then_some(*shard) + }); + if let Some(shard) = victim { + state.idle.get_mut(&shard).and_then(Vec::pop); + state.connections -= 1; + if let Some(count) = state.by_shard.get_mut(&shard) { + *count -= 1; + if *count == 0 { + state.by_shard.remove(&shard); + } + } + } + } + if state.connections >= self.inner.limits.max_connections { + return Err(quota(QuotaKind::InFlightRequests)); + } + state.connections += 1; + *state.by_shard.entry(endpoint.shard_id).or_default() += 1; + drop(state); + let mut session = ShardSession { + owner: Arc::clone(&self.inner), + endpoint, + revision, + client: None, + }; + session.reconnect()?; + Ok(session) + } + + /// Publish a fully validated snapshot after servers have been fenced. + /// Every shard must exceed the previous snapshot's highest epoch. Publication + /// invalidates all old leases; late replies are suppressed. This does not + /// fence a remote server and is not a distributed membership consensus. + pub fn replace_routes( + &self, + expected_revision: u64, + endpoints: Vec, + ) -> Result { + let endpoints = validate_endpoints(endpoints, self.inner.limits.max_shards)?; + let mut state = self.inner.lock(); + if state.revision != expected_revision { + return Err(CacheError::FencingRejected { + reason: "route revision", + }); + } + // A scalar high-water mark also fences removal followed by re-addition + // without retaining an unbounded table of retired shard IDs. + let epoch_floor = state + .endpoints + .iter() + .map(|endpoint| endpoint.epoch) + .max() + .unwrap_or(0); + if endpoints + .iter() + .any(|endpoint| endpoint.epoch <= epoch_floor) + { + return Err(CacheError::FencingRejected { + reason: "route epoch", + }); + } + let revision = state + .revision + .checked_add(1) + .ok_or_else(|| quota(QuotaKind::InFlightRequests))?; + let idle = std::mem::take(&mut state.idle); + for (shard, clients) in idle { + state.connections -= clients.len(); + if let Some(count) = state.by_shard.get_mut(&shard) { + *count -= clients.len(); + if *count == 0 { + state.by_shard.remove(&shard); + } + } + } + state.revision = revision; + state.endpoints = endpoints; + Ok(revision) + } +} + +impl Inner { + fn lock(&self) -> MutexGuard<'_, State> { + self.state + .lock() + .unwrap_or_else(|poison| poison.into_inner()) + } + + fn identity_hash(&self, identity: &CacheIdentity) -> Result<[u8; 32], CacheError> { + if identity.tenant != self.auth.tenant || identity.trust_domain != self.auth.trust_domain { + return Err(CacheError::Unauthorized); + } + let bytes = encode_physical_key( + identity, + &IdentityLimits::from_resources(&ResourceLimits::dev_defaults()), + )?; + Ok(content_checksum(&bytes).0) + } +} + +impl State { + fn expire(&mut self) { + let now = Instant::now(); + self.retries + .retain(|_, entry| entry.active || entry.expires > now); + } +} + +impl ShardSession { + pub fn shard_id(&self) -> u64 { + self.endpoint.shard_id + } + pub fn epoch(&self) -> u64 { + self.endpoint.epoch + } + pub fn deadline(&self, ttl_ms: u64) -> u64 { + self.client + .as_ref() + .map_or(0, |client| client.deadline(ttl_ms)) + } + + pub fn lookup( + &mut self, + items: Vec, + ) -> Result, CacheError> { + self.call(RequestBody::Lookup(items), |client, body| { + let RequestBody::Lookup(items) = body else { + unreachable!() + }; + client.lookup(items) + }) + } + pub fn reserve_put( + &mut self, + items: Vec, + ) -> Result, CacheError> { + self.call(RequestBody::ReservePut(items), |client, body| { + let RequestBody::ReservePut(items) = body else { + unreachable!() + }; + client.reserve_put(items) + }) + } + pub fn commit( + &mut self, + items: Vec, + ) -> Result, CacheError> { + self.call(RequestBody::Commit(items), |client, body| { + let RequestBody::Commit(items) = body else { + unreachable!() + }; + client.commit(items) + }) + } + pub fn get(&mut self, items: Vec) -> Result, CacheError> { + self.call(RequestBody::Get(items), |client, body| { + let RequestBody::Get(items) = body else { + unreachable!() + }; + client.get(items) + }) + } + pub fn release_cancel( + &mut self, + items: Vec, + ) -> Result, CacheError> { + self.call(RequestBody::ReleaseCancel(items), |client, body| { + let RequestBody::ReleaseCancel(items) = body else { + unreachable!() + }; + client.release_cancel(items) + }) + } + pub fn invalidate( + &mut self, + items: Vec, + ) -> Result, CacheError> { + self.call(RequestBody::Invalidate(items), |client, body| { + let RequestBody::Invalidate(items) = body else { + unreachable!() + }; + client.invalidate(items) + }) + } + + fn current(&self) -> Result<(), CacheError> { + if self.owner.lock().revision != self.revision { + return Err(CacheError::FencingRejected { + reason: "route revision", + }); + } + Ok(()) + } + + fn reconnect(&mut self) -> Result<(), CacheError> { + self.current()?; + self.client = None; + let client = TcpClient::connect( + self.endpoint.address, + self.owner.auth.clone(), + self.owner.limits.io_timeout, + self.owner.limits.max_frame_bytes, + )?; + if client.epoch() != self.endpoint.epoch { + return Err(CacheError::EpochMismatch { + store: client.epoch(), + request: self.endpoint.epoch, + }); + } + self.current()?; + self.client = Some(client); + Ok(()) + } + + fn call( + &mut self, + request: RequestBody, + invoke: impl Fn(&mut TcpClient, RequestBody) -> Result, CacheError>, + ) -> Result, CacheError> { + self.current()?; + if request.is_empty() || request.len() > 1024 { + return Err(quota(QuotaKind::BatchItems)); + } + let request_headers = headers(&request); + let mut unique = BTreeSet::new(); + for header in &request_headers { + if !unique.insert(header.request_id.0) { + return Err(CacheError::IdempotencyConflict); + } + if header.epoch != self.epoch() { + return Err(CacheError::EpochMismatch { + store: self.epoch(), + request: header.epoch, + }); + } + } + for identity in identities(&request) { + let hash = self.owner.identity_hash(identity)?; + let state = self.owner.lock(); + if select_endpoint(&state.endpoints, &hash).shard_id != self.shard_id() { + return Err(CacheError::FencingRejected { + reason: "wrong shard", + }); + } + } + if let RequestBody::ReservePut(items) = &request { + let payload_bytes = items.iter().try_fold(0usize, |n, item| { + n.checked_add(item.payload.as_ref().map_or(0, Vec::len)) + }); + if payload_bytes.is_none_or(|n| n > self.owner.limits.max_frame_bytes) { + return Err(quota(QuotaKind::InFlightBytes)); + } + } + let encoded = encode_request(&self.owner.auth, &request)?; + if encoded.len() > self.owner.limits.max_frame_bytes { + return Err(CacheError::ObjectTooLarge { + limit: self.owner.limits.max_frame_bytes as u64, + actual: encoded.len() as u64, + }); + } + let fingerprint = content_checksum(&encoded).0; + drop(encoded); + if self + .client + .as_ref() + .is_none_or(|client| !client.is_usable()) + { + self.reconnect()?; + } + let server_now = self + .client + .as_ref() + .ok_or(CacheError::Cancelled)? + .server_now(); + let now = Instant::now(); + for header in &request_headers { + if header.deadline_ms <= server_now { + return Err(CacheError::TimedOut); + } + if header.deadline_ms - server_now > self.owner.limits.max_request_ttl_ms { + return Err(CacheError::Malformed { + reason: "retry deadline", + }); + } + } + // Reads without side effects are retryable. Get creates a reader pin + // and cannot be retried until the server offers pin idempotency. + let retryable = matches!(request, RequestBody::Lookup(_)) + || (!matches!(request, RequestBody::Get(_)) + && self.owner.limits.mutation_retry_retention_ms.is_some()); + let attempt_limit = if retryable { + self.owner.limits.max_attempts + } else { + 1 + }; + let keys: Vec<_> = request_headers + .iter() + .map(|header| (self.shard_id(), self.epoch(), header.request_id.0)) + .collect(); + let mut state = self.owner.lock(); + state.expire(); + let mut new_entries = 0; + for key in &keys { + if let Some(entry) = state.retries.get(key) { + if entry.fingerprint != fingerprint || entry.active { + return Err(CacheError::IdempotencyConflict); + } + if entry.attempts >= attempt_limit { + return Err(quota(QuotaKind::IdempotencyTable)); + } + } else { + new_entries += 1; + } + } + if state.retries.len().saturating_add(new_entries) > self.owner.limits.max_retry_entries { + return Err(quota(QuotaKind::IdempotencyTable)); + } + for (key, header) in keys.iter().zip(request_headers) { + state + .retries + .entry(*key) + .or_insert(RetryEntry { + fingerprint, + attempts: 0, + active: false, + expires: now + Duration::from_millis(header.deadline_ms - server_now), + }) + .active = true; + } + drop(state); + let _guard = RetryGuard { + owner: Arc::clone(&self.owner), + keys: keys.clone(), + }; + loop { + self.current()?; + if self + .client + .as_ref() + .is_none_or(|client| !client.is_usable()) + { + self.reconnect()?; + } + let server_now = self + .client + .as_ref() + .ok_or(CacheError::Cancelled)? + .server_now(); + if headers(&request) + .iter() + .any(|header| header.deadline_ms <= server_now) + { + return Err(CacheError::TimedOut); + } + let mut state = self.owner.lock(); + if keys + .iter() + .any(|key| state.retries[key].attempts >= attempt_limit) + { + return Err(quota(QuotaKind::IdempotencyTable)); + } + for key in &keys { + let entry = state.retries.get_mut(key).ok_or(CacheError::Cancelled)?; + entry.attempts += 1; + } + drop(state); + let result = invoke( + self.client.as_mut().ok_or(CacheError::Cancelled)?, + request.clone(), + ); + if let Err(error) = self.current() { + let mut state = self.owner.lock(); + state.fenced_replies = state.fenced_replies.saturating_add(1); + return Err(error); + } + let retry = retryable + && matches!( + &result, + Err(CacheError::TimedOut + | CacheError::Storage { + reason: "transport io" + }) + ); + if !retry { + return result; + } + let mut state = self.owner.lock(); + if keys + .iter() + .any(|key| state.retries[key].attempts >= attempt_limit) + { + return result; + } + state.retry_attempts = state.retry_attempts.saturating_add(1); + drop(state); + // Closing the old socket precedes reconnect; no queued request and + // no retry on another shard, address, revision, or epoch. + self.client = None; + } + } +} + +impl Drop for ShardSession { + fn drop(&mut self) { + let mut state = self.owner.lock(); + if state.revision == self.revision && self.client.as_ref().is_some_and(TcpClient::is_usable) + { + if let Some(client) = self.client.take() { + state.idle.entry(self.shard_id()).or_default().push(client); + return; + } + } + state.connections -= 1; + if let Some(count) = state.by_shard.get_mut(&self.shard_id()) { + *count -= 1; + if *count == 0 { + state.by_shard.remove(&self.shard_id()); + } + } + } +} + +struct RetryGuard { + owner: Arc, + keys: Vec<(u64, u64, [u8; 16])>, +} +impl Drop for RetryGuard { + fn drop(&mut self) { + let mut state = self.owner.lock(); + for key in &self.keys { + if let Some(entry) = state.retries.get_mut(key) { + entry.active = false; + } + } + state.expire(); + } +} + +fn headers(request: &RequestBody) -> Vec<&ItemHeader> { + match request { + RequestBody::Lookup(items) => items.iter().map(|item| &item.header).collect(), + RequestBody::ReservePut(items) => items.iter().map(|item| &item.header).collect(), + RequestBody::Commit(items) => items.iter().map(|item| &item.header).collect(), + RequestBody::Get(items) => items.iter().map(|item| &item.header).collect(), + RequestBody::ReleaseCancel(items) => items.iter().map(|item| &item.header).collect(), + RequestBody::Invalidate(items) => items.iter().map(|item| &item.header).collect(), + } +} +fn identities(request: &RequestBody) -> Vec<&CacheIdentity> { + match request { + RequestBody::Lookup(items) => items.iter().map(|item| &item.identity).collect(), + RequestBody::ReservePut(items) => items.iter().map(|item| &item.identity).collect(), + RequestBody::Get(items) => items.iter().map(|item| &item.identity).collect(), + _ => Vec::new(), + } +} +fn quota(kind: QuotaKind) -> CacheError { + CacheError::QuotaExceeded { kind } +} +fn validate_endpoints( + mut endpoints: Vec, + max: usize, +) -> Result, CacheError> { + if endpoints.is_empty() || endpoints.len() > max { + return Err(quota(QuotaKind::InFlightRequests)); + } + endpoints.sort_by_key(|endpoint| endpoint.shard_id); + let mut addresses = BTreeSet::new(); + for (index, endpoint) in endpoints.iter().enumerate() { + if endpoint.epoch == 0 + || !addresses.insert(endpoint.address) + || index > 0 && endpoints[index - 1].shard_id == endpoint.shard_id + { + return Err(CacheError::Malformed { + reason: "shard endpoints", + }); + } + } + Ok(endpoints) +} +fn select_endpoint(endpoints: &[ShardEndpoint], hash: &[u8; 32]) -> ShardEndpoint { + // Rendezvous placement: stable for endpoint ordering and address changes. + *endpoints + .iter() + .max_by_key(|endpoint| { + let mut key = [0u8; 40]; + key[..32].copy_from_slice(hash); + key[32..].copy_from_slice(&endpoint.shard_id.to_le_bytes()); + content_checksum(&key).0 + }) + .expect("validated nonempty endpoints") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::frame::{read_envelope, write_envelope}; + use crate::pool::Pool; + use gpucache_core_types::{synthetic_identity, RequestId}; + use gpucache_protocol::{ + decode_request, encode_hello_ack, encode_reply, ItemResult, MissReason, ReplyBody, + TenantPolicy, WIRE_VERSION, + }; + use gpucache_storage::MemoryStore; + use std::net::{TcpListener, TcpStream}; + use std::sync::mpsc; + + fn auth() -> AuthContext { + let id = synthetic_identity(1, 1); + AuthContext::new(id.tenant, id.trust_domain, [9; 32]).unwrap() + } + fn accept(listener: &TcpListener, epoch: u64) -> (TcpStream, Arc) { + listener.set_nonblocking(true).unwrap(); + let start = Instant::now(); + let mut stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(error) + if error.kind() == std::io::ErrorKind::WouldBlock + && start.elapsed() < Duration::from_secs(3) => + { + std::thread::sleep(Duration::from_millis(1)) + } + Err(error) => panic!("accept: {error}"), + } + }; + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + stream + .set_write_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let pool = Arc::new(Pool::new(1)); + drop(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap()); + write_envelope(&mut stream, 1, &encode_hello_ack(WIRE_VERSION, epoch), 4096).unwrap(); + (stream, pool) + } + fn client(listener: &TcpListener, mutation_retry_retention_ms: Option) -> ShardedClient { + ShardedClient::new( + auth(), + vec![ShardEndpoint { + shard_id: 0, + address: listener.local_addr().unwrap(), + epoch: 1, + }], + ClientLimits { + max_frame_bytes: 4096, + io_timeout: Duration::from_secs(2), + mutation_retry_retention_ms, + ..ClientLimits::default() + }, + ) + .unwrap() + } + fn reserve(session: &ShardSession) -> ReservePutItem { + ReservePutItem { + header: ItemHeader { + request_id: RequestId::from_counter(1).unwrap(), + epoch: session.epoch(), + deadline_ms: session.deadline(1000), + }, + identity: synthetic_identity(1, 1), + generation: 0, + length: 4, + checksum: content_checksum(&[1, 2, 3, 4]), + payload: Some(vec![1, 2, 3, 4]), + lease_ttl_ms: 1000, + } + } + + #[test] + fn ambiguous_put_retries_identical_bytes_once_without_duplicate_allocation() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = client(&listener, Some(60_000)); + let worker = std::thread::spawn(move || { + let auth = auth(); + let policy = TenantPolicy::new(auth.tenant, auth.trust_domain, [9; 32]).unwrap(); + let mut store = MemoryStore::new(ResourceLimits::dev_defaults(), vec![policy]).unwrap(); + let mut original = Vec::new(); + for attempt in 0..2 { + let (mut stream, pool) = accept(&listener, 1); + let envelope = read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap(); + let bytes = envelope.body().to_vec(); + if attempt == 0 { + original = bytes.clone(); + } else { + assert_eq!(bytes, original); + } + let (auth, RequestBody::ReservePut(items)) = decode_request(&bytes).unwrap() else { + panic!("reserve") + }; + let reply = store.reserve_put(&auth, 2, items).unwrap(); + assert_eq!(reply.items[0].body.as_ref().unwrap().generation, 1); + if attempt == 1 { + write_envelope( + &mut stream, + 2, + &encode_reply(&ReplyBody::Reserve(reply)).unwrap(), + 4096, + ) + .unwrap(); + } + // First acknowledgement is deliberately lost after the mutation. + } + assert_eq!(store.stats().retained_bytes, 4); + assert_eq!(store.stats().in_flight_requests, 1); + assert_eq!(store.stats().idempotency_entries, 1); + }); + let mut session = client.shard(0).unwrap(); + let item = reserve(&session); + assert_eq!( + session.reserve_put(vec![item.clone()]).unwrap().items[0] + .body + .as_ref() + .unwrap() + .generation, + 1 + ); + assert_eq!(client.stats().retries, 1); + assert_eq!(client.stats().retry_entries, 1); + assert_eq!( + session.reserve_put(vec![item]), + Err(quota(QuotaKind::IdempotencyTable)) + ); + worker.join().unwrap(); + } + + #[test] + fn retry_reconnect_epoch_change_never_sends_the_old_mutation() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = client(&listener, Some(60_000)); + let worker = std::thread::spawn(move || { + let (mut stream, pool) = accept(&listener, 1); + drop(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap()); + drop(stream); + let (mut stream, pool) = accept(&listener, 2); + // The old request never reaches the new epoch after its hello. + assert!(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).is_err()); + }); + let mut session = client.shard(0).unwrap(); + let item = reserve(&session); + assert_eq!( + session.reserve_put(vec![item]), + Err(CacheError::EpochMismatch { + store: 2, + request: 1 + }) + ); + worker.join().unwrap(); + drop(session); + assert_eq!(client.stats().connections, 0); + } + + #[test] + fn default_mutations_and_all_gets_never_retry_after_lost_reply() { + for get in [false, true] { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = client(&listener, if get { Some(60_000) } else { None }); + let worker = std::thread::spawn(move || { + let (mut stream, pool) = accept(&listener, 1); + drop(read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap()); + // Drop both stream and listener. Any accidental retry is visible + // in the client's retry counter even though reconnect fails. + }); + let mut session = client.shard(0).unwrap(); + let item = reserve(&session); + let result = if get { + session + .get(vec![GetItem { + header: item.header, + generation: 1, + identity: item.identity, + pin_ttl_ms: 1000, + }]) + .map(|_| ()) + } else { + session.reserve_put(vec![item]).map(|_| ()) + }; + assert!(result.is_err()); + assert_eq!(client.stats().retries, 0); + worker.join().unwrap(); + } + } + + #[test] + fn late_reply_after_route_publication_is_suppressed_and_socket_closed() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = client(&listener, None); + let address = listener.local_addr().unwrap(); + let (arrived, wait_arrived) = mpsc::channel(); + let (release, wait_release) = mpsc::channel(); + let worker = std::thread::spawn(move || { + let (mut stream, pool) = accept(&listener, 1); + let envelope = read_envelope(&mut stream, pool.checkout().unwrap(), 4096).unwrap(); + let (_, RequestBody::Lookup(items)) = decode_request(envelope.body()).unwrap() else { + panic!("lookup") + }; + arrived.send(()).unwrap(); + wait_release.recv().unwrap(); + write_envelope( + &mut stream, + 2, + &encode_reply(&ReplyBody::Lookup(BatchReply { + store_epoch: 1, + items: vec![ItemResult { + request_id: items[0].header.request_id, + body: Ok(LookupOutcome::Miss { + reason: MissReason::Absent, + }), + }], + })) + .unwrap(), + 4096, + ) + .unwrap(); + }); + let mut session = client.shard(0).unwrap(); + let call = std::thread::spawn(move || { + let item = LookupItem { + header: reserve(&session).header, + identity: synthetic_identity(1, 1), + }; + session.lookup(vec![item]) + }); + wait_arrived.recv_timeout(Duration::from_secs(3)).unwrap(); + client + .replace_routes( + 1, + vec![ShardEndpoint { + shard_id: 0, + address, + epoch: 2, + }], + ) + .unwrap(); + release.send(()).unwrap(); + assert_eq!( + call.join().unwrap(), + Err(CacheError::FencingRejected { + reason: "route revision" + }) + ); + worker.join().unwrap(); + assert_eq!(client.stats().fenced_replies, 1); + assert_eq!(client.stats().connections, 0); + } + + #[test] + fn removing_then_readding_shard_cannot_reset_epoch_high_water_mark() { + let first: SocketAddr = "127.0.0.1:10001".parse().unwrap(); + let second: SocketAddr = "127.0.0.1:10002".parse().unwrap(); + let client = ShardedClient::new( + auth(), + vec![ShardEndpoint { + shard_id: 1, + address: first, + epoch: 8, + }], + ClientLimits::default(), + ) + .unwrap(); + client + .replace_routes( + 1, + vec![ShardEndpoint { + shard_id: 2, + address: second, + epoch: 9, + }], + ) + .unwrap(); + assert_eq!( + client.replace_routes( + 2, + vec![ShardEndpoint { + shard_id: 1, + address: first, + epoch: 8 + }] + ), + Err(CacheError::FencingRejected { + reason: "route epoch" + }) + ); + assert_eq!(client.stats().revision, 2); + assert_eq!(client.stats().shards, 1); + } +} diff --git a/crates/transport/tests/sharded_tcp.rs b/crates/transport/tests/sharded_tcp.rs new file mode 100644 index 0000000..760f830 --- /dev/null +++ b/crates/transport/tests/sharded_tcp.rs @@ -0,0 +1,524 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Barrier}; + +use gpucache_core_types::{ + content_checksum, synthetic_identity, Attestation, CacheError, CacheIdentity, QuotaKind, + RequestId, ResourceLimits, +}; +use gpucache_protocol::{ + AuthContext, CommitItem, GetItem, ItemHeader, LookupItem, LookupOutcome, ReleaseCancelItem, + ReservePutItem, TenantPolicy, +}; +use gpucache_storage::FileStore; +use gpucache_transport::{ + serve, ClientLimits, ServerConfig, ServerHandle, ShardEndpoint, ShardSession, ShardedClient, +}; + +struct TempDir(PathBuf); +impl TempDir { + fn new() -> Self { + static NEXT: AtomicU64 = AtomicU64::new(0); + Self(std::env::temp_dir().join(format!( + "gpucache-sharded-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + ))) + } +} +impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} +fn auth() -> AuthContext { + let id = synthetic_identity(1, 1); + AuthContext::new(id.tenant, id.trust_domain, [9; 32]).unwrap() +} +fn server(dir: &TempDir) -> ServerHandle { + let auth = auth(); + let store = FileStore::open( + &dir.0, + ResourceLimits::dev_defaults(), + vec![TenantPolicy::new(auth.tenant, auth.trust_domain, [9; 32]).unwrap()], + ) + .unwrap(); + serve(store, ServerConfig::default()).unwrap() +} +fn endpoint(shard_id: u64, server: &ServerHandle, epoch: u64) -> ShardEndpoint { + ShardEndpoint { + shard_id, + address: server.addr(), + epoch, + } +} +fn header(session: &ShardSession, id: u128) -> ItemHeader { + ItemHeader { + request_id: RequestId::from_counter(id).unwrap(), + epoch: session.epoch(), + deadline_ms: session.deadline(10_000), + } +} +fn identity_on(client: &ShardedClient, shard: u64, seed: u64) -> CacheIdentity { + for suffix in seed..seed + 1000 { + let mut id = synthetic_identity(1, 1); + id.prefix_chain[0].0[..8].copy_from_slice(&suffix.to_le_bytes()); + if client.shard_for(&id).unwrap() == shard { + return id; + } + } + panic!("rendezvous placement must cover two shards") +} +fn put(session: &mut ShardSession, identity: &CacheIdentity, id: u128) -> u64 { + let payload = vec![1, 2, 3, 4]; + let reserve = ReservePutItem { + header: header(session, id), + identity: identity.clone(), + generation: 0, + length: 4, + checksum: content_checksum(&payload), + payload: Some(payload), + lease_ttl_ms: 10_000, + }; + let generation = session.reserve_put(vec![reserve]).unwrap().items[0] + .body + .as_ref() + .unwrap() + .generation; + let committed = session + .commit(vec![CommitItem { + header: header(session, id + 1), + generation, + length: 4, + checksum: content_checksum(&[1, 2, 3, 4]), + }]) + .unwrap(); + assert!(committed.items[0].body.is_ok()); + generation +} + +#[test] +fn two_shards_keep_generation_scope_tenant_identity_and_pool_reuse() { + let dirs = [TempDir::new(), TempDir::new()]; + let servers = [server(&dirs[0]), server(&dirs[1])]; + let client = ShardedClient::new( + auth(), + vec![endpoint(1, &servers[0], 1), endpoint(2, &servers[1], 1)], + ClientLimits::default(), + ) + .unwrap(); + let ids = [identity_on(&client, 1, 0), identity_on(&client, 2, 100)]; + for (index, identity) in ids.iter().enumerate() { + let mut session = client.route(identity).unwrap(); + assert_eq!(session.shard_id(), index as u64 + 1); + // Both stores allocate the same local generation. No global remapping. + assert_eq!(put(&mut session, identity, 1), 1); + let reply = session + .get(vec![GetItem { + header: header(&session, 3), + generation: 1, + identity: identity.clone(), + pin_ttl_ms: 1000, + }]) + .unwrap(); + assert_eq!(reply.items[0].body.as_ref().unwrap().payload, [1, 2, 3, 4]); + let mut foreign = synthetic_identity(2, 1); + foreign.tenant = gpucache_core_types::TenantId::new(b"other-tenant".to_vec()).unwrap(); + assert!(matches!( + session.lookup(vec![LookupItem { + header: header(&session, 4), + identity: foreign + }]), + Err(CacheError::Unauthorized) + )); + let wrong = identity_on(&client, index as u64 + 1, 200); + let reply = session + .get(vec![GetItem { + header: header(&session, 5), + generation: 1, + identity: wrong, + pin_ttl_ms: 1000, + }]) + .unwrap(); + assert_eq!(reply.items[0].body, Err(CacheError::Incompatible)); + } + assert_eq!(client.stats().connections, 2); + assert_eq!(client.stats().idle_connections, 2); + let mut session = client.route(&ids[0]).unwrap(); + assert!(matches!( + session.lookup(vec![LookupItem { + header: header(&session, 6), + identity: ids[1].clone() + }]), + Err(CacheError::FencingRejected { + reason: "wrong shard" + }) + )); + assert_eq!(client.stats().connection_reuses, 1); +} + +#[test] +fn checked_out_connections_and_cancel_retry_entries_have_hard_bounds() { + let dir = TempDir::new(); + let server = server(&dir); + let limits = ClientLimits { + max_connections: 2, + max_connections_per_shard: 2, + max_retry_entries: 8, + ..ClientLimits::default() + }; + let client = ShardedClient::new(auth(), vec![endpoint(0, &server, 1)], limits).unwrap(); + let mut first = client.route(&synthetic_identity(1, 1)).unwrap(); + let second = client.shard(0).unwrap(); + assert!(matches!( + client.shard(0), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests + }) + )); + for n in 1..=200u128 { + let result = first.release_cancel(vec![ReleaseCancelItem { + header: header(&first, n), + generation: n as u64, + region_id: None, + pin_owner: None, + attestation: Attestation::Drained, + }]); + if n <= 8 { + assert!(result.is_ok()); + } else { + assert_eq!( + result, + Err(CacheError::QuotaExceeded { + kind: QuotaKind::IdempotencyTable + }) + ); + } + assert!(client.stats().retry_entries <= 8); + assert_eq!(client.stats().connections, 2); + } + drop(second); + for _ in 0..200 { + drop(client.shard(0).unwrap()); + } + assert_eq!(client.stats().connections, 2); + assert_eq!(client.stats().idle_connections, 1); +} + +#[test] +fn retry_admission_is_atomic_and_get_is_never_replayed() { + let dir = TempDir::new(); + let server = server(&dir); + let limits = ClientLimits { + max_retry_entries: 4, + ..ClientLimits::default() + }; + let client = ShardedClient::new(auth(), vec![endpoint(0, &server, 1)], limits).unwrap(); + let identity = synthetic_identity(1, 1); + let mut session = client.route(&identity).unwrap(); + put(&mut session, &identity, 1); + let item = GetItem { + header: header(&session, 3), + generation: 1, + identity: identity.clone(), + pin_ttl_ms: 1000, + }; + assert!(session.get(vec![item.clone()]).unwrap().items[0] + .body + .is_ok()); + assert_eq!( + session.get(vec![item]), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::IdempotencyTable + }) + ); + let too_large = vec![ + LookupItem { + header: header(&session, 4), + identity: identity.clone(), + }, + LookupItem { + header: header(&session, 5), + identity: identity.clone(), + }, + ]; + assert_eq!( + session.lookup(too_large), + Err(CacheError::QuotaExceeded { + kind: QuotaKind::IdempotencyTable + }) + ); + assert_eq!(client.stats().retry_entries, 3); + let item = LookupItem { + header: header(&session, 4), + identity, + }; + assert!(session.lookup(vec![item.clone()]).is_ok()); + let mut different = item; + different.header.deadline_ms += 1; + assert_eq!( + session.lookup(vec![different]), + Err(CacheError::IdempotencyConflict) + ); + assert_eq!(client.stats().retry_entries, 4); +} + +#[test] +fn stale_route_update_and_early_replay_retention_contract_are_rejected() { + let dir = TempDir::new(); + let server = server(&dir); + let routes = vec![endpoint(0, &server, 1)]; + let invalid = ClientLimits { + mutation_retry_retention_ms: Some(30_000), + ..ClientLimits::default() + }; + assert!(matches!( + ShardedClient::new(auth(), routes.clone(), invalid), + Err(CacheError::Malformed { + reason: "shard client config" + }) + )); + let client = ShardedClient::new(auth(), routes.clone(), ClientLimits::default()).unwrap(); + let mut old = client.shard(0).unwrap(); + assert_eq!( + client.replace_routes(1, routes), + Err(CacheError::FencingRejected { + reason: "route epoch" + }) + ); + assert_eq!(client.stats().revision, 1); + assert!(old + .lookup(vec![LookupItem { + header: header(&old, 1), + identity: synthetic_identity(1, 1) + }]) + .is_ok()); + client + .replace_routes(1, vec![endpoint(0, &server, 2)]) + .unwrap(); + assert_eq!( + old.lookup(vec![LookupItem { + header: header(&old, 2), + identity: synthetic_identity(1, 1) + }]), + Err(CacheError::FencingRejected { + reason: "route revision" + }) + ); + drop(old); + assert_eq!(client.stats().connections, 0); + // Merely publishing a route does not make an unfenced server current. + assert!(matches!( + client.shard(0), + Err(CacheError::EpochMismatch { + store: 1, + request: 2 + }) + )); + assert_eq!(client.stats().connections, 0); +} + +#[test] +fn two_shard_restart_fences_concurrent_lookup_get_and_put_and_inflight_reservation() { + let dirs = [TempDir::new(), TempDir::new()]; + let servers = [server(&dirs[0]), server(&dirs[1])]; + let client = ShardedClient::new( + auth(), + vec![endpoint(1, &servers[0], 1), endpoint(2, &servers[1], 1)], + ClientLimits::default(), + ) + .unwrap(); + let identities = [identity_on(&client, 1, 0), identity_on(&client, 2, 100)]; + for identity in &identities { + put(&mut client.route(identity).unwrap(), identity, 1); + } + let ready = Arc::new(Barrier::new(7)); + let resumed = Arc::new(Barrier::new(7)); + let mut workers = Vec::new(); + for (shard, identity) in identities.iter().enumerate() { + for operation in 0..3 { + let mut session = client.route(identity).unwrap(); + let identity = identity.clone(); + let ready = Arc::clone(&ready); + let resumed = Arc::clone(&resumed); + workers.push(std::thread::spawn(move || { + ready.wait(); + let id = 1000 + shard as u128 * 100 + operation; + let result = match operation { + 0 => session + .lookup(vec![LookupItem { + header: header(&session, id), + identity: identity.clone(), + }]) + .map(|_| ()), + 1 => session + .get(vec![GetItem { + header: header(&session, id), + generation: 1, + identity: identity.clone(), + pin_ttl_ms: 1000, + }]) + .map(|_| ()), + _ => session + .reserve_put(vec![ReservePutItem { + header: header(&session, id), + identity: identity.clone(), + generation: 0, + length: 4, + checksum: content_checksum(&[1, 2, 3, 4]), + payload: Some(vec![1, 2, 3, 4]), + lease_ttl_ms: 1000, + }]) + .map(|_| ()), + }; + // An operation racing shutdown may finish or fail to recompute. + if let Err(error) = result { + assert!(matches!( + error, + CacheError::Storage { .. } + | CacheError::TimedOut + | CacheError::EpochMismatch { .. } + | CacheError::FencingRejected { .. } + )); + } + resumed.wait(); + let result = session.lookup(vec![LookupItem { + header: header(&session, id + 10), + identity, + }]); + assert_eq!( + result, + Err(CacheError::FencingRejected { + reason: "route revision" + }) + ); + })); + } + } + // A writing generation must not survive the epoch change as a hit. + let absent_id = identity_on(&client, 1, 400); + let mut pending = client.route(&absent_id).unwrap(); + let reserved = pending + .reserve_put(vec![ReservePutItem { + header: header(&pending, 5000), + identity: absent_id.clone(), + generation: 0, + length: 4, + checksum: content_checksum(&[1, 2, 3, 4]), + payload: Some(vec![1, 2, 3, 4]), + lease_ttl_ms: 10_000, + }]) + .unwrap(); + let generation = reserved.items[0].body.as_ref().unwrap().generation; + ready.wait(); + drop(servers); + let restarted = [server(&dirs[0]), server(&dirs[1])]; + client + .replace_routes( + 1, + vec![endpoint(1, &restarted[0], 2), endpoint(2, &restarted[1], 2)], + ) + .unwrap(); + resumed.wait(); + for worker in workers { + worker.join().unwrap(); + } + assert_eq!( + pending.commit(vec![CommitItem { + header: header(&pending, 5001), + generation, + length: 4, + checksum: content_checksum(&[1, 2, 3, 4]) + }]), + Err(CacheError::FencingRejected { + reason: "route revision" + }) + ); + drop(pending); + let mut fresh = client.route(&absent_id).unwrap(); + assert_eq!(fresh.epoch(), 2); + assert!(matches!( + fresh + .lookup(vec![LookupItem { + header: header(&fresh, 6000), + identity: absent_id + }]) + .unwrap() + .items[0] + .body, + Ok(LookupOutcome::Miss { .. }) + )); +} + +#[test] +fn idle_connection_on_one_shard_does_not_starve_another_shard() { + let dirs = [TempDir::new(), TempDir::new()]; + let servers = [server(&dirs[0]), server(&dirs[1])]; + let limits = ClientLimits { + max_connections: 1, + max_connections_per_shard: 1, + ..ClientLimits::default() + }; + let client = ShardedClient::new( + auth(), + vec![endpoint(1, &servers[0], 1), endpoint(2, &servers[1], 1)], + limits, + ) + .unwrap(); + for shard in [1, 2, 1, 2, 1, 2] { + drop(client.shard(shard).unwrap()); + assert_eq!(client.stats().connections, 1); + assert_eq!(client.stats().idle_connections, 1); + } +} + +#[test] +fn prefix_on_another_shard_is_a_clean_miss_without_global_fanout() { + let dirs = [TempDir::new(), TempDir::new()]; + let servers = [server(&dirs[0]), server(&dirs[1])]; + let client = ShardedClient::new( + auth(), + vec![endpoint(1, &servers[0], 1), endpoint(2, &servers[1], 1)], + ClientLimits::default(), + ) + .unwrap(); + let short = identity_on(&client, 1, 0); + put(&mut client.route(&short).unwrap(), &short, 1); + let longer = (0..100u8) + .find_map(|nonce| { + let mut query = short.clone(); + query + .prefix_chain + .push(gpucache_core_types::Digest([nonce; 32])); + (client.shard_for(&query).unwrap() == 2).then_some(query) + }) + .expect("longer prefix can land on the other shard"); + let mut session = client.route(&longer).unwrap(); + let result = session + .lookup(vec![LookupItem { + header: header(&session, 10), + identity: longer, + }]) + .unwrap(); + assert!(matches!( + result.items[0].body, + Ok(LookupOutcome::Miss { .. }) + )); + // A wrong local generation cannot fetch bytes from the shorter prefix's + // owner. Neither lookup nor Get silently fans out or slices a block. + let result = session.get(vec![GetItem { + header: header(&session, 11), + generation: 1, + identity: short, + pin_ttl_ms: 1000, + }]); + assert_eq!( + result, + Err(CacheError::FencingRejected { + reason: "wrong shard" + }) + ); +} diff --git a/docs/adr/0011-bounded-shard-client.md b/docs/adr/0011-bounded-shard-client.md new file mode 100644 index 0000000..c6a25c5 --- /dev/null +++ b/docs/adr/0011-bounded-shard-client.md @@ -0,0 +1,156 @@ +# ADR 0011: Bounded TCP shard sessions and explicit retry contracts + +- Status: **Accepted for the CPU/TCP contract** +- Date: 2026-10-20 +- Milestone: V3 + +## Context + +`TcpClient` already checked reply operation, ordered request IDs, and store epoch, +and poisoned an incomplete stream. It had no shared connection pool or shard +placement. The directory bounded active fills and notices but exposed membership +changes one at a time. The V3 control plane's membership was not configurable +through a public batch operation. A distributed client must not turn an ambiguous +write into a write on another owner, reuse a poisoned stream, or duplicate a +reader pin after losing a Get reply. + +## Decision + +The GCK1/GCKR bytes and `CacheIdentity` remain unchanged. `TcpClient` remains +available. `ShardedClient` adds a tenant-scoped CPU/TCP connection pool and +rendezvous routing over the validated full physical identity. Endpoint order and +address changes do not change the placement hash. `route(identity)` checks the +client's tenant and trust domain before opening a connection and returns a +`ShardSession` with the same six RPC methods as `TcpClient`. + +Generations remain local to a shard. Retain `session.shard_id()` with a generation +and use `shard(shard_id)` to reopen that owner. Do not hash generation numbers or +move a Commit/Release/Invalidate to another shard: independent stores can both +allocate generation 1. Identity-bearing batches must belong to one session's +shard. Cross-shard batches fail before sending; callers partition them explicitly. +No generation bits, identity fields, or wire version are repurposed for routing. + +### Admission and retries + +- `ClientLimits` bounds shards, global and per-shard sockets, retry metadata, + frame size, request lifetime, and total attempts. Upper validation caps are + 4,096 shards/sockets, 1,048,576 retry entries, 32 MiB per frame, four attempts, + and 60 seconds request lifetime. Deployment defaults are smaller. There is no + waiting or retry queue: exhaustion returns a quota error for recomputation. +- Active and idle sockets share the connection ceiling. Checkout can close an + idle socket on another shard so it does not monopolize the global ceiling. + Poisoned sockets are closed. Dropping a healthy lease returns its socket. +- The retry table retains only checksums and bounded request metadata, not KV + payloads or replies. Admission validates the entire batch before inserting any + entry. Duplicate request IDs, a changed batch fingerprint, or a concurrently + active same-ID request fail. Live entries are not evicted to admit new work. + Entries expire at the original deadline; active entries survive until their + call finishes. Request IDs must remain unique across new logical requests. +- Every retry keeps the exact batch, authentication, request IDs, epochs, and + deadlines. Retry only follows an ambiguous I/O/timeout result; item-level + failures and epoch/identity/authorization failures are returned unchanged. + The original call and automatic or explicit replays share the attempt ceiling. +- Lookup can retry because it does not create a pin or mutation. **Get never + retries**, including when mutation retries are enabled. A lost Get reply means + recompute and later explicit pin drain according to the storage lease contract. +- Mutation retries default to disabled. Opt-in requires + `mutation_retry_retention_ms` to declare the deployment's minimum server + idempotency retention, and it must be strictly greater than the client's + maximum request lifetime. Operators must configure every server to retain + idempotency entries for at least that declared duration. GCK1 does not negotiate + this setting. If this contract is unknown, keep mutation retries disabled. + Reconnecting checks the expected epoch before resending; restart cannot replay + an old mutation into a new store. + +### Membership and fences + +`ShardDirectory::apply_membership` and `ControlPlane::apply_membership` apply a +bounded complete member snapshot. Duplicate IDs, zero/stale incarnations, empty +membership, capacity exhaustion, and epoch exhaustion fail before changing +membership, fault domains, credits, or notices. Reordering an unchanged set is a +no-op. A valid changed snapshot fences once. There is no membership update queue +in which a rejected update can block later independent snapshots. This is a +synchronous admission rule, not a distributed scheduler fairness guarantee. + +`ShardedClient::replace_routes` checks the expected local revision and preflights +the full snapshot. Every new endpoint must have an epoch above the highest epoch +in the previous snapshot, including newly added and re-added shard IDs. This +coordinated replacement rule keeps a scalar epoch high-water mark instead of an +unbounded retired-node history. It closes idle sockets, retires old sessions, +and suppresses replies that arrive after publication. Checked-out old leases +remain charged until dropped. + +**Publishing a client route does not fence a remote server.** The caller must +stop/drain old ownership and advance every destination server epoch before +publishing. The client refuses a handshake from an unexpected epoch, but the +unchanged wire format cannot prove that a disconnected old server has stopped +serving other clients. This tree has no distributed quorum, authenticated +membership distribution, or cross-host ownership authority. A local directory +fence is not a remote data-plane fence. + +## CPU restart and rebalance checklist + +1. Stop admitting fills on the old shard; retain generation/epoch/shard scope. +2. Drain known completions. Keep unknown completions quarantined and charged. +3. Stop the old TCP service and wait for its workers to exit before reopening + the same file store. Reopening advances the persisted store epoch and drops + uncommitted state. Coordinate independent hosts externally. +4. Start and verify all snapshot endpoints at epochs above the previous maximum. + Do not publish a snapshot if any endpoint or directory fence fails. +5. Apply directory/control-plane membership snapshots and publish client routes + with the expected revision. Drop retired sessions. Never rewrite an old + request's epoch or retry it on the new owner. +6. Treat absent uncommitted blocks, unavailable shards, rejected old generations, + and any unverified completion as recompute. Committed file blocks may remain + readable under the new epoch; they are not durability replicas. + +## Observability + +`ClientStats` exposes active-plus-idle connection count, idle connections, retry +entries, reuse/retry counters, route revision, and suppressed replies. +`DirectoryStats` reports flights, request entries, notices, charged tenants, +members, fences, and fenced fills. `RegisteredPool::stats` reports quarantined +slots, held bytes, fences, drained slots, and reclaimed slots. Lease expiry never +increments the drained/reclaimed counters. + +`TcpClient::io_stats` separates complete RPC body bytes from complete framed +bytes, including hello. Each complete application envelope adds 12 framing bytes. +These are application counters: TCP/IP headers, kernel retransmissions, and +partial failed I/O are excluded. `TransferDesc::bytes.len()` and TCP staging receipts +continue to describe submitted/staged payload size, not measured NIC +wire traffic. None of these values is a bandwidth benchmark. + +## Enforced limitations and evidence + +| Contract | CPU regression | +| --- | --- | +| A longer prefix routed away from a stored shorter prefix is a clean miss with no fanout | `prefix_on_another_shard_is_a_clean_miss_without_global_fanout` | +| Two servers can both allocate generation 1 without routing a wrong identity or tenant hit | `two_shards_keep_generation_scope_tenant_identity_and_pool_reuse` | +| Global socket capacity cannot be monopolized by idle connections on another shard | `idle_connection_on_one_shard_does_not_starve_another_shard` | +| Socket/retry tables stay bounded under a cancel storm; no hidden queue | `checked_out_connections_and_cancel_retry_entries_have_hard_bounds` | +| Retry batch admission has no partial metadata mutation; Get cannot be replayed | `retry_admission_is_atomic_and_get_is_never_replayed` | +| Lost mutation acknowledgement replays identical bytes with one allocation | `ambiguous_put_retries_identical_bytes_once_without_duplicate_allocation` | +| Get and default mutations do not retry an ambiguous loss | `default_mutations_and_all_gets_never_retry_after_lost_reply` | +| Reconnect to a new epoch never sends an old mutation | `retry_reconnect_epoch_change_never_sends_the_old_mutation` | +| Late replies cannot cross a route revision | `late_reply_after_route_publication_is_suppressed_and_socket_closed` | +| Re-added shard cannot reset the epoch high-water mark | `removing_then_readding_shard_cannot_reset_epoch_high_water_mark` | +| Concurrent Lookup/Get/Put racing two FileStore restarts, plus stale pending Commit, fail safely | `two_shard_restart_fences_concurrent_lookup_get_and_put_and_inflight_reservation` | +| Invalid membership batches leave notices, credits and members untouched | `membership_batch_rejects_every_invalid_snapshot_before_fencing` | +| Cancellation/restart storms keep directory metadata bounded | `cancel_restart_and_membership_storm_keep_every_table_bounded` | +| Registered cancellation requires explicit drain/fence and reports bounded held bytes | `cancellation_storm_requires_drain_and_reports_fence_reclaim_counts` | + +Whole-block prefix lookup remains correct within a shard. Full-identity routing +can place a longer prefix on a different shard, so discovery of shorter prefixes +across shards is deferred and returns a miss; no cross-shard search or intra-block +slicing is added. Generations are not globally unique. Independently incremental +route replacement is deferred in favor of the coordinated snapshot fence rule. +Client restart does not recover a retry ledger; only the server's current-epoch +idempotency table can recognize a still-valid exact request. No retry survives a +server epoch change. Error callers must recompute; I/O errors are not fabricated +as cache hits or successful mutations. + +Run the complete CPU reproduction with +`python3 scripts/ci/vm_regression.py --profile cpu --offline --output artifacts/astra-full-cpu`. +The runner discovers these tests through workspace tests. Live RDMA, NIC/DPU +completion, GDS, vLLM KV import/export and multi-host partitions remain +`NOT_RUN_HARDWARE`. This is CPU/TCP fault evidence, not PB-scale verification.