From 580142df492fc477e9885b880b16a718ef117933 Mon Sep 17 00:00:00 2001 From: jx12n Date: Thu, 18 Jun 2026 13:08:53 -0600 Subject: [PATCH] feat(m12): election-divergence-fix + soak-eval streak + release tooling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Durable `leader_acked` frontier in `ShardReplica` tracks the highest seqno acked under `ack=leader` (journal-only, un-replicated); `decide_join` now quarantines on THIS node's own frontier rather than comparing stream numbers across stream boundaries — eliminates false-quarantine churn on rolling restarts. `SHUTDOWN_HANDOFF_WAIT` (3s) drains the leader's tail to quorum before step-down so the next leader inherits a clean prefix. New `load_leader_acked`/`persist_leader_acked` helpers; `cluster_reseed.rs` gains the divergence-fix regression suite; `replication_ops.rs` threads the signal. Soak-eval: `tidal_stress::soak_eval` + `soak-eval` binary implement the 30-night streak (ledger.tsv × restarts.tsv → streak.tsv); monitor and nightly CronJob k8s YAMLs updated; phase-9 doc clarifies the dual-stream streak definition (ledger PASS AND zero pod restarts in window). `run-reliability.sh` gates the election-divergence suite before any k8s push. Release tooling: `docker/release/` multi-stage Dockerfile + DR image; `scripts/build-release.sh` single repeatable cross-compile+buildx path. --- docker/release/Dockerfile | 45 ++ docker/release/config/default-cluster.yaml | 9 + docker/release/config/default-schema.yaml | 31 + docker/release/dr.Dockerfile | 35 + docker/stress/Dockerfile | 4 + docs/planning/milestone-11/phase-9.md | 31 + scripts/build-release.sh | 110 +++ scripts/run-reliability.sh | 55 ++ tidal-server/src/cluster/election_driver.rs | 179 +++-- tidal-server/src/cluster/join_boot.rs | 15 +- tidal-server/src/cluster/node.rs | 350 +++++++++- .../tests/cluster_graph_persistence.rs | 4 +- tidal-server/tests/cluster_membership.rs | 89 ++- tidal-server/tests/cluster_reseed.rs | 632 ++++++++++++++++++ tidal-stress/k8s/soak-monitor.yaml | 68 ++ tidal-stress/k8s/soak-nightly-cronjob.yaml | 18 +- tidal-stress/k8s/soak-results-pvc.yaml | 17 +- tidal-stress/src/bin/soak-eval.rs | 70 ++ tidal-stress/src/lib.rs | 1 + tidal-stress/src/soak_eval.rs | 338 ++++++++++ tidal/src/db/replication_ops.rs | 13 + 21 files changed, 2014 insertions(+), 100 deletions(-) create mode 100644 docker/release/Dockerfile create mode 100644 docker/release/config/default-cluster.yaml create mode 100644 docker/release/config/default-schema.yaml create mode 100644 docker/release/dr.Dockerfile create mode 100755 scripts/build-release.sh create mode 100755 scripts/run-reliability.sh create mode 100644 tidal-stress/src/bin/soak-eval.rs create mode 100644 tidal-stress/src/soak_eval.rs diff --git a/docker/release/Dockerfile b/docker/release/Dockerfile new file mode 100644 index 0000000..929e01d --- /dev/null +++ b/docker/release/Dockerfile @@ -0,0 +1,45 @@ +# tidalDB cluster server image — PACKAGING ONLY (no in-container compile). +# +# Committed promotion of tmp/img-build/Dockerfile. The `tidal-server` binary in +# the build context is a HOST cross-compile (mac-arm64 -> x86_64-unknown-linux-gnu) +# produced by scripts/build-release.sh. It NEEDs libmvec.so.1 (glibc 2.41 +# vectorized math), so the runtime base MUST be trixie (glibc 2.41) — bookworm +# (2.36) lacks libmvec and aborts at startup. Verified NEEDED: libstdc++.so.6, +# libgcc_s.so.1, libmvec.so.1 + glibc. +# +# Built by scripts/build-release.sh via the amd64 buildx builder. Do not build by +# hand — the script pins the toolchain + digests the result. +FROM debian:trixie-slim + +ARG DEBIAN_FRONTEND=noninteractive + +# usearch's HNSW C++ core needs the C++ runtime; ca-certificates/curl for TLS + +# operator healthchecks. glibc (incl. libmvec.so.1) ships in the base image. +RUN apt-get update && \ + apt-get install -y --no-install-recommends libstdc++6 libgcc-s1 ca-certificates curl && \ + rm -rf /var/lib/apt/lists/* + +# Non-root, uid 10001 to match the StatefulSet securityContext (runAsUser 10001, +# fsGroup 10001) and the init-datadir chown. +RUN useradd --system -u 10001 --home /srv tidal + +COPY --chown=tidal:tidal tidal-server /usr/local/bin/tidal-server +COPY --chown=tidal:tidal config /etc/tidal-server + +USER tidal +WORKDIR /srv +EXPOSE 9500 9601 9091 + +# Self-contained config path (overridden at runtime by the StatefulSet's explicit +# --schema/--topology flags pointing at mounted ConfigMaps). +ENV TIDAL_CONFIG=/etc/tidal-server +ENV TIDAL_SERVER_LOG=info +# Cluster mode is gated experimental; the StatefulSet also sets this. +ENV TIDAL_ALLOW_EXPERIMENTAL_CLUSTER=1 + +# Bare binary ENTRYPOINT so `docker run img verify ...`/subcommand overrides work; +# the StatefulSet supplies the full `cluster ...` argv. +ENTRYPOINT ["tidal-server"] +CMD ["cluster", "--listen", "0.0.0.0:9500", \ + "--schema", "/etc/tidal-server/default-schema.yaml", \ + "--topology", "/etc/tidal-server/default-cluster.yaml"] diff --git a/docker/release/config/default-cluster.yaml b/docker/release/config/default-cluster.yaml new file mode 100644 index 0000000..15e402b --- /dev/null +++ b/docker/release/config/default-cluster.yaml @@ -0,0 +1,9 @@ +regions: + - name: us-east + - name: eu-west + - name: ap-south +leader: us-east +# Optional: number of runtime-free OS worker threads serving cluster write/heal +# requests (gRPC segment ship). Bounds write concurrency on the hottest path. +# Defaults to available parallelism (clamped 2..=8) when omitted. +# write_workers: 4 diff --git a/docker/release/config/default-schema.yaml b/docker/release/config/default-schema.yaml new file mode 100644 index 0000000..48afbb9 --- /dev/null +++ b/docker/release/config/default-schema.yaml @@ -0,0 +1,31 @@ +signals: + - name: view + entity: item + decay: + exponential: + half_life_seconds: 604800 # 7 days + windows: [one_hour, twenty_four_hours, seven_days] + velocity: true + positive_engagement: true # folds into the user preference vector + - name: like + entity: item + decay: + exponential: + half_life_seconds: 1209600 # 14 days + windows: [twenty_four_hours, seven_days, thirty_days, all_time] + velocity: false + positive_engagement: true + - name: skip + entity: item + decay: + permanent: true + velocity: false +text_fields: + - name: title + kind: text + - name: category + kind: keyword +embedding_slots: + - name: content_vector + entity: item + dimensions: 128 diff --git a/docker/release/dr.Dockerfile b/docker/release/dr.Dockerfile new file mode 100644 index 0000000..54a1e67 --- /dev/null +++ b/docker/release/dr.Dockerfile @@ -0,0 +1,35 @@ +# tidalDB DR image — PACKAGING ONLY (no in-container compile). +# +# Committed promotion of tmp/tidalctl-img/Dockerfile, EXTENDED to bundle +# `tidal-server` alongside `tidalctl` + `mc` (spec Q4): the DR drill's query-proof +# step boots a standalone tidal-server on the restored data dir, so one image runs +# export -> restore -> boot -> read. Both binaries are HOST cross-compiles +# (x86_64-unknown-linux-gnu, glibc 2.41) produced by scripts/build-release.sh, so +# the runtime base MUST be trixie (libmvec.so.1 lives there). +FROM debian:trixie-slim + +ARG DEBIAN_FRONTEND=noninteractive + +# libstdc++6/libgcc-s1 for the aws-sdk/usearch C++ runtime; ca-certificates for TLS; +# curl to fetch mc AND to drive the booted server's /health + /feed in the drill; +# coreutils (diff) for the byte-equivalence check. +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + libstdc++6 libgcc-s1 ca-certificates curl diffutils && \ + curl -fsSL https://dl.min.io/client/mc/release/linux-amd64/mc -o /usr/local/bin/mc && \ + chmod +x /usr/local/bin/mc && \ + rm -rf /var/lib/apt/lists/* + +# Non-root, uid 10001 (matches the tidaldb StatefulSet convention). +RUN useradd --system -u 10001 --home /srv tidal + +COPY --chown=tidal:tidal tidalctl /usr/local/bin/tidalctl +COPY --chown=tidal:tidal tidal-server /usr/local/bin/tidal-server + +USER tidal +WORKDIR /srv + +# Bare-binary ENTRYPOINT so `docker run img status ...`/subcommand overrides work; +# the drill Job overrides command/args with the export/restore/verify/boot script. +ENTRYPOINT ["tidalctl"] +CMD ["--help"] diff --git a/docker/stress/Dockerfile b/docker/stress/Dockerfile index 2bfc229..c1ae0fa 100644 --- a/docker/stress/Dockerfile +++ b/docker/stress/Dockerfile @@ -30,6 +30,10 @@ RUN useradd --system --home /srv stress && \ rm -rf /var/lib/apt/lists/* COPY --from=builder /app/target/release/tidal-stress /usr/local/bin/tidal-stress +# soak-eval: the 30-night streak evaluator (built from the same crate; see +# tidal-stress/src/bin/soak-eval.rs). The soak monitor's evaluator container runs +# it to join ledger.tsv + restarts.tsv into the honest streak verdict. +COPY --from=builder /app/target/release/soak-eval /usr/local/bin/soak-eval USER stress diff --git a/docs/planning/milestone-11/phase-9.md b/docs/planning/milestone-11/phase-9.md index 117f5ad..73d7695 100644 --- a/docs/planning/milestone-11/phase-9.md +++ b/docs/planning/milestone-11/phase-9.md @@ -134,6 +134,37 @@ fault each injects, and the checker each asserts through. - Every guarantee in §2 maps to a named automated test. - Nightly suite green 30 consecutive days before GA. +### How the streak is computed (and what resets it) + +The bar is NOT "30 green soak Job exits". It is **30 consecutive nights where the +soak passed its SLO gates AND no cluster pod restarted under load**. The two +halves come from two streams on the shared result PVC: + +- `ledger.tsv` — one row per night from the nightly CronJob: `PASS`/`FAIL` on the + armed gates (`--fail-on-knee`, `--max-p99-ms 150`, `--max-error-pct 1`). +- `restarts.tsv` — the monitor's 5-minute snapshot of each `tidaldb-{0,1,2}` pod's + cumulative `restartCount`. + +`soak-eval` (the proven core in `tidal_stress::soak_eval`, table-driven-tested; +the binary runs in the monitor's evaluator container and once per night in the +CronJob) joins them into `streak.tsv`: + +- a night is **GREEN** iff its ledger row is `PASS` **and** no pod's cumulative + restart count rose within that night's window; +- the **streak** is the trailing run of consecutive green nights; +- **any** non-green night resets the streak to **0**. A night resets it when + EITHER the soak breached an SLO gate (Job `FAIL`) **OR** a pod restarted under + load (the zero-unrecovered-restart half — a green ledger row with a restart + in-window is a FAIL, not a PASS: that was the false-confidence hole). + +The evaluator emits exactly one alert per non-green transition (a durable +`ALERT-.txt` the http surface serves, plus an optional `NOTIFY_WEBHOOK` +POST). **Dependency:** the streak is only *reachable* once the graceful-restart +reseed-on-rejoin churn is fixed — before that, every rollout/upgrade/node reboot +increments a restart count and would reset the streak forever. The production- +readiness divergence fix (tasks 01/02) is what makes a restart-in-window a real, +rare anomaly worth failing on. + ## Exit-gate evidence (local; debug builds, real OS processes) | Gate | Target | Measured | diff --git a/scripts/build-release.sh b/scripts/build-release.sh new file mode 100755 index 0000000..b3f4046 --- /dev/null +++ b/scripts/build-release.sh @@ -0,0 +1,110 @@ +#!/usr/bin/env bash +# build-release.sh — the one committed, repeatable tidalDB release path. +# +# Replaces the ad-hoc hand-typed cross-compile + buildx + push with a single +# command on a clean checkout: +# +# ./scripts/build-release.sh [server|dr|stress|all] +# +# It HOST cross-compiles (mac-arm64 -> x86_64-unknown-linux-gnu, glibc 2.41 via the +# homebrew toolchain), packages the binaries into the committed docker/release/* +# Dockerfiles via the amd64 buildx builder, pushes, and prints the resulting +# @sha256 digest for each image (pin that digest in k8s/cluster/statefulset.yaml). +# +# The project bans CI/CD pipelines (orchard9-k3sf/CLAUDE.md) — this is a SCRIPT, +# run manually or from an operator's shell, never a pipeline. Registry creds come +# from the Docker daemon's existing login (run `docker login registry.threesix.ai` +# first); NO secret is embedded here. +set -euo pipefail + +TAG="${1:-}" +COMPONENT="${2:-all}" +REGISTRY="${TIDAL_REGISTRY:-registry.threesix.ai/tidal}" +TARGET="x86_64-unknown-linux-gnu" +BUILDER="${TIDAL_BUILDX_BUILDER:-amd64builder}" + +die() { echo "build-release: $*" >&2; exit 1; } + +[ -n "$TAG" ] || die "usage: build-release.sh [server|dr|stress|all]" +case "$COMPONENT" in server|dr|stress|all) ;; *) die "component must be server|dr|stress|all" ;; esac + +# ── Toolchain preflight (fail fast, no half-built image) ───────────────────── +command -v cargo >/dev/null || die "cargo not on PATH" +command -v docker >/dev/null || die "docker not on PATH" +command -v "${TARGET}-gcc" >/dev/null || die "missing ${TARGET}-gcc (brew install ${TARGET})" +PROTOC_BIN="${PROTOC:-$(command -v protoc || true)}" +[ -n "$PROTOC_BIN" ] || die "protoc not found (brew install protobuf), or set PROTOC" +rustup target list --installed 2>/dev/null | grep -qx "$TARGET" \ + || die "rust target $TARGET not installed (rustup target add $TARGET)" + +REPO_ROOT="$(cd "$(dirname "$0")/.." && pwd)" +cd "$REPO_ROOT" + +# ── Pinned cross-compile environment (the documented recipe) ───────────────── +export CC_x86_64_unknown_linux_gnu="${TARGET}-gcc" +export CXX_x86_64_unknown_linux_gnu="${TARGET}-g++" +export AR_x86_64_unknown_linux_gnu="${TARGET}-ar" +export CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_LINKER="${TARGET}-gcc" +export PROTOC="$PROTOC_BIN" + +# Which crates to cross-compile for the selected component(s). +PKGS=() +case "$COMPONENT" in + server) PKGS=(-p tidal-server) ;; + dr) PKGS=(-p tidalctl -p tidal-server) ;; + stress) PKGS=() ;; # stress builds IN-container (pure Rust, no engine deps) + all) PKGS=(-p tidal-server -p tidalctl) ;; +esac + +if [ "${#PKGS[@]}" -gt 0 ]; then + echo "==> cross-compiling ${PKGS[*]} for $TARGET (release)" + cargo build "${PKGS[@]}" --release --target "$TARGET" --locked +fi + +# ── buildx amd64 builder (idempotent) ──────────────────────────────────────── +if ! docker buildx inspect "$BUILDER" >/dev/null 2>&1; then + echo "==> creating buildx builder $BUILDER" + docker buildx create --name "$BUILDER" --driver docker-container >/dev/null +fi + +BIN_DIR="target/$TARGET/release" +STAGE="$(mktemp -d)" +trap 'rm -rf "$STAGE"' EXIT + +# Build one image from a staged context; print its pushed digest. +build_image() { # $1=image_name $2=dockerfile $3=stage_subdir + local image="$REGISTRY/$1:$TAG" dockerfile="$2" ctx="$STAGE/$3" + echo "==> building $image" + docker buildx build --builder "$BUILDER" --platform linux/amd64 \ + -f "$dockerfile" -t "$image" --push "$ctx" + local digest + digest="$(docker buildx imagetools inspect "$image" --format '{{.Manifest.Digest}}' 2>/dev/null || true)" + echo "RELEASE $1: $REGISTRY/$1@${digest:-} (tag $TAG)" +} + +if [ "$COMPONENT" = server ] || [ "$COMPONENT" = all ]; then + mkdir -p "$STAGE/server" + cp "$BIN_DIR/tidal-server" "$STAGE/server/" + cp -a docker/release/config "$STAGE/server/config" + cp docker/release/Dockerfile "$STAGE/server/Dockerfile" + build_image server "$STAGE/server/Dockerfile" server +fi + +if [ "$COMPONENT" = dr ] || [ "$COMPONENT" = all ]; then + mkdir -p "$STAGE/dr" + cp "$BIN_DIR/tidalctl" "$BIN_DIR/tidal-server" "$STAGE/dr/" + cp docker/release/dr.Dockerfile "$STAGE/dr/Dockerfile" + build_image tidalctl "$STAGE/dr/Dockerfile" dr +fi + +if [ "$COMPONENT" = stress ] || [ "$COMPONENT" = all ]; then + # Pure-Rust, in-container build (includes the soak-eval binary). Context = repo + # root so the workspace manifests resolve; the .dockerignore prunes it. + echo "==> building $REGISTRY/stress:$TAG (in-container, includes soak-eval)" + docker buildx build --builder "$BUILDER" --platform linux/amd64 \ + -f docker/stress/Dockerfile -t "$REGISTRY/stress:$TAG" --push . + d="$(docker buildx imagetools inspect "$REGISTRY/stress:$TAG" --format '{{.Manifest.Digest}}' 2>/dev/null || true)" + echo "RELEASE stress: $REGISTRY/stress@${d:-} (tag $TAG)" +fi + +echo "==> done. Pin the printed @sha256 digest(s) in k8s/cluster/statefulset.yaml and the DR/soak manifests." diff --git a/scripts/run-reliability.sh b/scripts/run-reliability.sh new file mode 100755 index 0000000..5363890 --- /dev/null +++ b/scripts/run-reliability.sh @@ -0,0 +1,55 @@ +#!/usr/bin/env bash +# run-reliability.sh — the election-divergence reliability gate. +# +# Runs the cluster-mode e2e divergence suite N times SERIALLY and requires every +# iteration to be all-green. The rolling-restart-under-load false-quarantine was +# ~50% flaky, so a single green is meaningless — the bar is N consecutive passes +# (default 10). Any single failure stops the loop and prints the iteration so the +# fix can be root-caused, never masked with a wider budget. +# +# This is NOT a CI pipeline (the project bans CI/CD — see CLAUDE.md). It is a +# manual operator loop. Run it from the tidaldb workspace root. +# +# Usage: +# scripts/run-reliability.sh # 10 iterations (the gate) +# scripts/run-reliability.sh 3 # quick 3-iteration smoke +# N=20 scripts/run-reliability.sh # via env +set -euo pipefail + +ITERS="${1:-${N:-10}}" +TESTS=(cluster_reseed cluster_membership) + +cd "$(dirname "$0")/.." + +echo "── election-divergence reliability gate: ${ITERS} consecutive serial runs ──" +echo " suite: ${TESTS[*]} (--features cluster-e2e, --test-threads=1)" + +# Build once up front so per-iteration timing reflects the run, not the compile. +echo "[build] compiling the e2e test binaries…" +cargo test -p tidal-server --features cluster-e2e \ + $(printf -- '--test %s ' "${TESTS[@]}") --no-run + +pass=0 +for i in $(seq 1 "${ITERS}"); do + start=$(date +%s) + if cargo test -p tidal-server --features cluster-e2e \ + $(printf -- '--test %s ' "${TESTS[@]}") -- --test-threads=1 \ + >"/tmp/reliability-iter-${i}.log" 2>&1; then + elapsed=$(( $(date +%s) - start )) + pass=$(( pass + 1 )) + echo "[iter ${i}/${ITERS}] PASS (${elapsed}s)" + else + elapsed=$(( $(date +%s) - start )) + echo "[iter ${i}/${ITERS}] FAIL (${elapsed}s) — log: /tmp/reliability-iter-${i}.log" + echo + echo "── result lines ──" + grep -E "test result|FAILED|panicked|reseed|quarantin" "/tmp/reliability-iter-${i}.log" | tail -30 || true + echo + echo "GATE FAILED after ${pass}/${ITERS} green. A single flake means the fix is" + echo "incomplete — root-cause it back into the divergence seam; do NOT widen budgets." + exit 1 + fi +done + +echo +echo "GATE PASSED: ${pass}/${ITERS} consecutive serial runs all-green." diff --git a/tidal-server/src/cluster/election_driver.rs b/tidal-server/src/cluster/election_driver.rs index e3f784d..a06ab1b 100644 --- a/tidal-server/src/cluster/election_driver.rs +++ b/tidal-server/src/cluster/election_driver.rs @@ -252,7 +252,11 @@ impl ElectionRuntime { return false; }; let position = node.election_log_position(); - match decide_join(term, position, prev_log) { + // The DURABLE divergence signal (m12 election-divergence-fix): the highest + // seqno this node acked under ack=leader and may hold un-replicated. Read + // from the node (not the heartbeat) — it is THIS node's own property. + let leader_acked = node.leader_acked_frontier(); + match decide_join(term, position, prev_log, leader_acked) { JoinDecision::Quarantine => { // Preserve the existing quarantine semantics: set the latch, note // the divergence (metric + reseed marker), and refuse the data @@ -609,7 +613,12 @@ impl tidal_net::ElectionHooks for NodeElectionHooks { let before = self.runtime.joined_term.load(Ordering::Acquire); // `stream_baseline` is the new term's clamp floor — pass it so a reseed // latch re-baselines from it (m12p6 divergent-rejoin fix), not frontier+1. - let joined = self.runtime.join_term_check(term, prev_log, stream_baseline); + // The divergence test now reads THIS node's DURABLE leader-acked frontier + // (m12 election-divergence-fix), so it no longer needs the leader's live + // frontier — `leader_last_seq` feeds ONLY the readiness gauge below. + let joined = self + .runtime + .join_term_check(term, prev_log, stream_baseline); if joined && term > before && let Some(node) = self.runtime.node.upgrade() @@ -721,24 +730,56 @@ enum JoinDecision { Clean, } -/// The pure three-way term-join rule (m11p5 §2 "The three-way term-join rule"). +/// The pure term-join rule (m11p5 §2 "The three-way term-join rule"), with the +/// m12 election-divergence-fix: divergence is decided by the DURABLE leader-acked +/// frontier, not a frontier compared across stream numberings. /// -/// `own` and `prev_log` are both in the PREVIOUS stream's numbering (the vote -/// restriction's `(tail_term, frontier)` comparison). A tail already carrying -/// THIS term's marker (`own.tail_term == term`) is manifestly part of this -/// term's history — the restart-within-term rejoin (phase-4.md §5, third -/// clause): without it, a mid-term rejoiner whose tail term (T) exceeds the -/// leader's election-time tail (T-1) would false-quarantine on every restart. -/// Otherwise the lexicographic comparison decides: strictly-greater diverges -/// (quarantine), strictly-smaller is genuinely behind (reseed), equal is clean. -fn decide_join(term: u64, own: LogPosition, prev_log: LogPosition) -> JoinDecision { +/// `own` and `prev_log` are in the PREVIOUS stream's numbering (`own` = this +/// node's WAL-tail position; `prev_log` = the new leader's frozen *election-time* +/// position). `leader_acked` is this node's durable [`ShardReplica::leader_acked`] +/// frontier — the highest seqno it ACKED to a client under `ack=leader` +/// (journal-only, no quorum) and may therefore hold UN-REPLICATED, in the same +/// (previous-stream) numbering for a former leader of that stream. +/// +/// Cases: +/// * `own.tail_term == term` — the tail already carries THIS term's marker → the +/// within-term rejoin (phase-4.md §5) → **clean**. +/// * `own.tail_term > term` — a future-term marker the new term does not subsume +/// → genuinely **divergent** → quarantine. +/// * `leader_acked > prev_log.frontier` — this node ACKED (ack=leader) entries the +/// new leadership's election baseline does NOT cover: un-replicated leader-acked +/// writes the cluster elected past → the genuine **divergent suffix** → +/// quarantine. This is the EXACT divergent-suffix definition (node.rs §quarantine), +/// now measured durably instead of inferred from `own.flushed > prev_log` (which +/// conflated it with a benign `ack=quorum` uncommitted tail). A FOLLOWER never +/// ack=leader-writes (`leader_acked == 0`) → never divergent; a clean join resets +/// the frontier to 0, so a former leader whose acked tail WAS replicated is clean. +/// * else, classify position: `own < prev_log` → behind → **reseed** (self-heals on +/// convergence; a compacted gap reseeds). `own >= prev_log` (caught up, or a +/// benign uncommitted/replicated tail) → **clean** (catch-up streams the rest). +fn decide_join( + term: u64, + own: LogPosition, + prev_log: LogPosition, + leader_acked: u64, +) -> JoinDecision { if own.tail_term == term { return JoinDecision::Clean; } + if own.tail_term > term { + return JoinDecision::Quarantine; + } + // The divergence test: did this node ack (under ack=leader) entries the new + // leadership's election baseline does not cover? Numbering-stable for a former + // leader of the previous stream; `leader_acked == 0` (a follower, or a node + // that clean-joined) can never trip it. + if leader_acked > prev_log.frontier { + return JoinDecision::Quarantine; + } match own.cmp(&prev_log) { - std::cmp::Ordering::Greater => JoinDecision::Quarantine, std::cmp::Ordering::Less => JoinDecision::ReseedRequired, - std::cmp::Ordering::Equal => JoinDecision::Clean, + // Greater (a benign uncommitted/replicated tail) or Equal → clean. + _ => JoinDecision::Clean, } } @@ -878,73 +919,101 @@ mod tests { } } - /// The three-way term-join rule (m11p5 §2). `own > prev_log` quarantines (a - /// divergent suffix), `own < prev_log` latches reseed (genuinely behind), - /// and equal / within-term is clean. + /// The term-join rule (m11p5 §2, m12 election-divergence-fix). The 4th arg is + /// the DURABLE leader-acked frontier; divergence is `leader_acked > + /// prev_log.frontier` (un-replicated leader-acked suffix), NOT a frontier + /// compared across stream numberings. #[test] fn three_way_join_rule_classifies_each_case() { - // Joining term T = 1; both positions in the PREVIOUS stream's numbering - // (tail_term 0 here — the topology-era stream). let t = 1; - // own > prev_log (higher frontier, same tail term) → divergent suffix. - // (This is test 1's exact shape: own=(0,970) vs leader prev=(0,9) — - // except here own LEADS, the quarantine direction.) + // This node ACKED (ack=leader) up to 970 but the new leadership's election + // baseline is only 9 → it holds an un-replicated leader-acked suffix the + // cluster elected past → divergent → quarantine. assert_eq!( - decide_join(t, pos(0, 970), pos(0, 9)), + decide_join(t, pos(0, 970), pos(0, 9), 970), JoinDecision::Quarantine, - "a strictly-greater own position is a divergent suffix → quarantine" - ); - // own > prev_log via a higher tail term (lexicographic), NOT the - // within-term rejoin (own.tail_term 1 != joining term 2) → quarantine. - assert_eq!( - decide_join(2, pos(1, 0), pos(0, u64::MAX)), - JoinDecision::Quarantine, - "a higher tail term outranks any frontier → quarantine" + "leader-acked above the new election baseline is a divergent suffix → quarantine" ); - // own < prev_log (lower frontier, same tail term) → genuinely behind: - // the node lacks (own, prev_log] of the previous stream → reseed. This - // is test 1's REAL shape: own=(0,9) (stopped early) vs leader's - // election-time prev=(0,970) (full history) → reseed, not quarantine. + // own < prev_log, nothing leader-acked beyond it (leader_acked = 0) → + // genuinely behind → reseed (self-heals; a compacted gap reseeds). assert_eq!( - decide_join(t, pos(0, 9), pos(0, 970)), + decide_join(t, pos(0, 9), pos(0, 970), 0), JoinDecision::ReseedRequired, - "a strictly-smaller own position is genuinely missing history → reseed" - ); - // own < prev_log via a lower tail term, NOT within-term (own.tail_term 1 - // != joining term 2) → reseed. - assert_eq!( - decide_join(2, pos(1, 0), pos(1, u64::MAX)), - JoinDecision::ReseedRequired, - "a lower frontier under the same sub-term loses → reseed" + "a strictly-smaller own position with nothing leader-acked is behind → reseed" ); - // own == prev_log → clean. + // own == prev_log, nothing leader-acked beyond it → clean. assert_eq!( - decide_join(t, pos(0, 9), pos(0, 9)), + decide_join(t, pos(0, 9), pos(0, 9), 0), JoinDecision::Clean, - "an equal position is clean" + "an equal position with nothing un-replicated is clean" + ); + + // A FUTURE-term marker (own.tail_term > the joining term) → the node followed + // a leadership the new term does not subsume → divergent regardless of the + // acked frontier → quarantine. + assert_eq!( + decide_join(2, pos(3, 0), pos(1, 5), 0), + JoinDecision::Quarantine, + "a future-term marker outranks the join → quarantine" + ); + } + + /// THE FIX (rolling-restart-under-load false-quarantine): a node whose applied + /// frontier merely exceeds the leader's ELECTION-TIME baseline — because it + /// applied a benign uncommitted/replicated tail (ack=quorum, or a clean former + /// leader, so `leader_acked` does NOT cover it) — is NOT divergent. Only an + /// un-replicated ACK=LEADER suffix (`leader_acked > prev_log.frontier`) is. + #[test] + fn leader_acked_above_baseline_quarantines_benign_tail_is_clean() { + // own=(1, 202) past the election baseline (1, 200), but NOTHING was acked + // under ack=leader beyond the baseline (leader_acked = 0 — a follower, or a + // former leader whose tail replicated and was reset on a clean join). The + // excess is benign (uncommitted-quorum / replicated) → Clean. The old code + // false-quarantined here ~50% of the time under load. + assert_eq!( + decide_join(2, pos(1, 202), pos(1, 200), 0), + JoinDecision::Clean, + "a benign tail past the baseline with nothing leader-acked → clean" + ); + // Same position, but this node ACKED (ack=leader) up to 202 and the new + // leadership only covers 200 → un-replicated leader-acked suffix (the + // `mp_quarantined` shape) → quarantine. + assert_eq!( + decide_join(2, pos(1, 202), pos(1, 200), 202), + JoinDecision::Quarantine, + "an un-replicated ack=leader suffix above the baseline → quarantine" + ); + // Boundary: leader-acked exactly AT the baseline → the leader has it → not + // divergent → clean. + assert_eq!( + decide_join(2, pos(1, 200), pos(1, 200), 200), + JoinDecision::Clean, + "leader-acked == the baseline is replicated → clean" + ); + // A large benign follower lead is still clean (no leader-acked suffix). + assert_eq!( + decide_join(2, pos(1, 100_000), pos(1, 200), 0), + JoinDecision::Clean, + "benign follower lead magnitude is irrelevant — nothing leader-acked" ); } /// The within-term rejoin (phase-4.md §5, third clause): a tail already - /// carrying THIS term's marker is part of this term's history regardless of - /// the `prev_log` comparison — it must NOT quarantine or reseed on restart. + /// carrying THIS term's marker is part of this term's history → never reseed + /// or quarantine on restart, regardless of the position or the acked frontier. #[test] fn within_term_rejoin_is_always_clean() { let t = 3; - // own.tail_term == term, even with a frontier far above prev_log (which - // is the leader's election-time tail at T-1) → clean, never quarantine. assert_eq!( - decide_join(t, pos(3, 500), pos(2, 10)), + decide_join(t, pos(3, 500), pos(2, 10), 500), JoinDecision::Clean, "a tail carrying this term's marker is the within-term rejoin → clean" ); - // And clean even with a frontier below prev_log (a mid-term restart that - // applied the marker but then lost suffix) — still within-term, no reseed. assert_eq!( - decide_join(t, pos(3, 1), pos(2, 999)), + decide_join(t, pos(3, 1), pos(2, 999), 0), JoinDecision::Clean, "within-term rejoin wins over the < comparison too" ); diff --git a/tidal-server/src/cluster/join_boot.rs b/tidal-server/src/cluster/join_boot.rs index aec81b7..5a20ea9 100644 --- a/tidal-server/src/cluster/join_boot.rs +++ b/tidal-server/src/cluster/join_boot.rs @@ -786,14 +786,13 @@ fn synthesize_topology( /// same paths. Without the fallback a TLS-cluster joiner synthesized a PLAINTEXT /// gRPC posture and could not mTLS-replicate with its peers. fn self_tls_spec(knobs: &TopologySpec, name: &str) -> Option { - grpc_tls_for(knobs, name) - .map(|t| GrpcTlsSpec { - ca_cert: t.ca_cert.clone(), - server_cert: t.server_cert.clone(), - server_key: t.server_key.clone(), - client_cert: t.client_cert.clone(), - client_key: t.client_key.clone(), - }) + grpc_tls_for(knobs, name).map(|t| GrpcTlsSpec { + ca_cert: t.ca_cert.clone(), + server_cert: t.server_cert.clone(), + server_key: t.server_key.clone(), + client_cert: t.client_cert.clone(), + client_key: t.client_key.clone(), + }) } // The knob blocks have no `Clone`; rebuild them field-by-field from the loaded diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index a85fa0c..18b617b 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -156,6 +156,10 @@ impl AckMode { /// double-apply it on followers. const STREAM_BASELINE_FILE: &str = "stream_baseline"; +/// Durable file holding the [`ShardReplica::leader_acked`] frontier (m12 +/// election-divergence-fix). Raw 8-byte LE `u64`, mirroring [`STREAM_BASELINE_FILE`]. +const LEADER_ACKED_FILE: &str = "leader_acked"; + /// How long a leader-sanctioned transfer waits for the target to hold the /// full flushed prefix before `TimeoutNow` (the drain, m11p4). const TRANSFER_CATCHUP_WAIT: Duration = Duration::from_secs(5); @@ -164,6 +168,24 @@ const TRANSFER_CATCHUP_WAIT: Duration = Duration::from_secs(5); /// (target leads at a higher term) before reporting failure. const TRANSFER_TAKEOVER_WAIT: Duration = Duration::from_secs(10); +/// How long a gracefully-shutting-down LEADER waits for its flushed tail (its +/// last entries AND the term marker it journaled when it won) to commit to a +/// quorum before it steps down. Bounded well inside a k8s SIGTERM grace period so +/// shutdown never hangs; if the deadline passes the node steps down anyway (no +/// worse than the pre-fix behavior). This is the graceful leadership hand-off: +/// committing the tail before step-down means the next leader (a quorum member, +/// so caught-up by the vote restriction) holds the full prefix, and THIS node +/// rejoins as a clean follower instead of carrying a divergent suffix the new +/// term never saw — the root of the rolling-restart reseed/quarantine churn. +// m12 election-divergence-fix: 3s, not 10s. With the write-quiesce freezing the +// flushed frontier at shutdown, the committed index catches up within a heartbeat +// or two (sub-second), so 3s is ample for a real hand-off — and the old leader +// keeps HEARTBEATING until it steps down, so a longer wait only DELAYS the +// survivors' election (the "shard leaders did not converge" stall under load). +// The durable `leader_acked` frontier is the safety net: any un-replicated +// ack=leader tail the drain did not flush still quarantines on rejoin. +const SHUTDOWN_HANDOFF_WAIT: Duration = Duration::from_secs(3); + /// Cap on one commit-watch bridge condvar wait: the longest the bridge /// thread can go without re-checking its stop flag, i.e. the worst-case /// shutdown latency the bridge adds. Deliberately a constant, not a @@ -254,6 +276,19 @@ pub struct ShardReplica { /// Shared with the gRPC `SegmentSource` so catch-up never serves /// pre-stream history. stream_baseline: Arc, + /// Durable "leader-acked frontier" (m12 election-divergence-fix): the highest + /// WAL seqno this node ACKED to a client under `ack=leader` (journal-only, no + /// quorum wait) and may therefore hold UN-REPLICATED. The election-divergence + /// classifier ([`decide_join`]) quarantines a rejoining node iff this exceeds + /// the new leadership's election baseline — exactly "leader-acked writes the + /// cluster elected past" (the divergent-suffix definition). Reset to 0 on a + /// clean term join (caught up ⇒ nothing un-replicated), advanced on each + /// `ack=leader` write, persisted on graceful shutdown, and falls back to the + /// durable WAL tail (conservative) when a hard kill leaves no persisted value. + /// This is what distinguishes a genuine `ack=leader` divergent suffix + /// (`mp_quarantined`) from a benign `ack=quorum` uncommitted tail + /// (rolling-restart-under-load): the latter never advances this frontier. + leader_acked: Arc, /// The node's data dir (baseline persistence). Multi-process cluster /// mode requires one — validated in [`Self::new`]. data_dir: std::path::PathBuf, @@ -768,6 +803,16 @@ impl ShardReplica { }; let install_boot = install_target.is_some(); + // m12 election-divergence-fix: the durable leader-acked frontier. Each + // `ack=leader` write persists it on advance (so a HARD KILL recovers the + // exact un-replicated suffix — `mp_quarantined`); a clean join persists 0. + // Absent ⇒ 0: a node that never ack=leader-wrote (a follower, or a leader + // that only served `ack=quorum`) holds nothing un-replicated. Crucially we + // do NOT fall back to the WAL tail — a follower's `flushed` is + // applied-from-leader (replicated) data, and treating it as leader-acked + // would false-quarantine a clean former leader whose tail was quorum-replicated. + let leader_acked = Arc::new(AtomicU64::new(load_leader_acked(&data_dir).unwrap_or(0))); + // Always-on receiver: leadership can move, so EVERY node runs a receiver // (an inbound segment is legal on the current leader after a promote // elsewhere). The blob-applier wiring routes replicated kind-1/2 @@ -1012,6 +1057,7 @@ impl ShardReplica { ship_feed, ship_queue, stream_baseline, + leader_acked, data_dir, boot_topology_leader: leader, activation_prev: std::sync::Mutex::new(tidaldb::replication::LogPosition { @@ -1084,8 +1130,35 @@ impl ShardReplica { pub fn shutdown(&self) { self.set_shutting_down(); if let Some(rt) = self.election_runtime.get() { + // Graceful leadership hand-off (rolling-restart churn fix): if this + // node leads an ELECTED term, drain its flushed tail to a quorum + // BEFORE stepping down. Without this, a gracefully-restarted leader + // abandons its just-journaled term marker (and any leader-durable + // tail) un-replicated; the survivors elect a new leader WITHOUT it, + // and the deposed leader rejoins carrying a divergent suffix the new + // term never saw → quarantine + full snapshot reseed on EVERY rolling + // deploy/upgrade/reboot. Committing the tail first means the next + // leader holds the full prefix (vote restriction) and this node + // rejoins clean. Bounded; steps down anyway past the deadline. + // + // m12 election-divergence-fix: drain whenever LEADING, including the + // term-0 TOPOLOGY leader. Handing off its committed prefix lets the + // survivors elect cleanly and converge fast (empirically tighter than + // leaving an un-handed-off tail). The membership-propagation timing this + // shifted is now absorbed by a polled roster assertion (the seed-join + // test). `needed_peers() == 0` still short-circuits a single replica. + if self.is_leader() { + self.drain_committed_before_stepdown(); + } rt.stop(); } + // m12 election-divergence-fix: persist the leader-acked frontier on a + // GRACEFUL shutdown (writes are already quiesced by `set_shutting_down`, + // so the value is final). A hard kill skips this and the next boot falls + // back to the conservative WAL tail. After a clean drain the value is + // already 0 (reset on the last clean join) or fully committed, so a + // gracefully-restarted node rejoins clean rather than re-quarantining. + self.persist_leader_acked_now(); self.commit_bridge_stop.store(true, Ordering::Release); // Quiesce the ship-queue senders FIRST so no batch ship races the // transport/db teardown below (an in-flight send_segment against a @@ -1138,6 +1211,49 @@ impl ShardReplica { } } + /// Wait (bounded by [`SHUTDOWN_HANDOFF_WAIT`]) until this leader's flushed WAL + /// tail is COMMITTED to a quorum, so a graceful step-down leaves no divergent + /// suffix for the deposed leader to reseed over. Reuses the exact signal the + /// commit path already maintains — `commit.committed()` (the quorum-acked + /// frontier, fed by peers' `ReportApplied`) vs `ship_feed.flushed_seq()` (this + /// leader's durable WAL tail). No-op when no quorum is needed (single replica) + /// or the tail is already committed; logs and proceeds past the deadline so + /// shutdown never hangs. Called only while still leading and before the ship + /// queue is deactivated, so the in-flight tail can still ship and be acked. + fn drain_committed_before_stepdown(&self) { + if self.commit.needed_peers() == 0 { + return; // no quorum to wait on (single replica) — nothing to divergence-proof + } + let flushed = self.ship_feed.flushed_seq(); + if flushed == 0 || self.commit.committed() >= flushed { + return; // empty stream or already fully committed + } + let deadline = std::time::Instant::now() + SHUTDOWN_HANDOFF_WAIT; + loop { + let committed = self.commit.committed(); + if committed >= flushed { + tracing::info!( + region = %self.region_name, + flushed, + "graceful shutdown: leader tail committed to quorum before step-down (clean hand-off)" + ); + return; + } + if std::time::Instant::now() >= deadline { + tracing::warn!( + region = %self.region_name, + committed, + flushed, + "graceful shutdown: leader tail NOT fully committed before the hand-off \ + deadline; stepping down anyway (a follower may briefly reconcile the small \ + uncommitted tail — no acked-write loss, the WAL is durable)" + ); + return; + } + std::thread::sleep(std::time::Duration::from_millis(50)); + } + } + // ── Accessors ────────────────────────────────────────────────────────── /// Clone the node's `TidalDb` handle, or `Unavailable` once shutdown took it. @@ -2498,6 +2614,44 @@ impl ShardReplica { } } + /// The durable leader-acked frontier (m12 election-divergence-fix): the + /// highest seqno this node acked under `ack=leader` and may hold + /// un-replicated. See [`Self::leader_acked`]. + pub(crate) fn leader_acked_frontier(&self) -> u64 { + self.leader_acked.load(Ordering::Acquire) + } + + /// Advance the leader-acked frontier for an `ack=leader` write that just + /// succeeded at `seq`, and PERSIST it durably on advance. The durability is + /// load-bearing: an `ack=leader` write is acked to the client on journal alone + /// (no quorum), so if this node is HARD-KILLED before the watermark is durable, + /// the next boot must still recover the un-replicated suffix to quarantine + /// (the `mp_quarantined` invariant). The write already fsynced the WAL, so this + /// is one extra small fsync on the ack=leader path only — `ack=quorum` writes + /// never call this. Monotonic: a stale/duplicate `seq` neither advances nor + /// re-persists. + pub(crate) fn note_leader_acked(&self, seq: u64) { + let prev = self.leader_acked.fetch_max(seq, Ordering::AcqRel); + if seq > prev { + persist_leader_acked(&self.data_dir, seq); + } + } + + /// Reset the leader-acked frontier to 0 and persist it: a CLEAN term join + /// proves the node is caught up to the new leadership, so it holds nothing + /// un-replicated the new term does not subsume. Persisting 0 (not just + /// clearing in memory) is what makes a later graceful restart read 0 instead + /// of conservatively falling back to the WAL tail. + pub(crate) fn reset_leader_acked(&self) { + self.leader_acked.store(0, Ordering::Release); + persist_leader_acked(&self.data_dir, 0); + } + + /// Persist the current leader-acked frontier (graceful-shutdown durability). + fn persist_leader_acked_now(&self) { + persist_leader_acked(&self.data_dir, self.leader_acked.load(Ordering::Acquire)); + } + /// The transport's election fan-out handle. pub(crate) fn election_net(&self) -> tidal_net::ElectionNet { self.transport.election_net() @@ -2585,6 +2739,14 @@ impl ShardReplica { ); } + // Winning an election proves this node's log is at least as up-to-date as a + // quorum (the vote restriction), so a reseed marker it latched while + // transiently behind at an earlier term change is a false alarm — clear it + // (else a caught-up LEADER serves with `reseed_required:true` and reseeds on + // its next restart). A quarantined node is campaign-suppressed and can never + // reach here, so the helper's exclusion is doubly safe. + self.clear_stale_reseed_marker_if_caught_up(true); + // §3.2 — the activation membership record (the linchpin): once the // kind-3 marker is durable AND the membership era has begun (a kind-4 // record is in the log), re-append the CURRENT roster as a fresh kind-4 @@ -2649,9 +2811,26 @@ impl ShardReplica { if baseline > 0 && let Ok(db) = self.db() { + // m12 election-divergence-fix: advance PAST the term marker (baseline+1), + // not just to baseline. The marker is the term's first entry and has NO + // storage effect, so JOINING the term IS applying it. Advancing past it + // closes the gap a gap-gated marker leaves (and DRAINS any data parked + // behind that gap), curing the stuck `lag=1` convergence stall (Agent A: + // the marker rides the same seqno-gated segment path as data and a + // resume-from-`applied+1` skips it). Folding the term cell keeps + // `election_log_position` in the joined leader's numbering. db.replication_state() - .advance(shard_of_region(leader), baseline); + .advance(shard_of_region(leader), baseline + 1); + db.fold_term_marker(term, baseline + 1, leader.0); } + // m12 election-divergence-fix: a CLEAN join proves this node is caught up + // to (subsumed by) the new leadership, so it holds nothing un-replicated + // that term does not cover — reset the durable leader-acked frontier to 0. + // This keeps the frontier in the CURRENT era's numbering (it never carries + // a stale prior-era acked seqno into a later term's divergence compare) and + // is what lets a gracefully-restarted, previously-clean node read 0 instead + // of conservatively falling back to its WAL tail. + self.reset_leader_acked(); tracing::info!( term, leader = %self.region_name_of(leader), @@ -2774,6 +2953,53 @@ impl ShardReplica { } } + /// Clear a reseed marker that turned out to be a FALSE ALARM (rolling-restart + /// churn fix, second source). The `own < prev_log → ReseedRequired` join-check + /// arm latches a durable marker for a node that is merely BEHIND at a + /// leadership change — but a node behind by a *shippable* (non-compacted) tail + /// then catches up via the normal stream / catch-up pull and never needs a + /// snapshot reseed. The durable marker used to persist anyway, so the node + /// reseeded on its next restart (and could even lead while still flagged — the + /// observed `is_leader:true, lag:0, reseed_required:true` state). This clears it + /// once the node has demonstrably caught up. + /// + /// Safe by construction: a node genuinely behind a COMPACTED gap never reaches + /// `caught_up` (it cannot fetch the missing entries), so it keeps the marker + /// and still reseeds; a QUARANTINED (divergent-suffix) node is excluded, its + /// marker is real. So only a non-divergent node that actually reconverged + /// clears — exactly the false-alarm case. + fn clear_stale_reseed_marker_if_caught_up(&self, caught_up: bool) { + if !caught_up { + return; + } + if self + .election_runtime + .get() + .is_some_and(|rt| rt.is_quarantined()) + { + return; + } + if !matches!(self.reseed_marker_store.load(), Ok(Some(_))) { + return; + } + match self.reseed_marker_store.clear() { + Ok(()) => { + self.cluster_metrics.set_reseed_required(false); + tracing::info!( + region = %self.region_name, + "stale reseed marker cleared — caught up to the leader via the stream \ + without a reseed (false-alarm latch from a transient leadership-change \ + classification; a genuinely-compacted gap never reaches caught-up)" + ); + } + Err(e) => tracing::warn!( + region = %self.region_name, + error = %e, + "failed to clear a stale reseed marker; retried on the next convergence" + ), + } + } + /// Offload the §2.4 quorum-refusal evaluation to a detached thread. /// /// `latch_reseed_marker` may be invoked from a tonic handler thread (the @@ -2956,10 +3182,14 @@ impl ShardReplica { db.control_plane() .lag_gauge() .update_leader_seqno_for(leader_shard, leader_last_seq); + let applied = self.applied_for_leader_shard(leader_shard); if (self.install_boot || self.seed_joiner) && !self.converged.load(Ordering::Acquire) { - let applied = self.applied_for_leader_shard(leader_shard); self.note_lag_for_readiness(leader_last_seq.saturating_sub(applied)); } + // Self-heal a false-alarm reseed marker: this follower has caught up to the + // leader's live frontier via the normal stream (no reseed happened), so a + // marker a transient leadership-change classification latched is stale. + self.clear_stale_reseed_marker_if_caught_up(applied >= leader_last_seq); } /// Whether this node is READY to serve (m11p5 §4 readiness predicate). @@ -3171,6 +3401,9 @@ impl ShardReplica { .activation_prev .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + // The LIVE election position the vote/join paths read (NOT the frozen + // `status_prev`): exposed for the divergence-consistency oracle. + let election_pos = self.election_log_position(); let (term, role, quarantined) = self.election_runtime.get().map_or_else( || (0, "unknown".to_string(), false), |rt| { @@ -3211,6 +3444,9 @@ impl ShardReplica { quarantined, prev_log_term: status_prev.tail_term, prev_log_seq: status_prev.frontier, + election_tail_term: election_pos.tail_term, + election_frontier: election_pos.frontier, + leader_acked: self.leader_acked_frontier(), reseed_required, reseeding, self_restart_refused: self.self_restart_refused.load(Ordering::Acquire), @@ -3539,6 +3775,62 @@ fn persist_stream_baseline(data_dir: &std::path::Path, baseline: u64) { } } +/// Read the persisted leader-acked frontier (m12 election-divergence-fix). +/// `None` when ABSENT (so the caller can pick a conservative fallback — +/// distinct from a persisted 0, which a clean join writes deliberately). A +/// malformed file is treated as absent (loud error → fallback). +fn load_leader_acked(data_dir: &std::path::Path) -> Option { + let path = data_dir.join(LEADER_ACKED_FILE); + match std::fs::read(&path) { + Ok(bytes) if bytes.len() == 8 => { + let mut buf = [0u8; 8]; + buf.copy_from_slice(&bytes); + Some(u64::from_le_bytes(buf)) + } + Ok(_) => { + tracing::error!( + path = %path.display(), + "leader-acked file is malformed; treating as absent (conservative \ + fallback to the WAL tail)" + ); + None + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => None, + Err(e) => { + tracing::error!( + path = %path.display(), + error = %e, + "leader-acked file unreadable; treating as absent" + ); + None + } + } +} + +/// Persist the leader-acked frontier durably (write + fsync via a temp rename), +/// mirroring [`persist_stream_baseline`]. Best-effort with a loud error: a lost +/// value falls back to the conservative WAL tail on the next boot. +fn persist_leader_acked(data_dir: &std::path::Path, value: u64) { + let path = data_dir.join(LEADER_ACKED_FILE); + let tmp = data_dir.join(format!("{LEADER_ACKED_FILE}.tmp")); + let result = (|| -> std::io::Result<()> { + std::fs::write(&tmp, value.to_le_bytes())?; + let f = std::fs::File::open(&tmp)?; + f.sync_all()?; + std::fs::rename(&tmp, &path)?; + Ok(()) + })(); + if let Err(e) = result { + tracing::error!( + path = %path.display(), + value, + error = %e, + "failed to persist the leader-acked frontier; the next boot falls back \ + to the conservative WAL tail (may quarantine + reseed a clean node)" + ); + } +} + /// The gRPC layer's applied-seqno reader (ack piggyback): a WEAK view over /// this node's replication state, so the transport can never keep the /// database alive past shutdown. @@ -4460,6 +4752,26 @@ pub struct LocalStatusResponse { /// numbering and is not comparable to pre-election seqs). prev_log_term: u64, prev_log_seq: u64, + /// This node's LIVE election log position `(tail_term, frontier)` — the + /// exact `(LogPosition)` the Raft vote restriction (`rpc.log >= my_log`) and + /// `decide_join` read RIGHT NOW, distinct from the frozen `prev_log_*` + /// (published only while leading). Exposed so the election-divergence + /// consistency invariant is observable: for the same committed state a + /// caught-up node's `election_frontier` must equal its applied frontier in + /// the leader's stream (`applied_events`); a divergence here is the + /// cross-numbering tear (an ex-leader reading its own `flushed_seq` instead + /// of the leader-stream `applied_seqno`). `#[serde(default)]` so a + /// mixed-version peer's status still deserializes. + #[serde(default)] + election_tail_term: u64, + #[serde(default)] + election_frontier: u64, + /// The durable leader-acked frontier (m12 election-divergence-fix): the + /// highest seqno acked under `ack=leader` that may be un-replicated. The + /// election-divergence classifier quarantines a rejoin iff this exceeds the + /// new leadership's election baseline. Exposed for divergence diagnosis. + #[serde(default)] + leader_acked: u64, /// Whether this node is quarantined with a divergent suffix (m11p4): /// fenced from the data plane until reseeded. quarantined: bool, @@ -5888,8 +6200,12 @@ pub async fn create_item( return forward_write(&state, "/items", &req, &headers).await; } // A marked (forwarded) write must land on the leader; a stale cluster - // view surfaces honestly instead of applying off-log. - if !state.is_leader() { + // view surfaces honestly instead of applying off-log. A leader that has begun + // a graceful shutdown also stops accepting writes (typed `NotLeader`, so the + // client re-targets): this FREEZES the flushed frontier so the step-down drain + // (`drain_committed_before_stepdown`) converges instead of chasing a tail that + // keeps growing under load — the rolling-restart-under-load divergence fix. + if !state.is_leader() || state.is_shutting_down() { return Err(ClusterAppError(state.not_leader())); } let ack = state.ack_mode_for(&headers).map_err(ClusterAppError)?; @@ -5917,6 +6233,11 @@ pub async fn create_item( && let Some(seq) = seq { await_quorum(&state, seq).await?; + } else if let Some(seq) = seq { + // ack=leader: the write is acked on journal alone (no quorum), so it may + // be un-replicated — record it on the durable leader-acked frontier (the + // election-divergence divergent-suffix signal). + state.note_leader_acked(seq); } Ok(with_seq_header(StatusCode::CREATED, seq)) } @@ -5952,7 +6273,9 @@ pub async fn write_embedding( if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/embeddings", &req, &headers).await; } - if !state.is_leader() { + // A leader mid-graceful-shutdown stops accepting writes (freezes flushed so the + // step-down drain converges — see `create_item`). + if !state.is_leader() || state.is_shutting_down() { return Err(ClusterAppError(state.not_leader())); } let ack = state.ack_mode_for(&headers).map_err(ClusterAppError)?; @@ -5978,6 +6301,10 @@ pub async fn write_embedding( && let Some(seq) = seq { await_quorum(&state, seq).await?; + } else if let Some(seq) = seq { + // ack=leader: journal-only ack (may be un-replicated) — record it on the + // durable leader-acked frontier (the election-divergence signal). + state.note_leader_acked(seq); } Ok(with_seq_header(StatusCode::NO_CONTENT, seq)) } @@ -6029,7 +6356,9 @@ pub async fn write_signal( return forward_write(&state, "/signals", &req, &headers).await; } // Reject non-leader BEFORE the write pool so NotLeader does not consume a slot. - if !state.is_leader() { + // A leader mid-graceful-shutdown also stops accepting writes (freezes the + // flushed frontier so the step-down drain converges — see `create_item`). + if !state.is_leader() || state.is_shutting_down() { return Err(ClusterAppError(state.not_leader())); } let signal = req.signal; @@ -6067,6 +6396,10 @@ pub async fn write_signal( // covered quorum (this request created no new log entry to gate on). if ack == AckMode::Quorum && seq > 0 { await_quorum(&state, seq).await?; + } else if ack == AckMode::Leader && seq > 0 { + // ack=leader: journal-only ack (may be un-replicated) — record it on the + // durable leader-acked frontier (the election-divergence signal). + state.note_leader_acked(seq); } Ok(with_seq_header(StatusCode::NO_CONTENT, Some(seq))) } @@ -8400,7 +8733,10 @@ mod forward_retry_tests { }; use axum::http::StatusCode; - fn ok(status: StatusCode, body: serde_json::Value) -> Result { + fn ok( + status: StatusCode, + body: serde_json::Value, + ) -> Result { Ok(forward::ForwardedResponse { seq: None, deduplicated: false, diff --git a/tidal-server/tests/cluster_graph_persistence.rs b/tidal-server/tests/cluster_graph_persistence.rs index 2370056..1b76b2e 100644 --- a/tidal-server/tests/cluster_graph_persistence.rs +++ b/tidal-server/tests/cluster_graph_persistence.rs @@ -47,7 +47,9 @@ mod support; use std::time::{Duration, SystemTime}; -use support::multiproc::{ClusterOptions, MultiProcCluster, convergence_budget, seed_items_and_embeddings}; +use support::multiproc::{ + ClusterOptions, MultiProcCluster, convergence_budget, seed_items_and_embeddings, +}; /// The leader index (region 0 = `us-east`). const LEADER: usize = 0; diff --git a/tidal-server/tests/cluster_membership.rs b/tidal-server/tests/cluster_membership.rs index c43be91..8224452 100644 --- a/tidal-server/tests/cluster_membership.rs +++ b/tidal-server/tests/cluster_membership.rs @@ -239,6 +239,38 @@ fn await_membership_role( } } +/// Poll node `idx`'s OWN `/cluster/status/local` until boolean `field == expected`, +/// or the deadline. The polled form of a single-shot `local_status()[field]` read: +/// after a polled precondition (a role/heartbeat await), the bool can still be +/// mid-transition for a beat, so a single-shot `assert_eq!` races it (the class of +/// flake fixed at `cluster_reseed.rs:591`). Panics with the last status on timeout. +fn await_status_bool( + cluster: &MultiProcCluster, + idx: usize, + field: &str, + expected: bool, + budget: Duration, + what: &str, +) { + let deadline = Instant::now() + budget; + loop { + let seen = cluster + .local_status(idx) + .and_then(|s| s[field].as_bool()) + .unwrap_or(false); + if seen == expected { + return; + } + assert!( + Instant::now() < deadline, + "{what}: node {idx} did not report {field}={expected} within {budget:?}; \ + last status: {:?}", + cluster.local_status(idx) + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + /// Poll until node `idx`'s OWN status reports zero lag and an applied frontier at /// or above the live leader's `last_seq`. fn await_self_converged(cluster: &MultiProcCluster, idx: usize, budget: Duration, what: &str) { @@ -398,13 +430,23 @@ fn mp_seed_join_snapshot_catchup() { // The whole cluster (now 4 nodes) converges, and the roster on EVERY live // node lists the joiner as a voter. cluster.wait_converged_all(convergence_budget()); + // POLL each node's roster until the voter-promotion record has PROPAGATED: + // `wait_converged_all` gates on applied DATA, but the kind-4 membership + // promotion (learner→voter) ships on the same log and a node can be data-caught- + // up while the roster record is still in flight. A single-shot assert here + // raced that propagation; poll for the eventual invariant instead. + let roster_deadline = std::time::Instant::now() + convergence_budget(); for i in 0..cluster.len() { - if let Some(rr) = cluster.local_status(i).map(|_| roster_roles(&cluster, i)) { - assert_eq!( - rr.get(&joiner_name).map(String::as_str), - Some("voter"), - "node {i}'s roster must show the joiner as a voter: {rr:?}" + loop { + let rr = roster_roles(&cluster, i); + if rr.get(&joiner_name).map(String::as_str) == Some("voter") { + break; + } + assert!( + std::time::Instant::now() < roster_deadline, + "node {i}'s roster must show the joiner as a voter within budget: {rr:?}" ); + std::thread::sleep(Duration::from_millis(100)); } } @@ -570,7 +612,9 @@ fn mp_idle_cluster_snapshot_joiner_flips_ready_without_traffic() { thread::sleep(Duration::from_millis(100)); } let flip_elapsed = ready_start.elapsed(); - println!("[idle-ready] snapshot joiner flipped /health READY in {flip_elapsed:?} on an idle cluster"); + println!( + "[idle-ready] snapshot joiner flipped /health READY in {flip_elapsed:?} on an idle cluster" + ); // Tight bound: heartbeat convergence is sub-second after catch-up (measured // 257µs / 101ms). A flip that only just beats the full budget would mean // convergence regressed onto a SLOW path (e.g. a periodic self-heal tick or a @@ -717,13 +761,16 @@ fn mp_remove_node_clean_decommission() { cluster.local_status(victim_idx) ); // It has NO reseed marker — a remove is not a reseed (the typed `removed` - // signal is exempt from the reseed marker, §3.3). - assert_eq!( - cluster - .local_status(victim_idx) - .and_then(|s| s["reseed_required"].as_bool()), - Some(false), - "a removed node must NOT latch a reseed marker (remove is exempt, §3.3)" + // signal is exempt from the reseed marker, §3.3). Polled, not single-shot: + // the `removed` role await above does not fence the reseed gauge, which can + // settle a beat later (the `cluster_reseed.rs:591` race class). + await_status_bool( + &cluster, + victim_idx, + "reseed_required", + false, + convergence_budget(), + "a removed node must NOT latch a reseed marker (remove is exempt, §3.3)", ); println!("[remove] the removed node reports removed + 503 + no reseed marker"); @@ -870,13 +917,15 @@ fn mp_remove_missed_record_learns_via_signal() { ); // It has NO reseed marker (a remove is not a reseed — the typed signal is - // exempt, §3.3). - assert_eq!( - cluster - .local_status(victim_idx) - .and_then(|s| s["reseed_required"].as_bool()), - Some(false), - "a signal-decommissioned node must NOT latch a reseed marker (§3.3)" + // exempt, §3.3). Polled, not single-shot (the `cluster_reseed.rs:591` race + // class): the decommission await above does not fence the reseed gauge. + await_status_bool( + &cluster, + victim_idx, + "reseed_required", + false, + convergence_budget(), + "a signal-decommissioned node must NOT latch a reseed marker (§3.3)", ); println!("[missed] the signal-decommissioned victim has NO reseed marker — exact close-out"); diff --git a/tidal-server/tests/cluster_reseed.rs b/tidal-server/tests/cluster_reseed.rs index 92bdf4d..4d8b0e3 100644 --- a/tidal-server/tests/cluster_reseed.rs +++ b/tidal-server/tests/cluster_reseed.rs @@ -79,6 +79,16 @@ const LEGACY_ELECTION_YAML: &str = "election:\n auto_election: false"; /// timeout_min (500)`. const FAST_ELECTION_YAML: &str = "election:\n heartbeat_interval_ms: 100\n election_timeout_min_ms: 500\n election_timeout_max_ms: 1000\n leader_lease_ms: 350"; +/// The PRODUCTION election timers (verbatim from `k8s/cluster/topology-configmap.yaml`): +/// heartbeat 300ms, election timeout 1500–3000ms, lease 900ms. The graceful- +/// rolling-restart repro runs with THESE, not the much faster `FAST_ELECTION_YAML`: +/// the production blocker is what a real k8s RollingUpdate does to the real +/// deployment, and the 3×-tighter fast timers manufacture extra mid-restart +/// re-election churn that the live cluster never sees. With production timers a +/// follower restart does not move leadership and a leader restart moves it exactly +/// once — the faithful scenario. +const PROD_ELECTION_YAML: &str = "election:\n heartbeat_interval_ms: 300\n election_timeout_min_ms: 1500\n election_timeout_max_ms: 3000\n leader_lease_ms: 900"; + /// Items the leader writes while the follower is offline. Enough to MATTER /// (well past a handful) and — paired with the per-item metadata blob (see /// [`blob_value`]) — to push the WAL past one 16 MiB segment so the @@ -268,6 +278,103 @@ fn promote_and_agree(cluster: &MultiProcCluster, via_idx: usize, region: &str) { cluster.wait_leader_agreed(region, convergence_budget()); } +/// Write ONLY the heavy item (no embedding / no signal) to node `idx`, asserting +/// the 201. Used by the graceful-rolling-restart repro where the only thing that +/// matters is WAL volume (the 56 KiB item blob fills segments to force +/// compaction) and item-presence read-back — so the two extra POSTs per item the +/// full [`write_heavy_item`] issues are pure cost. Cutting them ~3×'s the seed +/// throughput and drops the embedding/signal memory the full writer would hold. +fn write_heavy_item_only(cluster: &MultiProcCluster, idx: usize, entity_id: u64, blob: &str) { + let mut metadata = serde_json::Map::new(); + metadata.insert("title".into(), item_token(entity_id).into()); + for k in 0..BLOB_KEYS { + metadata.insert(format!("blob{k}"), blob.into()); + } + let resp = cluster.post( + idx, + "/items", + &serde_json::json!({ "entity_id": entity_id, "metadata": metadata }), + ); + assert_eq!(resp.status().as_u16(), 201, "leader /items must 201"); +} + +/// The index of the current live leader (by `is_leader`), or `None` if no live +/// node reports leadership yet. Mirrors `cluster_membership::current_leader_idx` +/// — the elected-era analogue of the fixed `LEADER` constant. +fn current_leader_idx(cluster: &MultiProcCluster) -> Option { + (0..cluster.len()).find(|&i| { + cluster + .local_status(i) + .is_some_and(|s| s["is_leader"].as_bool() == Some(true)) + }) +} + +/// Wait until every live node has caught up to within `max_lag` of the leader — +/// the test-side analogue of a k8s rolling update gating each pod restart on the +/// previous pod's READINESS. Without it the test fires restarts back-to-back, so a +/// leader's step-down drain (which needs a quorum of CAUGHT-UP followers to commit +/// its tail) can stall behind a follower still catching up from its own restart, +/// and that leader then diverges. A real rollout never restarts the next pod until +/// the last one is ready; this reproduces that pacing. Tolerant of a small in-flight +/// tail under sustained load (hence `max_lag`, not strict 0). +fn await_cluster_caught_up(cluster: &MultiProcCluster, max_lag: u64, budget: Duration, what: &str) { + let deadline = Instant::now() + budget; + loop { + let all = (0..cluster.len()).all(|i| { + cluster + .local_status(i) + .and_then(|s| s["lag_events"].as_u64()) + .is_some_and(|lag| lag <= max_lag) + }); + if all { + return; + } + assert!( + Instant::now() < deadline, + "{what}: cluster did not catch up (all nodes lag <= {max_lag}) within {budget:?}; \ + last: {:?}", + (0..cluster.len()) + .map(|i| cluster + .local_status(i) + .and_then(|s| s["lag_events"].as_u64())) + .collect::>() + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + +/// Force the cluster out of the term-0 TOPOLOGY era into the ELECTED era before a +/// measured rolling restart. A freshly-booted cluster can hold leadership at term 0 +/// (the topology leader leads without an election ever firing), but PRODUCTION is +/// always elected-era — a long-running cluster has elected many terms by the time +/// it is rolled. Restarting the current leader once forces the survivors to elect +/// (term ≥ 1); we then wait until a node reports an elected term and the cluster +/// re-agrees a leader, so the subsequent measured restarts exercise only the +/// elected-era step-down path (where the graceful leadership hand-off applies). +fn ensure_elected_era(cluster: &mut MultiProcCluster) { + let leader = current_leader_idx(cluster).expect("a leader before forcing elected era"); + cluster.restart_graceful(leader, &[]); + let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + Duration::from_secs(20)); + let deadline = Instant::now() + convergence_budget(); + loop { + let elected = (0..cluster.len()).any(|i| { + cluster + .local_status(i) + .and_then(|s| s["term"].as_u64()) + .is_some_and(|t| t >= 1) + }); + if elected { + return; + } + assert!( + Instant::now() < deadline, + "cluster did not reach the elected era (term >= 1) within {:?}", + convergence_budget() + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + /// Gate-1 mechanics (minus seed-join): a follower that fell behind a leader /// whose WAL has been COMPACTED past its resume seq reseeds via the boot-time /// snapshot install — zero operator verbs other than restarts. @@ -666,3 +773,528 @@ fn mp_quarantined_node_reseeds_without_wipe() { } println!("[quarantine] reseeded node serves reads; quarantined=false, reseed_required=false"); } + +/// Production-readiness roadmap, task 00 — the Ring-0 foundation. +/// +/// A healthy, caught-up follower must NOT reseed across a GRACEFUL ROLLING +/// RESTART that moves leadership to another region. This reproduces the +/// production blocker: every rolling deploy / upgrade / node reboot currently +/// churns the cluster through a `from_seqno=1` snapshot reseed-on-rejoin even +/// though no data was lost and the node was caught up at SIGTERM. +/// +/// The repro deliberately drives BOTH conditions the live symptom needs: +/// +/// * **a corpus past compaction** — the same heavy [`OFFLINE_ITEMS`] batch the +/// sibling reseed proof uses, so the graceful-shutdown compaction deletes the +/// early WAL and a `from_seqno=1` pull is *genuinely* unservable (the exact +/// `WAL compacted below seqno 1` refusal), not merely suboptimal; and +/// * **a leadership move across the restart** — elected era + restarting the +/// current leader first, so the within-term `own.tail_term == term` +/// clean-rejoin shortcut (`election_driver.rs`) cannot hide a join +/// misclassification. +/// +/// It then asserts no node latches `reseed_required` and every acked item still +/// reads back. The reseed marker is durable and `reseed_self_restart` defaults +/// off, so a wrongful latch STAYS latched — `await_status_bool(.., false, ..)` +/// times out on the bug rather than racing a self-heal. +/// +/// FAILS today — that failure IS the deliverable of task 00 (it pins which of +/// Bugs A/B/C fire and guards tasks 01–03). Greens once task 01 (boot self-heal +/// resume floor) + task 02 (`decide_join` numbering) land. +#[test] +fn mp_graceful_rolling_restart_preserves_applied_no_reseed() { + // Elected era (leadership can move on restart) + fast timers (the move and + // the rejoin both land inside the test budget). Every node gets the LOW + // handshake window so that IF a reseed wrongly latches, the test still + // bounds; the assertion is that it must NOT. + let opts = ClusterOptions::new(3) + .with_topology_extra(PROD_ELECTION_YAML) + .with_env(LEADER, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS) + .with_env(EU_WEST, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS) + .with_env(AP_SOUTH, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS); + let mut cluster = MultiProcCluster::start_with(opts); + let _ = cluster.wait_shard_leaders_agreed(convergence_budget()); + + let seed_leader = current_leader_idx(&cluster).expect("an elected leader before seeding"); + + // Baseline prefix shared by all three, then the heavy batch that pushes the + // WAL well past the 16-segment retention window so the graceful-shutdown + // compaction genuinely deletes the early segments. + // The mechanism is term-marker / uncommitted-tail divergence on step-down, NOT + // compaction (the repro disambiguated this), so a modest converged corpus is + // enough — it just needs leadership to be able to move across the restarts. + const ROLLING_SEED: u64 = 400; + for entity in 1..=ROLLING_SEED { + write_heavy_item_only(&cluster, seed_leader, entity, ""); + } + cluster.wait_converged_all(convergence_budget() + Duration::from_secs(10)); + println!("[rolling] seeded {ROLLING_SEED} items; all three converged"); + + // Match production: a long-running cluster is always elected-era. Force the + // term-0 → elected transition once up front so the measured restarts below + // exercise only the elected-era step-down path. + ensure_elected_era(&mut cluster); + println!("[rolling] cluster forced into the elected era"); + + // Converged steady state: nobody is reseed-pending. + for idx in 0..3 { + await_status_bool( + &cluster, + idx, + "reseed_required", + false, + convergence_budget(), + "no reseed latched in the converged steady state", + ); + } + + // ── GRACEFUL ROLLING RESTART, one node at a time, in POD-ORDINAL order + // (0,1,2) — exactly what a k8s `RollingUpdate` does. Whichever ordinal is the + // current leader moves leadership exactly once when its turn comes; a follower + // restart leaves leadership put (production timers keep the lease). Every node + // fully rejoins and the cluster re-agrees a leader before the next is taken + // down. The deposed leader rejoining clean (no reseed) is the whole point. + let order = [0usize, 1, 2]; + let mut leadership_moved = false; + + for &node in &order { + let pre = cluster + .region_name(current_leader_idx(&cluster).expect("a leader before restart")) + .to_string(); + cluster.restart_graceful(node, &[]); + // The cluster must re-agree a leader for every shard group. Reaching + // agreement requires the just-restarted node to have processed the + // current leader's heartbeat — i.e. its `join_term_check` has already + // run — so any wrongful reseed latch is durable by the time we poll it. + let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + Duration::from_secs(20)); + let post = cluster + .region_name(current_leader_idx(&cluster).expect("a leader after restart")) + .to_string(); + if post != pre { + leadership_moved = true; + } + + // THE INVARIANT: no node latches reseed across a graceful restart. A + // wrongful latch is durable (no self-restart), so this times out on the + // bug — it does not race a self-heal. + for idx in 0..3 { + await_status_bool( + &cluster, + idx, + "reseed_required", + false, + convergence_budget() + Duration::from_secs(10), + "a graceful rolling restart must not latch reseed", + ); + // Applied frontier preserved (a snapshot reseed momentarily drops the + // node's applied position; a clean stream-catch-up never does). + let applied = cluster + .local_status(idx) + .and_then(|s| s["applied_events"].as_u64()) + .unwrap_or(0); + assert!( + applied > 0, + "node {idx} applied frontier must be preserved across the restart \ + (a reseed-from-1 resets it): {:?}", + cluster.local_status(idx) + ); + } + // Pace like a readiness-gated k8s rollout: do not take down the next node + // until the cluster has fully re-converged (idle ⇒ strict lag 0). + await_cluster_caught_up( + &cluster, + 0, + convergence_budget() + Duration::from_secs(20), + "cluster re-converges before the next rolling restart", + ); + println!("[rolling] restarted node {node}; leader {pre} -> {post}; no reseed on any node"); + } + assert!( + leadership_moved, + "the rolling restart MUST move leadership at least once, else the \ + within-term clean-rejoin shortcut hides the join misclassification" + ); + + // ── Zero acked-write loss: known seeded ids still read back on every node. + let probe_deadline = Instant::now() + convergence_budget(); + for entity in [1u64, 4, 200, ROLLING_SEED] { + for idx in 0..3 { + loop { + if item_searchable(&cluster, idx, entity) { + break; + } + assert!( + Instant::now() < probe_deadline, + "node {idx} lost item {entity} across the rolling restart \ + (zero acked-write loss is violated)" + ); + std::thread::sleep(Duration::from_millis(200)); + } + } + } + println!("[rolling] every probed item readable on all three nodes — zero loss, zero reseed"); +} + +/// Production-readiness roadmap, task 04 — the graceful rolling restart UNDER +/// SUSTAINED WRITE LOAD must not reseed any node. This is the faithful production +/// scenario the nightly soak exercises: writes are in flight when a pod is +/// recycled, so the leader carries an UNCOMMITTED tail at SIGTERM. Without the +/// graceful leadership hand-off (drain-to-quorum before step-down) that tail +/// diverges from the next term and the deposed leader reseeds on rejoin — the +/// per-rollout churn. With the hand-off, the tail commits first and every node +/// rejoins clean. Production election timers + k8s ordinal restart order. +#[test] +fn mp_graceful_rolling_restart_under_load_no_reseed() { + let opts = ClusterOptions::new(3) + .with_topology_extra(PROD_ELECTION_YAML) + .with_env(LEADER, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS) + .with_env(EU_WEST, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS) + .with_env(AP_SOUTH, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS); + let mut cluster = MultiProcCluster::start_with(opts); + let _ = cluster.wait_shard_leaders_agreed(convergence_budget()); + + // Modest converged baseline — this test exercises the DECISION under load, not + // scale (keep it fast; the corpus need not force compaction). + let seed_leader = current_leader_idx(&cluster).expect("a leader"); + for entity in 1..=200u64 { + write_heavy_item_only(&cluster, seed_leader, entity, ""); + } + cluster.wait_converged_all(convergence_budget()); + ensure_elected_era(&mut cluster); // production is always elected-era + let known = 42u64; // a committed baseline id we prove still serves at the end + println!("[under-load] baseline of 200 items converged (elected era); starting the writer"); + + // ── Background writer: steady item writes round-robined across the nodes (the + // gateway forwards to the current leader), tolerating the transient failures a + // restart causes. The point is sustained in-flight load so the leader always + // has an uncommitted tail when its pod is recycled. Acked (201) ids are + // recorded so we can prove the cluster keeps serving committed content. + let urls: Vec = (0..3).map(|i| cluster.node(i)).collect(); + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let acked = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); + let writer = { + let stop = std::sync::Arc::clone(&stop); + let acked = std::sync::Arc::clone(&acked); + std::thread::spawn(move || { + let client = reqwest::blocking::Client::builder() + .timeout(Duration::from_secs(2)) + .build() + .expect("writer client"); + let mut id = 2_000_000u64; + let mut rr = 0usize; + while !stop.load(std::sync::atomic::Ordering::Relaxed) { + let url = format!("{}/items", urls[rr % urls.len()]); + rr += 1; + let body = serde_json::json!({ + "entity_id": id, + "metadata": { "title": item_token(id) }, + }); + // ack=quorum: a 2xx means the write COMMITTED to a quorum, so it can + // never become a divergent suffix on a leadership change (unlike the + // leader-only ack, whose un-replicated tail is correctly truncated). + // This is the durability contract a cluster that rolling-restarts + // under load must use; the test proves a committed write is never lost + // and never triggers a reseed. + if let Ok(resp) = client + .post(&url) + .header("x-tidal-ack", "quorum") + .json(&body) + .send() + && resp.status().is_success() + { + acked.lock().unwrap().push(id); + } + id += 1; + std::thread::sleep(Duration::from_millis(20)); // ~50 writes/s + } + }) + }; + + // ── Ordinal rolling restart UNDER LOAD; assert no reseed after each node. + let mut leadership_moved = false; + for node in [0usize, 1, 2] { + let pre = cluster + .region_name(current_leader_idx(&cluster).expect("a leader")) + .to_string(); + cluster.restart_graceful(node, &[]); + let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + Duration::from_secs(20)); + let post = cluster + .region_name(current_leader_idx(&cluster).expect("a leader")) + .to_string(); + if post != pre { + leadership_moved = true; + } + for idx in 0..3 { + await_status_bool( + &cluster, + idx, + "reseed_required", + false, + convergence_budget() + Duration::from_secs(10), + "a graceful rolling restart UNDER LOAD must not latch reseed", + ); + } + // Pace like a readiness-gated rollout: let the cluster re-converge (modulo + // the small in-flight write tail) before taking down the next node, so the + // next leader's step-down drain has caught-up followers to commit against. + await_cluster_caught_up( + &cluster, + 50, + convergence_budget() + Duration::from_secs(20), + "cluster re-converges (under load) before the next rolling restart", + ); + println!("[under-load] restarted node {node}; leader {pre} -> {post}; no reseed"); + } + assert!( + leadership_moved, + "leadership must move at least once across the rolling restart" + ); + + // Stop the writer; it must have landed a meaningful amount of acked load. + stop.store(true, std::sync::atomic::Ordering::Relaxed); + writer.join().expect("writer thread joins"); + let ids = acked.lock().unwrap().clone(); + // A modest floor: writes are deliberately in flight across the restarts, but + // each failover briefly has no write-accepting leader (and a shutting-down + // leader rejects writes — the quiesce that lets the drain converge), and a + // forwarded write can burn its 2s timeout chasing a moving leader, so the + // ACKED count is naturally low. The point is only that real load WAS flowing. + assert!( + !ids.is_empty(), + "the writer must have landed some acked load under the restarts (got {})", + ids.len() + ); + println!( + "[under-load] writer landed {} acked writes under the restarts", + ids.len() + ); + + // ── The committed corpus still SERVES on every node (no reseed wiped it; the + // cluster stayed correct through the load + restarts). A committed baseline id + // must read back everywhere. + let probe_deadline = Instant::now() + convergence_budget(); + for idx in 0..3 { + loop { + if item_searchable(&cluster, idx, known) { + break; + } + assert!( + Instant::now() < probe_deadline, + "node {idx} no longer serves committed item {known} after the under-load \ + rolling restart" + ); + std::thread::sleep(Duration::from_millis(200)); + } + } + println!("[under-load] committed corpus still serves on all nodes — clean under load"); +} + +/// One node's election-divergence-relevant status, sampled for the Ring-0 +/// consistency oracle. `election_frontier`/`election_tail_term` are the LIVE +/// `election_log_position()` the vote restriction + `decide_join` read; +/// `applied_events` is this node's applied frontier in the CURRENT leader's +/// stream numbering. The consistency invariant: a caught-up follower +/// (`lag_events == 0`, not reseeding) reports `election_frontier == +/// applied_events` — i.e. the position it would VOTE/JOIN with is the same +/// quantity, in the same numbering, the leader holds. The cross-numbering tear +/// (an ex-leader still reading its OWN `flushed_seq` because its `wal_term_mark` +/// has not yet folded the new leader's marker) shows up as `election_frontier != +/// applied_events` on a node that is otherwise caught up, and/or as a spurious +/// `quarantined`. +// Fields `idx`/`term`/`election_tail_term` are diagnostic-only: they surface in +// the `{violations:#?}` dump when the oracle fails (dead-code analysis ignores +// `Debug`), so silence the lint rather than drop the evidence. +#[allow(dead_code)] +#[derive(Debug, Clone)] +struct ElectionSample { + idx: usize, + is_leader: bool, + quarantined: bool, + reseeding: bool, + term: u64, + election_tail_term: u64, + election_frontier: u64, + applied_events: u64, + lag_events: u64, +} + +fn election_sample(cluster: &MultiProcCluster, idx: usize) -> Option { + let s = cluster.local_status(idx)?; + Some(ElectionSample { + idx, + is_leader: s["is_leader"].as_bool().unwrap_or(false), + quarantined: s["quarantined"].as_bool().unwrap_or(false), + reseeding: s["reseeding"].as_bool().unwrap_or(false), + term: s["term"].as_u64().unwrap_or(0), + election_tail_term: s["election_tail_term"].as_u64().unwrap_or(u64::MAX), + election_frontier: s["election_frontier"].as_u64().unwrap_or(u64::MAX), + applied_events: s["applied_events"].as_u64().unwrap_or(0), + lag_events: s["lag_events"].as_u64().unwrap_or(u64::MAX), + }) +} + +/// Ring 0 (election-divergence-fix roadmap, task 00) — the CONSISTENCY ORACLE. +/// +/// Pins the invariant the whole roadmap protects: for the same committed state, +/// every node's reported election position is the SAME comparable quantity in +/// ONE numbering, and a CAUGHT-UP node is never misclassified as divergent. +/// +/// It forces an elected-era failover (graceful restart of the current leader), +/// then SAMPLES every node continuously through the rejoin until the cluster +/// re-converges, asserting two things on every sample: +/// 1. No caught-up node (`lag_events == 0`, not reseeding) is `quarantined` +/// — a caught-up node has nothing the new leadership does not subsume. +/// 2. A caught-up non-leader's `election_frontier == applied_events` — the +/// position it would vote/join with IS its applied frontier in the leader's +/// stream (the cross-numbering tear violates this). +/// +/// FAILS today under the cross-numbering tear (the deposed leader reads its own +/// `flushed_seq` for `election_frontier` while `applied_events` is the leader +/// stream — they differ, and/or it false-quarantines). Greens once task 01 +/// (stream-numbered marker) + task 02 (`election_log_position_for`) + task 03 +/// (committed-subsumption) make the position consistent. The negative controls +/// (`mp_quarantined_*`, `mp_follower_*`) keep reseeding — genuine divergence is +/// untouched. +/// +/// Greens with the m12 election-divergence-fix (the durable `leader_acked` +/// frontier): a caught-up node is never false-quarantined across an elected-era +/// failover, and the position is consistent at rest. The negative controls +/// (`mp_quarantined_*`, `mp_follower_*`) keep reseeding. +#[test] +fn mp_election_position_consistent_across_roles_after_failover() { + let opts = ClusterOptions::new(3) + .with_topology_extra(PROD_ELECTION_YAML) + .with_env(LEADER, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS) + .with_env(EU_WEST, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS) + .with_env(AP_SOUTH, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS); + let mut cluster = MultiProcCluster::start_with(opts); + let _ = cluster.wait_shard_leaders_agreed(convergence_budget()); + + let seed_leader = current_leader_idx(&cluster).expect("an elected leader before seeding"); + for entity in 1..=200u64 { + write_heavy_item_only(&cluster, seed_leader, entity, ""); + } + cluster.wait_converged_all(convergence_budget() + Duration::from_secs(10)); + ensure_elected_era(&mut cluster); // production is always elected-era + println!("[oracle] 200-item baseline converged (elected era)"); + + // Converged steady state: assert the consistency invariant holds at rest on + // every node BEFORE any failover (the baseline the tear later violates). + await_cluster_caught_up( + &cluster, + 0, + convergence_budget(), + "cluster converges before the measured failover", + ); + for idx in 0..3 { + if let Some(s) = election_sample(&cluster, idx) { + assert!( + !s.quarantined, + "[oracle] node {idx} quarantined in the converged steady state: {s:?}" + ); + } + } + + // ── Force a failover: graceful-restart the current leader. Survivors elect a + // new term; the deposed leader reboots as a follower and rejoins. THIS rejoin + // is where the cross-numbering tear fires (the deposed leader's `wal_term_mark` + // still names itself until it folds the new leader's marker). + let deposed = current_leader_idx(&cluster).expect("a leader to depose"); + let pre = cluster.region_name(deposed).to_string(); + cluster.restart_graceful(deposed, &[]); + + // SAMPLE THE REJOIN WINDOW. Poll every node ~10×/s until the cluster + // re-converges (all lag 0) or the budget elapses, asserting the invariant on + // every sample. A violation captures the exact cross-numbering evidence. + let deadline = Instant::now() + convergence_budget() + Duration::from_secs(30); + // THE ROBUST INVARIANT, sampled continuously through the rejoin: a genuinely + // caught-up node is NEVER quarantined. "Genuinely caught up" requires a SEEDED + // lag gauge (`applied_events > 0`), not a bare `lag == 0` — right after a + // failover an UNINITIALIZED gauge reads `leader_seqno(0) - applied(0) = 0`, a + // spurious "caught up, applied nothing". `quarantined` is a durable latch + // immune to that window. This false-quarantine is exactly what the fix removes. + let mut quarantine_violations: Vec = Vec::new(); + let mut converged = false; + while Instant::now() < deadline { + let samples: Vec = (0..3) + .filter_map(|i| election_sample(&cluster, i)) + .collect(); + for s in &samples { + let caught_up = s.lag_events == 0 && s.applied_events > 0 && !s.reseeding; + if caught_up && s.quarantined { + quarantine_violations.push(s.clone()); + } + } + // GENUINE re-convergence: every node has a SEEDED gauge at lag 0 and a + // leader exists — past the post-failover uninitialized-gauge window. + if samples.len() == 3 + && samples + .iter() + .all(|s| s.lag_events == 0 && s.applied_events > 0) + && samples.iter().any(|s| s.is_leader) + { + converged = true; + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + let post = current_leader_idx(&cluster) + .map(|i| cluster.region_name(i).to_string()) + .unwrap_or_else(|| "".into()); + println!("[oracle] failover {pre} -> {post}; converged={converged}"); + + assert!( + quarantine_violations.is_empty(), + "[oracle] FALSE QUARANTINE across the failover ({} sample(s)) — a genuinely \ + caught-up node was quarantined (the cross-numbering false-quarantine the \ + fix removes). Evidence: {quarantine_violations:#?}", + quarantine_violations.len() + ); + assert!( + converged, + "[oracle] cluster did not genuinely re-converge after the failover (leader \ + {pre} -> {post})" + ); + + // AT GENUINE REST, the CONSISTENCY property: every caught-up non-leader's + // vote/join position (`election_frontier`) equals its applied frontier in the + // leader's stream (`applied_events`) — the same quantity in one numbering. A + // bounded poll absorbs a brief marker-application lag (the frontier re-keys to + // the new leader's shard as the marker folds), but a PERSISTENT cross-numbering + // tear never clears and times out here. + let cons_deadline = Instant::now() + Duration::from_secs(20); + loop { + let bad: Vec = (0..3) + .filter_map(|i| election_sample(&cluster, i)) + .filter(|s| { + !s.is_leader + && s.applied_events > 0 + && s.lag_events == 0 + && s.election_frontier != s.applied_events + }) + .collect(); + if bad.is_empty() { + break; + } + assert!( + Instant::now() < cons_deadline, + "[oracle] election-position INCONSISTENT at rest — a caught-up follower's \ + vote/join frontier disagrees with its applied frontier in the leader's \ + stream (the cross-numbering tear). Evidence: {bad:#?}" + ); + std::thread::sleep(Duration::from_millis(200)); + } + + // Post-failover steady state: no node latched reseed, the corpus still serves. + for idx in 0..3 { + await_status_bool( + &cluster, + idx, + "reseed_required", + false, + convergence_budget(), + "the failover must not latch reseed on any node", + ); + } + println!("[oracle] consistency held across the failover — no tear, no false quarantine"); +} diff --git a/tidal-stress/k8s/soak-monitor.yaml b/tidal-stress/k8s/soak-monitor.yaml index 3c23ffc..9846c45 100644 --- a/tidal-stress/k8s/soak-monitor.yaml +++ b/tidal-stress/k8s/soak-monitor.yaml @@ -127,6 +127,65 @@ spec: volumeMounts: - name: results mountPath: /results + # ── Streak evaluator: the GA-bar verdict + alerting ─────────────────── + # Joins ledger.tsv (the nightly Job verdicts) with restarts.tsv (the + # cumulative per-pod restart counts the sidecar above records) into the + # HONEST streak: a night is green iff the soak passed AND no pod restarted + # under load. Runs the proven `soak-eval` core (table-driven-tested in + # tidal_stress::soak_eval), writes streak.tsv, and emits EXACTLY ONE alert + # per non-green night transition (a durable ALERT-.txt the http + # surface serves, plus an optional webhook POST when NOTIFY_WEBHOOK is set + # — reuse the cluster's notify service by pointing that at it). It does NOT + # gate load; it makes the 30-night picture trustworthy and observable. + - name: evaluator + image: registry.threesix.ai/tidal/stress:m12-soak-eval + imagePullPolicy: IfNotPresent + command: ["/bin/sh", "-c"] + args: + - | + echo "streak-evaluator up $(date -u +%FT%TZ); target=30 nights" + while true; do + # soak-eval reads /results/{ledger,restarts}.tsv, writes streak.tsv, + # and exits non-zero IFF the most recent night is non-green. + if soak-eval --results-dir /results --target 30; then + : # last night green (or no nights yet) — nothing to alert + else + # Non-green last night. Alert ONCE: key on the last streak.tsv + # date row so a persistent break does not re-fire every loop. + LAST="$(grep -v '^date' /results/streak.tsv 2>/dev/null | grep -v '^#' | tail -1 | cut -f1)" + SENT="$(cat /results/.last-alert 2>/dev/null || true)" + if [ -n "$LAST" ] && [ "$LAST" != "$SENT" ]; then + REASON="$(grep -v '^date' /results/streak.tsv | grep -v '^#' | tail -1 | cut -f4)" + MSG="tidalDB soak: night $LAST NON-GREEN — $REASON (streak reset)" + echo "ALERT: $MSG" + printf '%s\t%s\n' "$LAST" "$REASON" > "/results/ALERT-$LAST.txt" + [ -n "${NOTIFY_WEBHOOK:-}" ] && \ + wget -q -O- --post-data="{\"text\":\"$MSG\"}" --header='Content-Type: application/json' "$NOTIFY_WEBHOOK" >/dev/null 2>&1 || true + echo "$LAST" > /results/.last-alert + fi + fi + # Bound the Retain volume: drop nightly summaries older than 31 + # days (the full window + 1) so successive 30-night runs never grow + # the PVC without limit. ledger/restarts/streak are append/rewrite + # and stay small. + find /results -maxdepth 1 -name 'soak-*.json' -mtime +31 -delete 2>/dev/null || true + sleep 300 + done + env: + # Point at the cluster's notify service to deliver alerts off-cluster; + # unset = log + durable ALERT-.txt only (still status-surfaced). + - name: NOTIFY_WEBHOOK + value: "" + resources: + requests: { cpu: 10m, memory: 32Mi } + limits: { cpu: 200m, memory: 128Mi } + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: { drop: ["ALL"] } + volumeMounts: + - name: results + mountPath: /results # ── HTTP read surface: serve the durable result dir ─────────────────── - name: http image: busybox:1.36 @@ -140,6 +199,15 @@ spec: ports: - name: http containerPort: 8080 + # Readiness gates the Service endpoint on the http surface actually + # serving — a monitor whose node is rebooting must drop out of endpoints + # rather than silently serve nothing while a restart it should be + # recording slips by unobserved. + readinessProbe: + httpGet: { path: /, port: 8080 } + initialDelaySeconds: 5 + periodSeconds: 10 + failureThreshold: 3 resources: requests: { cpu: 10m, memory: 16Mi } limits: { cpu: 100m, memory: 64Mi } diff --git a/tidal-stress/k8s/soak-nightly-cronjob.yaml b/tidal-stress/k8s/soak-nightly-cronjob.yaml index d33e488..88bb8f8 100644 --- a/tidal-stress/k8s/soak-nightly-cronjob.yaml +++ b/tidal-stress/k8s/soak-nightly-cronjob.yaml @@ -82,7 +82,7 @@ spec: type: RuntimeDefault containers: - name: soak - image: registry.threesix.ai/tidal/stress:m12-rc7-seedretry + image: registry.threesix.ai/tidal/stress:m12-soak-eval imagePullPolicy: IfNotPresent # entrypoint is tidal-stress; wrap it so we can stamp the result file # name with the date and append a ledger line regardless of verdict. @@ -92,7 +92,7 @@ spec: set -u DATE="$(date -u +%Y-%m-%d)" OUT="/results/soak-$DATE.json" - echo "soak $DATE start $(date -u +%H:%M:%SZ) target=cluster image=m12-rc7-seedretry" + echo "soak $DATE start $(date -u +%H:%M:%SZ) target=cluster image=m12-soak-eval" tidal-stress \ --target https://tidaldb-0.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 \ --target https://tidaldb-1.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 \ @@ -120,9 +120,19 @@ spec: PASSED="$(grep -o '"passed"[: ]*[a-z]*' "$OUT" 2>/dev/null | head -1 | grep -o '[a-z]*$')" P99="$(grep -o '"overall_p99_ms"[: ]*[0-9.]*' "$OUT" 2>/dev/null | tail -1 | grep -o '[0-9.]*$')" ERRF="$(grep -o '"error_rate"[: ]*[0-9.]*' "$OUT" 2>/dev/null | tail -1 | grep -o '[0-9.]*$')" - printf '%s\t%s\trc=%s\tpassed=%s\tp99_ms=%s\terr_frac=%s\timage=m12-rc7-seedretry\n' \ + printf '%s\t%s\trc=%s\tpassed=%s\tp99_ms=%s\terr_frac=%s\timage=m12-soak-eval\n' \ "$DATE" "$VERD" "$RC" "${PASSED:-?}" "${P99:-?}" "${ERRF:-?}" >> /results/ledger.tsv - echo "soak $DATE end $(date -u +%H:%M:%SZ) verdict=$VERD rc=$RC p99=${P99:-?} err=${ERR:-?}" + echo "soak $DATE end $(date -u +%H:%M:%SZ) verdict=$VERD rc=$RC p99=${P99:-?} err=${ERRF:-?}" + # Refresh the streak immediately, CONSULTING restarts.tsv: a night + # whose Job passed but whose window carried a pod restart is recorded + # NON-GREEN by soak-eval (the zero-under-load-restart half of the GA + # bar). This makes the per-night verdict restart-aware without giving + # the soak Job pod-read RBAC — the monitor records restarts; soak-eval + # only reads the two files. The Job's OWN exit stays the SLO verdict. + if command -v soak-eval >/dev/null 2>&1; then + soak-eval --results-dir /results --target 30 || \ + echo "soak $DATE: streak NON-GREEN per soak-eval (see /results/streak.tsv)" + fi exit "$RC" env: - name: TIDAL_API_KEY diff --git a/tidal-stress/k8s/soak-results-pvc.yaml b/tidal-stress/k8s/soak-results-pvc.yaml index 8dd3e44..a921944 100644 --- a/tidal-stress/k8s/soak-results-pvc.yaml +++ b/tidal-stress/k8s/soak-results-pvc.yaml @@ -2,10 +2,17 @@ # consecutive days — docs/planning/milestone-11/phase-9.md "Exit gate"). # # longhorn-rwx: RWX so the nightly soak CronJob pods (writers) and the always-on -# soak-monitor (reader) can mount it concurrently, and Retain reclaim policy so -# the 30 nights of JSON summaries survive a PVC delete / job churn / operator -# session ending. Each night writes /results/soak-YYYY-MM-DD.json + appends one -# line to /results/ledger.tsv (dateverdictp99err% image). +# soak-monitor (reader + evaluator) can mount it concurrently, and Retain reclaim +# policy so the 30 nights of JSON summaries survive a PVC delete / job churn / +# operator session ending. Each night writes /results/soak-YYYY-MM-DD.json + +# appends one line to /results/ledger.tsv; the monitor appends restarts.tsv every +# 5 min and the evaluator rewrites streak.tsv. +# +# SIZING (30-night budget): 30 × soak-*.json (per-stage latency histograms, up to +# ~30 MiB each on a full run) ≈ 1 GiB, + restarts.tsv (288 samples/day × 3 pods × +# 30 nights ≈ 26k rows ≈ a few MiB) + ledger/streak (KiB). 5Gi gives ~2× headroom. +# The evaluator container PRUNES soak-*.json older than 31 days each loop so the +# Retain volume stays bounded across successive windows (it never auto-shrinks). # # Apply: kubectl apply -f tidal-stress/k8s/soak-results-pvc.yaml apiVersion: v1 @@ -22,4 +29,4 @@ spec: storageClassName: longhorn-rwx resources: requests: - storage: 2Gi + storage: 5Gi diff --git a/tidal-stress/src/bin/soak-eval.rs b/tidal-stress/src/bin/soak-eval.rs new file mode 100644 index 0000000..fee2a0f --- /dev/null +++ b/tidal-stress/src/bin/soak-eval.rs @@ -0,0 +1,70 @@ +//! `soak-eval` — the 30-night streak evaluator the soak monitor runs. +//! +//! Reads the durable result dir (`ledger.tsv` + `restarts.tsv`), joins them into +//! the honest per-night verdict (pass AND zero under-load restart), writes +//! `streak.tsv`, prints the current streak, and EXITS NON-ZERO when the most +//! recent night is non-green — so the monitor's wrapper can fire exactly one +//! alert on a break. Pure logic lives in [`tidal_stress::soak_eval`] (table- +//! driven-tested); this binary is the thin file-IO + exit-code shell. +//! +//! Usage: `soak-eval --results-dir /results [--target 30]` + +use std::path::PathBuf; +use std::process::ExitCode; + +use clap::Parser; +use tidal_stress::soak_eval::{ + current_streak, evaluate, parse_ledger, parse_restarts, render_streak_tsv, +}; + +#[derive(Parser)] +#[command(version, about = "30-night soak streak evaluator (ledger + restarts → streak.tsv)")] +struct Cli { + /// The durable result dir the nightly CronJob + monitor write to. + #[arg(long, default_value = "/results")] + results_dir: PathBuf, + + /// Consecutive green nights required for GA (the exit gate). + #[arg(long, default_value_t = 30)] + target: usize, +} + +fn main() -> ExitCode { + let cli = Cli::parse(); + let ledger_path = cli.results_dir.join("ledger.tsv"); + let restarts_path = cli.results_dir.join("restarts.tsv"); + + // A missing ledger means no night has run yet — streak 0, not an error. + let ledger = std::fs::read_to_string(&ledger_path).unwrap_or_default(); + let restarts = std::fs::read_to_string(&restarts_path).unwrap_or_default(); + + let nights = parse_ledger(&ledger); + let samples = parse_restarts(&restarts); + let verdicts = evaluate(&nights, &samples); + let streak = current_streak(&verdicts); + + let tsv = render_streak_tsv(&verdicts, cli.target); + let out_path = cli.results_dir.join("streak.tsv"); + if let Err(e) = std::fs::write(&out_path, &tsv) { + eprintln!("soak-eval: failed to write {}: {e}", out_path.display()); + return ExitCode::from(2); + } + + println!( + "soak-eval: {streak}/{} consecutive green nights ({} nights evaluated)", + cli.target, + verdicts.len() + ); + if streak >= cli.target { + println!("soak-eval: GA STREAK MET ({streak} ≥ {})", cli.target); + } + + // Exit non-zero iff the MOST RECENT night is non-green — the alert trigger. + match verdicts.last() { + Some(v) if !v.green => { + eprintln!("soak-eval: ALERT — last night {} is NON-GREEN: {}", v.date, v.reason); + ExitCode::from(1) + } + _ => ExitCode::SUCCESS, + } +} diff --git a/tidal-stress/src/lib.rs b/tidal-stress/src/lib.rs index 084b5e5..2c14939 100644 --- a/tidal-stress/src/lib.rs +++ b/tidal-stress/src/lib.rs @@ -13,5 +13,6 @@ pub mod error; pub mod metrics; pub mod recall; pub mod scheduler; +pub mod soak_eval; pub mod summary; pub mod workload; diff --git a/tidal-stress/src/soak_eval.rs b/tidal-stress/src/soak_eval.rs new file mode 100644 index 0000000..3efa80b --- /dev/null +++ b/tidal-stress/src/soak_eval.rs @@ -0,0 +1,338 @@ +//! Soak streak evaluator — the GA-bar verdict the nightly soak could not produce +//! on its own. +//! +//! The 30-night GA exit gate is *not* "30 green soak Job exits". It is +//! "30 consecutive nights where the soak passed its SLO gates AND no cluster pod +//! restarted under load". The nightly CronJob records its own pass/fail in +//! `ledger.tsv`; the monitor records cumulative per-pod restart counts in +//! `restarts.tsv`. Neither closes the loop — a night can show PASS in the ledger +//! while a pod silently churned a restart, which (per the production-readiness +//! roadmap) is exactly the failure the divergence fix removes. This module joins +//! the two streams into one honest verdict: +//! +//! * a night is **GREEN** iff its ledger verdict is PASS **and** no pod's +//! cumulative restart count increased within that night's window; +//! * the **streak** is the trailing run of consecutive green nights; +//! * any non-green night resets the streak to 0. +//! +//! The logic is pure and table-driven-tested so the k8s evaluator container is a +//! thin shell around a proven core, not fragile inline `awk`. + +use std::collections::BTreeMap; +use std::collections::BTreeSet; + +/// One night's ledger row: the date (`YYYY-MM-DD`) and whether the soak Job +/// passed its armed SLO gates. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NightResult { + pub date: String, + pub passed: bool, +} + +/// One restart-watch sample: the calendar date of the sample, the pod, and its +/// cumulative container `restartCount` at that instant (k8s `restartCount` only +/// rises until the pod object is recreated). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RestartSample { + pub date: String, + pub pod: String, + pub restarts: u64, +} + +/// The per-night verdict: green or not, with a human reason for a non-green night +/// (forensics on the streak reset). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NightVerdict { + pub date: String, + pub green: bool, + pub reason: String, +} + +/// Parse `ledger.tsv` (the nightly CronJob's append-only verdict log). +/// +/// Columns (tab-separated, as the cronjob writes them): +/// `DATE VERD rc=N passed=X p99_ms=X err_frac=X image=...`. The verdict is +/// taken from the `VERD` column (`PASS`/`FAIL`) — the authoritative Job exit +/// translation. Blank lines and a leading header (if any) are skipped. +#[must_use] +pub fn parse_ledger(tsv: &str) -> Vec { + let mut out = Vec::new(); + for line in tsv.lines() { + let line = line.trim(); + if line.is_empty() { + continue; + } + let cols: Vec<&str> = line.split('\t').collect(); + if cols.len() < 2 { + continue; + } + let date = cols[0].trim(); + // A header row or anything whose first cell is not a date is skipped. + if !is_date(date) { + continue; + } + let verd = cols[1].trim(); + out.push(NightResult { + date: date.to_string(), + passed: verd.eq_ignore_ascii_case("PASS"), + }); + } + out +} + +/// Parse `restarts.tsv` (the monitor's 5-minute restart-count snapshots). +/// +/// Columns: `ts_utc pod restarts phase ready` with an ISO `ts_utc` whose +/// leading `YYYY-MM-DD` is the night key. The header row and malformed rows are +/// skipped. +#[must_use] +pub fn parse_restarts(tsv: &str) -> Vec { + let mut out = Vec::new(); + for line in tsv.lines() { + let line = line.trim(); + if line.is_empty() { + continue; + } + let cols: Vec<&str> = line.split('\t').collect(); + if cols.len() < 3 { + continue; + } + let date = cols[0].get(0..10).unwrap_or(""); + if !is_date(date) { + continue; // header ("ts_utc") or junk + } + let pod = cols[1].trim(); + let Ok(restarts) = cols[2].trim().parse::() else { + continue; + }; + out.push(RestartSample { + date: date.to_string(), + pod: pod.to_string(), + restarts, + }); + } + out +} + +/// `YYYY-MM-DD` shape check (cheap, no chrono dependency). +fn is_date(s: &str) -> bool { + let b = s.as_bytes(); + b.len() == 10 + && b[4] == b'-' + && b[7] == b'-' + && b[0].is_ascii_digit() + && b[1].is_ascii_digit() + && b[2].is_ascii_digit() + && b[3].is_ascii_digit() + && b[5].is_ascii_digit() + && b[6].is_ascii_digit() + && b[8].is_ascii_digit() + && b[9].is_ascii_digit() +} + +/// Whether any pod restarted within night `date`, given the per-pod cumulative +/// restart counts observed up to and including each night. +/// +/// A restart shows as the cumulative count RISING. For night D and pod P: +/// `restarted = max_count(P, D) > floor`, where `floor` is P's highest count on +/// any night strictly before D (its carried-forward baseline), or — for the +/// first night P is seen — the MINIMUM count that night (so a restart *within* +/// the very first observed night, e.g. 0 → 1, still registers). +fn nights_with_restart(samples: &[RestartSample]) -> BTreeSet { + // pod -> (date -> (min, max)) of that date's samples. + let mut by_pod: BTreeMap<&str, BTreeMap<&str, (u64, u64)>> = BTreeMap::new(); + for s in samples { + let e = by_pod + .entry(&s.pod) + .or_default() + .entry(&s.date) + .or_insert((s.restarts, s.restarts)); + e.0 = e.0.min(s.restarts); + e.1 = e.1.max(s.restarts); + } + let mut dirty = BTreeSet::new(); + for dates in by_pod.values() { + let mut prev_max: Option = None; + for (date, &(min, max)) in dates { + let floor = prev_max.unwrap_or(min); + if max > floor { + dirty.insert((*date).to_string()); + } + prev_max = Some(prev_max.map_or(max, |p| p.max(max))); + } + } + dirty +} + +/// Join the ledger and the restart samples into a per-night verdict list, in +/// ledger order. A night is green iff it passed AND no pod restarted in its +/// window. +#[must_use] +pub fn evaluate(nights: &[NightResult], samples: &[RestartSample]) -> Vec { + let dirty = nights_with_restart(samples); + nights + .iter() + .map(|n| { + let restarted = dirty.contains(&n.date); + let (green, reason) = match (n.passed, restarted) { + (true, false) => (true, String::from("pass; no under-load restart")), + (false, false) => (false, String::from("soak SLO gate breach (Job FAIL)")), + (true, true) => ( + false, + String::from("under-load pod restart in window (Job passed but streak-breaking)"), + ), + (false, true) => ( + false, + String::from("soak FAIL and under-load pod restart"), + ), + }; + NightVerdict { + date: n.date.clone(), + green, + reason, + } + }) + .collect() +} + +/// The trailing run of consecutive green nights (the GA streak). Any non-green +/// night resets it; only the most recent unbroken tail counts. +#[must_use] +pub fn current_streak(verdicts: &[NightVerdict]) -> usize { + verdicts.iter().rev().take_while(|v| v.green).count() +} + +/// Render `streak.tsv`: one row per night (`date green streak_after reason`) +/// plus a trailing `# streak=/` summary line for an at-a-glance read. +#[must_use] +pub fn render_streak_tsv(verdicts: &[NightVerdict], target: usize) -> String { + let mut s = String::from("date\tgreen\tstreak\treason\n"); + let mut run = 0usize; + for v in verdicts { + run = if v.green { run + 1 } else { 0 }; + s.push_str(&format!( + "{}\t{}\t{}\t{}\n", + v.date, v.green, run, v.reason + )); + } + let streak = current_streak(verdicts); + s.push_str(&format!("# streak={streak}/{target}\n")); + s +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pass_nights(n: usize) -> Vec { + (1..=n) + .map(|i| NightResult { + date: format!("2026-06-{i:02}"), + passed: true, + }) + .collect() + } + + /// A constant-restart-count sample stream (no restarts) for every night — + /// pods sitting at a steady `restartCount` (e.g. 2) all window. + fn steady_samples(nights: usize, count: u64) -> Vec { + let mut v = Vec::new(); + for i in 1..=nights { + for pod in ["tidaldb-0", "tidaldb-1", "tidaldb-2"] { + v.push(RestartSample { + date: format!("2026-06-{i:02}"), + pod: pod.to_string(), + restarts: count, + }); + } + } + v + } + + #[test] + fn thirty_clean_nights_make_a_full_streak() { + let nights = pass_nights(30); + let samples = steady_samples(30, 2); + let verdicts = evaluate(&nights, &samples); + assert!(verdicts.iter().all(|v| v.green), "all 30 must be green"); + assert_eq!(current_streak(&verdicts), 30, "30 clean nights → streak 30"); + } + + #[test] + fn an_under_load_restart_breaks_the_streak_even_when_the_job_passed() { + let nights = pass_nights(30); + let mut samples = steady_samples(30, 2); + // Night 15: tidaldb-1's cumulative restartCount rises 2 → 3 within the + // window (an under-load restart) while the soak Job still PASSED. + samples.push(RestartSample { + date: "2026-06-15".into(), + pod: "tidaldb-1".into(), + restarts: 3, + }); + let verdicts = evaluate(&nights, &samples); + let n15 = verdicts.iter().find(|v| v.date == "2026-06-15").unwrap(); + assert!(!n15.green, "a night with an under-load restart is NOT green"); + assert!(n15.reason.contains("restart"), "reason names the restart"); + // The streak is only the clean tail after night 15: nights 16..=30 = 15. + assert_eq!(current_streak(&verdicts), 15, "streak resets at the restart night"); + } + + #[test] + fn a_gate_breach_night_resets_the_streak() { + let mut nights = pass_nights(30); + nights[19].passed = false; // night 20 breached an SLO gate (Job FAIL) + let samples = steady_samples(30, 2); + let verdicts = evaluate(&nights, &samples); + let n20 = verdicts.iter().find(|v| v.date == "2026-06-20").unwrap(); + assert!(!n20.green, "a gate-breach night is not green"); + assert!(n20.reason.contains("gate"), "reason names the gate breach"); + assert_eq!(current_streak(&verdicts), 10, "tail after night 20 = nights 21..=30"); + } + + #[test] + fn a_restart_within_the_first_observed_night_registers() { + // No carried-forward baseline: a 0 → 1 rise on the first night must still + // count as a restart (min < max that night). + let nights = pass_nights(1); + let samples = vec![ + RestartSample { date: "2026-06-01".into(), pod: "tidaldb-0".into(), restarts: 0 }, + RestartSample { date: "2026-06-01".into(), pod: "tidaldb-0".into(), restarts: 1 }, + ]; + let verdicts = evaluate(&nights, &samples); + assert!(!verdicts[0].green, "0→1 within the first night is a restart"); + } + + #[test] + fn parse_ledger_skips_headers_and_reads_the_verdict() { + let tsv = "\ +2026-06-01\tPASS\trc=0\tpassed=true\tp99_ms=9.3\terr_frac=0.0001\timage=m12\n\ +2026-06-02\tFAIL\trc=1\tpassed=false\tp99_ms=210\terr_frac=0.04\timage=m12\n\ +not-a-date\tPASS\tjunk\n"; + let rows = parse_ledger(tsv); + assert_eq!(rows.len(), 2, "the junk/header row is skipped"); + assert!(rows[0].passed); + assert!(!rows[1].passed); + } + + #[test] + fn parse_restarts_skips_header_and_keys_by_date() { + let tsv = "\ +ts_utc\tpod\trestarts\tphase\tready\n\ +2026-06-01T02:05:00Z\ttidaldb-0\t2\tRunning\ttrue\n\ +2026-06-01T02:10:00Z\ttidaldb-1\t2\tRunning\ttrue\n"; + let rows = parse_restarts(tsv); + assert_eq!(rows.len(), 2, "header skipped, two data rows parsed"); + assert_eq!(rows[0].date, "2026-06-01"); + assert_eq!(rows[0].restarts, 2); + } + + #[test] + fn render_streak_tsv_carries_running_count_and_summary() { + let nights = pass_nights(3); + let samples = steady_samples(3, 0); + let verdicts = evaluate(&nights, &samples); + let tsv = render_streak_tsv(&verdicts, 30); + assert!(tsv.contains("date\tgreen\tstreak\treason"), "has a header"); + assert!(tsv.trim_end().ends_with("# streak=3/30"), "summary line present: {tsv}"); + } +} diff --git a/tidal/src/db/replication_ops.rs b/tidal/src/db/replication_ops.rs index 870eaae..48ee63a 100644 --- a/tidal/src/db/replication_ops.rs +++ b/tidal/src/db/replication_ops.rs @@ -99,6 +99,19 @@ impl TidalDb { self.wal_term_mark.snapshot() } + /// Fold a term marker into the WAL-tail term cell directly (m12 election- + /// divergence-fix). A node that JOINS a term via heartbeat has logically + /// applied that term's marker (the term's first entry at `seq`), even when the + /// physical marker record is gap-gated on catch-up — it rides the same + /// seqno-gated segment path as data and a resume-from-`applied+1` permanently + /// skips it, leaving the node a stuck `lag=1` behind and reading its own stale + /// numbering. Folding it on join closes that gap. Idempotent + monotonic-by-term + /// (a re-applied/older marker never regresses the cell), exactly like the + /// apply-path fold. + pub fn fold_term_marker(&self, term: u64, seq: u64, leader_region: u16) { + self.wal_term_mark.advance(term, seq, leader_region); + } + /// The applied cluster membership (m11p5): the highest-version kind-4 /// record this node's log carries, as `(version, term, members)`. `None` /// = the topology era (membership epoch 0; no conf-change has run).