jordan/tidaldb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(m12): election-divergence-fix + soak-eval streak + release tooling

Browse Source 
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.
This commit is contained in:
jx12n 2026-06-18 13:08:53 -06:00
parent 46610cd411
commit 580142df49
21 changed files with 2014 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
docker/release/Dockerfile Normal file
View File

@ -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"]

9
docker/release/config/default-cluster.yaml Normal file
View File

@ -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

31
docker/release/config/default-schema.yaml Normal file
View File

@ -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

35
docker/release/dr.Dockerfile Normal file
View File

@ -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"]

4
docker/stress/Dockerfile
View File

@ -30,6 +30,10 @@ RUN useradd --system --home /srv stress && \
rm -rf /var/lib/apt/lists/* rm -rf /var/lib/apt/lists/*
COPY --from=builder /app/target/release/tidal-stress /usr/local/bin/tidal-stress 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 USER stress

31
docs/planning/milestone-11/phase-9.md
View File

@ -134,6 +134,37 @@ fault each injects, and the checker each asserts through.
- Every guarantee in §2 maps to a named automated test. - Every guarantee in §2 maps to a named automated test.
- Nightly suite green 30 consecutive days before GA. - 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-<date>.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) ## Exit-gate evidence (local; debug builds, real OS processes)
| Gate | Target | Measured | | Gate | Target | Measured |

110
scripts/build-release.sh Executable file
View File

@ -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 <tag> [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 <tag> [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:-<digest-unavailable>} (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:-<digest-unavailable>} (tag $TAG)"
fi
echo "==> done. Pin the printed @sha256 digest(s) in k8s/cluster/statefulset.yaml and the DR/soak manifests."

55
scripts/run-reliability.sh Executable file
View File

@ -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."

179
tidal-server/src/cluster/election_driver.rs
View File

@ -252,7 +252,11 @@ impl ElectionRuntime {
return false; return false;
}; };
let position = node.election_log_position(); 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 => { JoinDecision::Quarantine => {
// Preserve the existing quarantine semantics: set the latch, note // Preserve the existing quarantine semantics: set the latch, note
// the divergence (metric + reseed marker), and refuse the data // 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); let before = self.runtime.joined_term.load(Ordering::Acquire);
// `stream_baseline` is the new term's clamp floor — pass it so a reseed // `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. // 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 if joined
&& term > before && term > before
&& let Some(node) = self.runtime.node.upgrade() && let Some(node) = self.runtime.node.upgrade()
@ -721,24 +730,56 @@ enum JoinDecision {
Clean, 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 /// `own` and `prev_log` are in the PREVIOUS stream's numbering (`own` = this
/// restriction's `(tail_term, frontier)` comparison). A tail already carrying /// node's WAL-tail position; `prev_log` = the new leader's frozen *election-time*
/// THIS term's marker (`own.tail_term == term`) is manifestly part of this /// position). `leader_acked` is this node's durable [`ShardReplica::leader_acked`]
/// term's history — the restart-within-term rejoin (phase-4.md §5, third /// frontier — the highest seqno it ACKED to a client under `ack=leader`
/// clause): without it, a mid-term rejoiner whose tail term (T) exceeds the /// (journal-only, no quorum) and may therefore hold UN-REPLICATED, in the same
/// leader's election-time tail (T-1) would false-quarantine on every restart. /// (previous-stream) numbering for a former leader of that stream.
/// Otherwise the lexicographic comparison decides: strictly-greater diverges ///
/// (quarantine), strictly-smaller is genuinely behind (reseed), equal is clean. /// Cases:
fn decide_join(term: u64, own: LogPosition, prev_log: LogPosition) -> JoinDecision { /// * `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 { if own.tail_term == term {
return JoinDecision::Clean; 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) { match own.cmp(&prev_log) {
std::cmp::Ordering::Greater => JoinDecision::Quarantine,
std::cmp::Ordering::Less => JoinDecision::ReseedRequired, 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 /// The term-join rule (m11p5 §2, m12 election-divergence-fix). The 4th arg is
/// divergent suffix), `own < prev_log` latches reseed (genuinely behind), /// the DURABLE leader-acked frontier; divergence is `leader_acked >
/// and equal / within-term is clean. /// prev_log.frontier` (un-replicated leader-acked suffix), NOT a frontier
/// compared across stream numberings.
#[test] #[test]
fn three_way_join_rule_classifies_each_case() { 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; let t = 1;
// own > prev_log (higher frontier, same tail term) → divergent suffix. // This node ACKED (ack=leader) up to 970 but the new leadership's election
// (This is test 1's exact shape: own=(0,970) vs leader prev=(0,9) — // baseline is only 9 → it holds an un-replicated leader-acked suffix the
// except here own LEADS, the quarantine direction.) // cluster elected past → divergent → quarantine.
assert_eq!( assert_eq!(
decide_join(t, pos(0, 970), pos(0, 9)), decide_join(t, pos(0, 970), pos(0, 9), 970),
JoinDecision::Quarantine, JoinDecision::Quarantine,
"a strictly-greater own position is a divergent suffix → quarantine" "leader-acked above the new election baseline 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"
); );
// own < prev_log (lower frontier, same tail term) → genuinely behind: // own < prev_log, nothing leader-acked beyond it (leader_acked = 0) →
// the node lacks (own, prev_log] of the previous stream → reseed. This // genuinely behind → reseed (self-heals; a compacted gap reseeds).
// is test 1's REAL shape: own=(0,9) (stopped early) vs leader's
// election-time prev=(0,970) (full history) → reseed, not quarantine.
assert_eq!( assert_eq!(
decide_join(t, pos(0, 9), pos(0, 970)), decide_join(t, pos(0, 9), pos(0, 970), 0),
JoinDecision::ReseedRequired, JoinDecision::ReseedRequired,
"a strictly-smaller own position is genuinely missing history → reseed" "a strictly-smaller own position with nothing leader-acked is behind → 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"
); );
// own == prev_log → clean. // own == prev_log, nothing leader-acked beyond it → clean.
assert_eq!( assert_eq!(
decide_join(t, pos(0, 9), pos(0, 9)), decide_join(t, pos(0, 9), pos(0, 9), 0),
JoinDecision::Clean, 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 /// 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 /// carrying THIS term's marker is part of this term's history → never reseed
/// the `prev_log` comparison — it must NOT quarantine or reseed on restart. /// or quarantine on restart, regardless of the position or the acked frontier.
#[test] #[test]
fn within_term_rejoin_is_always_clean() { fn within_term_rejoin_is_always_clean() {
let t = 3; 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!( assert_eq!(
decide_join(t, pos(3, 500), pos(2, 10)), decide_join(t, pos(3, 500), pos(2, 10), 500),
JoinDecision::Clean, JoinDecision::Clean,
"a tail carrying this term's marker is the within-term rejoin → 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!( assert_eq!(
decide_join(t, pos(3, 1), pos(2, 999)), decide_join(t, pos(3, 1), pos(2, 999), 0),
JoinDecision::Clean, JoinDecision::Clean,
"within-term rejoin wins over the < comparison too" "within-term rejoin wins over the < comparison too"
); );

15
tidal-server/src/cluster/join_boot.rs
View File

@ -786,14 +786,13 @@ fn synthesize_topology(
/// same paths. Without the fallback a TLS-cluster joiner synthesized a PLAINTEXT /// same paths. Without the fallback a TLS-cluster joiner synthesized a PLAINTEXT
/// gRPC posture and could not mTLS-replicate with its peers. /// gRPC posture and could not mTLS-replicate with its peers.
fn self_tls_spec(knobs: &TopologySpec, name: &str) -> Option<GrpcTlsSpec> { fn self_tls_spec(knobs: &TopologySpec, name: &str) -> Option<GrpcTlsSpec> {
grpc_tls_for(knobs, name) grpc_tls_for(knobs, name).map(|t| GrpcTlsSpec {
.map(|t| GrpcTlsSpec { ca_cert: t.ca_cert.clone(),
ca_cert: t.ca_cert.clone(), server_cert: t.server_cert.clone(),
server_cert: t.server_cert.clone(), server_key: t.server_key.clone(),
server_key: t.server_key.clone(), client_cert: t.client_cert.clone(),
client_cert: t.client_cert.clone(), client_key: t.client_key.clone(),
client_key: t.client_key.clone(), })
})
} }
// The knob blocks have no `Clone`; rebuild them field-by-field from the loaded // The knob blocks have no `Clone`; rebuild them field-by-field from the loaded

350
tidal-server/src/cluster/node.rs
View File

@ -156,6 +156,10 @@ impl AckMode {
/// double-apply it on followers. /// double-apply it on followers.
const STREAM_BASELINE_FILE: &str = "stream_baseline"; 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 /// How long a leader-sanctioned transfer waits for the target to hold the
/// full flushed prefix before `TimeoutNow` (the drain, m11p4). /// full flushed prefix before `TimeoutNow` (the drain, m11p4).
const TRANSFER_CATCHUP_WAIT: Duration = Duration::from_secs(5); 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. /// (target leads at a higher term) before reporting failure.
const TRANSFER_TAKEOVER_WAIT: Duration = Duration::from_secs(10); 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 /// 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 /// thread can go without re-checking its stop flag, i.e. the worst-case
/// shutdown latency the bridge adds. Deliberately a constant, not a /// 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 /// Shared with the gRPC `SegmentSource` so catch-up never serves
/// pre-stream history. /// pre-stream history.
stream_baseline: Arc<AtomicU64>, stream_baseline: Arc<AtomicU64>,
/// 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<AtomicU64>,
/// The node's data dir (baseline persistence). Multi-process cluster /// The node's data dir (baseline persistence). Multi-process cluster
/// mode requires one — validated in [`Self::new`]. /// mode requires one — validated in [`Self::new`].
data_dir: std::path::PathBuf, data_dir: std::path::PathBuf,
@ -768,6 +803,16 @@ impl ShardReplica {
}; };
let install_boot = install_target.is_some(); 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 // Always-on receiver: leadership can move, so EVERY node runs a receiver
// (an inbound segment is legal on the current leader after a promote // (an inbound segment is legal on the current leader after a promote
// elsewhere). The blob-applier wiring routes replicated kind-1/2 // elsewhere). The blob-applier wiring routes replicated kind-1/2
@ -1012,6 +1057,7 @@ impl ShardReplica {
ship_feed, ship_feed,
ship_queue, ship_queue,
stream_baseline, stream_baseline,
leader_acked,
data_dir, data_dir,
boot_topology_leader: leader, boot_topology_leader: leader,
activation_prev: std::sync::Mutex::new(tidaldb::replication::LogPosition { activation_prev: std::sync::Mutex::new(tidaldb::replication::LogPosition {
@ -1084,8 +1130,35 @@ impl ShardReplica {
pub fn shutdown(&self) { pub fn shutdown(&self) {
self.set_shutting_down(); self.set_shutting_down();
if let Some(rt) = self.election_runtime.get() { 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(); 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); self.commit_bridge_stop.store(true, Ordering::Release);
// Quiesce the ship-queue senders FIRST so no batch ship races the // Quiesce the ship-queue senders FIRST so no batch ship races the
// transport/db teardown below (an in-flight send_segment against a // 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 ────────────────────────────────────────────────────────── // ── Accessors ──────────────────────────────────────────────────────────
/// Clone the node's `TidalDb` handle, or `Unavailable` once shutdown took it. /// 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. /// The transport's election fan-out handle.
pub(crate) fn election_net(&self) -> tidal_net::ElectionNet { pub(crate) fn election_net(&self) -> tidal_net::ElectionNet {
self.transport.election_net() 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 // §3.2 — the activation membership record (the linchpin): once the
// kind-3 marker is durable AND the membership era has begun (a kind-4 // 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 // record is in the log), re-append the CURRENT roster as a fresh kind-4
@ -2649,9 +2811,26 @@ impl ShardReplica {
if baseline > 0 if baseline > 0
&& let Ok(db) = self.db() && 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() 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!( tracing::info!(
term, term,
leader = %self.region_name_of(leader), 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. /// Offload the §2.4 quorum-refusal evaluation to a detached thread.
/// ///
/// `latch_reseed_marker` may be invoked from a tonic handler thread (the /// `latch_reseed_marker` may be invoked from a tonic handler thread (the
@ -2956,10 +3182,14 @@ impl ShardReplica {
db.control_plane() db.control_plane()
.lag_gauge() .lag_gauge()
.update_leader_seqno_for(leader_shard, leader_last_seq); .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) { 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.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). /// Whether this node is READY to serve (m11p5 §4 readiness predicate).
@ -3171,6 +3401,9 @@ impl ShardReplica {
.activation_prev .activation_prev
.lock() .lock()
.unwrap_or_else(std::sync::PoisonError::into_inner); .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( let (term, role, quarantined) = self.election_runtime.get().map_or_else(
|| (0, "unknown".to_string(), false), || (0, "unknown".to_string(), false),
|rt| { |rt| {
@ -3211,6 +3444,9 @@ impl ShardReplica {
quarantined, quarantined,
prev_log_term: status_prev.tail_term, prev_log_term: status_prev.tail_term,
prev_log_seq: status_prev.frontier, 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, reseed_required,
reseeding, reseeding,
self_restart_refused: self.self_restart_refused.load(Ordering::Acquire), 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<u64> {
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 /// 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 /// this node's replication state, so the transport can never keep the
/// database alive past shutdown. /// database alive past shutdown.
@ -4460,6 +4752,26 @@ pub struct LocalStatusResponse {
/// numbering and is not comparable to pre-election seqs). /// numbering and is not comparable to pre-election seqs).
prev_log_term: u64, prev_log_term: u64,
prev_log_seq: 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): /// Whether this node is quarantined with a divergent suffix (m11p4):
/// fenced from the data plane until reseeded. /// fenced from the data plane until reseeded.
quarantined: bool, quarantined: bool,
@ -5888,8 +6200,12 @@ pub async fn create_item(
return forward_write(&state, "/items", &req, &headers).await; return forward_write(&state, "/items", &req, &headers).await;
} }
// A marked (forwarded) write must land on the leader; a stale cluster // A marked (forwarded) write must land on the leader; a stale cluster
// view surfaces honestly instead of applying off-log. // view surfaces honestly instead of applying off-log. A leader that has begun
if !state.is_leader() { // 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())); return Err(ClusterAppError(state.not_leader()));
} }
let ack = state.ack_mode_for(&headers).map_err(ClusterAppError)?; let ack = state.ack_mode_for(&headers).map_err(ClusterAppError)?;
@ -5917,6 +6233,11 @@ pub async fn create_item(
&& let Some(seq) = seq && let Some(seq) = seq
{ {
await_quorum(&state, seq).await?; 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)) Ok(with_seq_header(StatusCode::CREATED, seq))
} }
@ -5952,7 +6273,9 @@ pub async fn write_embedding(
if !is_internal(&headers) && !state.is_leader() { if !is_internal(&headers) && !state.is_leader() {
return forward_write(&state, "/embeddings", &req, &headers).await; 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())); return Err(ClusterAppError(state.not_leader()));
} }
let ack = state.ack_mode_for(&headers).map_err(ClusterAppError)?; let ack = state.ack_mode_for(&headers).map_err(ClusterAppError)?;
@ -5978,6 +6301,10 @@ pub async fn write_embedding(
&& let Some(seq) = seq && let Some(seq) = seq
{ {
await_quorum(&state, seq).await?; 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)) 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; return forward_write(&state, "/signals", &req, &headers).await;
} }
// Reject non-leader BEFORE the write pool so NotLeader does not consume a slot. // 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())); return Err(ClusterAppError(state.not_leader()));
} }
let signal = req.signal; 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). // covered quorum (this request created no new log entry to gate on).
if ack == AckMode::Quorum && seq > 0 { if ack == AckMode::Quorum && seq > 0 {
await_quorum(&state, seq).await?; 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))) Ok(with_seq_header(StatusCode::NO_CONTENT, Some(seq)))
} }
@ -8400,7 +8733,10 @@ mod forward_retry_tests {
}; };
use axum::http::StatusCode; use axum::http::StatusCode;
fn ok(status: StatusCode, body: serde_json::Value) -> Result<forward::ForwardedResponse, String> { fn ok(
status: StatusCode,
body: serde_json::Value,
) -> Result<forward::ForwardedResponse, String> {
Ok(forward::ForwardedResponse { Ok(forward::ForwardedResponse {
seq: None, seq: None,
deduplicated: false, deduplicated: false,

4
tidal-server/tests/cluster_graph_persistence.rs
View File

@ -47,7 +47,9 @@ mod support;
use std::time::{Duration, SystemTime}; 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`). /// The leader index (region 0 = `us-east`).
const LEADER: usize = 0; const LEADER: usize = 0;

89
tidal-server/tests/cluster_membership.rs
View File

@ -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 /// Poll until node `idx`'s OWN status reports zero lag and an applied frontier at
/// or above the live leader's `last_seq`. /// or above the live leader's `last_seq`.
fn await_self_converged(cluster: &MultiProcCluster, idx: usize, budget: Duration, what: &str) { 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 // The whole cluster (now 4 nodes) converges, and the roster on EVERY live
// node lists the joiner as a voter. // node lists the joiner as a voter.
cluster.wait_converged_all(convergence_budget()); 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() { for i in 0..cluster.len() {
if let Some(rr) = cluster.local_status(i).map(|_| roster_roles(&cluster, i)) { loop {
assert_eq!( let rr = roster_roles(&cluster, i);
rr.get(&joiner_name).map(String::as_str), if rr.get(&joiner_name).map(String::as_str) == Some("voter") {
Some("voter"), break;
"node {i}'s roster must show the joiner as a voter: {rr:?}" }
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)); thread::sleep(Duration::from_millis(100));
} }
let flip_elapsed = ready_start.elapsed(); 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 // 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 // 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 // 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) cluster.local_status(victim_idx)
); );
// It has NO reseed marker — a remove is not a reseed (the typed `removed` // It has NO reseed marker — a remove is not a reseed (the typed `removed`
// signal is exempt from the reseed marker, §3.3). // signal is exempt from the reseed marker, §3.3). Polled, not single-shot:
assert_eq!( // the `removed` role await above does not fence the reseed gauge, which can
cluster // settle a beat later (the `cluster_reseed.rs:591` race class).
.local_status(victim_idx) await_status_bool(
.and_then(|s| s["reseed_required"].as_bool()), &cluster,
Some(false), victim_idx,
"a removed node must NOT latch a reseed marker (remove is exempt, §3.3)" "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"); 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 // It has NO reseed marker (a remove is not a reseed — the typed signal is
// exempt, §3.3). // exempt, §3.3). Polled, not single-shot (the `cluster_reseed.rs:591` race
assert_eq!( // class): the decommission await above does not fence the reseed gauge.
cluster await_status_bool(
.local_status(victim_idx) &cluster,
.and_then(|s| s["reseed_required"].as_bool()), victim_idx,
Some(false), "reseed_required",
"a signal-decommissioned node must NOT latch a reseed marker (§3.3)" 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"); println!("[missed] the signal-decommissioned victim has NO reseed marker — exact close-out");

632
tidal-server/tests/cluster_reseed.rs
View File

@ -79,6 +79,16 @@ const LEGACY_ELECTION_YAML: &str = "election:\n auto_election: false";
/// timeout_min (500)`. /// 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"; 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 /// 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 /// (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 /// [`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()); 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<usize> {
(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::<Vec<_>>()
);
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 /// 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 /// whose WAL has been COMPACTED past its resume seq reseeds via the boot-time
/// snapshot install — zero operator verbs other than restarts. /// 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"); 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<String> = (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::<u64>::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<ElectionSample> {
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<ElectionSample> = Vec::new();
let mut converged = false;
while Instant::now() < deadline {
let samples: Vec<ElectionSample> = (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(|| "<none>".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<ElectionSample> = (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");
}

68
tidal-stress/k8s/soak-monitor.yaml
View File

@ -127,6 +127,65 @@ spec:
volumeMounts: volumeMounts:
- name: results - name: results
mountPath: /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-<date>.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-<date>.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 ─────────────────── # ── HTTP read surface: serve the durable result dir ───────────────────
- name: http - name: http
image: busybox:1.36 image: busybox:1.36
@ -140,6 +199,15 @@ spec:
ports: ports:
- name: http - name: http
containerPort: 8080 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: resources:
requests: { cpu: 10m, memory: 16Mi } requests: { cpu: 10m, memory: 16Mi }
limits: { cpu: 100m, memory: 64Mi } limits: { cpu: 100m, memory: 64Mi }

18
tidal-stress/k8s/soak-nightly-cronjob.yaml
View File

@ -82,7 +82,7 @@ spec:
type: RuntimeDefault type: RuntimeDefault
containers: containers:
- name: soak - name: soak
image: registry.threesix.ai/tidal/stress:m12-rc7-seedretry image: registry.threesix.ai/tidal/stress:m12-soak-eval
imagePullPolicy: IfNotPresent imagePullPolicy: IfNotPresent
# entrypoint is tidal-stress; wrap it so we can stamp the result file # 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. # name with the date and append a ledger line regardless of verdict.
@ -92,7 +92,7 @@ spec:
set -u set -u
DATE="$(date -u +%Y-%m-%d)" DATE="$(date -u +%Y-%m-%d)"
OUT="/results/soak-$DATE.json" 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 \ tidal-stress \
--target https://tidaldb-0.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 \ --target https://tidaldb-0.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 \
--target https://tidaldb-1.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]*$')" 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.]*$')" 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.]*$')" 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 "$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" exit "$RC"
env: env:
- name: TIDAL_API_KEY - name: TIDAL_API_KEY

17
tidal-stress/k8s/soak-results-pvc.yaml
View File

@ -2,10 +2,17 @@
# consecutive days — docs/planning/milestone-11/phase-9.md "Exit gate"). # 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 # 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 # soak-monitor (reader + evaluator) can mount it concurrently, and Retain reclaim
# the 30 nights of JSON summaries survive a PVC delete / job churn / operator # policy so the 30 nights of JSON summaries survive a PVC delete / job churn /
# session ending. Each night writes /results/soak-YYYY-MM-DD.json + appends one # operator session ending. Each night writes /results/soak-YYYY-MM-DD.json +
# line to /results/ledger.tsv (date<TAB>verdict<TAB>p99<TAB>err% <TAB>image). # 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 # Apply: kubectl apply -f tidal-stress/k8s/soak-results-pvc.yaml
apiVersion: v1 apiVersion: v1
@ -22,4 +29,4 @@ spec:
storageClassName: longhorn-rwx storageClassName: longhorn-rwx
resources: resources:
requests: requests:
storage: 2Gi storage: 5Gi

70
tidal-stress/src/bin/soak-eval.rs Normal file
View File

@ -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,
}
}

1
tidal-stress/src/lib.rs
View File

@ -13,5 +13,6 @@ pub mod error;
pub mod metrics; pub mod metrics;
pub mod recall; pub mod recall;
pub mod scheduler; pub mod scheduler;
pub mod soak_eval;
pub mod summary; pub mod summary;
pub mod workload; pub mod workload;

338
tidal-stress/src/soak_eval.rs Normal file
View File

@ -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<NightResult> {
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<RestartSample> {
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::<u64>() 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<String> {
// 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<u64> = 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<NightVerdict> {
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=<n>/<target>` 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<NightResult> {
(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<RestartSample> {
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}");
}
}

13
tidal/src/db/replication_ops.rs
View File

@ -99,6 +99,19 @@ impl TidalDb {
self.wal_term_mark.snapshot() 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 /// The applied cluster membership (m11p5): the highest-version kind-4
/// record this node's log carries, as `(version, term, members)`. `None` /// record this node's log carries, as `(version, term, members)`. `None`
/// = the topology era (membership epoch 0; no conf-change has run). /// = the topology era (membership epoch 0; no conf-change has run).