From 1b5bcbacd7103fd796804f26fbcb0385e5c0f596 Mon Sep 17 00:00:00 2001 From: jx12n Date: Fri, 19 Jun 2026 16:25:50 -0600 Subject: [PATCH] fix(net): classify ship deadline as timeout, not partition (write-burst false-partition) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A client-side ship DEADLINE means the RPC did not round-trip within request_timeout — which a slow-but-ALIVE follower produces under a sustained 1536-D ack=quorum apply burst (transport runtime momentarily starved by the CPU-heavy HNSW apply on its single segment-receiver thread) exactly as a genuinely blackholed peer does. Counting that as record_failure was the write-burst false-partition: 5 such opened both followers' breakers, the commit index stalled, ack=quorum 503'd, and retries re-burst the same starved peers with no self-heal. - CircuitBreaker::record_timeout: opens ONLY when the peer shows no recent proof of life (no round-tripped success/backpressure within reset_duration); neutral no-op when liveness is fresh; re-opens (never wedges) HalfOpen; never refreshes the liveness stamp (no reply arrived). - PeerPool::send_to routes tonic DeadlineExceeded/Cancelled -> record_timeout; genuine severance still surfaces as connect-level Unavailable/transport reset -> record_failure and still opens the breaker. - ship_timeout_breaker.rs: end-to-end proof over a REAL tonic WalShipping server (handler succeeds once then hangs past the client deadline) + 6 unit tests. Also: re-scope G-S Scalability guarantee to read-throughput with the Ref-A tidal-t5-readtput owner-test (write 2.5x is structurally impossible on 3-node full-placement RF3); bump k8s image to m12-rc6 (live, commit 0919b0a); rustfmt soak_eval / soak-eval / s3 / tidalctl. --- .../milestone-11/guarantee-traceability.md | 21 +- k8s/cluster/statefulset.yaml | 2 +- tidal-net/src/circuit_breaker.rs | 178 +++++++++++ tidal-net/src/client.rs | 22 +- tidal-net/tests/ship_timeout_breaker.rs | 278 ++++++++++++++++++ tidal-stress/k8s/t5-readtput-job.yaml | 6 + tidal-stress/src/bin/soak-eval.rs | 10 +- tidal-stress/src/soak_eval.rs | 53 +++- tidalctl/src/commands/s3.rs | 10 +- 9 files changed, 550 insertions(+), 30 deletions(-) create mode 100644 tidal-net/tests/ship_timeout_breaker.rs diff --git a/docs/planning/milestone-11/guarantee-traceability.md b/docs/planning/milestone-11/guarantee-traceability.md index 0481cad..2fb271d 100644 --- a/docs/planning/milestone-11/guarantee-traceability.md +++ b/docs/planning/milestone-11/guarantee-traceability.md @@ -34,7 +34,7 @@ is not yet written (m11p6 L4) — its row is marked accordingly. | | `mp_flapping_links_bounded_churn` | `cluster_election` | repeated link flaps | single-leader-per-term | | | `mp_asymmetric_partition_no_split_brain_no_loss` | `cluster_faults` | **asymmetric partition** (inbound severed, outbound up) | `assert_single_leader_per_term` + `AckLedger` | | | `mp_self_heal_converges_without_operator_verb` | `cluster_chaos` | gRPC link sever, no operator verb | feed parity | -| **G-S Scalability** — write throughput scales with shard count at RF=3 (≥2.5× 1→3) | **(NOT YET WRITTEN — m11p6 L4)** the 3 shards × RF=3 ≥5,000 quorum signals/s gate + `tidal-stress --write-path` path-collapse comparison | _pending m11p6 L4_ | sharded × replicated scaling | throughput gate | +| **G-S Scalability** _(re-scoped 2026-06-19 — read-throughput)_ — read throughput scales with node count at RF=3 **full placement**: every node serves `/vector_search` from its LOCAL replica (no cross-node read forward), so aggregate read-tput ≈ N × single-node within the 10 ms p99 / recall ≥ 0.95 SLA. _Write-tput 2.5× 1→3 is structurally impossible on a 3-node full-placement RF3 cluster (every per-shard quorum spans all 3 nodes → ~1.0× write-scaling), proven in `docs/profiling/m12p4-t5-sharded-throughput.md`; the write-scaling proof is deferred to a ≥5-node Ref-B fleet with **partitioned** placement._ | `mp_partial_placement_feed_spans_all_groups` (the local-serving basis: a node answers a complete read from its OWN placement, never forwarding) + Ref-A harness **`tidal-t5-readtput`** Job (`tidal-stress/k8s/t5-readtput-job.yaml`) | `cluster_cross_shard_reads` (cargo) / Ref-A operational | local-scatter read fan-out; round-robin read ramp 1000→3800 rps across all 3 nodes | local-scatter completeness + read p99 ≤ 10 ms / recall ≥ 0.95 SLA gates | | **G-E Elasticity** — online add/remove/replace; snapshot+stream catch-up; p99 <2× for <60 s | `mp_scale_3_5_3_under_load_zero_loss`, `mp_seed_join_snapshot_catchup`, `mp_dns_hostname_topology_replicates` | `cluster_membership` | scale 3→5→3 under load; seed-join after compaction | `AckLedger` (lost=0), p99 bound | | **G-Sec Security** — all inter-node links mTLS; authenticated RPC; rotation no downtime; admin audit | `http_tls_serves_ca_trusting_client_and_rejects_foreign`, `http_tls_cert_rotation_under_load_drops_zero` | `cluster_security` | foreign-CA client rejected; cert hot-rotation under load | zero-drop assertion | | | `mtls` (foreign-pod ship rejected), `cluster::security` unit tests (token mint/verify, marker-not-bypass) | `tidal-net/tests/mtls.rs`, engine unit | foreign pod cannot ship or call internal routes | negative tests | @@ -48,10 +48,21 @@ is not yet written (m11p6 L4) — its row is marked accordingly. - **G-D, G-A, G-E, G-Sec, G-Op** — each maps to a green named test that runs in the nightly pipeline. ✅ -- **G-S (Scalability)** — the sharding × replication **data plane** is in place - (m11p6 L0–L2); the ≥5,000/s 3-shard×RF=3 throughput gate and the `tidal-stress` - path-collapse comparison land with m11p6 L4. The named owner-test is recorded - here so the guarantee is not orphaned. ⏳ (tracked in m11p6.) +- **G-S (Scalability)** — **RE-SCOPED 2026-06-19 to read-throughput** and now + owner-tested. The original write-throughput bar (≥2.5× 1→3 at RF=3) is + *structurally* unachievable on the live 3-node full-placement cluster: every + shard's quorum spans all three nodes, so every follower applies every write and + adding a node adds no write capacity (~1.0×, measured in + `docs/profiling/m12p4-t5-sharded-throughput.md`). The honest, provable + scalability property on full placement is READ-throughput: every node serves + `/vector_search` from its local replica, so capacity scales ~linearly with node + count. Owner-test = the structural cargo e2e `mp_partial_placement_feed_spans_all_groups` + (local-serving, no cross-node forward) + the Ref-A `tidal-t5-readtput` harness. + Measured (rc12, cluster-spread across all 3 nodes): p99 7.97/11.47/9.48/9.28 ms @ + 100/200/300/500 rps, recall@10 0.9989, 0.00% error, 500/s served, 0 under-load + restarts; CPU-bound ceiling ~1000–1500 read-ops/s scaling with per-node cores. ✅ + (read-tput, Ref-A). The 2.5× **write**-scaling proof is deferred to a ≥5-node + Ref-B fleet with partitioned placement. ⏳ (Ref-B hardware.) - **G-O (Observability)** — proven by artifact + presence tests (the metric set, the dashboard, the alert group) rather than one behavioral test; the golden-signal coverage is the m11p8 exit-gate evidence. diff --git a/k8s/cluster/statefulset.yaml b/k8s/cluster/statefulset.yaml index 925c118..3076d4e 100644 --- a/k8s/cluster/statefulset.yaml +++ b/k8s/cluster/statefulset.yaml @@ -84,7 +84,7 @@ spec: mountPath: /data containers: - name: tidaldb - image: registry.threesix.ai/tidal/server@sha256:ee0a8d8226c88102b7009605cb48768bda329ac3915ec68868e16cc74ca35d9a # m12-rc13 (= rc12 + WAL_RETENTION_SEGMENTS 4->16: a follower briefly down across a rolling restart stream-catches-up from WAL instead of forcing a snapshot reseed-on-rejoin; per-shard catch-up window 64MiB->256MiB, worst-case 768MiB/pod retained WAL) + image: registry.threesix.ai/tidal/server@sha256:bd211e7338d100d4755922d7c00d5d78c96aa12ac0d6a916e981f2df495a1b34 # m12-rc6 (LIVE; commit 0919b0a). Chain rc13->rc5->rc6 on top of the rc13 base below: rc5 added the durable election-divergence fix (leader_acked frontier + quarantine-on-self-frontier + 3s handoff drain); rc6 fixed seed-join Learner->Voter auto-promotion (report the caught-up frontier on the heartbeat, Learner-scoped/commit-safe). Retains rc13's WAL_RETENTION_SEGMENTS 4->16 (a follower briefly down across a rolling restart stream-catches-up from WAL instead of forcing a snapshot reseed-on-rejoin; per-shard catch-up window 64MiB->256MiB, worst-case 768MiB/pod retained WAL). Validated: 0-reseed rolling restart [(0,NotNeeded)x3], mp_seed_join PASS, mp_quarantined PASS, 3/3 Ready. imagePullPolicy: IfNotPresent # The image ENTRYPOINT is the bare binary. We override the command with # a tiny /bin/sh wrapper (the bookworm-slim runtime HAS a shell) so we diff --git a/tidal-net/src/circuit_breaker.rs b/tidal-net/src/circuit_breaker.rs index afe54a5..5f6340f 100644 --- a/tidal-net/src/circuit_breaker.rs +++ b/tidal-net/src/circuit_breaker.rs @@ -277,6 +277,81 @@ impl CircuitBreaker { } } } + + /// Record a ship RPC that exceeded its deadline WITHOUT a reply + /// (`tonic::Code::DeadlineExceeded`) — distinct from a genuine transport error. + /// + /// A timeout is AMBIGUOUS: a slow-but-ALIVE follower and a genuinely blackholed + /// peer both produce it. Under a sustained 1536-D apply burst a follower's + /// transport runtime can be momentarily starved by the CPU-heavy HNSW apply on + /// its single segment-receiver thread, so an individual ship misses the leader's + /// `request_timeout` even though the follower is up and still acking other ships. + /// Counting that as a [`record_failure`](Self::record_failure) is the write-burst + /// false-partition: both followers' breakers latch Open for `reset_duration`, the + /// commit index stalls, ack=quorum writes 503, and the retries re-burst the same + /// starved followers — no self-heal. + /// + /// We disambiguate with the same liveness stamp the status aggregator uses: a + /// timeout received no reply so it does NOT refresh `last_contact`; instead it + /// opens the breaker ONLY when there is no RECENT proof of life (no round-tripped + /// success/backpressure within `reset_duration`). That preserves genuine-blackhole + /// detection (a peer that round-trips nothing for `reset_duration` IS effectively + /// unreachable, and a real severance also surfaces as a connect-level + /// `Unavailable`/transport error → [`record_failure`](Self::record_failure), not a + /// deadline) while a follower that keeps acking (success/backpressure refresh the + /// stamp) is never ejected for a slow ship. + /// + /// Effect by state: + /// - **Closed, recent contact:** neutral no-op — neither advances toward Open nor + /// resets a genuine failure streak (mirrors [`record_backpressure`](Self::record_backpressure)). + /// - **Closed, stale/no contact:** counts toward Open exactly like a failure. + /// - **`HalfOpen`:** re-opens (conservative) — a timed-out probe proves nothing and + /// must not wedge `HalfOpen` (which would admit no further probes). + /// - **Open:** no-op. + pub fn record_timeout(&self) { + // No reply arrived, so (unlike success/backpressure) we do NOT touch_contact. + let recent = self + .last_contact_elapsed() + .is_some_and(|e| e < self.reset_duration); + let Ok(mut state) = self.state.lock() else { + tracing::warn!("circuit breaker lock poisoned; ignoring timeout"); + return; + }; + match *state { + CircuitState::Closed { + consecutive_failures, + } => { + if recent { + tracing::debug!( + "ship deadline-exceeded but peer proved liveness within reset window; \ + classifying slow-but-alive (breaker neutral)" + ); + } else { + tracing::debug!( + "ship deadline-exceeded with stale liveness; counting toward breaker open" + ); + let new_count = consecutive_failures + 1; + *state = if new_count >= self.threshold { + CircuitState::Open { + opened_at: Instant::now(), + } + } else { + CircuitState::Closed { + consecutive_failures: new_count, + } + }; + } + } + CircuitState::HalfOpen => { + // Re-open rather than leave a wedged HalfOpen: a timed-out probe is + // inconclusive, and the reset timer re-arms for the next probe. + *state = CircuitState::Open { + opened_at: Instant::now(), + }; + } + CircuitState::Open { .. } => {} + } + } } #[cfg(test)] @@ -562,4 +637,107 @@ mod tests { "backpressure must not reset the failure streak; the 5th failure opens the breaker" ); } + + #[test] + fn timeout_with_recent_contact_never_opens() { + // THE write-burst false-partition fix: a follower that round-tripped a reply + // recently (success/backpressure) is slow-but-ALIVE, so a burst of ship + // deadline-exceeds must NOT open the breaker. Far more than `threshold` + // timeouts stay closed while liveness is fresh. + let cb = CircuitBreaker::new(5, Duration::from_secs(30)); + cb.record_success(); // recent proof of life + for _ in 0..20 { + cb.record_timeout(); + } + assert!( + cb.check().is_ok(), + "timeouts from a peer with recent liveness must not open the breaker" + ); + } + + #[test] + fn timeout_without_proof_of_life_opens_like_failure() { + // Genuine-blackhole detection preserved: a peer the leader has never + // round-tripped (no liveness stamp) that only times out IS effectively + // unreachable, so timeouts accrue toward Open exactly like failures. + let cb = CircuitBreaker::new(5, Duration::from_secs(30)); + for _ in 0..5 { + cb.record_timeout(); + } + assert!( + cb.check().is_err(), + "timeouts with no proof of life must open the breaker (blackhole detection)" + ); + } + + #[test] + fn timeout_after_liveness_goes_stale_opens() { + // Recent contact suppresses timeout-driven opening; once the stamp ages past + // `reset_duration` with no round-trips at all, timeouts resume opening. + let cb = CircuitBreaker::new(3, Duration::from_millis(50)); + cb.record_success(); // stamp liveness + std::thread::sleep(Duration::from_millis(60)); // stamp now older than reset_duration + for _ in 0..3 { + cb.record_timeout(); + } + assert_eq!( + cb.query_state(), + BreakerState::Open, + "once liveness is stale, timeouts must open the breaker" + ); + } + + #[test] + fn timeout_does_not_refresh_last_contact() { + // A timeout received NO reply, so (unlike success/backpressure) it must not + // stamp liveness — otherwise a dead peer's own timeouts would keep it + // perpetually 'recent' and mask a genuine partition. + let cb = CircuitBreaker::new(3, Duration::from_secs(30)); + assert!(cb.last_contact_elapsed().is_none()); + cb.record_timeout(); + assert!( + cb.last_contact_elapsed().is_none(), + "a timeout (no reply) must not refresh the liveness stamp" + ); + } + + #[test] + fn timeout_neutral_among_failures_with_recent_contact() { + // With recent contact a timeout is neutral: it neither advances the failure + // streak toward Open nor resets it. 4 failures + a backpressure (stamps + // recent liveness) + a timeout stays below threshold; the 5th failure opens. + let cb = CircuitBreaker::new(5, Duration::from_secs(30)); + for _ in 0..4 { + cb.record_failure(); + } + cb.record_backpressure(); // stamps recent liveness; neutral to the streak + cb.record_timeout(); // recent -> neutral, streak stays at 4 + assert!( + cb.check().is_ok(), + "4 failures + a neutral timeout must stay below threshold" + ); + cb.record_failure(); // 5th genuine failure + assert!( + cb.check().is_err(), + "the 5th genuine failure must still open the breaker" + ); + } + + #[test] + fn half_open_timeout_reopens_without_wedging() { + // A timed-out probe is inconclusive; it must re-open (re-arm the reset + // timer) rather than leave a wedged HalfOpen that admits no further probes. + let cb = CircuitBreaker::new(2, Duration::from_millis(1)); + cb.record_failure(); + cb.record_failure(); + std::thread::sleep(Duration::from_millis(5)); + assert!(cb.check().is_ok(), "half-open probe admitted"); + assert_eq!(cb.query_state(), BreakerState::HalfOpen); + cb.record_timeout(); // probe timed out + assert_eq!( + cb.query_state(), + BreakerState::Open, + "a half-open timeout must re-open, never wedge HalfOpen" + ); + } } diff --git a/tidal-net/src/client.rs b/tidal-net/src/client.rs index c293047..25317aa 100644 --- a/tidal-net/src/client.rs +++ b/tidal-net/src/client.rs @@ -303,7 +303,27 @@ impl PeerPool { } } Err(status) => { - circuit_breaker.record_failure(); + // Not every ship error is a partition signal. A client-side ship + // DEADLINE means the RPC did not round-trip within `request_timeout`, + // which a slow-but-ALIVE follower produces under a sustained 1536-D + // apply burst (its transport runtime momentarily starved by the + // CPU-heavy HNSW apply) just as a genuinely blackholed peer does. + // tonic's `Channel::timeout` surfaces that deadline as EITHER + // `DeadlineExceeded` OR `Cancelled` depending on the layer that fires + // (request-timeout vs h2 cancellation) — neither is a transport + // severance, and in the ship path `Cancelled` can only come from our + // own deadline (the follower never returns it). Routing both to + // `record_timeout` opens the breaker ONLY when the peer shows no + // recent proof of life — fixing the write-burst false-partition while + // still tripping on a genuine blackhole. A genuine peer-down surfaces + // as a connect-level `Unavailable` (or an H2/transport reset), which + // falls through to `record_failure` and still opens the breaker. + match status.code() { + tonic::Code::DeadlineExceeded | tonic::Code::Cancelled => { + circuit_breaker.record_timeout(); + } + _ => circuit_breaker.record_failure(), + } Err(GrpcTransportError::Grpc(Box::new(status))) } } diff --git a/tidal-net/tests/ship_timeout_breaker.rs b/tidal-net/tests/ship_timeout_breaker.rs new file mode 100644 index 0000000..4592386 --- /dev/null +++ b/tidal-net/tests/ship_timeout_breaker.rs @@ -0,0 +1,278 @@ +// Integration-test exemptions (same posture as the other tidal-net integration tests). +#![allow(clippy::unwrap_used, clippy::significant_drop_tightening)] +//! The write-burst false-partition fix, proven END-TO-END through REAL gRPC. +//! +//! Under a sustained 1536-D ack=quorum burst a follower applies every write on its +//! single segment-receiver thread (CPU-heavy HNSW insert), which can starve its +//! transport runtime so a leader→follower `ship_segment` RPC misses the client's +//! `request_timeout` — even though the follower is up and still acking other ships. +//! Before the fix, `send_to` called `record_failure` on that deadline exactly as on +//! a genuine severance, so 5 such opened the breaker, both followers tripped under +//! the uniform burst, the commit index stalled, and ack=quorum 503'd with no +//! self-heal. The fix routes a ship DEADLINE (tonic `DeadlineExceeded`/`Cancelled`) +//! to `record_timeout`, which opens the breaker ONLY when the peer shows no recent +//! proof of life. +//! +//! These tests use a REAL tonic `WalShipping` server over a REAL established HTTP/2 +//! connection (not a synthetic `tonic::Status`): the handler succeeds once (proving +//! liveness) then hangs past the client deadline, so the client observes a genuine +//! request timeout on a live connection — exactly the production shape. + +use std::{ + collections::HashMap, + net::SocketAddr, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use tidal_net::{ + circuit_breaker::BreakerState, + client::PeerPool, + config::GrpcTransportConfig, + error::GrpcTransportError, + proto::{ + AppliedReport, AppliedReportAck, HeartbeatRequest, HeartbeatResponse, JoinRequest, + JoinResponse, ShipSegmentRequest, ShipSegmentResponse, SnapshotChunk, SnapshotRequest, + StreamRequest, TimeoutNowRequest, TimeoutNowResponse, VoteRequest, VoteResponse, + wal_shipping_server::{WalShipping, WalShippingServer}, + }, +}; +use tidaldb::replication::{ + WalSegmentId, + shard::{RegionId, ShardId}, + transport::WalSegmentPayload, +}; +use tokio_stream::wrappers::ReceiverStream; +use tonic::{Request, Response, Status, transport::Server}; + +/// A real WAL-shipping server whose `ship_segment` handler succeeds for the first +/// `hang_after` calls (fast `accepted=true`, refreshing the leader's liveness stamp) +/// and then HANGS `hang_for` on every subsequent call — long enough to exceed the +/// client's `request_timeout`, so the leader observes a genuine deadline on a LIVE, +/// already-established connection. Every other RPC is unimplemented (the client only +/// ships in these tests). +struct FlakyShipServer { + calls: Arc, + hang_after: usize, + hang_for: Duration, +} + +#[tonic::async_trait] +impl WalShipping for FlakyShipServer { + async fn ship_segment( + &self, + _request: Request, + ) -> Result, Status> { + let n = self.calls.fetch_add(1, Ordering::SeqCst); + if n >= self.hang_after { + // Slow-but-ALIVE: the server is up and processing, just late — the + // client's request_timeout fires first and surfaces a real deadline. + tokio::time::sleep(self.hang_for).await; + } + Ok(Response::new(ShipSegmentResponse { + accepted: true, + ..Default::default() + })) + } + + type StreamSegmentsStream = ReceiverStream>; + async fn stream_segments( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("mock")) + } + + type FetchSnapshotStream = ReceiverStream>; + async fn fetch_snapshot( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("mock")) + } + + async fn heartbeat( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("mock")) + } + + async fn report_applied( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("mock")) + } + + async fn request_vote( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("mock")) + } + + async fn timeout_now( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("mock")) + } + + async fn join_cluster( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("mock")) + } +} + +fn free_addr() -> SocketAddr { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() +} + +fn make_payload(seqno: u64) -> WalSegmentPayload { + WalSegmentPayload { + id: WalSegmentId::new(RegionId::SINGLE, ShardId(0), seqno), + bytes: vec![0xCD; 50], + event_count: 1, + leader_last_seq: seqno, + stream_baseline: 0, + term: 0, + leader_region: 0, + } +} + +fn config_for(peer: SocketAddr, threshold: u32) -> GrpcTransportConfig { + let mut peers = HashMap::new(); + peers.insert(ShardId(0), peer.to_string()); + GrpcTransportConfig { + local_shard: ShardId(0), + listen_addr: free_addr(), + peers, + insecure: true, + circuit_breaker_threshold: threshold, + circuit_breaker_reset: Duration::from_secs(30), + // Short request deadline so a hung handler trips it fast; generous connect + // timeout so the first (fast) ship establishes the connection cleanly. + request_timeout: Duration::from_millis(500), + connect_timeout: Duration::from_secs(5), + ..Default::default() + } +} + +/// Spawn the flaky server and wait until it is actually accepting TCP. +async fn spawn_server(addr: SocketAddr, server: FlakyShipServer) { + tokio::spawn(async move { + let _ = Server::builder() + .add_service(WalShippingServer::new(server)) + .serve(addr) + .await; + }); + for _ in 0..50 { + if tokio::net::TcpStream::connect(addr).await.is_ok() { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("mock WAL server never started listening on {addr}"); +} + +fn assert_ship_timeout(result: &Result<(u64, u64), GrpcTransportError>) { + match result { + Err(GrpcTransportError::Grpc(status)) => assert!( + matches!( + status.code(), + tonic::Code::DeadlineExceeded | tonic::Code::Cancelled + ), + "an established-connection ship deadline must surface as DeadlineExceeded|Cancelled \ + (the codes routed to record_timeout), got {:?}: {status}", + status.code() + ), + other => panic!("expected a gRPC deadline status from a hung ship, got {other:?}"), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn recent_liveness_keeps_breaker_closed_under_real_ship_timeouts() { + // THE fix, end-to-end: a follower that proved liveness (ship #1 succeeded) then + // times out a burst of ships must NOT trip the breaker — a slow-but-alive peer + // is not a partition. + let addr = free_addr(); + let calls = Arc::new(AtomicUsize::new(0)); + spawn_server( + addr, + FlakyShipServer { + calls: calls.clone(), + hang_after: 1, // first ship fast; the rest hang + hang_for: Duration::from_secs(3), + }, + ) + .await; + + let pool = PeerPool::new(&config_for(addr, 3)).unwrap(); + + // Ship #1: real success over a freshly-established H2 connection → record_success. + let first = pool.send_to(ShardId(0), make_payload(1)).await; + assert!(first.is_ok(), "the first ship must succeed: {first:?}"); + assert!( + pool.peer_grpc_fresh(ShardId(0), Duration::from_secs(5)), + "an accepted ship must refresh the peer's liveness stamp" + ); + + // Ships #2..#6: each hangs past the 500ms deadline → REAL ship timeouts (far more + // than the threshold of 3) on the SAME live connection. + for seq in 2..=6 { + let r = pool.send_to(ShardId(0), make_payload(seq)).await; + assert_ship_timeout(&r); + } + + assert_eq!( + pool.breaker_state(ShardId(0)), + BreakerState::Closed, + "5 real ship timeouts from a peer with recent liveness must NOT open the breaker \ + (the write-burst false-partition fix)" + ); + // And replication is not wedged: the breaker still admits the next ship. + assert!( + pool.peer_grpc_fresh(ShardId(0), Duration::from_secs(60)), + "the slow-but-alive peer's liveness stamp is still fresh from ship #1" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn ship_timeouts_with_no_proof_of_life_still_open_breaker() { + // Genuine-blackhole detection preserved END-TO-END: a peer that only ever times + // out (never round-trips a reply) has no liveness stamp, so real ship timeouts + // still open the breaker at the threshold. + let addr = free_addr(); + let calls = Arc::new(AtomicUsize::new(0)); + spawn_server( + addr, + FlakyShipServer { + calls: calls.clone(), + hang_after: 0, // every ship hangs — no proof of life ever + hang_for: Duration::from_secs(3), + }, + ) + .await; + + let pool = PeerPool::new(&config_for(addr, 3)).unwrap(); + + for seq in 1..=3 { + let r = pool.send_to(ShardId(0), make_payload(seq)).await; + assert_ship_timeout(&r); + } + + assert_eq!( + pool.breaker_state(ShardId(0)), + BreakerState::Open, + "ship timeouts with no proof of life must still open the breaker (blackhole detection)" + ); +} diff --git a/tidal-stress/k8s/t5-readtput-job.yaml b/tidal-stress/k8s/t5-readtput-job.yaml index d2e1d26..c6e168b 100644 --- a/tidal-stress/k8s/t5-readtput-job.yaml +++ b/tidal-stress/k8s/t5-readtput-job.yaml @@ -2,6 +2,12 @@ # 3-node cluster sustains within SLA, spread round-robin across all 3 pods. # (T5-as-written's 2.5x WRITE-scaling is structurally impossible on 3-node RF3 # full placement — proven in docs/profiling/m12p4-t5-sharded-throughput.md.) +# +# THIS JOB IS THE Ref-A OWNER-TEST for the re-scoped G-S Scalability guarantee +# (read-throughput) — see docs/planning/milestone-11/guarantee-traceability.md +# (G-S row, re-scoped 2026-06-19). The structural basis (a node serves reads from +# its LOCAL placement, no cross-node forward → per-node capacity adds) is proven by +# the cargo e2e `mp_partial_placement_feed_spans_all_groups` (cluster_cross_shard_reads). apiVersion: batch/v1 kind: Job metadata: diff --git a/tidal-stress/src/bin/soak-eval.rs b/tidal-stress/src/bin/soak-eval.rs index fee2a0f..c4b3175 100644 --- a/tidal-stress/src/bin/soak-eval.rs +++ b/tidal-stress/src/bin/soak-eval.rs @@ -18,7 +18,10 @@ use tidal_stress::soak_eval::{ }; #[derive(Parser)] -#[command(version, about = "30-night soak streak evaluator (ledger + restarts → streak.tsv)")] +#[command( + version, + about = "30-night soak streak evaluator (ledger + restarts → streak.tsv)" +)] struct Cli { /// The durable result dir the nightly CronJob + monitor write to. #[arg(long, default_value = "/results")] @@ -62,7 +65,10 @@ fn main() -> ExitCode { // Exit non-zero iff the MOST RECENT night is non-green — the alert trigger. match verdicts.last() { Some(v) if !v.green => { - eprintln!("soak-eval: ALERT — last night {} is NON-GREEN: {}", v.date, v.reason); + eprintln!( + "soak-eval: ALERT — last night {} is NON-GREEN: {}", + v.date, v.reason + ); ExitCode::from(1) } _ => ExitCode::SUCCESS, diff --git a/tidal-stress/src/soak_eval.rs b/tidal-stress/src/soak_eval.rs index 3efa80b..13374f4 100644 --- a/tidal-stress/src/soak_eval.rs +++ b/tidal-stress/src/soak_eval.rs @@ -179,12 +179,11 @@ pub fn evaluate(nights: &[NightResult], samples: &[RestartSample]) -> Vec (false, String::from("soak SLO gate breach (Job FAIL)")), (true, true) => ( false, - String::from("under-load pod restart in window (Job passed but streak-breaking)"), - ), - (false, true) => ( - false, - String::from("soak FAIL and under-load pod restart"), + String::from( + "under-load pod restart in window (Job passed but streak-breaking)", + ), ), + (false, true) => (false, String::from("soak FAIL and under-load pod restart")), }; NightVerdict { date: n.date.clone(), @@ -210,10 +209,7 @@ pub fn render_streak_tsv(verdicts: &[NightVerdict], target: usize) -> String { let mut run = 0usize; for v in verdicts { run = if v.green { run + 1 } else { 0 }; - s.push_str(&format!( - "{}\t{}\t{}\t{}\n", - v.date, v.green, run, v.reason - )); + s.push_str(&format!("{}\t{}\t{}\t{}\n", v.date, v.green, run, v.reason)); } let streak = current_streak(verdicts); s.push_str(&format!("# streak={streak}/{target}\n")); @@ -271,10 +267,17 @@ mod tests { }); let verdicts = evaluate(&nights, &samples); let n15 = verdicts.iter().find(|v| v.date == "2026-06-15").unwrap(); - assert!(!n15.green, "a night with an under-load restart is NOT green"); + assert!( + !n15.green, + "a night with an under-load restart is NOT green" + ); assert!(n15.reason.contains("restart"), "reason names the restart"); // The streak is only the clean tail after night 15: nights 16..=30 = 15. - assert_eq!(current_streak(&verdicts), 15, "streak resets at the restart night"); + assert_eq!( + current_streak(&verdicts), + 15, + "streak resets at the restart night" + ); } #[test] @@ -286,7 +289,11 @@ mod tests { let n20 = verdicts.iter().find(|v| v.date == "2026-06-20").unwrap(); assert!(!n20.green, "a gate-breach night is not green"); assert!(n20.reason.contains("gate"), "reason names the gate breach"); - assert_eq!(current_streak(&verdicts), 10, "tail after night 20 = nights 21..=30"); + assert_eq!( + current_streak(&verdicts), + 10, + "tail after night 20 = nights 21..=30" + ); } #[test] @@ -295,11 +302,22 @@ mod tests { // count as a restart (min < max that night). let nights = pass_nights(1); let samples = vec![ - RestartSample { date: "2026-06-01".into(), pod: "tidaldb-0".into(), restarts: 0 }, - RestartSample { date: "2026-06-01".into(), pod: "tidaldb-0".into(), restarts: 1 }, + RestartSample { + date: "2026-06-01".into(), + pod: "tidaldb-0".into(), + restarts: 0, + }, + RestartSample { + date: "2026-06-01".into(), + pod: "tidaldb-0".into(), + restarts: 1, + }, ]; let verdicts = evaluate(&nights, &samples); - assert!(!verdicts[0].green, "0→1 within the first night is a restart"); + assert!( + !verdicts[0].green, + "0→1 within the first night is a restart" + ); } #[test] @@ -333,6 +351,9 @@ ts_utc\tpod\trestarts\tphase\tready\n\ let verdicts = evaluate(&nights, &samples); let tsv = render_streak_tsv(&verdicts, 30); assert!(tsv.contains("date\tgreen\tstreak\treason"), "has a header"); - assert!(tsv.trim_end().ends_with("# streak=3/30"), "summary line present: {tsv}"); + assert!( + tsv.trim_end().ends_with("# streak=3/30"), + "summary line present: {tsv}" + ); } } diff --git a/tidalctl/src/commands/s3.rs b/tidalctl/src/commands/s3.rs index e2445cc..4704c1f 100644 --- a/tidalctl/src/commands/s3.rs +++ b/tidalctl/src/commands/s3.rs @@ -30,9 +30,9 @@ use std::path::{Path, PathBuf}; use aws_credential_types::Credentials; +use aws_sdk_s3::Client; use aws_sdk_s3::config::{BehaviorVersion, Region}; use aws_sdk_s3::primitives::ByteStream; -use aws_sdk_s3::Client; use crate::CliError; @@ -143,8 +143,9 @@ async fn build_client(target: &S3Target) -> Result { AWS_SECRET_ACCESS_KEY to your R2 token credentials)", ) })?; - let secret_key = std::env::var("AWS_SECRET_ACCESS_KEY") - .map_err(|_| CliError::new("AWS_SECRET_ACCESS_KEY is not set (required for the S3/R2 export)"))?; + let secret_key = std::env::var("AWS_SECRET_ACCESS_KEY").map_err(|_| { + CliError::new("AWS_SECRET_ACCESS_KEY is not set (required for the S3/R2 export)") + })?; let creds = Credentials::from_keys(access_key, secret_key, None); @@ -431,8 +432,7 @@ mod tests { #[test] fn relative_from_key_inverts_object_key() { assert_eq!( - relative_from_key("backups/node-A", "backups/node-A/wal/seg-1.seg") - .as_deref(), + relative_from_key("backups/node-A", "backups/node-A/wal/seg-1.seg").as_deref(), Some("wal/seg-1.seg") ); // Empty prefix: the whole key is the relative path.