fix(net): classify ship deadline as timeout, not partition (write-burst false-partition)
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.
This commit is contained in:
parent
7cb9724911
commit
1b5bcbacd7
@ -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.
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@ -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)))
|
||||
}
|
||||
}
|
||||
|
||||
278
tidal-net/tests/ship_timeout_breaker.rs
Normal file
278
tidal-net/tests/ship_timeout_breaker.rs
Normal file
@ -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<AtomicUsize>,
|
||||
hang_after: usize,
|
||||
hang_for: Duration,
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl WalShipping for FlakyShipServer {
|
||||
async fn ship_segment(
|
||||
&self,
|
||||
_request: Request<ShipSegmentRequest>,
|
||||
) -> Result<Response<ShipSegmentResponse>, 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<Result<ShipSegmentRequest, Status>>;
|
||||
async fn stream_segments(
|
||||
&self,
|
||||
_request: Request<StreamRequest>,
|
||||
) -> Result<Response<Self::StreamSegmentsStream>, Status> {
|
||||
Err(Status::unimplemented("mock"))
|
||||
}
|
||||
|
||||
type FetchSnapshotStream = ReceiverStream<Result<SnapshotChunk, Status>>;
|
||||
async fn fetch_snapshot(
|
||||
&self,
|
||||
_request: Request<SnapshotRequest>,
|
||||
) -> Result<Response<Self::FetchSnapshotStream>, Status> {
|
||||
Err(Status::unimplemented("mock"))
|
||||
}
|
||||
|
||||
async fn heartbeat(
|
||||
&self,
|
||||
_request: Request<HeartbeatRequest>,
|
||||
) -> Result<Response<HeartbeatResponse>, Status> {
|
||||
Err(Status::unimplemented("mock"))
|
||||
}
|
||||
|
||||
async fn report_applied(
|
||||
&self,
|
||||
_request: Request<AppliedReport>,
|
||||
) -> Result<Response<AppliedReportAck>, Status> {
|
||||
Err(Status::unimplemented("mock"))
|
||||
}
|
||||
|
||||
async fn request_vote(
|
||||
&self,
|
||||
_request: Request<VoteRequest>,
|
||||
) -> Result<Response<VoteResponse>, Status> {
|
||||
Err(Status::unimplemented("mock"))
|
||||
}
|
||||
|
||||
async fn timeout_now(
|
||||
&self,
|
||||
_request: Request<TimeoutNowRequest>,
|
||||
) -> Result<Response<TimeoutNowResponse>, Status> {
|
||||
Err(Status::unimplemented("mock"))
|
||||
}
|
||||
|
||||
async fn join_cluster(
|
||||
&self,
|
||||
_request: Request<JoinRequest>,
|
||||
) -> Result<Response<JoinResponse>, 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)"
|
||||
);
|
||||
}
|
||||
@ -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:
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -179,12 +179,11 @@ pub fn evaluate(nights: &[NightResult], samples: &[RestartSample]) -> Vec<NightV
|
||||
(false, false) => (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)"),
|
||||
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"),
|
||||
),
|
||||
(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}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@ -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<Client, CliError> {
|
||||
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.
|
||||
|
||||
Loading…
Reference in New Issue
Block a user