From 005e292cbb25518bf38c62ad5f92505085706677 Mon Sep 17 00:00:00 2001 From: jx12n Date: Sat, 13 Jun 2026 12:28:04 -0600 Subject: [PATCH] fix(m11): review remediation + tidal-stress perf sweep + perf wave 2 Resolve all BLOCKER/CRITICAL/WARNING findings from the m11p7/p8 review: - tidalctl restore: safe_join path-traversal/Zip-Slip guard + fsync on write - corrupt-WAL checkpoint_seq guard; PITR archive-before-delete - cluster: x-tidal-relayed audit-dedup marker; forward_failures counts 5xx - mTLS/HTTP-TLS handshake hardening; accept-loop EMFILE backoff - per-principal rate-limit + node-token marker-pinning tests - self-heal tier-3 coverage; 5 router-auth tests tidal-stress: measurement-fidelity fixes (schedule-lag p99/max, exact feed-over-SLO verdict, shed annotation) + typed Body, workload.next 184ns->68ns, RoundRobin len==1 short-circuit, HeaderValue cache; new benches/hotpath.rs + lib.rs. perf wave 2: signal_snapshot SmallVec/SignalKey carrier; one-get-per-type ranking pre-pass. --- .woodpecker.yaml | 6 +- Cargo.lock | 1 + docs/planning/milestone-11/phase-8.md | 4 +- docs/runbooks/cluster.md | 28 ++- scripts/gen-cluster-certs.sh | 25 ++- tidal-net/src/client.rs | 2 +- tidal-net/src/config.rs | 48 +++++ tidal-net/src/lib.rs | 12 +- tidal-net/src/server.rs | 42 +++- tidal-net/src/tls.rs | 6 +- tidal-server/src/cluster/audit.rs | 43 ++-- tidal-server/src/cluster/forward.rs | 62 ++++-- tidal-server/src/cluster/http_tls.rs | 35 +++- tidal-server/src/cluster/node.rs | 271 ++++++++++++++++++++++---- tidal-server/src/cluster/security.rs | 54 ++++- tidal-server/tests/cluster_chaos.rs | 63 ++++++ tidal-stress/Cargo.toml | 10 + tidal-stress/benches/hotpath.rs | 82 ++++++++ tidal-stress/src/client.rs | 82 ++++++-- tidal-stress/src/error.rs | 21 +- tidal-stress/src/lib.rs | 15 ++ tidal-stress/src/main.rs | 29 ++- tidal-stress/src/metrics.rs | 67 ++++++- tidal-stress/src/scheduler.rs | 44 +++-- tidal-stress/src/workload.rs | 123 ++++++++++-- tidal/src/db/metrics/cluster.rs | 21 +- tidal/src/ranking/executor/scoring.rs | 6 + tidal/src/wal/compaction.rs | 21 +- tidalctl/src/commands/backup.rs | 166 +++++++++++++--- tidalctl/tests/cli.rs | 94 +++++++++ 30 files changed, 1265 insertions(+), 218 deletions(-) create mode 100644 tidal-stress/benches/hotpath.rs create mode 100644 tidal-stress/src/lib.rs diff --git a/.woodpecker.yaml b/.woodpecker.yaml index e72790f..e47ff51 100644 --- a/.woodpecker.yaml +++ b/.woodpecker.yaml @@ -30,9 +30,13 @@ steps: repo: tidal/server dockerfile: docker/standalone/Dockerfile context: . + # Tag by immutable identity only: `latest` plus the commit SHA (the durable + # handle). A per-milestone literal tag drifts every milestone and lies about + # the image's vintage (a hand-edited `m8p10` once tagged m11 code) — the SHA + # never goes stale and the version-skew machinery keys off the binary's + # CARGO_PKG_VERSION/BUILD_HASH, not the image tag. tags: - latest - - m8p10 - ${CI_COMMIT_SHA} registry: registry.threesix.ai build_args: diff --git a/Cargo.lock b/Cargo.lock index 848417f..a613e53 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3620,6 +3620,7 @@ name = "tidal-stress" version = "0.1.0" dependencies = [ "clap", + "criterion", "rand 0.9.2", "reqwest", "serde", diff --git a/docs/planning/milestone-11/phase-8.md b/docs/planning/milestone-11/phase-8.md index cf7820a..78f0549 100644 --- a/docs/planning/milestone-11/phase-8.md +++ b/docs/planning/milestone-11/phase-8.md @@ -123,7 +123,7 @@ verb still exists as an immediate nudge. Convergence transitions are counted `BACKUP_MANIFEST.json` (BLAKE3 per file + the recovered WAL checkpoint cursor). Restore verifies EVERY file's BLAKE3 against the manifest BEFORE writing anything, and refuses a non-empty target (the destructive-op guard). -- **Coordinated cluster backup** is the manifest + procedure (runbook §10): under +- **Coordinated cluster backup** is the manifest + procedure (runbook §13): under `ack=quorum`, any committed replica's data dir holds the quorum-durable log, so it is a cluster-consistent snapshot at its recorded `checkpoint_seq`. Back up one committed replica per shard group; restore re-seeds each group's leader and @@ -167,7 +167,7 @@ verb still exists as an immediate nudge. Convergence transitions are counted - [x] `tidalctl backup` / `restore` (BLAKE3 manifest, integrity-verified round-trip) - [x] Version handshake (heartbeat `build_version` + status `version`) + N/N+1 policy - [x] `mp_rolling_upgrade_no_loss_no_stall` promoted to a Woodpecker release gate -- [x] Docs (runbook §10/§11 + rolling upgrade, monitoring cluster metrics + alerts) + CHANGELOG +- [x] Docs (runbook §12 security / §13 backup+PITR / §14 rolling upgrade, monitoring cluster metrics + alerts) + CHANGELOG ## Exit-gate evidence (local; release builds where noted) diff --git a/docs/runbooks/cluster.md b/docs/runbooks/cluster.md index 3d7559c..7976db3 100644 --- a/docs/runbooks/cluster.md +++ b/docs/runbooks/cluster.md @@ -1186,8 +1186,11 @@ it all on. - `TIDAL_RATE_LIMIT_RPS` (+ optional `TIDAL_RATE_LIMIT_BURST`, default 2×) caps per-principal request rate; a deny is **429 + `Retry-After`**. Off by default. - Verified sibling nodes are EXEMPT — replication/forward traffic is never - throttled by the external-client budget. Today external callers share one - bucket (the shared bearer); a future multi-key registry gives per-key buckets. + throttled by the external-client budget. Today external callers share ONE + bucket: `TIDAL_RATE_LIMIT_RPS` is an AGGREGATE cap across all external clients, + not a per-client budget — set it to your total external ceiling, not a + per-client one (one noisy client can consume it). A future multi-key registry + adds per-key buckets. ### 12.6 Foreign-pod / negative behavior (what an attacker on the network sees) @@ -1208,8 +1211,8 @@ group is a **cluster-consistent** snapshot at its recorded `checkpoint_seq`. ### 13.1 Enable the WAL archive (point-in-time recovery) -Set `wal.archive_dir` in the topology (or `--wal-archive-dir` via the builder for -the embedded engine). Each sealed WAL segment is copied there — durably, before +Set `wal.archive_dir` in the topology (or `TidalDb::builder().wal_archive_dir(path)` +for the embedded engine). Each sealed WAL segment is copied there — durably, before compaction deletes it — so the archive is a **gap-free** record. Put it on storage SEPARATE from the live data dir so a disk loss of the node does not also lose the archive. Segment filenames encode `shard + first_seq`, so co-located groups share @@ -1240,10 +1243,19 @@ wal: 2. Point a stopped node at the restored dir and boot it. Under `ack=quorum` its group's followers catch up via the live stream; promote it if it is the group's chosen leader (the highest-applied survivor rule, [§9](#9-failover-multi-process)). -3. **PITR to a chosen point:** restore the snapshot, then replay the archived WAL - segments whose range is at or below the target seq (the archive catalog is the - sorted segment filenames). Replay stops at the target — events above it are not - applied. + +**Recovery granularity today is full-snapshot-to-frontier, NOT arbitrary +point-in-time.** A restore recovers a node to the backup's `checkpoint_seq`, then +the live stream reconverges it to the cluster's CURRENT frontier. The WAL archive +(`wal.archive_dir`) is the durable, gap-free **primitive** that *backs* a future +point-in-time replay — but it is write-only today: no tool replays archived +segments up to a chosen target seq. `tidalctl recover` is verify-only +(`--verify-only`; an in-place replay mode is reserved for a future release), and +`tidalctl restore` copies a full snapshot with no seq bound. So the inputs you +retain for a future PITR are the per-shard `checkpoint_seq` set + the archive; +treat them as the **window**, not a one-command restore-to-instant. Do not plan an +incident around seq-bounded replay until a `tidalctl replay --until ` verb +ships. Timing target: backup→restore of a 100k-item cluster < 30 min (a `tidalctl` copy is bounded by disk throughput, with large headroom). The Ref-A timed figure is a diff --git a/scripts/gen-cluster-certs.sh b/scripts/gen-cluster-certs.sh index af7a03f..4d4d3a4 100755 --- a/scripts/gen-cluster-certs.sh +++ b/scripts/gen-cluster-certs.sh @@ -30,6 +30,15 @@ N="${5:-3}" mkdir -p "$OUT_DIR" cd "$OUT_DIR" +# The CA ROOT private key is the cluster's crown jewel — anyone holding it can +# mint any node identity and forge the whole mTLS + node-token trust. Keep it in a +# SEPARATE, more-protected directory so it never lands in the secret-mount glob +# (the deployable Secret is only ca.crt / tls.crt / tls.key). Move ca-private/ to +# offline storage immediately after issuance. +CA_DIR="ca-private" +mkdir -p "$CA_DIR" +chmod 700 "$CA_DIR" + # Build the SAN list: every pod's stable headless-Service DNS + the headless and # client Services + loopback (so a local test cluster on 127.0.0.1 also verifies). SANS="DNS:${HEADLESS}.${NAMESPACE}.svc.cluster.local,DNS:${STS}.${NAMESPACE}.svc.cluster.local,DNS:localhost,IP:127.0.0.1" @@ -37,9 +46,9 @@ for i in $(seq 0 $((N - 1))); do SANS="${SANS},DNS:${STS}-${i}.${HEADLESS}.${NAMESPACE}.svc.cluster.local" done -echo "==> cluster CA" -openssl genrsa -out ca.key 4096 2>/dev/null -openssl req -x509 -new -nodes -key ca.key -sha256 -days 3650 \ +echo "==> cluster CA (private key -> ${CA_DIR}/ca.key, NOT a deployable file)" +openssl genrsa -out "${CA_DIR}/ca.key" 4096 2>/dev/null +openssl req -x509 -new -nodes -key "${CA_DIR}/ca.key" -sha256 -days 3650 \ -subj "/CN=tidaldb-cluster-ca" -out ca.crt echo "==> node leaf (SANs: ${SANS})" @@ -47,11 +56,13 @@ openssl genrsa -out tls.key 4096 2>/dev/null openssl req -new -key tls.key -subj "/CN=tidaldb-cluster" -out node.csr # serverAuth + clientAuth so the same leaf is the gRPC mTLS client identity AND # the gRPC/HTTP server identity. -openssl x509 -req -in node.csr -CA ca.crt -CAkey ca.key -CAcreateserial \ +openssl x509 -req -in node.csr -CA ca.crt -CAkey "${CA_DIR}/ca.key" -CAcreateserial \ -days 825 -sha256 -out tls.crt \ -extfile <(printf 'subjectAltName=%s\nextendedKeyUsage=serverAuth,clientAuth\n' "$SANS") -rm -f node.csr ca.srl +rm -f node.csr "${CA_DIR}/ca.srl" -chmod 600 ca.key tls.key -echo "==> wrote ${OUT_DIR}/{ca.crt,tls.crt,tls.key} (+ ca.key — keep offline)" +chmod 600 "${CA_DIR}/ca.key" tls.key +echo "==> wrote DEPLOYABLE ${OUT_DIR}/{ca.crt,tls.crt,tls.key}" +echo " CA ROOT KEY at ${OUT_DIR}/${CA_DIR}/ca.key — move it to OFFLINE storage now;" +echo " it must NEVER be copied to a pod or included in the cluster Secret." echo " grpc_tls: ca_cert=ca.crt server_cert=client_cert=tls.crt server_key=client_key=tls.key" diff --git a/tidal-net/src/client.rs b/tidal-net/src/client.rs index 3dfdfb9..155edd5 100644 --- a/tidal-net/src/client.rs +++ b/tidal-net/src/client.rs @@ -444,7 +444,7 @@ impl PeerPool { // every heartbeat at the transport boundary (no engine threading — all // workspace crates share one version). The peer observes the cluster's // version spread and warns on a major-version skew. - request.build_version = env!("CARGO_PKG_VERSION").to_owned(); + env!("CARGO_PKG_VERSION").clone_into(&mut request.build_version); let mut client = peer.client.clone(); client .heartbeat(request) diff --git a/tidal-net/src/config.rs b/tidal-net/src/config.rs index e17454c..bd52173 100644 --- a/tidal-net/src/config.rs +++ b/tidal-net/src/config.rs @@ -95,6 +95,19 @@ pub struct GrpcTransportConfig { /// — a mode inotify watchers routinely miss. Default 30s; tests set it low /// to exercise rotation-under-load fast. pub rotation_poll_interval: Duration, + /// Maximum wall-clock time a single INBOUND TLS handshake may take before it + /// is abandoned (m11p7 hardening). The TCP accept happens before any client + /// cert is verified, so a peer that completes the connect but stalls the TLS + /// `ClientHello` (a slowloris) would otherwise pin a task + socket + /// indefinitely, pre-auth. Default 10s; only used on the mTLS path. + pub handshake_timeout: Duration, + /// Maximum number of inbound TLS handshakes allowed in flight at once + /// (m11p7 hardening). Bounds the task/fd cost of a connection flood against + /// the replication port: when the limit is reached, a freshly-accepted + /// connection is dropped (load-shed) before its handshake is attempted, + /// rather than each spawning an unbounded task. Default 256; only the mTLS + /// path enforces it. + pub max_concurrent_handshakes: usize, } impl GrpcTransportConfig { @@ -176,6 +189,20 @@ impl GrpcTransportConfig { .into(), )); } + if self.handshake_timeout.is_zero() { + return Err(GrpcTransportError::Internal( + "handshake_timeout must be > 0 (a zero deadline rejects every \ + inbound TLS handshake instantly)" + .into(), + )); + } + if self.max_concurrent_handshakes == 0 { + return Err(GrpcTransportError::Internal( + "max_concurrent_handshakes must be > 0 (a zero bound load-sheds \ + every inbound connection, refusing all peers)" + .into(), + )); + } for (shard, addr) in &self.peers { validate_peer_addr(*shard, addr)?; } @@ -240,6 +267,8 @@ impl Default for GrpcTransportConfig { keep_alive_timeout: Duration::from_secs(5), catchup_retry_interval: Duration::from_secs(30), rotation_poll_interval: Duration::from_secs(30), + handshake_timeout: Duration::from_secs(10), + max_concurrent_handshakes: 256, } } } @@ -304,6 +333,25 @@ mod tests { } } + #[test] + fn zero_handshake_bounds_are_rejected() { + // A zero handshake timeout rejects every inbound TLS handshake instantly; + // a zero concurrency bound load-sheds every connection — both refuse all + // peers, so the typed config check must catch them. + let mutators: [fn(&mut GrpcTransportConfig); 2] = [ + |c| c.handshake_timeout = Duration::ZERO, + |c| c.max_concurrent_handshakes = 0, + ]; + for mutate in mutators { + let mut cfg = GrpcTransportConfig::default(); + mutate(&mut cfg); + assert!( + cfg.validate().is_err(), + "a zero handshake bound must be rejected" + ); + } + } + #[test] fn zero_circuit_breaker_reset_is_rejected() { let cfg = GrpcTransportConfig { diff --git a/tidal-net/src/lib.rs b/tidal-net/src/lib.rs index 7a5246b..2353a2c 100644 --- a/tidal-net/src/lib.rs +++ b/tidal-net/src/lib.rs @@ -45,11 +45,13 @@ pub use sources::{ SnapshotRequiredSink, SnapshotSource, SnapshotStageError, SnapshotStaging, }; pub use transport::{ElectionNet, ElectionNetEvent, GrpcTransport, GrpcTransportFactory}; -// m11p7: the inter-node HTTP listener (in tidal-server) reuses these to serve -// TLS over the SAME hot-swappable cert as the gRPC plane, so one rotation covers -// both. `build_http_server_config` is server-auth only (external bearer clients -// share this listener); per-node HTTP identity is the signed token, not a client -// cert. +// m11p7: the inter-node HTTP listener (in tidal-server) reuses these to serve TLS +// over the SAME cert FILES the gRPC plane uses — via its OWN `DynamicCertResolver` +// + reload poller, not a shared resolver instance. A single secret rotation is +// therefore picked up by both planes independently (each within one poll interval +// of the change). `build_http_server_config` is server-auth only (external bearer +// clients share this listener); per-node HTTP identity is the signed token, not a +// client cert. pub use tls::{ DynamicCertResolver, ServerCertReloader, build_http_server_config, load_certified_key, }; diff --git a/tidal-net/src/server.rs b/tidal-net/src/server.rs index 759d3fb..7dbf811 100644 --- a/tidal-net/src/server.rs +++ b/tidal-net/src/server.rs @@ -988,6 +988,8 @@ pub(crate) fn start_server( wal_service, shutdown, addr, + config.handshake_timeout, + config.max_concurrent_handshakes, )); return Ok(handle); } @@ -1042,6 +1044,8 @@ async fn serve_mtls( service: WalShippingServer, shutdown: Arc, addr: std::net::SocketAddr, + handshake_timeout: std::time::Duration, + max_concurrent_handshakes: usize, ) -> Result<(), tonic::transport::Error> { let listener = match tokio::net::TcpListener::from_std(std_listener) { Ok(listener) => listener, @@ -1064,6 +1068,9 @@ async fn serve_mtls( >(128); let accept_shutdown = Arc::clone(&shutdown); + // Bound the number of in-flight handshakes so a connection flood against the + // (pre-auth) accept path cannot spawn unbounded tasks / exhaust fds. + let handshake_limiter = Arc::new(tokio::sync::Semaphore::new(max_concurrent_handshakes)); let accept_loop = tokio::spawn(async move { loop { let accepted = tokio::select! { @@ -1074,24 +1081,40 @@ async fn serve_mtls( let (tcp, peer) = match accepted { Ok(pair) => pair, Err(e) => { - // A transient accept error (fd limit, reset during accept): - // log and keep accepting. A persistent one floods debug logs, - // which is the visibility a wedged listener deserves. - tracing::debug!(%addr, error = %e, "mTLS accept error; continuing"); + // A transient accept error (a reset during accept) is fine to + // retry immediately. A PERSISTENT one — most importantly + // EMFILE/ENFILE (fd-table exhaustion) — returns Err without + // consuming the pending connection, so retrying with no pause + // pegs a core. A brief backoff caps the spin while preserving + // liveness: the loop resumes the instant fds free up. (tonic's + // own AddrIncoming sleeps on accept errors for the same reason; + // the hand-rolled mTLS loop must do it explicitly.) + tracing::debug!(%addr, error = %e, "mTLS accept error; backing off"); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; continue; } }; + // Load-shed under a handshake flood: if the limiter is saturated, drop + // this connection now rather than queue an unbounded task behind it. + let Ok(permit) = Arc::clone(&handshake_limiter).try_acquire_owned() else { + tracing::debug!(%peer, "inbound handshake limiter saturated; dropping connection"); + drop(tcp); + continue; + }; let acceptor = acceptor.clone(); let conn_tx = conn_tx.clone(); // Handshake off the accept path so one slow/foreign handshake cannot // head-of-line block the next connection. tokio::spawn(async move { - match acceptor.accept(tcp).await { - Ok(stream) => { + // Held for the whole handshake; released (load capacity returned) + // the instant this task ends, success or failure. + let _permit = permit; + match tokio::time::timeout(handshake_timeout, acceptor.accept(tcp)).await { + Ok(Ok(stream)) => { // Channel closed = serve loop gone; drop the stream. let _ = conn_tx.send(Ok(stream)).await; } - Err(e) => { + Ok(Err(e)) => { // Foreign pod (no/invalid client cert), a cert from // another CA, or a non-TLS probe: the handshake fails // HERE and the connection NEVER reaches tonic — the @@ -1103,6 +1126,11 @@ async fn serve_mtls( "rejected inbound gRPC handshake (no/invalid client cert or non-TLS probe)" ); } + Err(_elapsed) => { + // A peer that connected but stalled the ClientHello + // (slowloris): abandon it so it cannot pin a task/socket. + tracing::debug!(%peer, "inbound gRPC handshake timed out; dropping connection"); + } } }); } diff --git a/tidal-net/src/tls.rs b/tidal-net/src/tls.rs index fb38602..eede874 100644 --- a/tidal-net/src/tls.rs +++ b/tidal-net/src/tls.rs @@ -41,6 +41,10 @@ use crate::{config::TlsConfig, error::GrpcTransportError}; /// same or h2 negotiation fails and every gRPC handshake is rejected. const ALPN_H2: &[u8] = b"h2"; +/// ALPN protocol identifier for HTTP/1.1 (the HTTP plane also serves external +/// bearer clients that may speak h1). +const ALPN_HTTP11: &[u8] = b"http/1.1"; + /// Build a tonic [`ClientTlsConfig`] for outbound peer channels. /// /// Always pins the cluster CA. Attaches this node's client identity (for mutual @@ -216,7 +220,7 @@ pub fn build_http_server_config(resolver: Arc) -> Arc>, + /// itself visible. Behind an [`Arc`] so the append can be moved onto a + /// blocking pool thread (see [`AuditSink::record`]) instead of blocking the + /// async reactor worker the admin handler runs on. + file: Option>>, } impl AuditSink { @@ -57,7 +59,7 @@ impl AuditSink { { Ok(f) => { tracing::info!(path = %p.display(), "admin audit log opened (append JSONL)"); - Some(Mutex::new(f)) + Some(Arc::new(Mutex::new(f))) } Err(e) => { tracing::error!(path = %p.display(), error = %e, @@ -102,14 +104,31 @@ impl AuditSink { "target": target, "term": term, "outcome": outcome, - }); - let mut guard = file - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - // A trailing newline makes it line-delimited JSON; a write failure is - // logged (not fatal — the verb already ran). - if let Err(e) = writeln!(guard, "{line}") { - tracing::error!(error = %e, "admin audit file write failed"); + }) + .to_string(); + let file = Arc::clone(file); + // The append is a blocking write(2) under a std Mutex. Admin handlers + // run on the async reactor, so doing it inline would block a reactor + // worker (and serialize concurrent verbs on the lock) — exactly when + // it matters least: an operator healing/promoting while the audit + // volume is degraded. Move it to the blocking pool. The always-on + // `tidal_audit` tracing event above is the in-band record, so the file + // append lagging never loses a record. A trailing newline makes it + // line-delimited JSON; a write failure is logged (not fatal). + let append = move || { + let mut guard = file + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Err(e) = writeln!(guard, "{line}") { + tracing::error!(error = %e, "admin audit file write failed"); + } + }; + match tokio::runtime::Handle::try_current() { + Ok(_) => { + tokio::task::spawn_blocking(append); + } + // Not on a runtime (a unit test, or a sync caller): write inline. + Err(_) => append(), } } } diff --git a/tidal-server/src/cluster/forward.rs b/tidal-server/src/cluster/forward.rs index 9f0f327..aace55a 100644 --- a/tidal-server/src/cluster/forward.rs +++ b/tidal-server/src/cluster/forward.rs @@ -73,6 +73,21 @@ pub const INTERNAL_MARKER: &str = "x-tidal-internal"; /// The marker value siblings set. pub const INTERNAL_MARKER_VALUE: &str = "1"; +/// Header name for the relayed-operator-hop marker (m11p7 audit dedup). +/// +/// Distinct from [`INTERNAL_MARKER`]: a request carrying `x-tidal-relayed: 1` is +/// an operator action that one node RELAYED to another (e.g. a `/cluster/promote` +/// forwarded from the node the operator hit to the target/leader that performs the +/// fenced transfer). The receiver runs the FULL protocol (it is NOT the legacy +/// term-0 fan-out the internal marker triggers), but it must NOT re-audit the verb +/// — the operator's entry node already recorded it once. Like the internal marker, +/// it is honored only from a verified sibling (it is pinned to a node token when a +/// cluster key is configured), so a forged header cannot suppress an audit record. +pub const RELAY_MARKER: &str = "x-tidal-relayed"; + +/// The value set on [`RELAY_MARKER`]. +pub const RELAY_MARKER_VALUE: &str = "1"; + /// Request header overriding the deployment's write-acknowledgment mode /// (m11p3): `leader` or `quorum`. Forwarded verbatim with the write so the /// leader honors the CALLER's choice, not the gateway's default. @@ -167,6 +182,19 @@ pub fn is_internal(headers: &HeaderMap) -> bool { .is_some_and(|v| v == INTERNAL_MARKER_VALUE) } +/// True iff the request carries the relayed-operator-hop marker ([`RELAY_MARKER`]). +/// +/// A handler that sees this runs the full protocol but does NOT re-audit the verb +/// (the operator's entry node already recorded it). The value must be exactly +/// `"1"`; any other value is treated as absent. +#[must_use] +pub fn is_relayed(headers: &HeaderMap) -> bool { + headers + .get(RELAY_MARKER) + .and_then(|v| v.to_str().ok()) + .is_some_and(|v| v == RELAY_MARKER_VALUE) +} + /// Extract the verbatim `Authorization` header value, if present, so a forward /// can pass the caller's bearer token straight through to the peer. #[must_use] @@ -234,11 +262,15 @@ pub fn ack_passthrough(headers: &HeaderMap) -> Vec<(&'static str, String)> { out } -/// Forward a JSON request to one peer and relay its status + body back. +/// Forward a JSON request to one peer and relay its status + body back, with +/// extra request headers passed through verbatim. /// /// `auth` is the caller's verbatim `Authorization` header (passed through so the /// peer's bearer middleware accepts the forward); `internal` sets the -/// propagation marker so the peer applies the op locally and never re-forwards. +/// propagation marker so the peer applies the op locally and never re-forwards; +/// `passthrough` carries any additional headers the caller must relay (m11p3: the +/// `x-tidal-ack` override; m11p7: the node token + relay marker). Pass `&[]` when +/// none are needed. /// /// A connect/timeout/transport failure is returned as `Err(connect_error_string)` /// so the caller can degrade it into a 503 naming the unreachable peer — the @@ -248,18 +280,6 @@ pub fn ack_passthrough(headers: &HeaderMap) -> Vec<(&'static str, String)> { /// /// Returns the stringified transport error when the peer cannot be reached or /// the exchange fails before a status is received. -pub async fn forward_json( - client: &reqwest::Client, - url: &str, - body: &B, - auth: Option<&str>, - internal: bool, -) -> Result { - forward_json_with_headers(client, url, body, auth, internal, &[]).await -} - -/// [`forward_json`] with extra request headers passed through verbatim -/// (m11p3: the caller's `x-tidal-ack` override must reach the leader). pub async fn forward_json_with_headers( client: &reqwest::Client, url: &str, @@ -437,6 +457,20 @@ mod tests { assert!(!is_internal(&headers), "only the literal 1 counts"); } + #[test] + fn relay_marker_detected_only_for_exact_value() { + let mut headers = HeaderMap::new(); + assert!(!is_relayed(&headers), "absent marker is not relayed"); + headers.insert(RELAY_MARKER, HeaderValue::from_static("1")); + assert!(is_relayed(&headers), "x-tidal-relayed: 1 is relayed"); + headers.insert(RELAY_MARKER, HeaderValue::from_static("0")); + assert!(!is_relayed(&headers), "any other value is not relayed"); + // The relay marker is distinct from the internal marker. + let mut only_relay = HeaderMap::new(); + only_relay.insert(RELAY_MARKER, HeaderValue::from_static("1")); + assert!(is_relayed(&only_relay) && !is_internal(&only_relay)); + } + #[test] fn forwarded_auth_passes_through_verbatim() { let mut headers = HeaderMap::new(); diff --git a/tidal-server/src/cluster/http_tls.rs b/tidal-server/src/cluster/http_tls.rs index bf60d2f..ac1f843 100644 --- a/tidal-server/src/cluster/http_tls.rs +++ b/tidal-server/src/cluster/http_tls.rs @@ -21,12 +21,23 @@ use std::time::Duration; use tidal_net::{DynamicCertResolver, TlsConfig}; use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::Semaphore; use tokio_rustls::TlsAcceptor; use tokio_rustls::rustls::ServerConfig; use tokio_rustls::server::TlsStream; use crate::error::{Result, ServerError}; +/// Maximum wall-clock time a single inbound TLS handshake may take before it is +/// abandoned. The TCP accept precedes any auth, so a peer that stalls the +/// `ClientHello` (slowloris) would otherwise pin a task + socket indefinitely. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Maximum inbound TLS handshakes allowed in flight at once. Bounds the task/fd +/// cost of a connection flood against the HTTP control plane: excess connections +/// are load-shed before their handshake is attempted. +const MAX_CONCURRENT_HANDSHAKES: usize = 256; + /// The TLS material for the inter-node HTTP listener: the rustls server config /// to serve with, plus the hot-swappable resolver + cert file paths the rotation /// poller re-reads. @@ -86,6 +97,9 @@ impl TlsListener { let local_addr = listener.local_addr().map_err(ServerError::Network)?; let acceptor = TlsAcceptor::from(config); let (tx, rx) = tokio::sync::mpsc::channel(128); + // Bound in-flight handshakes so a connection flood against the (pre-auth) + // accept path cannot spawn unbounded tasks / exhaust fds. + let handshake_limiter = Arc::new(Semaphore::new(MAX_CONCURRENT_HANDSHAKES)); tokio::spawn(async move { loop { let (tcp, peer) = match listener.accept().await { @@ -94,25 +108,38 @@ impl TlsListener { // A transient accept error (fd pressure): brief backoff so // we never hot-loop, then keep accepting. tracing::debug!(error = %e, "HTTP TLS accept error; continuing"); - tokio::time::sleep(Duration::from_millis(5)).await; + tokio::time::sleep(Duration::from_millis(50)).await; continue; } }; + // Load-shed under a handshake flood rather than queue an unbounded + // task behind a saturated limiter. + let Ok(permit) = Arc::clone(&handshake_limiter).try_acquire_owned() else { + tracing::debug!(%peer, "HTTP TLS handshake limiter saturated; dropping connection"); + drop(tcp); + continue; + }; let acceptor = acceptor.clone(); let tx = tx.clone(); tokio::spawn(async move { - match acceptor.accept(tcp).await { - Ok(stream) => { + // Held for the handshake; released the instant this task ends. + let _permit = permit; + match tokio::time::timeout(HANDSHAKE_TIMEOUT, acceptor.accept(tcp)).await { + Ok(Ok(stream)) => { // Send failure = the listener was dropped (server // draining); drop the connection. let _ = tx.send((stream, peer)).await; } - Err(e) => { + Ok(Err(e)) => { // A foreign client with no/invalid cert chain, or a // plaintext probe against the TLS port: rejected at the // handshake, never reaches axum. tracing::debug!(%peer, error = %e, "HTTP TLS handshake rejected"); } + Err(_elapsed) => { + // A peer that connected but stalled the ClientHello. + tracing::debug!(%peer, "HTTP TLS handshake timed out; dropping connection"); + } } }); } diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 6337a17..a10e984 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -88,8 +88,8 @@ use utoipa::ToSchema; use super::{ forward::{ - self, broadcast_marked, forward_json, forward_json_with_headers, forwarded_auth, - is_internal, peer_url, + self, broadcast_marked, forward_json_with_headers, forwarded_auth, is_internal, is_relayed, + peer_url, }, reseed, routes::{ClusterAppError, ScatterGatherInfo, ShardedFeedResponse, ShardedSearchResponse}, @@ -1168,6 +1168,23 @@ impl ShardReplica { .unwrap_or_default() } + /// Forward headers for a RELAYED operator hop (m11p7): the node token (so the + /// receiving sibling honors the marker) plus the [`RELAY_MARKER`] that tells + /// the receiver to run the full protocol but NOT re-audit the verb — the + /// operator's entry node already recorded it once. Used by `/cluster/promote` + /// when a node forwards an operator promote to the target/leader that performs + /// the fenced transfer. + /// + /// [`RELAY_MARKER`]: crate::cluster::forward::RELAY_MARKER + fn relay_passthrough(&self) -> Vec<(&'static str, String)> { + let mut headers = self.node_token_passthrough(); + headers.push(( + super::forward::RELAY_MARKER, + super::forward::RELAY_MARKER_VALUE.to_string(), + )); + headers + } + /// Every region (self + peers) as `(RegionId, name, http_addr_or_none)`, in /// id order, for status aggregation. Self's `http_addr` is `None` (served /// in-process). A peer with no declared `http_addr` is `None` too @@ -2247,7 +2264,11 @@ impl ShardReplica { /// required for convergence. pub(crate) fn tick_self_heal(&self) { // Coarse cadence: act ~every SELF_HEAL_TICKS election ticks. - if self.heal_tick.fetch_add(1, Ordering::Relaxed) % SELF_HEAL_TICKS != 0 { + if !self + .heal_tick + .fetch_add(1, Ordering::Relaxed) + .is_multiple_of(SELF_HEAL_TICKS) + { return; } if !self.is_leader() { @@ -4029,40 +4050,12 @@ pub fn build_region_router( )) .with_state(node); - // m11p7 cluster auth, in one layer so `next` runs at most once: - // 1. Bearer gate, read PER REQUEST from `creds` (rotatable; open when unset). - // 2. Marker-pinning: when a cluster key is configured, a request that sets - // the `x-tidal-internal` marker WITHOUT a valid node token is rejected - // (403). The marker stays a routing hint — only a verified sibling may - // set it — so it can never be a standalone authorization bypass. + // m11p7 cluster auth, in one layer so `next` runs at most once. Extracted to + // `cluster_auth_middleware` so the assembled-router composition (bearer -> + // marker-pinning -> rate limit) is exercised by tests, not just the pure + // `ClusterCreds` helpers. let protected = protected.layer(middleware::from_fn(move |req: Request, next: Next| { - let creds = Arc::clone(&creds); - async move { - if let Some(key) = creds.bearer() - && !crate::router::bearer_token_ok(req.headers(), &key) - { - return crate::router::unauthorized_response(); - } - let marked = super::forward::is_internal(req.headers()); - if creds.marker_without_node_identity(req.headers(), marked) { - return ( - StatusCode::FORBIDDEN, - Json(serde_json::json!({ - "error": "x-tidal-internal requires a valid x-tidal-node-token; \ - the marker is honored only from a verified cluster sibling" - })), - ) - .into_response(); - } - // m11p7 per-principal rate limit: verified sibling NODES are exempt - // (replication/forwards must not be throttled); external principals - // consume their bucket. - let principal = creds.principal(req.headers()); - if let Err((retry_after_ms, limit)) = creds.check_rate(&principal) { - return crate::router::too_many_requests(retry_after_ms, limit); - } - next.run(req).await - } + cluster_auth_middleware(Arc::clone(&creds), req, next) })); let protected = protected.layer( @@ -4080,6 +4073,50 @@ pub fn build_region_router( crate::router::with_request_id_tracing(public.merge(protected)) } +/// The m11p7 cluster-auth middleware, applied to every protected region route in +/// one layer so `next` runs at most once. Three gates in order: +/// +/// 1. **Bearer**, read PER REQUEST from `creds` (rotatable; open when unset). +/// 2. **Marker-pinning**: when a cluster key is configured, a request that sets a +/// sibling-only marker (`x-tidal-internal` or `x-tidal-relayed`) WITHOUT a +/// valid node token is rejected (403). The markers stay routing/audit hints — +/// only a verified sibling may set them — so neither is a standalone bypass. +/// 3. **Per-principal rate limit**: verified sibling NODES are exempt (replication +/// must not be throttled); external principals consume their bucket → 429. +/// +/// Extracted from [`build_region_router`] so this composition is testable through +/// the real layer (a layer-order regression would otherwise compile and pass the +/// pure-`ClusterCreds` unit tests while reopening the marker bypass). +async fn cluster_auth_middleware( + creds: Arc, + req: Request, + next: Next, +) -> Response { + if let Some(key) = creds.bearer() + && !crate::router::bearer_token_ok(req.headers(), &key) + { + return crate::router::unauthorized_response(); + } + let marked = + super::forward::is_internal(req.headers()) || super::forward::is_relayed(req.headers()); + if creds.marker_without_node_identity(req.headers(), marked) { + return ( + StatusCode::FORBIDDEN, + Json(serde_json::json!({ + "error": "x-tidal-internal / x-tidal-relayed require a valid \ + x-tidal-node-token; these markers are honored only from a \ + verified cluster sibling" + })), + ) + .into_response(); + } + let principal = creds.principal(req.headers()); + if let Err((retry_after_ms, limit)) = creds.check_rate(&principal) { + return crate::router::too_many_requests(retry_after_ms, limit); + } + next.run(req).await +} + // ── Health ────────────────────────────────────────────────────────────────── #[allow(clippy::significant_drop_tightening)] @@ -4677,10 +4714,19 @@ pub async fn cluster_promote( let url = peer_url(&addr, "/cluster/promote"); let auth = forwarded_auth(&headers); let body = serde_json::json!({ "region": req.region }); - forward_json(&state.client, &url, &body, auth.as_deref(), false) - .await - .map(|resp| resp.status.is_success()) - .unwrap_or(false) + // A relayed operator hop: the leader runs the full fenced + // transfer but does NOT re-audit (this node audits once below). + forward_json_with_headers( + &state.client, + &url, + &body, + auth.as_deref(), + false, + &state.relay_passthrough(), + ) + .await + .map(|resp| resp.status.is_success()) + .unwrap_or(false) } else { false }; @@ -4699,8 +4745,17 @@ pub async fn cluster_promote( let url = peer_url(&addr, "/cluster/promote"); let auth = forwarded_auth(&headers); let body = serde_json::json!({ "region": req.region }); - if let Err(e) = - forward_json(&state.client, &url, &body, auth.as_deref(), false).await + // A relayed operator hop: the target runs the full protocol + // (sanction-through-leader or campaign) but does NOT re-audit. + if let Err(e) = forward_json_with_headers( + &state.client, + &url, + &body, + auth.as_deref(), + false, + &state.relay_passthrough(), + ) + .await { return Err(ClusterAppError(ServerError::RegionUnreachable { region: req.region, @@ -4832,7 +4887,14 @@ pub async fn cluster_promote( // `ClusterAppError` is a newtype over the `Display` `ServerError`. Err(e) => format!("error: {}", e.0), }; - node.audit_admin_outcome(&headers, "promote", &target_name, term, &outcome_str); + // Audit EXACTLY ONCE, at the node the operator's request first hit. The marked + // fan-out leg already returned above (never reaches here). A RELAYED hop (a + // sanction/forward to the target or leader, `x-tidal-relayed`) ran the full + // fenced protocol but must NOT re-audit — the operator's entry node records + // it, with the operator's principal, not the relaying node's. + if !is_relayed(&headers) { + node.audit_admin_outcome(&headers, "promote", &target_name, term, &outcome_str); + } result } @@ -6531,3 +6593,126 @@ where { offload_read(f).await.map_err(ClusterAppError) } + +/// m11p7 — the cluster auth middleware exercised through the real layer +/// composition (`middleware::from_fn`), not just the pure `ClusterCreds` +/// helpers. A layer-order or marker-set regression in [`cluster_auth_middleware`] +/// is caught HERE (the helper unit tests in `security.rs` would still pass). +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod auth_middleware_tests { + use super::*; + use crate::cluster::forward::{INTERNAL_MARKER, RELAY_MARKER}; + use crate::cluster::security::{ClusterCreds, NODE_TOKEN_HEADER}; + use axum::body::Body; + use axum::routing::get; + use tower::ServiceExt; + + fn app(creds: Arc) -> Router { + Router::new() + .route("/cluster/promote", get(|| async { StatusCode::OK })) + .layer(middleware::from_fn(move |req: Request, next: Next| { + cluster_auth_middleware(Arc::clone(&creds), req, next) + })) + } + + fn req(headers: &[(&str, String)]) -> Request { + let mut b = Request::builder().method("GET").uri("/cluster/promote"); + for (k, v) in headers { + b = b.header(*k, v.clone()); + } + b.body(Body::empty()).unwrap() + } + + #[tokio::test] + async fn sibling_markers_without_token_are_403_when_key_configured() { + // Both the internal-propagation and relayed-operator-hop markers are + // pinned: a request that sets one WITHOUT a valid node token is rejected. + for marker in [INTERNAL_MARKER, RELAY_MARKER] { + let creds = Arc::new(ClusterCreds::with_keys(None, Some("cluster-secret"))); + let resp = app(creds) + .oneshot(req(&[(marker, "1".to_string())])) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::FORBIDDEN, + "marker {marker} without a node token must be 403" + ); + } + } + + #[tokio::test] + async fn sibling_markers_with_valid_token_pass() { + let creds = Arc::new(ClusterCreds::with_keys(None, Some("cluster-secret"))); + let token = creds + .mint_node_token("region-a") + .expect("token mints with a key"); + for marker in [INTERNAL_MARKER, RELAY_MARKER] { + let resp = app(Arc::clone(&creds)) + .oneshot(req(&[ + (marker, "1".to_string()), + (NODE_TOKEN_HEADER, token.clone()), + ])) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::OK, + "marker {marker} with a valid node token is honored" + ); + } + } + + #[tokio::test] + async fn markers_are_hint_only_without_a_cluster_key() { + // No cluster key ⇒ pre-m11p7 hint-only behavior (no 403), backward compat. + let creds = Arc::new(ClusterCreds::unauthenticated()); + let resp = app(creds) + .oneshot(req(&[(RELAY_MARKER, "1".to_string())])) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + } + + #[tokio::test] + async fn external_over_budget_is_429_with_retry_after() { + // 1 rps / burst 1: the first external request passes, the second (issued + // immediately, before any refill) is denied with a Retry-After header. + let creds = Arc::new(ClusterCreds::with_rate_limit(1.0, 1.0)); + let app = app(creds); + let first = app.clone().oneshot(req(&[])).await.unwrap(); + assert_eq!(first.status(), StatusCode::OK); + let second = app.oneshot(req(&[])).await.unwrap(); + assert_eq!(second.status(), StatusCode::TOO_MANY_REQUESTS); + assert!( + second.headers().get("retry-after").is_some(), + "a 429 must carry Retry-After" + ); + } + + #[tokio::test] + async fn verified_node_is_exempt_from_the_rate_limit() { + // A cluster key (mint/verify tokens) AND a 1-rps bucket: a verified sibling + // NODE is never throttled, even well past the external budget. + let creds = Arc::new(ClusterCreds::with_cluster_key_and_rate_limit( + "cluster-secret", + 1.0, + 1.0, + )); + let token = creds.mint_node_token("region-a").expect("token"); + let app = app(creds); + for i in 0..5 { + let resp = app + .clone() + .oneshot(req(&[(NODE_TOKEN_HEADER, token.clone())])) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::OK, + "request {i}: a verified node is exempt from the external budget" + ); + } + } +} diff --git a/tidal-server/src/cluster/security.rs b/tidal-server/src/cluster/security.rs index adc6558..7d55a6b 100644 --- a/tidal-server/src/cluster/security.rs +++ b/tidal-server/src/cluster/security.rs @@ -197,6 +197,24 @@ impl ClusterCreds { } } + /// Build test creds with BOTH a cluster key (so node tokens mint/verify) and + /// an explicit per-principal rate limit — the combination needed to exercise + /// the node-exemption path (a verified sibling is never throttled) through the + /// assembled auth middleware. No bearer (auth-open); supply one separately if + /// the bearer gate is also under test. + #[must_use] + pub fn with_cluster_key_and_rate_limit(cluster_key: &str, rps: f64, burst: f64) -> Self { + Self { + bearer: ArcSwapOption::from(None), + bearer_file: None, + cluster_key: ArcSwapOption::from(Some(Arc::new(derive_key(cluster_key.as_bytes())))), + cluster_key_file: None, + rate_limiter: tidaldb::RateLimiter::new(tidaldb::RateLimiterConfig::limited( + rps, burst, + )), + } + } + /// Check the per-principal HTTP rate limit (m11p7). Verified sibling NODES /// are EXEMPT (inter-node replication/forward traffic must never be throttled /// by the external-client budget). External principals consume one token from @@ -295,16 +313,18 @@ impl ClusterCreds { .map_or(Principal::External, Principal::Node) } - /// Whether this request must be rejected because it claims the internal + /// Whether this request must be rejected because it claims a sibling-only /// marker WITHOUT proving sibling identity (m11p7 marker-pinning). /// - /// True iff a cluster key is configured AND the request carries the - /// `x-tidal-internal` marker AND it does NOT carry a valid node token. When - /// no cluster key is configured this is always `false` (the marker keeps its - /// pre-m11p7 hint-only behavior — backward compatible). + /// `marked` is set when the request carries the `x-tidal-internal` + /// propagation marker OR the `x-tidal-relayed` audit-dedup marker — both are + /// sibling-only signals. True iff a cluster key is configured AND `marked` + /// AND the request does NOT carry a valid node token. When no cluster key is + /// configured this is always `false` (the markers keep their pre-m11p7 + /// hint-only behavior — backward compatible). #[must_use] - pub fn marker_without_node_identity(&self, headers: &HeaderMap, marked_internal: bool) -> bool { - if !marked_internal || !self.cluster_key_enabled() { + pub fn marker_without_node_identity(&self, headers: &HeaderMap, marked: bool) -> bool { + if !marked || !self.cluster_key_enabled() { return false; } let verified = headers @@ -373,12 +393,24 @@ fn read_file_secret(path: &std::path::Path) -> Option { } } +/// Read an inline secret from `env`, trimming trailing whitespace/newline the +/// SAME way [`read_file_secret`] does so the env and `*_FILE` forms normalize +/// identically — an operator moving a secret between an inline var and a file +/// mount gets the same key, and an all-whitespace value is treated as unset (not +/// a one-space key). +fn read_env_secret(env: &str) -> Option { + std::env::var(env).ok().and_then(|v| { + let trimmed = v.trim_end_matches(['\n', '\r', ' ', '\t']).to_string(); + (!trimmed.is_empty()).then_some(trimmed) + }) +} + /// Read the bearer key from its file source else the inline env var. fn read_bearer(file: Option<&std::path::Path>, env: &str) -> Option { if let Some(path) = file { return read_file_secret(path); } - std::env::var(env).ok().filter(|v| !v.is_empty()) + read_env_secret(env) } /// Read + derive the 32-byte cluster key from its file source else the inline @@ -387,7 +419,7 @@ fn read_cluster_key(file: Option<&std::path::Path>, env: &str) -> Option<[u8; 32 let secret = if let Some(path) = file { read_file_secret(path)? } else { - std::env::var(env).ok().filter(|v| !v.is_empty())? + read_env_secret(env)? }; Some(derive_key(secret.as_bytes())) } @@ -428,7 +460,9 @@ fn decode_and_verify_token(key: &[u8; 32], token: &str, now: u64) -> Option exp.saturating_add(NODE_TOKEN_SKEW_SECS) { return None; // expired diff --git a/tidal-server/tests/cluster_chaos.rs b/tidal-server/tests/cluster_chaos.rs index ed8c2b8..0915047 100644 --- a/tidal-server/tests/cluster_chaos.rs +++ b/tidal-server/tests/cluster_chaos.rs @@ -782,3 +782,66 @@ fn mp_items_ride_the_log_and_catchup_stream() { items {offline_first}..={offline_last} present with feed parity 1e-6" ); } + +// ── m11p8-D: self-driving heal (the operator is OUT of the loop) ───────────────── + +/// The m11p8-D headline: a behind follower converges with NO `/cluster/heal` verb +/// AND no new writes — the leader's standing `tick_self_heal` duty re-delivers the +/// backlog by itself (incident §1.4-3, "the breaker eats the first heal", made an +/// operator non-event). +/// +/// Isolation of the behavior under test: `catchup_retry_ms` is pushed far past the +/// test budget so the FOLLOWER's own timer-pull cannot mask the leader-side heal, +/// and we stop writing after the link heals so no eager-ship probe fires. Once the +/// breaker half-opens, the ONLY thing that can re-deliver ap-south's gap is the +/// leader's `tick_self_heal` → `resume_from`. `learner_promote_lag` is lowered so +/// self-heal trips on a small lag (no need to write thousands of events). +#[test] +fn mp_self_heal_converges_without_operator_verb() { + // ap-south behind a TCP proxy so ONLY its replication link can be severed. + let (rewrite, proxies) = proxied_rewrite(&["ap-south"]); + let cluster = MultiProcCluster::start_with( + ClusterOptions::new(3) + .with_rewrite(rewrite) + .with_topology_extra( + "replication:\n learner_promote_lag: 4\n catchup_retry_ms: 600000", + ), + ); + + // Baseline: seed and let the whole cluster converge via the normal stream. + seed_items_and_embeddings(&cluster, LEADER, 5); + cluster.wait_converged_all(convergence_budget()); + + // SEVER ap-south's replication link. The leader's ship breaker to it opens + // after the repeated failed ships of the burst below. eu-west stays current. + proxies.region("ap-south").sever_grpc(); + println!("[self-heal] severed ap-south's gRPC replication link"); + + // Burst writes while severed: ap-south falls > learner_promote_lag behind and + // the leader's breaker to it trips. + seed_items_and_embeddings(&cluster, LEADER, 20); + let target = leader_seq(&cluster); + poll_until( + convergence_budget(), + "ap-south must fall behind while its link is severed", + || applied(&cluster, AP_SOUTH) < target, + ); + println!( + "[self-heal] ap-south is behind: applied={} < leader hwm={target}", + applied(&cluster, AP_SOUTH) + ); + + // HEAL THE LINK ONLY. Deliberately: NO `/cluster/heal`, and NO further writes. + // The operator is entirely out of the loop from here. + proxies.region("ap-south").heal_all(); + println!("[self-heal] healed the link; issuing NO heal verb and NO further writes"); + + // Convergence can now come ONLY from the leader's tick_self_heal re-arming the + // backlog re-ship from ap-south's durable mark once the breaker half-opens. + poll_until( + BREAKER_RESET + convergence_budget(), + "ap-south must self-heal to the leader hwm with the operator out of the loop", + || converged(&cluster, &[AP_SOUTH]), + ); + println!("[self-heal] ap-south converged to hwm={target} purely via tick_self_heal"); +} diff --git a/tidal-stress/Cargo.toml b/tidal-stress/Cargo.toml index 5ebebcc..f0fcff2 100644 --- a/tidal-stress/Cargo.toml +++ b/tidal-stress/Cargo.toml @@ -54,3 +54,13 @@ rand = "0.9" # tidal-net — drops the OpenSSL/native-tls system dependency for a clean static # build. `json` for the typed request/response bodies. reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } + +[dev-dependencies] +# Honest before/after for the generator's OWN per-request cost (no server, no +# network): bench workload.next() construction and the latency-histogram record +# path in isolation. Matches the engine crate's criterion posture. +criterion = { version = "0.5", features = ["html_reports"] } + +[[bench]] +name = "hotpath" +harness = false diff --git a/tidal-stress/benches/hotpath.rs b/tidal-stress/benches/hotpath.rs new file mode 100644 index 0000000..bd22410 --- /dev/null +++ b/tidal-stress/benches/hotpath.rs @@ -0,0 +1,82 @@ +//! Hot-path micro-benchmarks for the load generator's OWN per-request cost. +//! +//! These run with NO server and NO network, so they isolate exactly the work +//! the generator does per dispatched request — request construction +//! ([`Workload::next`]) and latency aggregation ([`LatencyHistogram::record`]). +//! That is the only honest way to answer "is this allocation material?": the +//! number here is the generator's CPU floor, the thing that must stay far below +//! the network round-trip it measures, and the before/after the perf sweep is +//! graded against. + +use std::hint::black_box; +use std::time::Duration; + +use criterion::{Criterion, criterion_group, criterion_main}; + +use tidal_stress::metrics::LatencyHistogram; +use tidal_stress::workload::{Workload, WritePath, parse_mix}; + +fn make_workload(mix: &str) -> Workload { + Workload::new( + // Three read gateways (round-robin), leader-pinned writes => write_bases + // len == 1, the headline `--leader-url` path (exercises the F4 short-circuit). + vec![ + "http://10.0.0.1:9500".into(), + "http://10.0.0.2:9500".into(), + "http://10.0.0.3:9500".into(), + ], + vec!["http://10.0.0.1:9500".into()], + WritePath::Leader, + parse_mix(mix).expect("mix preset parses"), + 10_000, // corpus + 50_000, // users + 1.3, // hot_skew + 24, // feed_limit + 128, // embedding_dim + ) +} + +/// `Workload::next` — the per-dispatched-request construction cost (URL String + +/// body). `peach` is the headline signal-dominated mix; `writes` forces the body +/// path on every call; `reads` isolates the URL-only feed/search path. +fn bench_next(c: &mut Criterion) { + let wl_peach = make_workload("peach"); + let wl_writes = make_workload("writes"); + let wl_reads = make_workload("reads"); + let mut rng = rand::rng(); + + let mut g = c.benchmark_group("workload_next"); + g.bench_function("peach", |b| b.iter(|| black_box(wl_peach.next(&mut rng)))); + g.bench_function("writes", |b| b.iter(|| black_box(wl_writes.next(&mut rng)))); + g.bench_function("reads", |b| b.iter(|| black_box(wl_reads.next(&mut rng)))); + g.finish(); +} + +/// `LatencyHistogram::record` — the per-OK-request collector cost (single-writer, +/// allocation-free), plus the per-stage `percentile` report cost. +fn bench_histogram(c: &mut Criterion) { + c.bench_function("histogram_record", |b| { + let mut h = LatencyHistogram::default(); + // Cheap LCG so successive records land in different buckets rather than + // hammering one — closer to a real latency distribution. + let mut state = 0x2545_F491_4F6C_DD1Du64; + b.iter(|| { + state = state + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + let ns = (state >> 33) % 200_000_000 + 1; // ~1ns..200ms + h.record(black_box(Duration::from_nanos(ns))); + }); + }); + + let mut h = LatencyHistogram::default(); + for i in 1..=100_000u64 { + h.record(Duration::from_nanos(i * 1_500)); + } + c.bench_function("histogram_p99", |b| { + b.iter(|| black_box(h.percentile(black_box(0.99)))); + }); +} + +criterion_group!(benches, bench_next, bench_histogram); +criterion_main!(benches); diff --git a/tidal-stress/src/client.rs b/tidal-stress/src/client.rs index 4ab3295..ed8dbba 100644 --- a/tidal-stress/src/client.rs +++ b/tidal-stress/src/client.rs @@ -11,17 +11,22 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use rand::Rng; +use reqwest::header::{AUTHORIZATION, HeaderValue}; use tokio::sync::Semaphore; use crate::error::{Result, StressError}; use crate::metrics::StatusClass; -use crate::workload::{HttpMethod, Plan}; +use crate::workload::{Body, HttpMethod, ItemMetadata, Plan}; /// Async client with optional bearer auth. Cheap to clone (Arc inside reqwest). #[derive(Clone)] pub struct HttpClient { inner: reqwest::Client, - api_key: Option, + /// The `Authorization: Bearer ` header value, built ONCE at construction + /// (the token never changes) so the per-request path is a cheap refcounted + /// clone instead of `format!` + header re-validation on every send. `None` when + /// no key is configured. Marked sensitive so it is redacted from any debug output. + auth_header: Option, /// `x-tidal-ack` value sent with every write (m11p3: `leader`/`quorum`); /// `None` = the deployment's topology default. ack: Option, @@ -30,6 +35,12 @@ pub struct HttpClient { impl HttpClient { /// `request_timeout` should sit ABOVE the server's 30s request timeout so the /// server's own 408 surfaces as a Timeout class rather than a client abort. + /// + /// # Errors + /// + /// Returns [`StressError::Client`] if the underlying reqwest client cannot be + /// built (TLS/builder fault), and [`StressError::Auth`] if `api_key` contains + /// characters that are not valid in an HTTP `Authorization` header value. pub fn new( request_timeout: Duration, api_key: Option, @@ -46,16 +57,31 @@ impl HttpClient { .tcp_nodelay(true) .build() .map_err(StressError::Client)?; + // Build the Authorization header once. reqwest's `bearer_auth` does a fresh + // `format!("Bearer {token}")` allocation plus a HeaderValue validation pass + // on EVERY call, over a token that never changes — pure per-request waste. + let auth_header = match api_key { + Some(k) => { + let mut v = HeaderValue::from_str(&format!("Bearer {k}")) + .map_err(|_| StressError::Auth("contains invalid header characters".into()))?; + v.set_sensitive(true); + Some(v) + } + None => None, + }; Ok(Self { inner, - api_key, + auth_header, ack, }) } fn apply_auth(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder { - match &self.api_key { - Some(k) => rb.bearer_auth(k), + // A HeaderValue clone is a cheap refcount bump (or inline copy), not a + // realloc+revalidate. `cluster_status` deliberately does NOT call this, so + // its status poll stays unauthenticated as before. + match &self.auth_header { + Some(v) => rb.header(AUTHORIZATION, v.clone()), None => rb, } } @@ -107,7 +133,7 @@ impl HttpClient { /// One-off POST returning the raw status, for the seeder (which needs to know /// success vs 429-retry rather than a capacity class). - async fn post_json(&self, url: &str, body: &serde_json::Value) -> Option { + async fn post_json(&self, url: &str, body: &Body) -> Option { let rb = self.inner.post(url).json(body); match self.apply_auth(rb).send().await { Ok(resp) => { @@ -121,9 +147,16 @@ impl HttpClient { } /// Register `count` items (id 1..=count) plus a `dim`-wide content embedding for -/// each, against `base` (use the LEADER url so items broadcast to every region -/// and `/feed` on any region can rank them). Bounded-concurrency, retries 429. -/// Returns the number of items confirmed registered. +/// each, against `base`. +/// +/// Use the LEADER url so items broadcast to every region and `/feed` on any +/// region can rank them. Bounded-concurrency, retries 429. Returns the number of +/// items confirmed registered. +/// +/// # Errors +/// +/// Returns [`StressError::Seed`] if the bounded-concurrency semaphore is closed +/// while acquiring a permit (the run cannot proceed). pub async fn seed_corpus( client: &HttpClient, base: &str, @@ -154,17 +187,30 @@ pub async fn seed_corpus( let (item, emb) = { let mut rng = rand::rng(); let values: Vec = (0..dim).map(|_| rng.random::() - 0.5).collect(); - let item = serde_json::json!({ - "entity_id": id, - "metadata": { "title": format!("post-{id}"), "category": category }, - }); - let emb = serde_json::json!({ "entity_id": id, "values": values }); + let item = Body::Item { + entity_id: id, + metadata: ItemMetadata { + title: format!("post-{id}"), + category, + }, + }; + let emb = Body::Embedding { + entity_id: id, + values, + }; (item, emb) }; - if retry_write(&client, &items_url, &item).await - && retry_write(&client, &emb_url, &emb).await - { + // The item and embedding writes are independent — the engine keys the + // embedding purely on entity id with no item-before-embedding + // precondition (verified: db/items.rs write_item_embedding) — so issue + // them concurrently rather than serially, roughly halving each id's + // seed wall-clock. `done` still requires BOTH to succeed. + let (item_ok, emb_ok) = tokio::join!( + retry_write(&client, &items_url, &item), + retry_write(&client, &emb_url, &emb), + ); + if item_ok && emb_ok { done.fetch_add(1, Ordering::Relaxed); } })); @@ -193,7 +239,7 @@ const SEED_CATEGORIES: [&str; 12] = [ /// POST with backpressure-aware retry: a 2xx is success; a 429 backs off and /// retries (the seed must not give up just because the write pool is busy); any /// other code / transport fault fails fast (a usage bug, not transient load). -async fn retry_write(client: &HttpClient, url: &str, body: &serde_json::Value) -> bool { +async fn retry_write(client: &HttpClient, url: &str, body: &Body) -> bool { const MAX_ATTEMPTS: u32 = 8; for attempt in 0..MAX_ATTEMPTS { match client.post_json(url, body).await { diff --git a/tidal-stress/src/error.rs b/tidal-stress/src/error.rs index d0e1e33..26b14f5 100644 --- a/tidal-stress/src/error.rs +++ b/tidal-stress/src/error.rs @@ -1,10 +1,14 @@ -//! Crate error type. Mirrors the `tidal-server` convention: a flat `thiserror` -//! enum + `Result` alias, surfaced by `main` as `error: {err}` + exit 1. No -//! `anyhow` (workspace convention is explicit error enums). +//! Crate error type. +//! +//! Mirrors the `tidal-server` convention: a flat `thiserror` enum + `Result` +//! alias, surfaced by `main` as `error: {err}` + exit 1. No `anyhow` (workspace +//! convention is explicit error enums). -/// Anything that can go wrong while *setting up or driving* a load run. Per-request -/// failures during a run are NOT errors — they are recorded as outcomes (see -/// [`crate::metrics::Outcome`]); this type is only for fatal setup/teardown faults. +/// Anything that can go wrong while *setting up or driving* a load run. +/// +/// Per-request failures during a run are NOT errors — they are recorded as +/// outcomes (see [`crate::metrics::Outcome`]); this type is only for fatal +/// setup/teardown faults. #[derive(Debug, thiserror::Error)] pub enum StressError { /// A target URL was missing or unparseable. @@ -26,6 +30,11 @@ pub enum StressError { /// The HTTP client could not be constructed (TLS/builder fault). #[error("http client build failed: {0}")] Client(#[source] reqwest::Error), + + /// The provided API key could not be turned into an HTTP `Authorization` + /// header value (contains control or non-visible characters). + #[error("invalid api key: {0}")] + Auth(String), } /// Crate-local result alias. diff --git a/tidal-stress/src/lib.rs b/tidal-stress/src/lib.rs new file mode 100644 index 0000000..fbfde37 --- /dev/null +++ b/tidal-stress/src/lib.rs @@ -0,0 +1,15 @@ +//! `tidal-stress` library surface. +//! +//! The crate is a binary (the `tidal-stress` load generator), but its hot paths +//! — request construction ([`workload`]) and latency aggregation ([`metrics`]) — +//! are exposed here so `benches/` can drive them directly and measure the +//! generator's own per-request cost in isolation (no server, no network), which +//! is the only honest way to answer "is this allocation material?". +//! +//! `main.rs` is the thin CLI binary that consumes this surface. + +pub mod client; +pub mod error; +pub mod metrics; +pub mod scheduler; +pub mod workload; diff --git a/tidal-stress/src/main.rs b/tidal-stress/src/main.rs index cd56eaa..b36d39b 100644 --- a/tidal-stress/src/main.rs +++ b/tidal-stress/src/main.rs @@ -10,23 +10,17 @@ //! Methodology lives in [`scheduler`] (coordinated-omission-corrected open loop); //! the workload shape and its thepeach derivation live in [`workload`]. -mod client; -mod error; -mod metrics; -mod scheduler; -mod workload; - use std::io::Write as _; use std::sync::Arc; use std::time::Duration; use clap::Parser; -use crate::client::{HttpClient, seed_corpus}; -use crate::error::{Result, StressError}; -use crate::metrics::StageStats; -use crate::scheduler::{Stage, parse_ramp, run_stage}; -use crate::workload::{OpKind, Workload, WritePath, parse_mix}; +use tidal_stress::client::{HttpClient, seed_corpus}; +use tidal_stress::error::{Result, StressError}; +use tidal_stress::metrics::{self, StageStats}; +use tidal_stress::scheduler::{Stage, parse_ramp, run_stage}; +use tidal_stress::workload::{OpKind, Workload, WritePath, parse_mix}; #[derive(Parser)] #[command( @@ -128,9 +122,9 @@ struct Cli { } // SLO thresholds for the per-stage verdict. tidalDB's own RETRIEVE SLA is p99 -// <50ms in-process; over a cluster network hop we allow 150ms before calling a -// stage degraded, and treat >1% errors (429/408/503/5xx/transport) as the knee. -const SLO_FEED_P99: Duration = Duration::from_millis(150); +// <50ms in-process; over a cluster network hop we allow 150ms (the feed-read p99 +// SLO lives in `metrics` as `FEED_P99_SLO` so the stage can count EXACT +// violations), and treat >1% errors (429/408/503/5xx/transport) as the knee. const SLO_ERROR_RATE: f64 = 0.01; // The thepeach 100k-DAU model: ~5 sessions/user/day × ~5 feed pages × ~15.4 @@ -324,7 +318,12 @@ fn stage_verdict(stats: &StageStats) -> (bool, f64) { let signal_ok_rps = signal_ok as f64 / secs; let feed = &stats.ops[OpKind::FeedRead.idx()]; - let feed_p99_ok = feed.ok() == 0 || feed.hist.percentile(0.99) <= SLO_FEED_P99; + let feed_ok = feed.ok(); + // Exact p99-vs-SLO: p99 ≤ SLO ⟺ at most 1% of OK feed reads breached it. Using + // the exact violation count (`StageStats::feed_over_slo`) rather than the + // bucket-interpolated percentile means a true tail sitting inside the SLO's + // histogram bucket [147.0ms, 158.4ms] cannot interpolate its way to a pass. + let feed_p99_ok = feed_ok == 0 || (stats.feed_over_slo as f64) <= 0.01 * feed_ok as f64; let passed = stats.error_rate() <= SLO_ERROR_RATE && feed_p99_ok && stats.client_shed == 0; (passed, signal_ok_rps) } diff --git a/tidal-stress/src/metrics.rs b/tidal-stress/src/metrics.rs index 974179c..bda6b39 100644 --- a/tidal-stress/src/metrics.rs +++ b/tidal-stress/src/metrics.rs @@ -25,6 +25,16 @@ const MIN_NS: f64 = 1_000.0; // 1µs — finer than that is noise over a network const GROWTH: f64 = 1.0772; // ~30 buckets/decade const BUCKETS: usize = 300; // 1µs * 1.0772^300 ≈ 9.4e12ns ≈ 2.6h — far past any real tail +/// Feed-read p99 latency SLO. tidalDB's in-process RETRIEVE SLA is p99 <50ms; +/// over a cluster network hop we allow 150ms before calling a stage degraded. +/// +/// It lives HERE (not in `main`) so [`StageStats::record`] can keep an EXACT count +/// of feed reads that breached it. The bucketed-histogram p99 is only ~3-4% +/// accurate and a 150ms threshold lands mid-bucket ([147.0ms, 158.4ms] at this +/// `GROWTH`), so a true 155ms p99 could interpolate to a false pass — the capacity +/// verdict must turn on the exact violation fraction, not the interpolated estimate. +pub const FEED_P99_SLO: Duration = Duration::from_millis(150); + /// A single operation's latency distribution. Exact count/min/max/sum, bucketed /// percentiles. #[derive(Clone)] @@ -71,6 +81,7 @@ impl LatencyHistogram { /// Estimated p-th percentile (`p` in 0.0..=1.0), linearly interpolated within /// the crossing bucket. Returns 0 for an empty histogram. + #[must_use] pub fn percentile(&self, p: f64) -> Duration { if self.total == 0 { return Duration::ZERO; @@ -99,6 +110,7 @@ impl LatencyHistogram { /// Exact worst-case latency (not bucket-estimated) — the tail a load report /// must surface alongside p999. + #[must_use] pub const fn max(&self) -> Duration { Duration::from_nanos(if self.total == 0 { 0 } else { self.max_ns }) } @@ -126,6 +138,7 @@ pub enum StatusClass { } impl StatusClass { + #[must_use] pub const fn from_status(code: u16) -> Self { match code { 200..=299 => Self::Ok, @@ -180,6 +193,7 @@ impl OpStats { } } + #[must_use] pub const fn total(&self) -> u64 { let mut sum = 0u64; let mut i = 0; @@ -190,27 +204,35 @@ impl OpStats { sum } + #[must_use] pub const fn ok(&self) -> u64 { self.status[0] } + #[must_use] pub const fn backpressure(&self) -> u64 { self.status[1] } + #[must_use] pub const fn timeout(&self) -> u64 { self.status[2] } + #[must_use] pub const fn unavailable(&self) -> u64 { self.status[3] } + #[must_use] pub const fn client_error(&self) -> u64 { self.status[4] } + #[must_use] pub const fn server_error(&self) -> u64 { self.status[5] } + #[must_use] pub const fn transport(&self) -> u64 { self.status[6] } + #[must_use] pub const fn errors(&self) -> u64 { self.total() - self.ok() } @@ -227,20 +249,49 @@ pub struct StageStats { /// Mean delay between a request's intended send time and its actual dispatch. /// Growing scheduling delay ⇒ the generator is falling behind the target rate. pub mean_schedule_lag: Duration, + /// p99 and max of that dispatch delay. The lag is folded into every + /// CO-corrected latency (it is measured from the *intended* send time), so a + /// single catch-up stall inflates the server's apparent tail — the mean hides + /// exactly the burst this tool exists to expose. The tail surfaces it. + pub p99_schedule_lag: Duration, + pub max_schedule_lag: Duration, + /// Exact count of OK feed reads whose CO-corrected latency exceeded + /// [`FEED_P99_SLO`]. The verdict turns on this exact fraction, not the + /// bucket-interpolated p99, so a true tail sitting inside the SLO's histogram + /// bucket cannot sneak a false pass. + pub feed_over_slo: u64, +} + +impl Default for StageStats { + fn default() -> Self { + Self::new() + } } impl StageStats { + #[must_use] pub fn new() -> Self { Self { ops: vec![OpStats::default(); OpKind::COUNT], elapsed: Duration::ZERO, client_shed: 0, mean_schedule_lag: Duration::ZERO, + p99_schedule_lag: Duration::ZERO, + max_schedule_lag: Duration::ZERO, + feed_over_slo: 0, } } pub fn record(&mut self, outcome: &Outcome) { self.ops[outcome.op.idx()].record(outcome.class, outcome.latency); + // Exact feed-read SLO violation count for the verdict (the bucketed p99 in + // the report is ~3-4% fuzzy and would decide pass/fail by interpolation). + if outcome.op == OpKind::FeedRead + && outcome.class == StatusClass::Ok + && outcome.latency > FEED_P99_SLO + { + self.feed_over_slo += 1; + } } pub fn total(&self) -> u64 { @@ -252,6 +303,7 @@ impl StageStats { pub fn total_errors(&self) -> u64 { self.ops.iter().map(OpStats::errors).sum() } + #[must_use] pub fn error_rate(&self) -> f64 { let t = self.total(); if t == 0 { @@ -261,6 +313,7 @@ impl StageStats { } } /// Achieved throughput: completed requests per second over the stage. + #[must_use] pub fn achieved_rps(&self) -> f64 { let s = self.elapsed.as_secs_f64(); if s <= 0.0 { @@ -350,11 +403,23 @@ pub fn render_stage(label: &str, target_rps: f64, stats: &StageStats) -> String stats.class_total(OpStats::transport), )); out.push_str(&format!( - "error rate {:.2}% | client-shed {} | schedule-lag {}\n", + "error rate {:.2}% | client-shed {} | schedule-lag mean {} / p99 {} / max {}\n", stats.error_rate() * 100.0, stats.client_shed, fmt_dur(stats.mean_schedule_lag), + fmt_dur(stats.p99_schedule_lag), + fmt_dur(stats.max_schedule_lag), )); + // A non-zero shed means the in-flight cap was hit: those requests were never + // timed, so the percentiles above are tail-under-measured (optimistic). The + // verdict already auto-fails the stage; this stops the printed table from + // being read as clean. + if stats.client_shed > 0 { + out.push_str(&format!( + " ⚠ percentiles above are tail-under-measured: {} request(s) shed (in-flight cap hit, never sent/timed)\n", + stats.client_shed, + )); + } out } diff --git a/tidal-stress/src/scheduler.rs b/tidal-stress/src/scheduler.rs index a166302..fc2e52d 100644 --- a/tidal-stress/src/scheduler.rs +++ b/tidal-stress/src/scheduler.rs @@ -13,7 +13,6 @@ //! reported as "the generator, not the server, is the limit here". use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use tokio::sync::{Semaphore, mpsc}; @@ -21,7 +20,7 @@ use tokio::time::Instant; use crate::client::HttpClient; use crate::error::{Result, StressError}; -use crate::metrics::{Outcome, StageStats}; +use crate::metrics::{LatencyHistogram, Outcome, StageStats}; use crate::workload::Workload; /// One step of the ramp: hold `target_rps` for `duration`. @@ -47,9 +46,14 @@ pub async fn run_stage( }); let sem = Arc::new(Semaphore::new(max_inflight)); - let shed = Arc::new(AtomicU64::new(0)); - let lag_sum_ns = Arc::new(AtomicU64::new(0)); - let lag_count = Arc::new(AtomicU64::new(0)); + // These are touched ONLY by this single dispatcher task — they are never moved + // into the spawned request tasks — so they are plain locals, not atomics. The + // local histogram captures the lag DISTRIBUTION so the generator's own catch-up + // stalls surface as p99/max instead of being averaged away into the mean. + let mut shed: u64 = 0; + let mut lag_sum_ns: u64 = 0; + let mut lag_count: u64 = 0; + let mut lag_hist = LatencyHistogram::default(); let start = Instant::now(); let deadline = start + stage.duration; @@ -57,15 +61,22 @@ pub async fn run_stage( let mut next = start; let mut dispatched: u64 = 0; - while Instant::now() < deadline { + loop { + // One clock read per iteration drives both the deadline check and the + // dispatch decision (the value was previously read twice with no work + // between the reads). let now = Instant::now(); + if now >= deadline { + break; + } if next <= now { match sem.clone().try_acquire_owned() { Ok(permit) => { let intended = next; let lag = now.saturating_duration_since(intended); - lag_sum_ns.fetch_add(lag.as_nanos() as u64, Ordering::Relaxed); - lag_count.fetch_add(1, Ordering::Relaxed); + lag_sum_ns += lag.as_nanos() as u64; + lag_count += 1; + lag_hist.record(lag); let workload = workload.clone(); let client = client.clone(); @@ -88,7 +99,7 @@ pub async fn run_stage( }); } Err(_) => { - shed.fetch_add(1, Ordering::Relaxed); + shed += 1; } } next += period; @@ -110,13 +121,14 @@ pub async fn run_stage( drop(tx); let mut stats = collector.await.unwrap_or_else(|_| StageStats::new()); stats.elapsed = elapsed; - stats.client_shed = shed.load(Ordering::Relaxed); - let lc = lag_count.load(Ordering::Relaxed); - stats.mean_schedule_lag = if lc == 0 { + stats.client_shed = shed; + stats.mean_schedule_lag = if lag_count == 0 { Duration::ZERO } else { - Duration::from_nanos(lag_sum_ns.load(Ordering::Relaxed) / lc) + Duration::from_nanos(lag_sum_ns / lag_count) }; + stats.p99_schedule_lag = lag_hist.percentile(0.99); + stats.max_schedule_lag = lag_hist.max(); stats } @@ -128,6 +140,12 @@ pub async fn run_stage( /// `peach` mix (~90% signals) a 100k-DAU TikTok-style evening peak is ≈3,900 /// total req/s, so `peach-100k` brackets and then doubles past that to answer /// "can we handle more?". `secs` defaults from the preset. +/// +/// # Errors +/// +/// Returns [`StressError::Ramp`] if a custom `rps:secs,...` spec is malformed: +/// a part missing the `:` separator, an unparseable rps or secs value, or a +/// spec that yields no stages at all. pub fn parse_ramp(spec: &str, stage_secs: u64) -> Result> { let mk = |rates: &[f64]| -> Vec { rates diff --git a/tidal-stress/src/workload.rs b/tidal-stress/src/workload.rs index 984cd84..f61294e 100644 --- a/tidal-stress/src/workload.rs +++ b/tidal-stress/src/workload.rs @@ -12,6 +12,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use rand::Rng; +use serde::Serialize; use crate::error::{Result, StressError}; @@ -47,6 +48,7 @@ impl OpKind { ]; pub const COUNT: usize = Self::ALL.len(); + #[must_use] pub const fn idx(self) -> usize { match self { Self::FeedRead => 0, @@ -59,6 +61,7 @@ impl OpKind { } } + #[must_use] pub const fn label(self) -> &'static str { match self { Self::FeedRead => "feed", @@ -119,7 +122,41 @@ pub struct Plan { pub op: OpKind, pub method: HttpMethod, pub url: String, - pub body: Option, + pub body: Option, +} + +/// A request body, serialized straight to the wire by reqwest's `.json()` with NO +/// intermediate `serde_json::Value` tree. +/// +/// The `Signal` variant — the hot ~90% of +/// the peach mix — is entirely stack-allocated (a `&'static str` signal name, no +/// owned `String` keys); only the rare item write allocates its title `String`. +/// `#[serde(untagged)]` emits just the inner fields, so the wire JSON is identical +/// to the previous `json!({...})` (asserted byte-for-byte in the tests below). +#[derive(Serialize)] +#[serde(untagged)] +pub enum Body { + Signal { + entity_id: u64, + signal: &'static str, + weight: f64, + }, + Item { + entity_id: u64, + metadata: ItemMetadata, + }, + Embedding { + entity_id: u64, + values: Vec, + }, +} + +/// Item metadata — `title` is the one per-write `String` allocation; `category` is +/// a `&'static str` drawn from the fixed [`CATEGORIES`] vocabulary. +#[derive(Serialize)] +pub struct ItemMetadata { + pub title: String, + pub category: &'static str, } /// Round-robin over a set of base URLs (e.g. the three region gateways). @@ -136,6 +173,13 @@ impl RoundRobin { } } fn pick(&self) -> &str { + // The headline `--leader-url` write path pins writes to a single base; with + // one base the round-robin answer is constant, so skip the process-global + // atomic RMW that would otherwise exist only to be contended across every + // worker core (every spawned request task calls this). + if self.bases.len() == 1 { + return &self.bases[0]; + } let i = self.next.fetch_add(1, Ordering::Relaxed) % self.bases.len(); &self.bases[i] } @@ -206,6 +250,7 @@ const CATEGORIES: [&str; 12] = [ ]; impl Workload { + #[must_use] #[allow(clippy::too_many_arguments)] pub fn new( read_bases: Vec, @@ -325,11 +370,11 @@ impl Workload { _ => ("view", 1.0), }; let url = format!("{}{}", self.writes.pick(), self.write_path.signal_path()); - let body = serde_json::json!({ - "entity_id": self.pick_item(rng), - "signal": name, - "weight": weight, - }); + let body = Body::Signal { + entity_id: self.pick_item(rng), + signal: name, + weight, + }; Plan { op, method: HttpMethod::Post, @@ -340,10 +385,13 @@ impl Workload { OpKind::RegisterItem => { let id = self.pick_item(rng); let url = format!("{}{}", self.writes.pick(), self.write_path.item_path()); - let body = serde_json::json!({ - "entity_id": id, - "metadata": { "title": format!("post-{id}"), "category": self.pick_category(rng) }, - }); + let body = Body::Item { + entity_id: id, + metadata: ItemMetadata { + title: format!("post-{id}"), + category: self.pick_category(rng), + }, + }; Plan { op, method: HttpMethod::Post, @@ -353,10 +401,10 @@ impl Workload { } OpKind::RegisterEmbedding => { let url = format!("{}{}", self.writes.pick(), self.write_path.embedding_path()); - let body = serde_json::json!({ - "entity_id": self.pick_item(rng), - "values": self.random_embedding(rng), - }); + let body = Body::Embedding { + entity_id: self.pick_item(rng), + values: self.random_embedding(rng), + }; Plan { op, method: HttpMethod::Post, @@ -376,6 +424,12 @@ impl Workload { /// user views ~12 of 24 tiles, likes ~0.6, skips ~1.5, searches ~0.25, and a /// trickle of creator posts. The result is ~78% views, ~10% skips, ~6.5% feed /// reads — the signal-dominated shape thepeach's user-graph spec describes. +/// +/// # Errors +/// +/// Returns [`StressError::Mix`] if a custom `op=weight,...` spec is malformed: +/// a part missing the `=` separator, an unknown op label, an unparseable weight, +/// or a spec that yields no weighted ops at all. pub fn parse_mix(spec: &str) -> Result> { match spec { "peach" | "tiktok" | "default" => Ok(vec![ @@ -469,4 +523,45 @@ mod tests { // concentrates ~34% (≈3400) there — clearly biased toward hot content. assert!(low > 3_000, "hot-skew not biasing low ids: {low}"); } + + #[test] + fn body_serializes_byte_identical_to_the_legacy_json() { + // The wire payload MUST be unchanged by the json!()->typed-struct rewrite. + // Assert each Body variant serializes to exactly the object the old + // serde_json::json!({...}) produced. Values chosen to be exactly + // representable in both f32 and f64 so the comparison is not float-fuzzy. + let signal = Body::Signal { + entity_id: 42, + signal: "view", + weight: 1.0, + }; + assert_eq!( + serde_json::to_value(&signal).expect("serialize signal"), + serde_json::json!({ "entity_id": 42, "signal": "view", "weight": 1.0 }), + ); + + let item = Body::Item { + entity_id: 7, + metadata: ItemMetadata { + title: "post-7".to_string(), + category: "anime", + }, + }; + assert_eq!( + serde_json::to_value(&item).expect("serialize item"), + serde_json::json!({ + "entity_id": 7, + "metadata": { "title": "post-7", "category": "anime" } + }), + ); + + let emb = Body::Embedding { + entity_id: 3, + values: vec![0.5_f32, -0.25, 0.0], + }; + assert_eq!( + serde_json::to_value(&emb).expect("serialize embedding"), + serde_json::json!({ "entity_id": 3, "values": [0.5, -0.25, 0.0] }), + ); + } } diff --git a/tidal/src/db/metrics/cluster.rs b/tidal/src/db/metrics/cluster.rs index f25dc75..89ceaba 100644 --- a/tidal/src/db/metrics/cluster.rs +++ b/tidal/src/db/metrics/cluster.rs @@ -130,9 +130,12 @@ pub struct ClusterMetrics { /// shard leader over HTTP. High forward rate at one gateway = client routing /// imbalance, not a fault. forwards_total: AtomicU64, - /// Total cross-node write forwards that ERRORED (m11p8) — the leader was - /// unreachable, returned 5xx, or the relay timed out. A non-zero rate means - /// a gateway cannot reach the current leader (election in flight, partition). + /// Total cross-node write forwards that hit a TRANSPORT failure (m11p8) — the + /// leader was unreachable (connect refused) or the relay timed out. A relayed + /// HTTP *status* (a 5xx such as `NotLeader`/`QuorumTimeout`) is a real verdict + /// the client retries and is relayed verbatim, NOT counted here. A non-zero + /// rate means a gateway cannot establish a connection to the current leader + /// (the leader is down or partitioned, not merely mid-election). forward_failures_total: AtomicU64, /// Total per-peer circuit-breaker OPEN transitions observed across all peers /// (m11p8). Each open is a `reset_duration` ship stall to that peer; a rising @@ -226,9 +229,15 @@ impl ClusterMetrics { /// 2 half-open. A closed/half-open → open transition also bumps the global /// `breaker_opens_total` counter so a dashboard can `rate()` the flap. pub fn set_peer_breaker_state(&self, peer: ShardId, state: u8) { + // Gauge encoding 0 closed / 1 open / 2 half-open. SOURCE OF TRUTH is + // tidal-net `CircuitBreaker::as_gauge`; tidal cannot depend on tidal-net + // (the dep runs the other way), so the OPEN value is pinned here by name + // rather than left as a bare literal that could silently desync. + const BREAKER_OPEN: u64 = 1; let cell = self.peer(peer); - let prev = cell.breaker_state.swap(u64::from(state), Ordering::Relaxed); - if state == 1 && prev != 1 { + let state = u64::from(state); + let prev = cell.breaker_state.swap(state, Ordering::Relaxed); + if state == BREAKER_OPEN && prev != BREAKER_OPEN { self.breaker_opens_total.fetch_add(1, Ordering::Relaxed); } } @@ -485,7 +494,7 @@ impl ClusterMetrics { emit_scalar( out, "tidaldb_cluster_forward_failures_total", - "Cross-node write forwards that errored — leader unreachable / 5xx / timeout (m11p8)", + "Cross-node write forwards that hit a transport failure — leader unreachable (connect) or timeout; a relayed 5xx status is NOT counted (m11p8)", "counter", self.forward_failures_total.load(Ordering::Relaxed) as f64, &extra, diff --git a/tidal/src/ranking/executor/scoring.rs b/tidal/src/ranking/executor/scoring.rs index 8c8a687..62fef80 100644 --- a/tidal/src/ranking/executor/scoring.rs +++ b/tidal/src/ranking/executor/scoring.rs @@ -75,6 +75,12 @@ impl ProfileExecutor<'_> { /// [`read_agg_for_sort`], so a signal the schema omits contributes 0.0 /// rather than erroring; any other read error (notably an unimplemented /// aggregation) still propagates. + /// + /// PERF MIRROR: `signal_values::sort_signal_reads` enumerates the same + /// (signal, agg, window) tuples each Sort variant reads here, so the one- + /// get-per-type pre-pass can collapse them. The two MUST stay in sync — a new + /// ledger read added below should be added there too. Drift only costs the + /// optimization (a miss falls back to a direct read), never correctness. #[allow(clippy::too_many_lines)] pub(super) fn score_by_sort( &self, diff --git a/tidal/src/wal/compaction.rs b/tidal/src/wal/compaction.rs index 87c3875..43598b3 100644 --- a/tidal/src/wal/compaction.rs +++ b/tidal/src/wal/compaction.rs @@ -279,11 +279,22 @@ fn archive_segment(archive_dir: &Path, seg_path: &Path) -> Result<(), WalError> return Ok(()); } let tmp = archive_dir.join(format!("{}.tmp", file_name.to_string_lossy())); - std::fs::copy(seg_path, &tmp)?; - std::fs::File::open(&tmp)?.sync_all()?; - std::fs::rename(&tmp, &dest)?; - crate::wal::sync_dir_durable(archive_dir)?; - Ok(()) + // On any failure after the copy begins (a full/flapping archive volume), the + // partial `.tmp` is best-effort removed so orphan temps don't accrete in the + // catalog `list_segments` must skip. A `.tmp` is never promoted (only `rename` + // publishes it) so this is hygiene, not a durability requirement. + let archive = || -> Result<(), WalError> { + std::fs::copy(seg_path, &tmp)?; + std::fs::File::open(&tmp)?.sync_all()?; + std::fs::rename(&tmp, &dest)?; + crate::wal::sync_dir_durable(archive_dir)?; + Ok(()) + }; + let result = archive(); + if result.is_err() { + let _ = std::fs::remove_file(&tmp); + } + result } #[cfg(test)] diff --git a/tidalctl/src/commands/backup.rs b/tidalctl/src/commands/backup.rs index d85ed2b..02c7237 100644 --- a/tidalctl/src/commands/backup.rs +++ b/tidalctl/src/commands/backup.rs @@ -7,6 +7,20 @@ //! per file + the recovered WAL checkpoint cursor); `restore` verifies every //! file's hash against the manifest before placing it into a FRESH target dir. //! +//! Both paths fsync their output (files + the directories whose dirents changed) +//! before returning: a backup that returned before its bytes and dirents were +//! durable would be crash-consistent only against a process crash, not a host or +//! power crash — the exact crash a backup exists to survive. This mirrors the +//! engine's in-process backup (`tidal/src/db/backup.rs`). +//! +//! Restore treats the manifest as UNTRUSTED input: a backup directory may have +//! been tampered with, and BLAKE3 verification does not help against a path +//! attack (the attacker controls the path string, the bytes, AND the hash +//! together). Every manifest path is therefore resolved through [`safe_join`], +//! which rejects any component that would escape the target (Zip-Slip / tar +//! traversal), and the whole restore fails before a single byte is written if any +//! path is unsafe. +//! //! Coordinated cluster backup (runbook): back up one committed replica per shard //! group — under `ack=quorum` any committed replica's data dir holds the //! quorum-durable log, so it is a cluster-consistent snapshot at its recorded @@ -15,7 +29,7 @@ //! group's leader from its backup; followers catch up via the live stream. use std::{ - path::{Path, PathBuf}, + path::{Component, Path, PathBuf}, time::{SystemTime, UNIX_EPOCH}, }; @@ -41,9 +55,13 @@ struct FileEntry { } /// The backup manifest, written at the backup root and verified on restore. +#[allow(clippy::struct_field_names)] // `manifest_version` is the on-disk wire field name; keep it explicit. #[derive(Debug, Serialize, Deserialize)] struct Manifest { - /// Format version (bump on an incompatible manifest change). + /// Format version. A restore accepts any version at or below + /// [`MANIFEST_VERSION`] (a newer binary can still read an older backup) and + /// refuses a newer one it cannot understand. Additive fields use + /// `#[serde(default)]` so a forward-compatible change does not bump this. manifest_version: u32, /// Unix-epoch seconds the backup was taken (operator audit trail). created_unix_secs: u64, @@ -85,13 +103,23 @@ pub(crate) fn run_backup(src: &Path, out: &Path, pretty: bool) -> Result<(String std::fs::create_dir_all(out) .map_err(|e| CliError::new(format!("create dest {}: {e}", out.display())))?; - // The cluster cursor this backup is consistent to (recovered offline). - let checkpoint_seq = wal_state::gather_wal_state(&src.join("wal")) - .map(|s| s.checkpoint_seq) - .unwrap_or(0); + // The cluster cursor this backup is consistent to (recovered offline). A read + // ERROR means the WAL dir exists but is corrupt/unreadable — refuse the backup + // rather than stamp a fabricated `checkpoint_seq: 0` that is indistinguishable + // from a genuinely fresh store (which still resolves to 0 via the Ok arm). + let checkpoint_seq = match wal_state::gather_wal_state(&src.join("wal")) { + Ok(state) => state.checkpoint_seq, + Err(e) => { + return Err(CliError::new(format!( + "cannot recover the WAL checkpoint from {} ({e}); refusing to write a backup \ + with an unknown cursor — repair or drain the source first", + src.join("wal").display() + ))); + } + }; let mut rel_files = Vec::new(); - collect_files(src, PathBuf::new(), &mut rel_files) + collect_files(src, Path::new(""), &mut rel_files) .map_err(|e| CliError::new(format!("walk source: {e}")))?; let mut files = Vec::with_capacity(rel_files.len()); @@ -108,6 +136,8 @@ pub(crate) fn run_backup(src: &Path, out: &Path, pretty: bool) -> Result<(String } std::fs::write(&dest, &bytes) .map_err(|e| CliError::new(format!("write {}: {e}", dest.display())))?; + // Durability: flush this file's bytes before it counts as backed up. + fsync_path(&dest).map_err(|e| CliError::new(format!("fsync {}: {e}", dest.display())))?; total_bytes += bytes.len() as u64; files.push(FileEntry { path: rel_to_slash(rel), @@ -128,8 +158,14 @@ pub(crate) fn run_backup(src: &Path, out: &Path, pretty: bool) -> Result<(String files, }; let manifest_json = serialize(&manifest, pretty)?; - std::fs::write(out.join(BACKUP_MANIFEST), &manifest_json) + let manifest_path = out.join(BACKUP_MANIFEST); + std::fs::write(&manifest_path, &manifest_json) .map_err(|e| CliError::new(format!("write manifest: {e}")))?; + // The manifest is the verification root — flush it, then flush every directory + // whose dirents we created so the new files survive a host/power crash. + fsync_path(&manifest_path).map_err(|e| CliError::new(format!("fsync manifest: {e}")))?; + fsync_dirs_recursive(out) + .map_err(|e| CliError::new(format!("fsync backup dirs {}: {e}", out.display())))?; let summary = serde_json::json!({ "backed_up": src.display().to_string(), @@ -158,9 +194,11 @@ pub(crate) fn run_restore( })?; let manifest: Manifest = serde_json::from_slice(&manifest_bytes) .map_err(|e| CliError::new(format!("parse manifest: {e}")))?; - if manifest.manifest_version != MANIFEST_VERSION { + // Accept any manifest at or below our version (a newer tidalctl reads an older + // backup); refuse a newer format we cannot interpret. + if manifest.manifest_version > MANIFEST_VERSION { return Err(CliError::new(format!( - "unsupported backup manifest version {} (this tidalctl writes/reads v{MANIFEST_VERSION})", + "backup manifest version {} is newer than this tidalctl supports (v{MANIFEST_VERSION}); upgrade tidalctl", manifest.manifest_version ))); } @@ -179,12 +217,25 @@ pub(crate) fn run_restore( ))); } + // Resolve EVERY manifest path through the traversal guard up front. A single + // unsafe entry (`..`, an absolute/prefix component, a backslash) fails the + // whole restore here — before any byte is read or written — so a tampered + // backup can never write outside `target` or read outside `from`. + let resolved: Vec<(PathBuf, PathBuf, &FileEntry)> = manifest + .files + .iter() + .map(|entry| { + let src = safe_join(from, &entry.path)?; + let dest = safe_join(target, &entry.path)?; + Ok((src, dest, entry)) + }) + .collect::>()?; + // Phase 1: verify EVERY file's hash against the manifest BEFORE writing // anything, so a corrupt backup fails the restore whole rather than leaving a // half-written target. - for entry in &manifest.files { - let src = from.join(slash_to_rel(&entry.path)); - let bytes = std::fs::read(&src) + for (src, _dest, entry) in &resolved { + let bytes = std::fs::read(src) .map_err(|e| CliError::new(format!("read backup file {}: {e}", src.display())))?; let hash = blake3::hash(&bytes).to_hex().to_string(); if hash != entry.blake3 { @@ -199,17 +250,19 @@ pub(crate) fn run_restore( std::fs::create_dir_all(target) .map_err(|e| CliError::new(format!("create target {}: {e}", target.display())))?; let mut restored = 0usize; - for entry in &manifest.files { - let src = from.join(slash_to_rel(&entry.path)); - let dest = target.join(slash_to_rel(&entry.path)); + for (src, dest, _entry) in &resolved { if let Some(parent) = dest.parent() { std::fs::create_dir_all(parent) .map_err(|e| CliError::new(format!("mkdir {}: {e}", parent.display())))?; } - std::fs::copy(&src, &dest) + std::fs::copy(src, dest) .map_err(|e| CliError::new(format!("copy {}: {e}", dest.display())))?; + fsync_path(dest).map_err(|e| CliError::new(format!("fsync {}: {e}", dest.display())))?; restored += 1; } + // Flush the target's directory tree so the restored dirents are durable. + fsync_dirs_recursive(target) + .map_err(|e| CliError::new(format!("fsync target dirs {}: {e}", target.display())))?; let summary = serde_json::json!({ "restored": target.display().to_string(), @@ -224,15 +277,15 @@ pub(crate) fn run_restore( /// Recursively collect file paths (relative to `root`), skipping the lock file /// and any previously-written backup manifest. Directories are recursed; only /// regular files are recorded. -fn collect_files(root: &Path, rel: PathBuf, out: &mut Vec) -> std::io::Result<()> { - let dir = root.join(&rel); +fn collect_files(root: &Path, rel: &Path, out: &mut Vec) -> std::io::Result<()> { + let dir = root.join(rel); for entry in std::fs::read_dir(&dir)? { let entry = entry?; let name = entry.file_name(); let child_rel = rel.join(&name); let file_type = entry.file_type()?; if file_type.is_dir() { - collect_files(root, child_rel, out)?; + collect_files(root, &child_rel, out)?; } else if file_type.is_file() { // Skip process-ownership + a stale manifest from a prior backup-in-place. if name == LOCK_FILE || name == BACKUP_MANIFEST { @@ -244,6 +297,74 @@ fn collect_files(root: &Path, rel: PathBuf, out: &mut Vec) -> std::io:: Ok(()) } +/// Resolve an UNTRUSTED forward-slash manifest path against a trusted `base`, +/// rejecting anything that would escape `base`. +/// +/// Restore reads the path from a backup directory that may have been tampered +/// with; `Path::join` does not normalize, so a `..` or absolute component would +/// otherwise let a crafted manifest read or write outside `base` (the canonical +/// Zip-Slip / tar-traversal hole). Only plain name components are accepted — `..`, +/// a root/prefix component, or a backslash separator is an error; empty and `.` +/// segments are skipped. An absolute-looking entry (a leading `/`) is therefore +/// neutralized to a contained relative path rather than replacing `base`. +fn safe_join(base: &Path, rel: &str) -> Result { + if rel.contains('\\') { + return Err(CliError::new(format!( + "unsafe manifest path {rel:?}: backslash separator" + ))); + } + let mut out = base.to_path_buf(); + let mut pushed = false; + // Empty (leading/trailing/duplicate slash) and `.` segments are no-ops. + for seg in rel.split('/').filter(|s| !s.is_empty() && *s != ".") { + if seg == ".." { + return Err(CliError::new(format!( + "unsafe manifest path {rel:?}: parent-directory ('..') component" + ))); + } + // A single segment must itself be exactly one Normal component (rejects a + // Windows drive prefix like `C:` or any root marker). + let mut comps = Path::new(seg).components(); + match (comps.next(), comps.next()) { + (Some(Component::Normal(name)), None) => { + out.push(name); + pushed = true; + } + _ => { + return Err(CliError::new(format!( + "unsafe manifest path {rel:?}: non-name component {seg:?}" + ))); + } + } + } + if !pushed { + return Err(CliError::new(format!( + "unsafe manifest path {rel:?}: empty after normalization" + ))); + } + Ok(out) +} + +/// fsync a file or directory so its bytes/dirents survive a host or power crash, +/// not just a process crash. Opening a directory read-only and `sync_all`-ing it +/// flushes its dirents on both Linux and macOS; for a file it flushes data + +/// metadata. Mirrors the engine's `tidal/src/db/backup.rs` durability barrier. +fn fsync_path(path: &Path) -> std::io::Result<()> { + std::fs::File::open(path)?.sync_all() +} + +/// fsync every directory under `root` (children before parents) and `root` +/// itself, so a freshly-written/restored tree's dirents are all durable. +fn fsync_dirs_recursive(root: &Path) -> std::io::Result<()> { + for entry in std::fs::read_dir(root)? { + let entry = entry?; + if entry.file_type()?.is_dir() { + fsync_dirs_recursive(&entry.path())?; + } + } + fsync_path(root) +} + /// Render a relative path with forward slashes for a portable manifest. fn rel_to_slash(rel: &Path) -> String { rel.components() @@ -252,11 +373,6 @@ fn rel_to_slash(rel: &Path) -> String { .join("/") } -/// Inverse of [`rel_to_slash`]: a manifest path → a platform `PathBuf`. -fn slash_to_rel(path: &str) -> PathBuf { - path.split('/').collect() -} - fn serialize(manifest: &Manifest, pretty: bool) -> Result { if pretty { serde_json::to_string_pretty(manifest) diff --git a/tidalctl/tests/cli.rs b/tidalctl/tests/cli.rs index aee4225..3598b4d 100644 --- a/tidalctl/tests/cli.rs +++ b/tidalctl/tests/cli.rs @@ -1315,3 +1315,97 @@ fn restore_rejects_corrupted_backup() { "error must name the integrity failure: {stderr}" ); } + +/// Restore must REFUSE a manifest whose path escapes the target (Zip-Slip / tar +/// traversal). A tampered backup controls the path, the bytes, AND the hash, so +/// BLAKE3 verification cannot catch it — `safe_join` rejects the path up front and +/// nothing is written outside the target. +#[test] +fn restore_rejects_path_traversal_manifest() { + let home = home_with_wal_data(); + let scratch = TempTidalHome::new().unwrap(); + let backup_dir = scratch.path().join("backup"); + let restore_parent = scratch.path().join("restore_parent"); + let restore_dir = restore_parent.join("target"); + std::fs::create_dir_all(&restore_dir).unwrap(); + // Remove it again so the empty-target guard passes (we just wanted the parent). + std::fs::remove_dir(&restore_dir).unwrap(); + + let out = tidalctl_bin() + .args(["backup", "--path"]) + .arg(home.path()) + .args(["--out"]) + .arg(&backup_dir) + .output() + .unwrap(); + assert!(out.status.success(), "backup failed: {out:?}"); + + // Tamper the manifest: inject an entry that escapes the target via `..`. + let manifest_path = backup_dir.join("BACKUP_MANIFEST.json"); + let mut manifest: serde_json::Value = + serde_json::from_slice(&std::fs::read(&manifest_path).unwrap()).unwrap(); + manifest["files"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({ + "path": "../../pwned.txt", + "size": 0, + "blake3": "00", + })); + std::fs::write(&manifest_path, serde_json::to_vec(&manifest).unwrap()).unwrap(); + + let out = tidalctl_bin() + .args(["restore", "--from"]) + .arg(&backup_dir) + .args(["--path"]) + .arg(&restore_dir) + .output() + .unwrap(); + assert!( + !out.status.success(), + "restore must reject a traversal manifest" + ); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!( + stderr.contains("unsafe manifest path"), + "error must name the unsafe path: {stderr}" + ); + // Nothing escaped the target: the `..`-relative file was never written. + assert!( + !restore_parent.join("pwned.txt").exists() && !scratch.path().join("pwned.txt").exists(), + "traversal file must not exist outside the target" + ); +} + +/// Backup must REFUSE a source whose WAL is corrupt/unreadable rather than stamp a +/// fabricated `checkpoint_seq: 0` (indistinguishable from a fresh store) into a +/// "successful" backup — the cursor is the coordinated-restore point. +#[test] +fn backup_refuses_corrupt_source_wal() { + let home = home_with_wal_data(); + corrupt_wal_checkpoint(&home); + let scratch = TempTidalHome::new().unwrap(); + let backup_dir = scratch.path().join("backup"); + + let out = tidalctl_bin() + .args(["backup", "--path"]) + .arg(home.path()) + .args(["--out"]) + .arg(&backup_dir) + .output() + .unwrap(); + assert!( + !out.status.success(), + "backup must refuse a corrupt source WAL" + ); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!( + stderr.contains("WAL checkpoint") || stderr.contains("unknown cursor"), + "error must name the unreadable cursor: {stderr}" + ); + // No manifest claiming a fabricated cursor was left behind. + assert!( + !backup_dir.join("BACKUP_MANIFEST.json").exists(), + "a refused backup must not write a manifest" + ); +}