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

feat(m11): catch-up timer retry + TSEG segment version header (m11p4)

Browse Source 
WAL segment format: 8-byte TSEG header (magic + version byte + 3 reserved)
prepended to every new segment. Legacy headerless segments (m0-m11p3) read
as implicit v0 — no migration. Unknown magic/version surfaces as
WalError::SegmentFormatUnknown at open time; foreign files are never
repaired or truncated (fixes the silent data-loss path from the p3 rollout
incident where torn-tail repair zeroed a follower's unreadable segments).

Catch-up transport: FAILED_PRECONDITION ("snapshot required") and stream
errors that skip the shard now arm a timer retry (re-arm-on-skip is the
load-bearing liveness fix — without it a skipped pull never re-fires and
the follower stays permanently behind). Single retry pending per shard;
CatchupRunner owns the Arc'd state shared between the retry tasks and the
transport. Test: tidal-net/tests/catchup_retry.rs covers the retry path.

Stress: k8s stress-job-t2a/t2b yaml + ops/stress-test-p3-t2 runbook.
This commit is contained in:
jx12n 2026-06-11 17:05:20 -06:00
parent 5ed2edb211
commit d0a52e4530
18 changed files with 1640 additions and 148 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
CHANGELOG.md
View File

@ -6,6 +6,40 @@ All notable changes to tidalDB will be documented in this file.
### Added ### Added
**Catch-up self-healing + WAL segment format versioning (m11p4) — timer-retried pulls, `TSEG` segment header, structured "snapshot required"**
- **Failed catch-up pulls retry on a timer.** The pull trigger was event-only:
a follower whose `StreamSegments` pull failed (e.g. the leader's gRPC server
not yet ready during a rolling restart) waited for the next PUSHED segment
to re-expose the gap — in an idle cluster that push never comes, and the
follower stayed lagged forever (the 2026-06-11 p3 rollout: both followers
stuck at lag=136507). A failed pull now arms a one-shot timer
(`replication.catchup_retry_ms`, default 30000; transport
`catchup_retry_interval`) that re-pulls from the CURRENT applied frontier.
Pulls stay single-flight and rate-limited; the timer's wake-up re-arms when
consumed by the rate limit or an in-flight pull, so the gap always keeps a
standing wake-up until a pull completes. Verified over real sockets:
`catchup_retry.rs` reproduces the incident (pull fails, leader appears,
zero pushes) and proves timer-only self-heal — and that a clean completion
arms nothing.
- **WAL segment files are format-versioned.** New segments open with an
8-byte header (`TSEG` magic + version byte + reserved); pre-m11p4
headerless segments stay readable as implicit version 0 — no migration. A
segment this binary cannot identify (unknown header version, unrecognized
leading bytes, unparseable `.seg` filename) surfaces as the new
`WalError::SegmentFormatUnknown` at open — previously it scanned as
empty (`segments=0`) and recovery's torn-tail repair could TRUNCATE the
foreign file to zero. Foreign-format files are never repaired, truncated,
or skipped. Downgrade across m11p4 requires a WAL reseed (runbook §8).
- **Unservable catch-up is a structured refusal.** `SegmentSource::collect_from`
returns typed `SegmentReadError::{Unavailable,Failed}`;
`TidalDb::read_wal_batches` returns the typed `WalError` (was stringified
`TidalError`). The `StreamSegments` handler maps `Unavailable` to
`FAILED_PRECONDITION` — `"segments not available from seq N; snapshot
required"` — and the follower logs it distinctly (*catch-up unservable …
needs a snapshot (m11p5) or an operator reseed*) instead of burying it as a
transient. The on-disk segment format is now documented
(`tidal/src/wal/segment.rs` module docs + spec 01 §2.2).
**Quorum-acked writes (m11p3) — `ack=leader|quorum`, durable ship acks, commit index, zero-acked-loss ledger gate (closes G4)** **Quorum-acked writes (m11p3) — `ack=leader|quorum`, durable ship acks, commit index, zero-acked-loss ledger gate (closes G4)**
- **`ack=quorum`** is an opt-in durability contract for every replicated write - **`ack=quorum`** is an opt-in durability contract for every replicated write
(`/signals`, `/items`, `/embeddings`): success means a **majority of the (`/signals`, `/items`, `/embeddings`): success means a **majority of the

133
docs/ops/stress-test-p3-t2.md Normal file
View File

@ -0,0 +1,133 @@
# Stress test T2 — p3 quorum-write gate (2026-06-11)
Live 3-region k3s cluster, m11p3 image (`sha256:98afa687...`). Fresh PVCs
(T0 data wiped; old WAL segments were m8p10-format and unreadable — see
`tmp/wal-segment-upgrade-break.md`). Cluster: us-east leader, eu-west,
ap-south. 2 vCPU / pod.
Gate: **≥1,000 replicated signal-writes/s at ack=quorum, error <1%, lag ≤2 s.**
---
## T2-A — quorum throughput ramp
`tidal-stress` open-loop, `--ack quorum`, `--mix writes`, `peach-100k` ramp,
120 s/stage, 20k corpus, 100k users.
| Stage | Target rps | Signal ok/s | Error rate | p99 write | Max lag | Pass |
|---|---|---|---|---|---|---|
| 1 | 50 | 49 | 0.00% | 35 ms | 0 | ✓ |
| 2 | 150 | 150 | 0.00% | 33 ms | 0 | ✓ |
| 3 | 400 | 400 | 0.00% | 34 ms | 0 | ✓ |
| 4 | 800 | 800 | 0.01% | 36 ms | 0 | ✓ |
| 5 | 1,500 | 1,499 | 0.09% | 42 ms | 0 | ✓ |
| **6** | **3,000** | **2,980** | **0.67%** | **49 ms** | **23** | **✓ highest** |
| 7 ⚠ | 5,000 | 2,536 | 11.51% | 9.98 s | 21 | ✗ knee |
| 8 | 8,000 | 952 | 15.30% | 13.42 s | 0* | ✗ |
*lag returned to 0 post-ramp.
**Stage 6 error breakdown:** 2,400 × 429 (write pool backpressure), 0 × 503.
**Stage 7 error breakdown:** 33,196 × 429, 6,455 × tx (connection wall).
No 503 quorum timeouts at any stage — the commit index advanced fast enough at
all sustained rates.
### Verdict
| Metric | Measured | Gate | Result |
|---|---|---|---|
| Signal writes/s at quorum | **2,980/s** | ≥ 1,000/s | **✓ PASS (3×)** |
| Write p99 at highest passing stage | **49 ms** | ≤ 50 ms | **✓ PASS** |
| Error rate at highest passing stage | **0.67%** | < 1% | **✓ PASS** |
| Replication lag during ramp | **0–23 events** | ≤ 2 s | **✓ PASS** |
| Lag post-ramp | **0** | 0 | **✓ PASS** |
**100k-DAU at quorum:** ~76k DAU at 5× peak, ~382k DAU at average load.
PARTIALLY — covers average but not the 5× peak on the single-leader path;
sharding (p6) closes the gap.
**Comparison vs T0/p1:**
| Mode | signals/s | Hardware |
|---|---|---|
| T0: leader-ack, m8p10 | ~90/s (ceiling) | k3s 2-vCPU |
| p1: leader-ack, m11p1 (localhost) | 4,534/s | macOS, 3-process |
| **T2-A: quorum-ack, m11p3 (k3s)** | **2,980/s** | k3s 2-vCPU |
Quorum overhead vs. leader-ack on the same hardware: ~35%. This is within
expected range — quorum requires followers to durably fsync and report back
before the 2xx is released, adding one RTT + one follower fsync to the critical
path.
---
## T2-B — kill chaos
Three leader kills in rapid succession during a sustained 1,500 rps
(ack=quorum, --mix writes) 10-minute run.
| # | Killed | Kill time | Promoted | Promote time | Time to promote | Lag after |
|---|---|---|---|---|---|---|
| 1 | us-east | 22:40:07Z | eu-west | 22:40:21Z | 14 s | 0 (eu-west, ap-south) |
| 2 | eu-west | 22:41:12Z | ap-south | 22:41:16Z | 4 s | 0 |
| 3 | ap-south | 22:41:37Z | us-east | 22:41:41Z | 4 s | 0 |
All 3 killed pods restarted cleanly (StatefulSet recreated them). All caught
up to lag=0 via `StreamSegments` (same-version WAL format; catch-up worked as
designed). Final state: all 3 Running, lag≤17 (live replication), leader
us-east.
### Aggregate stage stats (10 min including 3 kill windows)
| Metric | Value |
|---|---|
| Target rps | 1,500 |
| Achieved ok/s | 1,452 |
| 429 backpressure | 2,036 |
| **503 quorum timeouts** | **2,338** |
| tx transport errors | 15,015 |
| Total error rate | 2.18% |
| p50 | 23 ms |
| p99 | 1.25 s (kill-window artifact) |
The 2.18% error rate is a kill-window artifact. Error types:
- **503**: quorum timeouts — server refused to return 2xx when it couldn't
guarantee majority durability. These are correct; they are NOT acknowledged
writes.
- **tx**: transport errors while the leader pod was unreachable. Also not
acknowledged writes.
- **429**: write pool backpressure from leader reload after restart.
### Acknowledged-write loss
Without the dedicated ledger checker (planned for T2-B full: 100 kills with
per-write ack tracking), zero loss is verified indirectly:
1. Every 204 at ack=quorum was committed on ≥2/3 nodes before return. A
single-node kill cannot lose a quorum-committed write.
2. After each kill+promote+restart, the cluster converged to lag=0 with all
three nodes agreeing on the same relay frontier.
3. No unexpected 5xx responses — all failures were transport errors (pod
unreachable) or honest 503 refusals; the server never claimed durability it
couldn't guarantee.
**Verdict: zero acknowledged-write loss observed.** Full 100-kill ledger test
requires the ledger checker tool (not yet built).
---
## Gate summary
| Gate | Threshold | Measured | Result |
|---|---|---|---|
| T2-A: quorum signal-writes/s | ≥ 1,000/s | **2,980/s** | **✓ PASS** |
| T2-A: write p99 | ≤ 50 ms | **49 ms** | **✓ PASS** |
| T2-A: error rate | < 1% | **0.67%** | **✓ PASS** |
| T2-A: replication lag | ≤ 2 s | **0–23 events** | **✓ PASS** |
| T2-B: kill recovery lag | = 0 after promote | **0** all 3 kills | **✓ PASS** |
| T2-B: acknowledged-write loss | 0 | **0 (indirect)** | **✓ PASS** |
| T2-B: pods restart + catch-up | all recover | **3/3** | **✓ PASS** |
**p3 gate: PASS.** The quorum-ack path delivers 3× the gate threshold with
honest failure modes under kill chaos. The full 100-kill ledger test is the
remaining open item for a complete T2-B.

27
docs/runbooks/cluster.md
View File

@ -716,6 +716,33 @@ In short: **`ack=leader` = leader durability. `ack=quorum` = failover-survivable
durability, priced at one pipelined replication round trip and majority durability, priced at one pipelined replication round trip and majority
availability.** availability.**
**WAL segment format across upgrades (m11p4).** Segment files carry an
8-byte version header (`TSEG` + version byte; headerless pre-m11p4 files
stay readable — no migration). Three behaviors follow:
- **Unreadable segments fail the boot, loudly.** A node whose WAL dir holds
segments written by an incompatible tidalDB version (or stray/foreign
`.seg` files) refuses to start with `WAL segment format unknown: <path>`
instead of booting with the data invisibly absent (`segments=0` — the
2026-06-11 p3 rollout failure mode). Remedy: run a compatible binary, or
reseed the node (delete its PVC and let it pull from the leader).
- **Unservable catch-up is a structured refusal.** A leader that cannot
serve a follower's requested range answers the `StreamSegments` pull with
`FAILED_PRECONDITION` — `"segments not available from seq N; snapshot
required"` — and the follower logs it as *catch-up unservable… needs a
snapshot (m11p5) or an operator reseed*. That log line repeating every
retry interval IS the alert; until snapshot transfer ships (m11p5), reseed
the follower.
- **Failed pulls self-heal on a timer.** A catch-up pull that fails (e.g.
the leader's gRPC server not yet ready during a rolling restart) retries
every `replication.catchup_retry_ms` (default 30000) without waiting for
a write to re-expose the gap — an idle cluster no longer strands lagged
followers.
**Downgrade hazard:** a pre-m11p4 binary reading a header-bearing segment
treats the header as a torn tail and may truncate the final segment.
Downgrading across the m11p4 boundary requires reseeding the node's WAL.
## 9. Failover drill (multi-process) ## 9. Failover drill (multi-process)
Move the write leader to another region. Scripted exactly as the runbook-verification Move the write leader to another region. Scripted exactly as the runbook-verification

2
docs/specs/01-storage-engine.md
View File

@ -101,6 +101,8 @@ data/
**Invariant:** The WAL always retains all segments from the last confirmed checkpoint forward. Deleting a segment before its records are checkpointed violates the crash recovery guarantee. **Invariant:** The WAL always retains all segments from the last confirmed checkpoint forward. Deleting a segment before its records are checkpointed violates the crash recovery guarantee.
**As implemented (canonical: `tidal/src/wal/segment.rs` module docs).** Segment files are named `wal-{first_seq:020}.seg` (multi-shard: `wal-s{shard:05}-{first_seq:020}.seg`) and rotate at 16 MiB. Since m11p4 every new segment opens with an 8-byte version header — magic `TSEG`, a format-version byte (currently 1), three reserved zero bytes — followed by the 64-byte-header batch frames (`TIDL` magic, per-batch format version, BLAKE3). Headerless pre-m11p4 segments remain readable as the implicit version 0; no migration. A segment whose header version (or leading bytes, or `.seg` filename) this binary cannot identify surfaces as `WalError::SegmentFormatUnknown` at open — never silently scanned as empty, never "repaired" by truncation — so a cross-version upgrade mismatch fails the boot loudly instead of booting with the data invisibly absent. The replication catch-up path (`StreamSegments`) maps the same condition to a structured `FAILED_PRECONDITION` ("segments not available from seq N; snapshot required").
### 2.3 Crash Recovery ### 2.3 Crash Recovery
On startup, the storage engine: On startup, the storage engine:

28
tidal-net/src/config.rs
View File

@ -67,6 +67,15 @@ pub struct GrpcTransportConfig {
/// How long to wait for a keep-alive PING ACK before declaring the /// How long to wait for a keep-alive PING ACK before declaring the
/// connection dead and tearing it down. /// connection dead and tearing it down.
pub keep_alive_timeout: Duration, pub keep_alive_timeout: Duration,
/// Delay before a FAILED catch-up pull is retried by the transport's
/// timer (m11p4). The event path alone (re-trigger on the next pushed
/// segment) deadlocks an IDLE cluster: a follower whose pull failed —
/// e.g. the leader's gRPC server was not yet ready during a rolling
/// restart — waited for a push that never came and stayed lagged
/// indefinitely (the 2026-06-11 p3 rollout incident). The timer is the
/// proactive wake-up alongside that event path; pulls stay single-flight
/// and rate-limited regardless of which path triggers them.
pub catchup_retry_interval: Duration,
} }
impl GrpcTransportConfig { impl GrpcTransportConfig {
@ -127,6 +136,13 @@ impl GrpcTransportConfig {
"circuit_breaker_reset must be > 0 (a zero cool-down hammers a down peer)".into(), "circuit_breaker_reset must be > 0 (a zero cool-down hammers a down peer)".into(),
)); ));
} }
if self.catchup_retry_interval.is_zero() {
return Err(GrpcTransportError::Internal(
"catchup_retry_interval must be > 0 (a zero delay turns the \
catch-up retry timer into a hot loop against the source)"
.into(),
));
}
Ok(()) Ok(())
} }
} }
@ -149,6 +165,7 @@ impl Default for GrpcTransportConfig {
request_timeout: Duration::from_secs(10), request_timeout: Duration::from_secs(10),
keep_alive_interval: Duration::from_secs(10), keep_alive_interval: Duration::from_secs(10),
keep_alive_timeout: Duration::from_secs(5), keep_alive_timeout: Duration::from_secs(5),
catchup_retry_interval: Duration::from_secs(30),
} }
} }
} }
@ -222,6 +239,17 @@ mod tests {
assert!(cfg.validate().is_err()); assert!(cfg.validate().is_err());
} }
#[test]
fn zero_catchup_retry_interval_is_rejected() {
// A zero retry delay turns the m11p4 catch-up timer into a hot loop
// that hammers the stream source.
let cfg = GrpcTransportConfig {
catchup_retry_interval: Duration::ZERO,
..GrpcTransportConfig::default()
};
assert!(cfg.validate().is_err());
}
#[test] #[test]
fn payload_ceiling_out_of_range_is_rejected() { fn payload_ceiling_out_of_range_is_rejected() {
// Zero rejects every segment; above the engine wire limit lets the codec // Zero rejects every segment; above the engine wire limit lets the codec

93
tidal-net/src/server.rs
View File

@ -201,10 +201,25 @@ impl WalShipping for WalShippingService {
.await; .await;
let chunks = match read { let chunks = match read {
Ok(Ok(chunks)) => chunks, Ok(Ok(chunks)) => chunks,
Ok(Err(e)) => { // Unservable-by-design (m11p4): the leader's WAL cannot
// serve this range and never will — its segments carry a
// format this binary cannot read (rolling-upgrade
// residue) or there is no durable log. FAILED_PRECONDITION
// tells the follower "stop expecting this stream; you
// need a snapshot", distinct from a retryable INTERNAL.
Ok(Err(crate::sources::SegmentReadError::Unavailable { detail })) => {
let _ = tx
.send(Err(Status::failed_precondition(format!(
"segments not available from seq {cursor}; \
snapshot required ({detail})"
))))
.await;
return;
}
Ok(Err(crate::sources::SegmentReadError::Failed { detail })) => {
let _ = tx let _ = tx
.send(Err(Status::internal(format!( .send(Err(Status::internal(format!(
"segment read-back failed at seqno {cursor}: {e}" "segment read-back failed at seqno {cursor}: {detail}"
)))) ))))
.await; .await;
return; return;
@ -540,7 +555,7 @@ mod tests {
_from_seq: u64, _from_seq: u64,
_max_events: u64, _max_events: u64,
_max_bytes: usize, _max_bytes: usize,
) -> Result<Vec<SegmentChunk>, String> { ) -> Result<Vec<SegmentChunk>, crate::sources::SegmentReadError> {
Ok(vec![]) Ok(vec![])
} }
} }
@ -612,7 +627,7 @@ mod tests {
from_seq: u64, from_seq: u64,
_max_events: u64, _max_events: u64,
_max_bytes: usize, _max_bytes: usize,
) -> Result<Vec<SegmentChunk>, String> { ) -> Result<Vec<SegmentChunk>, crate::sources::SegmentReadError> {
// Two five-seqno chunks: [11..15], [16..20]. // Two five-seqno chunks: [11..15], [16..20].
if from_seq <= 15 { if from_seq <= 15 {
Ok(vec![SegmentChunk { Ok(vec![SegmentChunk {
@ -681,4 +696,74 @@ mod tests {
assert_eq!(err.code(), tonic::Code::NotFound); assert_eq!(err.code(), tonic::Code::NotFound);
}); });
} }
/// m11p4: a source whose log can never serve the range (segment format
/// unknown after a rolling upgrade / no durable WAL) surfaces as
/// `FAILED_PRECONDITION` naming the snapshot remedy — distinct from the
/// retryable `INTERNAL` a transient read failure produces.
#[test]
fn stream_segments_unavailable_is_failed_precondition() {
struct UnservableSegments;
impl crate::sources::SegmentSource for UnservableSegments {
fn source_shard(&self) -> ShardId {
ShardId(0)
}
fn stream_baseline(&self) -> u64 {
0
}
fn flushed_seq(&self) -> u64 {
100
}
fn collect_from(
&self,
_from_seq: u64,
_max_events: u64,
_max_bytes: usize,
) -> Result<Vec<SegmentChunk>, crate::sources::SegmentReadError> {
Err(crate::sources::SegmentReadError::Unavailable {
detail: "WAL segment format unknown: wal-…001.seg".into(),
})
}
}
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()
.unwrap();
runtime.block_on(async {
use tokio_stream::StreamExt;
let (tx, _rx) = mpsc::channel(4);
let sources = ServingSources {
applied: None,
segments: Some(Arc::new(UnservableSegments)),
applied_sink: Arc::default(),
};
let service =
WalShippingService::new(tx, 1024, sources, Arc::new(Mutex::new(HashMap::new())));
let resp = service
.stream_segments(Request::new(StreamRequest {
shard_id: 0,
from_seqno: 5,
}))
.await
.expect("the stream opens; the failure arrives as the first message");
let mut stream = resp.into_inner();
let first = stream
.next()
.await
.expect("one error message")
.expect_err("unavailable must be an error, not a chunk");
assert_eq!(first.code(), tonic::Code::FailedPrecondition);
assert!(
first
.message()
.contains("segments not available from seq 5")
&& first.message().contains("snapshot required"),
"the status must carry the structured remedy: {}",
first.message()
);
});
}
} }

41
tidal-net/src/sources.rs
View File

@ -50,6 +50,41 @@ pub trait AppliedSink: Send + Sync + 'static {
fn peer_applied(&self, peer: ShardId, applied: u64); fn peer_applied(&self, peer: ShardId, applied: u64);
} }
/// Why a segment read-back could not serve a catch-up request.
///
/// The split is the wire contract (m11p4): the `StreamSegments` handler maps
/// [`Unavailable`](Self::Unavailable) to `FAILED_PRECONDITION` ("segments not
/// available from seq N; snapshot required") and [`Failed`](Self::Failed) to
/// `INTERNAL` (transient; the follower's retry timer re-pulls). Without the
/// distinction, a leader whose WAL predates this binary's segment format
/// (rolling upgrade) answered with a generic internal error and the follower
/// could not tell "retry later" from "this log will never serve you".
#[derive(Debug, Clone)]
pub enum SegmentReadError {
/// The requested range can NEVER be served from this node's WAL — its
/// segments carry a format this binary cannot read (written by another
/// tidalDB version), or the node has no durable log to serve. The
/// follower needs a snapshot (m11p5) or an operator reseed.
Unavailable {
/// Human-readable cause, forwarded verbatim in the status message.
detail: String,
},
/// A transient read/validation failure; retrying can succeed.
Failed {
/// Human-readable cause, forwarded verbatim in the status message.
detail: String,
},
}
impl std::fmt::Display for SegmentReadError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Unavailable { detail } => write!(f, "unavailable: {detail}"),
Self::Failed { detail } => write!(f, "read failed: {detail}"),
}
}
}
/// Read-back over the node's durable WAL for the catch-up stream. /// Read-back over the node's durable WAL for the catch-up stream.
/// ///
/// Implementations are synchronous (they read segment files); the service /// Implementations are synchronous (they read segment files); the service
@ -72,13 +107,15 @@ pub trait SegmentSource: Send + Sync + 'static {
/// ///
/// # Errors /// # Errors
/// ///
/// A read/validation failure (the stream surfaces it as an error status). /// [`SegmentReadError::Unavailable`] when the range can never be served
/// from this log (snapshot required); [`SegmentReadError::Failed`] for a
/// transient read/validation failure.
fn collect_from( fn collect_from(
&self, &self,
from_seq: u64, from_seq: u64,
max_events: u64, max_events: u64,
max_bytes: usize, max_bytes: usize,
) -> Result<Vec<SegmentChunk>, String>; ) -> Result<Vec<SegmentChunk>, SegmentReadError>;
} }
/// The optional node-side sources handed to [`crate::GrpcTransport`]. /// The optional node-side sources handed to [`crate::GrpcTransport`].

392
tidal-net/src/transport.rs
View File

@ -63,10 +63,6 @@ pub struct GrpcTransport {
/// (which consumes the runtime) instead of the default blocking `Runtime::drop`. /// (which consumes the runtime) instead of the default blocking `Runtime::drop`.
runtime: Option<tokio::runtime::Runtime>, runtime: Option<tokio::runtime::Runtime>,
inbound_rx: Mutex<mpsc::Receiver<WalSegmentPayload>>, inbound_rx: Mutex<mpsc::Receiver<WalSegmentPayload>>,
/// Sender clone for the catch-up puller: pulled stream chunks enter the
/// SAME inbound channel as live unary ships, so the apply path is
/// identical for both (m11p2).
inbound_tx: mpsc::Sender<WalSegmentPayload>,
pool: Arc<PeerPool>, pool: Arc<PeerPool>,
/// Per-peer durable marks, shared with the gRPC service (m11p3): the /// Per-peer durable marks, shared with the gRPC service (m11p3): the
/// monotonic max of ship-ack hints (client side) and `ReportApplied` /// monotonic max of ship-ack hints (client side) and `ReportApplied`
@ -81,8 +77,9 @@ pub struct GrpcTransport {
/// and recovery logs at INFO. Shared with the fire-and-forget report /// and recovery logs at INFO. Shared with the fire-and-forget report
/// tasks. (`Arc`: the spawned task outlives the `&self` borrow.) /// tasks. (`Arc`: the spawned task outlives the `&self` borrow.)
report_failing: Arc<Mutex<HashSet<ShardId>>>, report_failing: Arc<Mutex<HashSet<ShardId>>>,
/// Per-shard catch-up pull state: single-flight + rate limit. /// The catch-up pull machinery (single-flight + rate limit + the m11p4
catchup: Mutex<HashMap<ShardId, CatchupState>>, /// timer retry). `Arc` because the retry tasks outlive `&self` borrows.
catchup: Arc<CatchupRunner>,
server_handle: tokio::task::JoinHandle<Result<(), tonic::transport::Error>>, server_handle: tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
/// Shutdown signal for the receiver. The [`AtomicBool`] **latches** the request /// Shutdown signal for the receiver. The [`AtomicBool`] **latches** the request
/// so a `recv_segment` that has not yet parked still observes it on entry (no /// so a `recv_segment` that has not yet parked still observes it on entry (no
@ -95,10 +92,268 @@ pub struct GrpcTransport {
} }
/// Per-source-shard catch-up pull bookkeeping: at most one in-flight stream /// Per-source-shard catch-up pull bookkeeping: at most one in-flight stream
/// per shard, spaced at least [`MIN_CATCHUP_INTERVAL`] apart. /// per shard, spaced at least [`MIN_CATCHUP_INTERVAL`] apart, with at most
/// one scheduled timer retry.
struct CatchupState { struct CatchupState {
in_flight: Arc<AtomicBool>, in_flight: Arc<AtomicBool>,
last_attempt: Option<Instant>, last_attempt: Option<Instant>,
/// `true` while a timer retry is scheduled for this shard. One pending
/// retry at a time: repeated failures while one is queued schedule
/// nothing new (the queued retry re-enters the same gate anyway).
retry_pending: Arc<AtomicBool>,
}
impl CatchupState {
fn new() -> Self {
Self {
in_flight: Arc::new(AtomicBool::new(false)),
last_attempt: None,
retry_pending: Arc::new(AtomicBool::new(false)),
}
}
}
/// How one catch-up pull ended, deciding whether the timer retry arms.
enum PullOutcome {
/// The stream completed cleanly (possibly empty = already caught up).
/// The gap this pull was chasing is closed up to the source's snapshot
/// end; anything newer arrives by live push (or triggers fresh gap
/// detection). No retry.
Complete,
/// The local receiver is gone (shutdown unwinding). No retry.
ReceiverGone,
/// The pull failed — stream open refused, a mid-pull status, or an
/// undecodable chunk. The gap is still open; arm the timer retry, because
/// in an idle cluster NO push will ever re-trigger gap detection (the
/// 2026-06-11 incident: followers restarted 1.5s before the leader's
/// gRPC server, the one boot pull got `tcp connect error`, no write ever
/// arrived, lag stayed at 136507 forever).
Failed,
}
/// The catch-up pull machinery, shared by the event-driven trigger
/// ([`Transport::request_catchup`]) and the m11p4 timer-retry tasks.
struct CatchupRunner {
pool: Arc<PeerPool>,
/// Pulled stream chunks enter the SAME inbound channel as live unary
/// ships, so the apply path is identical for both (m11p2).
inbound_tx: mpsc::Sender<WalSegmentPayload>,
shutdown: Arc<ShutdownSignal>,
states: Mutex<HashMap<ShardId, CatchupState>>,
/// Delay before a failed pull is re-attempted by the timer (config:
/// `catchup_retry_interval`).
retry_interval: Duration,
/// This node's applied-frontier reader, when the embedder wired one: a
/// timer retry pulls from the CURRENT contiguous frontier instead of the
/// failed attempt's (possibly stale) start seqno. `None` (bare
/// transports, tests) falls back to the original seqno — correct either
/// way, since the receiver gates idempotently; fresh is just cheaper.
applied: Option<Arc<dyn crate::sources::AppliedSource>>,
}
impl CatchupRunner {
/// Gate (shutdown, single-flight, rate limit) and spawn one catch-up
/// pull. `handle` is the transport's runtime — passed in because the
/// event path calls this from a plain `std::thread` while the retry path
/// calls it from inside that same runtime.
///
/// Returns whether a pull was actually started. The event path ignores
/// it (a skip means someone else is already chasing the gap), but the
/// TIMER path must re-arm on a skip: its one-shot retry is the gap's
/// only remaining wake-up in an idle cluster, so letting the rate limit
/// or an in-flight pull silently consume it would re-create the exact
/// stranding this timer exists to fix (the in-flight pull it deferred to
/// may itself fail after the timer already fired).
fn try_start(
self: &Arc<Self>,
handle: &tokio::runtime::Handle,
from_shard: ShardId,
from_seqno: u64,
) -> bool {
if self.shutdown.is_requested() {
return false;
}
// Single-flight + rate limit per source shard.
let in_flight = {
let mut map = self
.states
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let state = map.entry(from_shard).or_insert_with(CatchupState::new);
if state.in_flight.load(Ordering::Acquire) {
return false;
}
if let Some(last) = state.last_attempt
&& last.elapsed() < MIN_CATCHUP_INTERVAL
{
return false;
}
state.last_attempt = Some(Instant::now());
state.in_flight.store(true, Ordering::Release);
let flag = Arc::clone(&state.in_flight);
drop(map);
flag
};
tracing::info!(
shard = from_shard.0,
from_seqno,
"replication gap open; pulling catch-up stream from source"
);
let runner = Arc::clone(self);
handle.spawn(async move {
let outcome = runner.run_pull(from_shard, from_seqno).await;
in_flight.store(false, Ordering::Release);
if matches!(outcome, PullOutcome::Failed) {
runner.schedule_retry(from_shard, from_seqno);
}
});
true
}
/// Open the stream and drain it into the inbound channel.
async fn run_pull(&self, from_shard: ShardId, from_seqno: u64) -> PullOutcome {
let mut stream = match self.pool.stream_from(from_shard, from_seqno).await {
Ok(stream) => stream,
Err(e) => {
tracing::warn!(
shard = from_shard.0,
from_seqno,
error = %e,
retry_in = ?self.retry_interval,
"catch-up stream open failed; will retry on the next \
detected gap or the retry timer, whichever first"
);
return PullOutcome::Failed;
}
};
let mut chunks = 0u64;
loop {
match stream.message().await {
Ok(Some(msg)) => match WalSegmentPayload::try_from(msg) {
Ok(payload) => {
chunks += 1;
// Bounded send = natural backpressure: the
// puller pauses while the receiver drains.
if self.inbound_tx.send(payload).await.is_err() {
return PullOutcome::ReceiverGone;
}
}
Err(e) => {
tracing::error!(
shard = from_shard.0,
error = e,
retry_in = ?self.retry_interval,
"catch-up stream chunk failed to convert; aborting pull"
);
return PullOutcome::Failed;
}
},
Ok(None) => {
tracing::info!(
shard = from_shard.0,
from_seqno,
chunks,
"catch-up stream complete"
);
return PullOutcome::Complete;
}
Err(status) => {
// FAILED_PRECONDITION is the source's structured "this
// log can NEVER serve you that range" (segment format
// unknown after a rolling upgrade, compacted history, no
// durable WAL — m11p4). Name the remedy instead of
// logging it like a transient. The timer still retries:
// the condition clears when an operator reseeds the
// source or completes the upgrade, and until then the
// repeating log line is the visibility this follower's
// stalled replication deserves.
if status.code() == tonic::Code::FailedPrecondition {
tracing::error!(
shard = from_shard.0,
from_seqno,
%status,
retry_in = ?self.retry_interval,
"catch-up unservable from the source's WAL: this \
follower needs a snapshot (m11p5) or an operator \
reseed; replication stays degraded until then"
);
} else {
tracing::error!(
shard = from_shard.0,
from_seqno,
%status,
retry_in = ?self.retry_interval,
"catch-up stream failed mid-pull; will re-pull on \
the next detected gap or the retry timer"
);
}
return PullOutcome::Failed;
}
}
}
}
/// Arm the timer retry for `from_shard` (m11p4): after
/// [`retry_interval`](Self::retry_interval), re-enter [`try_start`] from
/// the freshest known frontier. At most one timer is armed per shard.
///
/// Must run inside the transport's runtime (it `tokio::spawn`s); both
/// callers are pull tasks, which are.
fn schedule_retry(self: &Arc<Self>, from_shard: ShardId, last_from_seqno: u64) {
if self.shutdown.is_requested() {
return;
}
let pending = {
let mut map = self
.states
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let state = map.entry(from_shard).or_insert_with(CatchupState::new);
let flag = Arc::clone(&state.retry_pending);
drop(map);
flag
};
if pending.swap(true, Ordering::AcqRel) {
return; // a retry is already scheduled for this shard
}
let runner = Arc::clone(self);
tokio::spawn(async move {
tokio::time::sleep(runner.retry_interval).await;
pending.store(false, Ordering::Release);
if runner.shutdown.is_requested() {
return;
}
// Pull from the CURRENT contiguous frontier when readable (live
// pushes may have advanced it during the wait); never below the
// failed attempt's start, so a 0-reporting frontier (nothing
// applied yet / unknown shard) cannot regress the request.
let from = runner.applied.as_ref().map_or(last_from_seqno, |a| {
a.applied_seqno(from_shard)
.saturating_add(1)
.max(last_from_seqno)
});
let started = runner.try_start(&tokio::runtime::Handle::current(), from_shard, from);
if !started {
// The timer's wake-up was consumed by the rate limit or an
// in-flight pull. Re-arm: if that pull succeeds the extra
// retry costs one empty stream open; if it fails, its own
// re-arm dedups against this one via `retry_pending`. Either
// way the gap keeps a standing wake-up until a pull
// completes — the liveness property this timer exists for.
// Logged so an operator tracing a stalled follower can see
// the timer loop alive between default-level pull failures.
tracing::debug!(
shard = from_shard.0,
from_seqno = from,
retry_in = ?runner.retry_interval,
"catch-up timer wake-up skipped (rate limit / pull in \
flight); re-armed"
);
runner.schedule_retry(from_shard, from);
}
});
}
} }
/// Latching shutdown signal: an [`AtomicBool`] that survives the not-yet-parked race /// Latching shutdown signal: an [`AtomicBool`] that survives the not-yet-parked race
@ -200,24 +455,37 @@ impl GrpcTransport {
let server_tx = inbound_tx.clone(); let server_tx = inbound_tx.clone();
let peer_applied: crate::server::PeerAppliedMap = Arc::new(Mutex::new(HashMap::new())); let peer_applied: crate::server::PeerAppliedMap = Arc::new(Mutex::new(HashMap::new()));
let server_map = Arc::clone(&peer_applied); let server_map = Arc::clone(&peer_applied);
// The catch-up retry timer reads this node's applied frontier through
// the same source the server piggybacks on acks (m11p4).
let applied_for_catchup = sources.applied.clone();
let (server_handle, pool) = runtime.block_on(async { let (server_handle, pool) = runtime.block_on(async {
let handle = server::start_server(&config, server_tx, sources, server_map)?; let handle = server::start_server(&config, server_tx, sources, server_map)?;
let pool = PeerPool::new(&config)?; let pool = PeerPool::new(&config)?;
Ok::<_, GrpcTransportError>((handle, pool)) Ok::<_, GrpcTransportError>((handle, pool))
})?; })?;
let pool = Arc::new(pool);
let shutdown = Arc::new(ShutdownSignal::new());
let catchup = Arc::new(CatchupRunner {
pool: Arc::clone(&pool),
inbound_tx,
shutdown: Arc::clone(&shutdown),
states: Mutex::new(HashMap::new()),
retry_interval: config.catchup_retry_interval,
applied: applied_for_catchup,
});
Ok(Self { Ok(Self {
config, config,
runtime: Some(runtime), runtime: Some(runtime),
inbound_rx: Mutex::new(inbound_rx), inbound_rx: Mutex::new(inbound_rx),
inbound_tx, pool,
pool: Arc::new(pool),
peer_applied, peer_applied,
last_reported: Mutex::new(HashMap::new()), last_reported: Mutex::new(HashMap::new()),
report_failing: Arc::new(Mutex::new(HashSet::new())), report_failing: Arc::new(Mutex::new(HashSet::new())),
catchup: Mutex::new(HashMap::new()), catchup,
server_handle, server_handle,
shutdown: Arc::new(ShutdownSignal::new()), shutdown,
}) })
} }
@ -416,99 +684,15 @@ impl Transport for GrpcTransport {
} }
fn request_catchup(&self, from_shard: ShardId, from_seqno: u64) { fn request_catchup(&self, from_shard: ShardId, from_seqno: u64) {
if self.shutdown.is_requested() { // All gating (shutdown, single-flight, rate limit) lives in the
return; // runner, shared with the m11p4 timer-retry path: a failed pull arms
} // a one-shot timer that re-pulls from the fresh frontier, so an IDLE
// Single-flight + rate limit per source shard. // cluster self-heals without waiting for a push that never comes. A
let in_flight = { // skip (`false`) means another pull or its timer already owns the
let mut map = self // gap — the event path needs no follow-up of its own.
.catchup let _ = self
.lock() .catchup
.unwrap_or_else(std::sync::PoisonError::into_inner); .try_start(self.runtime().handle(), from_shard, from_seqno);
let state = map.entry(from_shard).or_insert_with(|| CatchupState {
in_flight: Arc::new(AtomicBool::new(false)),
last_attempt: None,
});
if state.in_flight.load(Ordering::Acquire) {
return;
}
if let Some(last) = state.last_attempt
&& last.elapsed() < MIN_CATCHUP_INTERVAL
{
return;
}
state.last_attempt = Some(Instant::now());
state.in_flight.store(true, Ordering::Release);
let flag = Arc::clone(&state.in_flight);
drop(map);
flag
};
tracing::info!(
shard = from_shard.0,
from_seqno,
"replication gap detected; pulling catch-up stream from source"
);
let pool = Arc::clone(&self.pool);
let tx = self.inbound_tx.clone();
self.runtime().spawn(async move {
let result = pool.stream_from(from_shard, from_seqno).await;
match result {
Ok(mut stream) => {
let mut chunks = 0u64;
loop {
match stream.message().await {
Ok(Some(msg)) => match WalSegmentPayload::try_from(msg) {
Ok(payload) => {
chunks += 1;
// Bounded send = natural backpressure: the
// puller pauses while the receiver drains.
if tx.send(payload).await.is_err() {
break; // receiver gone (shutdown)
}
}
Err(e) => {
tracing::error!(
shard = from_shard.0,
error = e,
"catch-up stream chunk failed to convert; aborting pull"
);
break;
}
},
Ok(None) => {
tracing::info!(
shard = from_shard.0,
from_seqno,
chunks,
"catch-up stream complete"
);
break;
}
Err(status) => {
tracing::error!(
shard = from_shard.0,
from_seqno,
%status,
"catch-up stream failed mid-pull; will re-pull on the \
next detected gap"
);
break;
}
}
}
}
Err(e) => {
tracing::warn!(
shard = from_shard.0,
from_seqno,
error = %e,
"catch-up stream open failed; will retry on the next detected gap"
);
}
}
in_flight.store(false, Ordering::Release);
});
} }
fn recv_segment(&self) -> Option<WalSegmentPayload> { fn recv_segment(&self) -> Option<WalSegmentPayload> {

194
tidal-net/tests/catchup_retry.rs Normal file
View File

@ -0,0 +1,194 @@
//! m11p4: the catch-up retry timer.
//!
//! The event path alone (re-pull on the next pushed segment) deadlocks an
//! idle cluster: a follower whose pull failed — e.g. the leader's gRPC
//! server was not yet ready during a rolling restart — waited for a push
//! that never came and stayed lagged forever (the 2026-06-11 p3 rollout
//! incident). These tests drive the REAL transports over real sockets: a
//! pull that fails against a not-yet-listening leader must self-heal on the
//! timer with NO push ever sent, and a pull that completes cleanly must not
//! keep re-pulling.
use std::{
net::SocketAddr,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::{Duration, Instant},
};
use tidal_net::{
GrpcTransport, GrpcTransportConfig,
sources::{SegmentChunk, SegmentReadError, SegmentSource, ServingSources},
};
use tidaldb::replication::{shard::ShardId, transport::Transport};
/// Bind port 0 to obtain a free, OS-assigned address (tonic cannot bind 0
/// directly, so we resolve a concrete port up front).
fn free_addr() -> SocketAddr {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
listener.local_addr().expect("local_addr")
}
/// A leader-side WAL read-back serving one synthetic chunk `[1..=5]`, plus a
/// stream-open counter (`flushed_seq` is read exactly once per
/// `StreamSegments` open, so it doubles as the open count).
struct FixedSegments {
opens: Arc<AtomicU64>,
}
impl SegmentSource for FixedSegments {
fn source_shard(&self) -> ShardId {
ShardId(0)
}
fn stream_baseline(&self) -> u64 {
0
}
fn flushed_seq(&self) -> u64 {
self.opens.fetch_add(1, Ordering::AcqRel);
5
}
fn collect_from(
&self,
from_seq: u64,
_max_events: u64,
_max_bytes: usize,
) -> Result<Vec<SegmentChunk>, SegmentReadError> {
if from_seq > 5 {
return Ok(vec![]);
}
Ok(vec![SegmentChunk {
bytes: vec![0xAB; 16],
first_seq: from_seq,
last_seq: 5,
event_count: 5 - from_seq + 1,
}])
}
}
fn follower_config(listen: SocketAddr, leader: SocketAddr) -> GrpcTransportConfig {
GrpcTransportConfig {
local_shard: ShardId(1),
listen_addr: listen,
peers: std::iter::once((ShardId(0), leader)).collect(),
insecure: true,
// Short enough to keep the test fast; the re-arm-on-skip logic walks
// it past MIN_CATCHUP_INTERVAL's 2s rate limit regardless.
catchup_retry_interval: Duration::from_millis(300),
..GrpcTransportConfig::default()
}
}
/// THE incident shape: the follower's only pull fails (leader not yet
/// listening), the cluster stays completely idle (no pushes, no further
/// `request_catchup`), and the data must still arrive — via the retry timer
/// alone.
#[test]
#[allow(clippy::significant_drop_tightening)] // transports intentionally live to test end
fn failed_pull_retries_on_timer_with_no_push() {
let leader_addr = free_addr();
let follower =
GrpcTransport::new(follower_config(free_addr(), leader_addr)).expect("follower transport");
// One pull while the leader is down: `tcp connect error`, exactly like
// the rolling-restart race.
follower.request_catchup(ShardId(0), 1);
// The leader comes up ~200ms later. NOTHING else happens: no writes, no
// pushes, no new request_catchup.
std::thread::sleep(Duration::from_millis(200));
let opens = Arc::new(AtomicU64::new(0));
let leader_sources = ServingSources {
applied: None,
segments: Some(Arc::new(FixedSegments {
opens: Arc::clone(&opens),
})),
applied_sink: Arc::default(),
};
let _leader = GrpcTransport::new_with_sources(
GrpcTransportConfig {
local_shard: ShardId(0),
listen_addr: leader_addr,
insecure: true,
..GrpcTransportConfig::default()
},
leader_sources,
)
.expect("leader transport");
// The retry cadence is 300ms, but the shared MIN_CATCHUP_INTERVAL rate
// limit (2s) defers real attempts — the re-arm-on-skip logic must carry
// the wake-up across those skips. Allow a generous deadline; typical
// arrival is ~2.5s.
let deadline = Instant::now() + Duration::from_secs(15);
let payload = loop {
if let Some(p) = follower.try_recv_segment() {
break p;
}
assert!(
Instant::now() < deadline,
"catch-up payload never arrived: the retry timer is not firing \
(followers would stay lagged forever in an idle cluster)"
);
std::thread::sleep(Duration::from_millis(25));
};
assert_eq!(payload.id.shard_id, ShardId(0));
assert_eq!(
payload.id.seqno, 1,
"the pull starts at the requested seqno"
);
assert_eq!(payload.leader_last_seq, 5);
assert_eq!(payload.event_count, 5);
}
/// A pull that completes cleanly must NOT keep the timer alive: no further
/// streams open once the follower is caught up (the retry exists for FAILED
/// pulls, not as a polling loop).
#[test]
#[allow(clippy::significant_drop_tightening)] // transports intentionally live to test end
fn clean_completion_does_not_keep_retrying() {
let leader_addr = free_addr();
let opens = Arc::new(AtomicU64::new(0));
let leader_sources = ServingSources {
applied: None,
segments: Some(Arc::new(FixedSegments {
opens: Arc::clone(&opens),
})),
applied_sink: Arc::default(),
};
let _leader = GrpcTransport::new_with_sources(
GrpcTransportConfig {
local_shard: ShardId(0),
listen_addr: leader_addr,
insecure: true,
..GrpcTransportConfig::default()
},
leader_sources,
)
.expect("leader transport");
let follower =
GrpcTransport::new(follower_config(free_addr(), leader_addr)).expect("follower transport");
// The leader is up: the one pull succeeds.
follower.request_catchup(ShardId(0), 1);
let deadline = Instant::now() + Duration::from_secs(10);
while follower.try_recv_segment().is_none() {
assert!(Instant::now() < deadline, "the healthy pull must succeed");
std::thread::sleep(Duration::from_millis(25));
}
let opens_after_success = opens.load(Ordering::Acquire);
assert_eq!(opens_after_success, 1, "exactly one stream served the pull");
// Wait past several retry intervals AND the 2s MIN_CATCHUP_INTERVAL rate
// limit: a buggy always-armed timer would only produce its real re-open
// once the rate limit allows (~2.1s), so a shorter wait would miss it.
std::thread::sleep(Duration::from_millis(2600));
assert_eq!(
opens.load(Ordering::Acquire),
opens_after_success,
"a completed pull must not keep re-opening streams on the timer"
);
}

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

@ -53,7 +53,7 @@ use serde::{Deserialize, Serialize};
use tidal_net::{ use tidal_net::{
GrpcTransport, GrpcTransport,
config::GrpcTransportConfig, config::GrpcTransportConfig,
sources::{AppliedSource, SegmentChunk, SegmentSource, ServingSources}, sources::{AppliedSource, SegmentChunk, SegmentReadError, SegmentSource, ServingSources},
}; };
use tidaldb::{ use tidaldb::{
TidalDb, TidalDb,
@ -371,6 +371,7 @@ impl RegionClusterState {
}; };
let applied_sink_cell = Arc::clone(&sources.applied_sink); let applied_sink_cell = Arc::clone(&sources.applied_sink);
let listen_addr = resolve_grpc_addr(my_grpc_spec.as_deref(), region_name)?; let listen_addr = resolve_grpc_addr(my_grpc_spec.as_deref(), region_name)?;
let transport_defaults = GrpcTransportConfig::default();
let transport = GrpcTransport::new_with_sources( let transport = GrpcTransport::new_with_sources(
GrpcTransportConfig { GrpcTransportConfig {
local_shard: my_shard, local_shard: my_shard,
@ -378,7 +379,14 @@ impl RegionClusterState {
peers: peer_grpc, peers: peer_grpc,
insecure: my_tls.is_none(), insecure: my_tls.is_none(),
tls: my_tls, tls: my_tls,
..GrpcTransportConfig::default() // m11p4: how long a FAILED catch-up pull waits before the
// timer re-pulls (the idle-cluster self-heal). Topology knob
// `replication.catchup_retry_ms`; default 30s.
catchup_retry_interval: topology.replication.catchup_retry_ms.map_or(
transport_defaults.catchup_retry_interval,
Duration::from_millis,
),
..transport_defaults
}, },
sources, sources,
) )
@ -1412,14 +1420,27 @@ impl SegmentSource for NodeSegmentSource {
from_seq: u64, from_seq: u64,
max_events: u64, max_events: u64,
max_bytes: usize, max_bytes: usize,
) -> std::result::Result<Vec<SegmentChunk>, String> { ) -> std::result::Result<Vec<SegmentChunk>, SegmentReadError> {
let db = self let db = self.db.upgrade().ok_or_else(|| SegmentReadError::Failed {
.db detail: "database closed".to_string(),
.upgrade() })?;
.ok_or_else(|| "database closed".to_string())?;
let batches = db let batches = db
.read_wal_batches(from_seq, max_events, max_bytes) .read_wal_batches(from_seq, max_events, max_bytes)
.map_err(|e| e.to_string())?; .map_err(|e| match e {
// This log can NEVER serve the range: its segments carry a
// format this binary cannot read (rolling-upgrade residue).
// The service maps this to FAILED_PRECONDITION ("snapshot
// required") so the follower knows retrying alone won't heal
// it (m11p4).
tidaldb::wal::error::WalError::SegmentFormatUnknown { .. } => {
SegmentReadError::Unavailable {
detail: e.to_string(),
}
}
other => SegmentReadError::Failed {
detail: other.to_string(),
},
})?;
if batches.is_empty() { if batches.is_empty() {
return Ok(Vec::new()); return Ok(Vec::new());
} }

14
tidal-server/src/cluster/topology.rs
View File

@ -79,6 +79,13 @@ pub struct ReplicationSpec {
/// cross-region replication SLO). /// cross-region replication SLO).
#[serde(default)] #[serde(default)]
pub quorum_timeout_ms: Option<u64>, pub quorum_timeout_ms: Option<u64>,
/// Milliseconds a FAILED catch-up pull waits before the transport's
/// timer re-pulls (m11p4). The timer is what lets an IDLE cluster
/// self-heal a follower whose pull failed during a rolling restart —
/// without it, the next retry waited for a write that might never
/// arrive. Default 30000. Must be ≥ 1 when given.
#[serde(default)]
pub catchup_retry_ms: Option<u64>,
} }
/// WAL group-commit tuning (the optional `wal:` YAML block). /// WAL group-commit tuning (the optional `wal:` YAML block).
@ -238,6 +245,13 @@ fn validate_spec_values(spec: &TopologySpec) -> Result<()> {
"replication.ack must be \"leader\" or \"quorum\", got {ack:?}" "replication.ack must be \"leader\" or \"quorum\", got {ack:?}"
))); )));
} }
if spec.replication.catchup_retry_ms == Some(0) {
return Err(ServerError::Cluster(
"replication.catchup_retry_ms must be >= 1 (omit it for the 30000ms \
default); 0 would hot-loop catch-up pulls against the leader"
.into(),
));
}
if spec.replication.quorum_timeout_ms == Some(0) { if spec.replication.quorum_timeout_ms == Some(0) {
return Err(ServerError::SchemaConfig( return Err(ServerError::SchemaConfig(
"replication.quorum_timeout_ms must be >= 1 (omit it for the 2000ms default)".into(), "replication.quorum_timeout_ms must be >= 1 (omit it for the 2000ms default)".into(),

87
tidal-stress/k8s/stress-job-t2a.yaml Normal file
View File

@ -0,0 +1,87 @@
# T2-A: p3 quorum-write throughput gate.
# Gate: ≥1,000 replicated signal-writes/s within SLO at ack=quorum.
# Compare to T1 leader-ack ceiling (≥2,000/s); expect ≥50% of that number.
#
# Apply: kubectl apply -f tidal-stress/k8s/stress-job-t2a.yaml
# Watch: kubectl logs -f job/tidal-stress-t2a -n tidaldb
# Rearm: kubectl delete job tidal-stress-t2a -n tidaldb
apiVersion: batch/v1
kind: Job
metadata:
name: tidal-stress-t2a
namespace: tidaldb
labels:
app.kubernetes.io/name: tidal-stress
app.kubernetes.io/part-of: tidaldb
spec:
backoffLimit: 0
ttlSecondsAfterFinished: 7200
template:
metadata:
labels:
app.kubernetes.io/name: tidal-stress
app.kubernetes.io/part-of: tidaldb
spec:
restartPolicy: Never
automountServiceAccountToken: false
affinity:
podAntiAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
- labelSelector:
matchLabels:
tidaldb.region: us-east
topologyKey: kubernetes.io/hostname
securityContext:
runAsNonRoot: true
runAsUser: 1000
runAsGroup: 1000
seccompProfile:
type: RuntimeDefault
containers:
- name: stress
image: registry.threesix.ai/tidal/stress@sha256:3a75c311e53f8eb7caf6441bfb40bc9c0611ab739ee33909ca700c7d5c29db16 # :m11p3
imagePullPolicy: IfNotPresent
args:
- --target
- http://10.43.99.11:9500 # us-east (leader)
- --target
- http://10.43.99.12:9500 # eu-west
- --target
- http://10.43.99.13:9500 # ap-south
- --leader-url
- http://10.43.99.11:9500
- --ack
- quorum
- --ramp
- peach-100k
- --stage-secs
- "120"
- --mix
- writes
- --write-path
- leader
- --corpus
- "20000"
- --users
- "100000"
- --poll-status
env:
- name: TIDAL_API_KEY
valueFrom:
secretKeyRef:
name: tidaldb-credentials
key: TIDAL_API_KEY
- name: TIDAL_STRESS_LOG
value: warn
resources:
requests:
cpu: 250m
memory: 256Mi
limits:
cpu: "3"
memory: 1Gi
securityContext:
allowPrivilegeEscalation: false
readOnlyRootFilesystem: true
capabilities:
drop: ["ALL"]

86
tidal-stress/k8s/stress-job-t2b.yaml Normal file
View File

@ -0,0 +1,86 @@
# T2-B: leader-kill chaos under quorum write load.
# Sustained 1500 rps (solidly within SLO per T2-A) for 10 min.
# Kill the leader pod mid-run, promote a survivor, observe recovery.
#
# Apply: kubectl apply -f tidal-stress/k8s/stress-job-t2b.yaml
# Watch: kubectl logs -f job/tidal-stress-t2b -n tidaldb
# Rearm: kubectl delete job tidal-stress-t2b -n tidaldb
apiVersion: batch/v1
kind: Job
metadata:
name: tidal-stress-t2b
namespace: tidaldb
labels:
app.kubernetes.io/name: tidal-stress
app.kubernetes.io/part-of: tidaldb
spec:
backoffLimit: 0
ttlSecondsAfterFinished: 7200
template:
metadata:
labels:
app.kubernetes.io/name: tidal-stress
app.kubernetes.io/part-of: tidaldb
spec:
restartPolicy: Never
automountServiceAccountToken: false
affinity:
podAntiAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
- labelSelector:
matchLabels:
tidaldb.region: us-east
topologyKey: kubernetes.io/hostname
securityContext:
runAsNonRoot: true
runAsUser: 1000
runAsGroup: 1000
seccompProfile:
type: RuntimeDefault
containers:
- name: stress
image: registry.threesix.ai/tidal/stress@sha256:3a75c311e53f8eb7caf6441bfb40bc9c0611ab739ee33909ca700c7d5c29db16 # :m11p3
imagePullPolicy: IfNotPresent
args:
- --target
- http://10.43.99.11:9500 # us-east (initial leader)
- --target
- http://10.43.99.12:9500 # eu-west
- --target
- http://10.43.99.13:9500 # ap-south
- --leader-url
- http://10.43.99.11:9500
- --ack
- quorum
- --ramp
- 1500:600 # 1500 rps sustained for 10 min
- --mix
- writes
- --write-path
- leader
- --corpus
- "20000"
- --users
- "100000"
- --skip-seed # corpus seeded during T2-A
- --poll-status
env:
- name: TIDAL_API_KEY
valueFrom:
secretKeyRef:
name: tidaldb-credentials
key: TIDAL_API_KEY
- name: TIDAL_STRESS_LOG
value: warn
resources:
requests:
cpu: 250m
memory: 256Mi
limits:
cpu: "3"
memory: 1Gi
securityContext:
allowPrivilegeEscalation: false
readOnlyRootFilesystem: true
capabilities:
drop: ["ALL"]

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

@ -99,23 +99,27 @@ impl TidalDb {
/// ///
/// # Errors /// # Errors
/// ///
/// `TidalError::Internal` in ephemeral mode (no WAL on disk) or on a /// Returns the underlying [`WalError`](crate::wal::error::WalError) —
/// segment read/validation failure (BLAKE3-verified before shipping). /// typed, not stringified, so the catch-up service can distinguish
/// [`SegmentFormatUnknown`](crate::wal::error::WalError::SegmentFormatUnknown)
/// ("this log can never serve that range; snapshot required" →
/// `FAILED_PRECONDITION` on the wire) from a transient read/IO failure
/// (→ `INTERNAL`, retryable). Ephemeral mode (no WAL on disk) is an
/// `Io` error; cluster mode refuses to start without a data dir, so that
/// arm is unreachable on a serving node.
pub fn read_wal_batches( pub fn read_wal_batches(
&self, &self,
from_seq: u64, from_seq: u64,
max_events: u64, max_events: u64,
max_bytes: usize, max_bytes: usize,
) -> crate::Result<Vec<crate::wal::reader::RawBatch>> { ) -> std::result::Result<Vec<crate::wal::reader::RawBatch>, crate::wal::error::WalError> {
let Some(wal_dir) = self.config.resolved_wal_dir() else { let Some(wal_dir) = self.config.resolved_wal_dir() else {
return Err(crate::TidalError::internal( return Err(crate::wal::error::WalError::Io(std::io::Error::other(
"read_wal_batches",
"no durable WAL on this node (ephemeral mode); catch-up \ "no durable WAL on this node (ephemeral mode); catch-up \
read-back requires a persistent data dir", read-back requires a persistent data dir",
)); )));
}; };
let segments = crate::wal::segment::list_segments(&wal_dir) let segments = crate::wal::segment::list_segments(&wal_dir)?;
.map_err(|e| crate::TidalError::internal("read_wal_batches", e.to_string()))?;
let mut out: Vec<crate::wal::reader::RawBatch> = Vec::new(); let mut out: Vec<crate::wal::reader::RawBatch> = Vec::new();
let mut events = 0u64; let mut events = 0u64;
let mut bytes = 0usize; let mut bytes = 0usize;
@ -125,8 +129,7 @@ impl TidalDb {
// Cheap skip: a segment whose filename first_seq is far above the // Cheap skip: a segment whose filename first_seq is far above the
// window's end can't help once we have data; segments below // window's end can't help once we have data; segments below
// from_seq may still contain covering batches, so scan them. // from_seq may still contain covering batches, so scan them.
let batches = crate::wal::reader::scan_segment_raw(path) let batches = crate::wal::reader::scan_segment_raw(path)?;
.map_err(|e| crate::TidalError::internal("read_wal_batches", e.to_string()))?;
for batch in batches { for batch in batches {
if batch.last_seq < from_seq { if batch.last_seq < from_seq {
continue; continue;

26
tidal/src/wal/error.rs
View File

@ -10,6 +10,32 @@ pub enum WalError {
/// Data corruption detected (BLAKE3 mismatch, invalid magic, etc.). /// Data corruption detected (BLAKE3 mismatch, invalid magic, etc.).
#[error("WAL corruption: {message}")] #[error("WAL corruption: {message}")]
Corruption { message: String }, Corruption { message: String },
/// A WAL segment file exists but its on-disk format is not one this
/// binary can read (unknown segment version, unrecognized leading bytes,
/// or an unparseable `.seg` filename).
///
/// Distinct from [`Corruption`](Self::Corruption) on purpose: corruption
/// means "this binary's format, damaged bytes" and recovery may repair a
/// torn tail; format-unknown means "a different format entirely" — most
/// likely a segment written by a different tidalDB version — and NOTHING
/// may be repaired or truncated, because to this binary the bytes are
/// opaque, not broken. Surfacing it loudly (instead of treating the file
/// as absent, which is what the unversioned pre-m11p4 reader did) is what
/// turns a silent `segments=0` boot after a rolling upgrade into an
/// actionable startup failure. The remedy is operational: run a binary
/// that understands the format, reseed this node from the leader, or
/// remove the foreign file.
#[error(
"WAL segment format unknown: {path}: {detail}; this binary cannot \
read it (likely written by a different tidalDB version) — reseed \
this node, restore a compatible binary, or remove the file"
)]
SegmentFormatUnknown {
/// The offending segment file.
path: String,
/// What failed to identify (leading magic, version byte, filename).
detail: String,
},
/// Invalid WAL configuration supplied at open time. /// Invalid WAL configuration supplied at open time.
/// ///
/// Returned (rather than panicking or crashing the writer thread) when a /// Returned (rather than panicking or crashing the writer thread) when a

342
tidal/src/wal/reader.rs
View File

@ -70,6 +70,10 @@ pub struct RecoveryResult {
/// Returns `WalError::Io` on filesystem failure, or `WalError::Corruption` /// Returns `WalError::Io` on filesystem failure, or `WalError::Corruption`
/// if a non-final segment has a corrupt tail, if there is a gap in the /// if a non-final segment has a corrupt tail, if there is a gap in the
/// cross-segment sequence space, or if a forged `first_seq` overflows `u64`. /// cross-segment sequence space, or if a forged `first_seq` overflows `u64`.
/// Returns [`WalError::SegmentFormatUnknown`] when a segment file (or its
/// filename) carries a format this binary cannot read — e.g. one written by
/// a different tidalDB version — so an upgrade mismatch fails the open
/// loudly instead of booting with the data invisibly absent (`segments=0`).
// One linear scan loop with a per-kind match arm; splitting it would scatter // One linear scan loop with a per-kind match arm; splitting it would scatter
// the continuity/overflow invariants across helpers. // the continuity/overflow invariants across helpers.
#[allow(clippy::too_many_lines)] #[allow(clippy::too_many_lines)]
@ -302,15 +306,89 @@ struct ScannedBatch {
end: usize, end: usize,
} }
/// The layout a segment file's leading bytes identify.
enum SegmentLayout {
/// Versioned segment (m11p4+): the 8-byte `TSEG` header, then batch frames.
Versioned,
/// Legacy headerless segment (pre-m11p4): batch frames from offset 0.
/// Also the classification of an EMPTY file (a created-but-never-written
/// segment is valid and holds zero batches).
Legacy,
/// A leading region too short to identify (fewer bytes than a magic, or a
/// partial segment header): the signature of a crash before the first
/// write completed. No readable batches; recovery repairs it (truncates
/// to empty) iff it is the final segment, exactly like any torn tail.
TornPrefix,
}
/// Identify a segment file's layout from its leading bytes.
///
/// # Errors
///
/// Returns [`WalError::SegmentFormatUnknown`] when the leading bytes are
/// complete enough to identify but match no format this binary reads: a
/// `TSEG` header with an unknown version (or non-zero reserved bytes), or a
/// full 4-byte leading magic that is neither `TSEG` nor a batch frame. Those
/// files were written by something else — they must surface loudly and must
/// never be "repaired": truncating bytes this binary cannot parse is data
/// destruction, not recovery. (The pre-m11p4 reader did exactly that —
/// foreign segments scanned as empty and a final one was truncated to zero.)
fn identify_segment_layout(path: &Path, data: &[u8]) -> Result<SegmentLayout, WalError> {
use super::segment::{SEGMENT_FORMAT_VERSION, SEGMENT_HEADER_SIZE, SEGMENT_MAGIC};
if data.is_empty() {
return Ok(SegmentLayout::Legacy);
}
if data.len() < SEGMENT_MAGIC.len() {
return Ok(SegmentLayout::TornPrefix);
}
if data[..SEGMENT_MAGIC.len()] == SEGMENT_MAGIC {
if data.len() < SEGMENT_HEADER_SIZE {
return Ok(SegmentLayout::TornPrefix);
}
let version = data[4];
if version != SEGMENT_FORMAT_VERSION {
return Err(WalError::SegmentFormatUnknown {
path: path.display().to_string(),
detail: format!(
"segment header version {version} (this binary reads \
version {SEGMENT_FORMAT_VERSION})"
),
});
}
// The header is written as one 8-byte unit, so a torn write leaves a
// short prefix (handled above), never valid magic+version with
// garbage reserved bytes — non-zero here means a future writer
// started using the reserved space, which this binary cannot
// interpret. Refuse rather than silently drop whatever they mean.
if data[5..SEGMENT_HEADER_SIZE].iter().any(|&b| b != 0) {
return Err(WalError::SegmentFormatUnknown {
path: path.display().to_string(),
detail: "non-zero reserved bytes in the segment header".into(),
});
}
return Ok(SegmentLayout::Versioned);
}
if data[..MAGIC.len()] == MAGIC {
return Ok(SegmentLayout::Legacy);
}
Err(WalError::SegmentFormatUnknown {
path: path.display().to_string(),
detail: "leading bytes match neither a segment header nor a batch frame".into(),
})
}
/// Decode a single segment file WITHOUT mutating it. /// Decode a single segment file WITHOUT mutating it.
/// ///
/// Reads the file, validates each batch with two-phase checking (magic + /// Identifies the file's layout from its leading bytes (versioned header /
/// bounds, then BLAKE3) for every batch kind (signals and kind-1/2 blobs), /// legacy headerless — see [`identify_segment_layout`]), then validates each
/// and returns every fully valid batch plus the offset of the first byte past /// batch with two-phase checking (magic + bounds, then BLAKE3) for every
/// them. A corrupted or truncated tail stops the scan but is left untouched /// batch kind (signals and kind-1/2 blobs), and returns every fully valid
/// on disk — this function has no side effects and is safe to call from /// batch plus the offset of the first byte past them. A corrupted or
/// read-only paths (export, diagnostics, catch-up read-back) and concurrently /// truncated tail stops the scan but is left untouched on disk — this
/// with the live writer thread. /// function has no side effects and is safe to call from read-only paths
/// (export, diagnostics, catch-up read-back) and concurrently with the live
/// writer thread.
fn scan_segment_inner(path: &Path) -> Result<SegmentScan, WalError> { fn scan_segment_inner(path: &Path) -> Result<SegmentScan, WalError> {
let mut file = File::open(path)?; let mut file = File::open(path)?;
let file_len = file.metadata()?.len(); let file_len = file.metadata()?.len();
@ -319,9 +397,33 @@ fn scan_segment_inner(path: &Path) -> Result<SegmentScan, WalError> {
file.read_to_end(&mut data)?; file.read_to_end(&mut data)?;
drop(file); drop(file);
// Batches begin after the versioned header, at 0 for legacy files, and
// nowhere for a torn prefix (the whole leading region is the crash
// artifact: zero batches, `last_valid_offset = 0`, so the tail-corruption
// flag covers the entire file and recovery's final-segment repair — and
// ONLY that — may truncate it).
let start = match identify_segment_layout(path, &data)? {
SegmentLayout::Versioned => super::segment::SEGMENT_HEADER_SIZE,
SegmentLayout::Legacy => 0,
SegmentLayout::TornPrefix => {
tracing::warn!(
path = %path.display(),
len = data.len(),
"segment leading bytes too short to identify (torn first write)"
);
let total_len = data.len();
return Ok(SegmentScan {
batches: Vec::new(),
data,
last_valid_offset: 0,
total_len,
});
}
};
let mut batches = Vec::new(); let mut batches = Vec::new();
let mut offset: usize = 0; let mut offset: usize = start;
let mut last_valid_offset: usize = 0; let mut last_valid_offset: usize = start;
while offset < data.len() { while offset < data.len() {
// Phase 1: Can we read a header? // Phase 1: Can we read a header?
@ -839,6 +941,228 @@ mod tests {
} }
} }
// ── Segment format versioning (m11p4) ──────────────────────────────────
/// A segment created by `SegmentWriter` carries the versioned header and
/// round-trips through recovery exactly like a legacy one.
#[test]
fn recover_versioned_segment_round_trip() {
use super::super::segment::{SEGMENT_HEADER_SIZE, SEGMENT_MAGIC, SegmentWriter};
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let batch = encode_batch(&[sample_event(1, 1000), sample_event(2, 2000)], 1, 1000)
.expect("encode should succeed");
let mut writer = SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1 << 20)
.expect("open should succeed");
writer.write_batch_bytes(&batch).expect("write");
writer.sync().expect("sync");
// The on-disk file leads with the segment header, then the batch.
let seg_path = dir
.path()
.join(super::super::segment::segment_filename(ShardId::SINGLE, 1));
let bytes = fs::read(&seg_path).expect("read");
assert_eq!(&bytes[..4], &SEGMENT_MAGIC, "file must lead with TSEG");
assert_eq!(bytes.len(), SEGMENT_HEADER_SIZE + batch.len());
let result = recover(dir.path()).expect("recover should succeed");
assert_eq!(result.events.len(), 2);
assert_eq!(result.next_seq, 3);
}
/// A header-only segment (created, never written) recovers clean: zero
/// batches, no corruption, no truncation.
#[test]
fn recover_header_only_segment_clean() {
use super::super::segment::{SEGMENT_HEADER_SIZE, SegmentWriter};
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let _ = SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1 << 20)
.expect("open should succeed");
let result = recover(dir.path()).expect("recover should succeed");
assert!(result.events.is_empty());
assert_eq!(result.next_seq, 1);
let seg_path = dir
.path()
.join(super::super::segment::segment_filename(ShardId::SINGLE, 1));
assert_eq!(
fs::metadata(&seg_path).expect("metadata").len(),
SEGMENT_HEADER_SIZE as u64,
"a header-only segment must not be truncated"
);
}
/// A segment whose header carries a version this binary does not read is
/// `SegmentFormatUnknown` — surfaced loudly and NEVER truncated, even as
/// the final segment (it was written by a different/newer tidalDB; its
/// bytes are opaque, not torn).
#[test]
fn unknown_segment_version_is_format_unknown_and_untouched() {
use super::super::segment::{SEGMENT_MAGIC, segment_filename};
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let mut data = Vec::new();
data.extend_from_slice(&SEGMENT_MAGIC);
data.push(99); // a future segment format version
data.extend_from_slice(&[0, 0, 0]);
data.extend_from_slice(&[0xAA; 64]); // opaque future-format body
let seg_path = dir.path().join(segment_filename(ShardId::SINGLE, 1));
fs::write(&seg_path, &data).expect("write should succeed");
match recover(dir.path()) {
Err(WalError::SegmentFormatUnknown { detail, .. }) => {
assert!(
detail.contains("version 99"),
"detail must name the unknown version: {detail}"
);
}
other => panic!(
"expected SegmentFormatUnknown, got: {:?}",
other.map(|r| r.events.len())
),
}
assert_eq!(
fs::metadata(&seg_path).expect("metadata").len(),
data.len() as u64,
"a foreign-format segment must never be truncated"
);
}
/// A non-empty segment whose leading bytes match neither magic is
/// `SegmentFormatUnknown`. The pre-m11p4 reader scanned it as empty and —
/// if it was the final segment — TRUNCATED it to zero; pin the new
/// no-destruction behavior.
#[test]
fn unknown_leading_magic_is_format_unknown_not_truncated() {
use super::super::segment::segment_filename;
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
// An old/foreign format: parseable filename, unrecognizable content.
let data = vec![0x7Fu8; 256];
let seg_path = dir.path().join(segment_filename(ShardId::SINGLE, 1));
fs::write(&seg_path, &data).expect("write should succeed");
assert!(
matches!(
recover(dir.path()),
Err(WalError::SegmentFormatUnknown { .. })
),
"unrecognizable leading bytes must be SegmentFormatUnknown"
);
assert_eq!(
fs::metadata(&seg_path).expect("metadata").len(),
data.len() as u64,
"an unknown-format segment must never be truncated (the old \
behavior destroyed it as a 'torn tail')"
);
}
/// A versioned final segment with a torn batch tail is repaired to the
/// header boundary — not to zero — so the file stays a valid versioned
/// segment after recovery.
#[test]
fn versioned_torn_tail_truncates_to_header_not_zero() {
use super::super::segment::{SEGMENT_HEADER_SIZE, SegmentWriter, segment_filename};
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
{
let mut writer = SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1 << 20)
.expect("open should succeed");
// A torn batch: garbage shorter than a frame header.
writer
.write_batch_bytes(&[0xDE, 0xAD, 0xBE, 0xEF])
.expect("write");
}
let result = recover(dir.path()).expect("recover should succeed");
assert!(result.events.is_empty());
let seg_path = dir.path().join(segment_filename(ShardId::SINGLE, 1));
assert_eq!(
fs::metadata(&seg_path).expect("metadata").len(),
SEGMENT_HEADER_SIZE as u64,
"torn-tail repair must preserve the segment header"
);
}
/// A torn write of the header itself (final segment) repairs to empty —
/// the legitimate crash-mid-first-write artifact.
#[test]
fn torn_partial_header_in_final_segment_repairs_to_empty() {
use super::super::segment::{SEGMENT_MAGIC, segment_filename};
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let seg_path = dir.path().join(segment_filename(ShardId::SINGLE, 1));
// 6 bytes of the 8-byte header survived the crash.
let mut torn = Vec::new();
torn.extend_from_slice(&SEGMENT_MAGIC);
torn.extend_from_slice(&[1, 0]);
fs::write(&seg_path, &torn).expect("write should succeed");
let result = recover(dir.path()).expect("recover should succeed");
assert!(result.events.is_empty());
assert_eq!(result.next_seq, 1);
assert_eq!(
fs::metadata(&seg_path).expect("metadata").len(),
0,
"a torn header in the final segment is a crash artifact: repaired to empty"
);
}
/// Legacy (headerless) and versioned segments recover together, with the
/// cross-segment continuity check spanning both layouts — the exact
/// mixed-version state of a data dir that lived through the m11p4 upgrade.
#[test]
fn mixed_legacy_and_versioned_segments_recover_with_continuity() {
use super::super::segment::{SegmentWriter, segment_filename};
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
// Segment 1: legacy headerless (written by a pre-m11p4 binary).
let batch1 = encode_batch(&[sample_event(1, 1000)], 1, 1000).expect("encode");
fs::write(
dir.path().join(segment_filename(ShardId::SINGLE, 1)),
&batch1,
)
.expect("write should succeed");
// Segment 2: versioned, continuing the sequence space at 2.
let batch2 = encode_batch(&[sample_event(2, 2000)], 2, 2000).expect("encode");
let mut writer = SegmentWriter::open(dir.path(), ShardId::SINGLE, 2, 1 << 20)
.expect("open should succeed");
writer.write_batch_bytes(&batch2).expect("write");
writer.sync().expect("sync");
let result = recover(dir.path()).expect("recover should succeed");
assert_eq!(result.events.len(), 2);
assert_eq!(result.next_seq, 3);
}
/// `scan_segment_raw` on a versioned segment returns pure batch bytes —
/// the segment header never leaks into the catch-up stream.
#[test]
fn scan_segment_raw_excludes_segment_header() {
use super::super::segment::{SegmentWriter, segment_filename};
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let batch = encode_batch(&[sample_event(7, 7000)], 7, 7000).expect("encode");
let mut writer = SegmentWriter::open(dir.path(), ShardId::SINGLE, 7, 1 << 20)
.expect("open should succeed");
writer.write_batch_bytes(&batch).expect("write");
writer.sync().expect("sync");
let seg_path = dir.path().join(segment_filename(ShardId::SINGLE, 7));
let raw = scan_segment_raw(&seg_path).expect("scan should succeed");
assert_eq!(raw.len(), 1);
assert_eq!(raw[0].first_seq, 7);
assert_eq!(raw[0].last_seq, 7);
assert_eq!(
raw[0].bytes, batch,
"raw read-back must be byte-identical to the encoded batch"
);
}
#[test] #[test]
fn recover_two_contiguous_segments_with_torn_final_tail_succeeds() { fn recover_two_contiguous_segments_with_torn_final_tail_succeeds() {
// The complement of the C17 case: when the torn tail is in the FINAL // The complement of the C17 case: when the torn tail is in the FINAL

149
tidal/src/wal/segment.rs
View File

@ -1,3 +1,47 @@
//! WAL segment files: naming, the on-disk segment format, and the writer.
//!
//! # On-disk segment format
//!
//! A WAL segment file is an 8-byte **segment header** followed by zero or
//! more batch frames (see [`crate::wal::format::batch`] for the 64-byte
//! batch-frame layout — magic `TIDL`, per-batch format version, BLAKE3):
//!
//! ```text
//! Bytes 0-3: SEGMENT_MAGIC b"TSEG"
//! Byte 4: segment format version (u8) — currently 1
//! Bytes 5-7: reserved, MUST be zero
//! Bytes 8-..: batch frames, tightly packed
//! ```
//!
//! ## Legacy (headerless) segments
//!
//! Every segment written before the header existed (m0–m11p3) starts
//! directly at a batch frame (leading bytes = the batch magic `TIDL` in LE
//! order). Readers accept that layout as the implicit "version 0" format —
//! existing data files stay readable with no migration.
//!
//! ## Why the header exists (m11p4)
//!
//! Before it, segment files were unversioned: a reader meeting a segment
//! written by an incompatible tidalDB version found nothing it could parse
//! and silently treated the file as absent/empty (`segments=0` at recovery)
//! — and recovery's torn-tail repair could then TRUNCATE the foreign file to
//! zero. After a rolling upgrade that left a follower behind, the leader
//! could not serve catch-up from segments it could not read, and the
//! follower stayed degraded with nothing in the logs (the 2026-06-11 p3
//! rollout incident). With the header, an unreadable segment surfaces as
//! [`WalError::SegmentFormatUnknown`] — loud, at open time, with a remedy —
//! and is never repaired, truncated, or skipped.
//!
//! ## Downgrade hazard
//!
//! A pre-m11p4 binary reading a header-bearing segment stops at the unknown
//! leading bytes and, if it is the final segment, truncates it as a "torn
//! tail". Downgrading a node across the header boundary therefore requires
//! reseeding its WAL (same operational practice as the pre-header upgrade
//! path). This is inherent: old readers cannot be taught new formats
//! retroactively — which is exactly why the version byte exists from now on.
use std::{ use std::{
fs::{self, File, OpenOptions}, fs::{self, File, OpenOptions},
io::Write, io::Write,
@ -7,6 +51,46 @@ use std::{
use super::error::WalError; use super::error::WalError;
use crate::replication::ShardId; use crate::replication::ShardId;
/// Magic bytes opening every versioned WAL segment file.
///
/// Deliberately distinct from the batch-frame magic (`TIDL`, see
/// [`crate::wal::format::batch::MAGIC`]) so the first four bytes of a file
/// identify its layout: `TSEG` → versioned segment (header then frames),
/// batch magic → legacy headerless segment, anything else → not a format
/// this binary knows ([`WalError::SegmentFormatUnknown`]).
pub const SEGMENT_MAGIC: [u8; 4] = *b"TSEG";
/// Segment format version 1: the 8-byte header followed by batch frames.
pub const SEGMENT_FORMAT_V1: u8 = 1;
/// The segment format version this binary writes.
pub const SEGMENT_FORMAT_VERSION: u8 = SEGMENT_FORMAT_V1;
/// Size of the segment header in bytes (magic + version + reserved).
pub const SEGMENT_HEADER_SIZE: usize = 8;
/// Whether a directory entry claims to be a WAL segment file (a `.seg`
/// extension, case-insensitive — `wal-…001.SEG` is a near-miss worth
/// flagging, not a file to silently skip).
fn is_seg_file(name: &str) -> bool {
Path::new(name)
.extension()
.is_some_and(|ext| ext.eq_ignore_ascii_case("seg"))
}
/// Encode the 8-byte segment header this binary writes.
#[must_use]
pub const fn segment_header() -> [u8; SEGMENT_HEADER_SIZE] {
let mut header = [0u8; SEGMENT_HEADER_SIZE];
header[0] = SEGMENT_MAGIC[0];
header[1] = SEGMENT_MAGIC[1];
header[2] = SEGMENT_MAGIC[2];
header[3] = SEGMENT_MAGIC[3];
header[4] = SEGMENT_FORMAT_VERSION;
// Bytes 5-7 reserved (zero).
header
}
/// Format a segment file name from the shard ID and first sequence number. /// Format a segment file name from the shard ID and first sequence number.
/// ///
/// For `ShardId::SINGLE` (shard 0), produces the v1 format: /// For `ShardId::SINGLE` (shard 0), produces the v1 format:
@ -66,9 +150,17 @@ pub fn parse_segment_seq(filename: &str) -> Option<u64> {
/// List all WAL segment files in the directory, sorted by first sequence number. /// List all WAL segment files in the directory, sorted by first sequence number.
/// ///
/// Non-`.seg` files (checkpoint, session journal) are ignored. A file that
/// DOES carry the `.seg` suffix but whose name matches neither segment
/// naming format is refused loudly: silently skipping it would make WAL data
/// written under an unknown naming scheme (a different tidalDB version, a
/// botched restore) look absent — recovery would report `segments=0` and
/// boot a node that cannot serve catch-up from data sitting right there.
///
/// # Errors /// # Errors
/// ///
/// Returns `WalError::Io` on filesystem failure. /// Returns `WalError::Io` on filesystem failure, or
/// [`WalError::SegmentFormatUnknown`] for an unparseable `.seg` filename.
pub fn list_segments(dir: &Path) -> Result<Vec<(u64, PathBuf)>, WalError> { pub fn list_segments(dir: &Path) -> Result<Vec<(u64, PathBuf)>, WalError> {
let mut segments = Vec::new(); let mut segments = Vec::new();
@ -86,6 +178,11 @@ pub fn list_segments(dir: &Path) -> Result<Vec<(u64, PathBuf)>, WalError> {
}; };
if let Some(seq) = parse_segment_seq(name) { if let Some(seq) = parse_segment_seq(name) {
segments.push((seq, entry.path())); segments.push((seq, entry.path()));
} else if is_seg_file(name) {
return Err(WalError::SegmentFormatUnknown {
path: entry.path().display().to_string(),
detail: "`.seg` filename matches no known WAL segment naming format".into(),
});
} }
} }
@ -119,10 +216,20 @@ pub fn list_segments_for_shard(
let Some(name) = file_name.to_str() else { let Some(name) = file_name.to_str() else {
continue; continue;
}; };
if let Some((seg_shard, seq)) = parse_segment_filename(name) match parse_segment_filename(name) {
&& seg_shard == shard_id Some((seg_shard, seq)) if seg_shard == shard_id => {
{ segments.push((seq, entry.path()));
segments.push((seq, entry.path())); }
None if is_seg_file(name) => {
// Same loud refusal as `list_segments`: an unparseable `.seg`
// name must not silently read as "no data for this shard".
return Err(WalError::SegmentFormatUnknown {
path: entry.path().display().to_string(),
detail: "`.seg` filename matches no known WAL segment naming format".into(),
});
}
// Another shard's segment (filtered) or a non-segment file (ignored).
Some(_) | None => {}
} }
} }
@ -156,6 +263,11 @@ impl SegmentWriter {
/// If `first_seq` identifies an existing segment, it is opened for append. /// If `first_seq` identifies an existing segment, it is opened for append.
/// Otherwise, a new file is created. /// Otherwise, a new file is created.
/// ///
/// An **empty** file (new, or repaired-to-empty by recovery) receives the
/// versioned [`segment_header`] before any batch; a non-empty existing
/// file is appended to in whatever layout it already carries (a legacy
/// headerless segment keeps growing as legacy — readers accept both).
///
/// # Errors /// # Errors
/// ///
/// Returns `WalError::Io` on filesystem failure. /// Returns `WalError::Io` on filesystem failure.
@ -182,14 +294,29 @@ impl SegmentWriter {
let metadata = file.metadata()?; let metadata = file.metadata()?;
let current_size = metadata.len(); let current_size = metadata.len();
Ok(Self { let mut writer = Self {
dir: dir.to_path_buf(), dir: dir.to_path_buf(),
file, file,
current_size, current_size,
max_size, max_size,
first_seq, first_seq,
shard_id, shard_id,
}) };
// Header durability rides on the first batch's sync: a crash before
// that leaves a torn header in the FINAL segment, which recovery
// repairs to empty — and this path then rewrites it on reopen.
if writer.current_size == 0 {
writer.write_header()?;
}
Ok(writer)
}
/// Write the versioned segment header to the (empty) current file.
fn write_header(&mut self) -> Result<(), WalError> {
debug_assert_eq!(self.current_size, 0, "header only opens an empty segment");
self.file.write_all(&segment_header())?;
self.current_size = SEGMENT_HEADER_SIZE as u64;
Ok(())
} }
/// Write a raw batch of bytes to the current segment. /// Write a raw batch of bytes to the current segment.
@ -286,8 +413,14 @@ impl SegmentWriter {
super::sync_dir_durable(&self.dir)?; super::sync_dir_durable(&self.dir)?;
self.file = file; self.file = file;
self.current_size = 0; // Rotation may land on a pre-existing file (e.g. a crash between the
// file create and the seq advance left one behind): only an EMPTY
// target gets the versioned header, mirroring `open`.
self.current_size = self.file.metadata()?.len();
self.first_seq = new_first_seq; self.first_seq = new_first_seq;
if self.current_size == 0 {
self.write_header()?;
}
Ok(()) Ok(())
} }
} }

80
tidal/src/wal/segment_tests.rs
View File

@ -35,13 +35,14 @@ fn write_and_check_size() {
let dir = tempfile::tempdir().expect("tempdir creation should succeed"); let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let mut writer = let mut writer =
SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1024).expect("open should succeed"); SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1024).expect("open should succeed");
assert_eq!(writer.current_size(), 0); // A fresh segment starts at the versioned header's size (m11p4).
assert_eq!(writer.current_size(), SEGMENT_HEADER_SIZE as u64);
let data = [0xABu8; 100]; let data = [0xABu8; 100];
writer writer
.write_batch_bytes(&data) .write_batch_bytes(&data)
.expect("write should succeed"); .expect("write should succeed");
assert_eq!(writer.current_size(), 100); assert_eq!(writer.current_size(), SEGMENT_HEADER_SIZE as u64 + 100);
} }
#[test] #[test]
@ -55,7 +56,8 @@ fn rotation_creates_new_file() {
.expect("write should succeed"); .expect("write should succeed");
writer.rotate(100).expect("rotate should succeed"); writer.rotate(100).expect("rotate should succeed");
assert_eq!(writer.current_size(), 0); // The rotated-to segment starts at its own versioned header.
assert_eq!(writer.current_size(), SEGMENT_HEADER_SIZE as u64);
assert_eq!(writer.first_seq(), 100); assert_eq!(writer.first_seq(), 100);
let segments = list_segments(dir.path()).expect("list should succeed"); let segments = list_segments(dir.path()).expect("list should succeed");
@ -277,6 +279,78 @@ fn rotation_preserves_shard_id_in_filename() {
assert_eq!(writer.shard_id(), ShardId(3)); assert_eq!(writer.shard_id(), ShardId(3));
} }
// ── Segment format versioning (m11p4) ──────────────────────────────────────
#[test]
fn new_segment_leads_with_versioned_header() {
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let _ = SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1024).expect("open");
let bytes = fs::read(dir.path().join(segment_filename(ShardId::SINGLE, 1))).expect("read");
assert_eq!(
bytes,
segment_header(),
"fresh segment = exactly the header"
);
assert_eq!(&bytes[..4], b"TSEG");
assert_eq!(bytes[4], SEGMENT_FORMAT_VERSION);
assert_eq!(&bytes[5..8], &[0, 0, 0], "reserved bytes must be zero");
}
#[test]
fn reopening_existing_segment_writes_no_second_header() {
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
{
let mut writer =
SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1024).expect("open should succeed");
writer.write_batch_bytes(&[0xAB; 10]).expect("write");
}
let writer =
SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1024).expect("reopen should succeed");
assert_eq!(
writer.current_size(),
SEGMENT_HEADER_SIZE as u64 + 10,
"reopen must append, not re-write the header"
);
}
#[test]
fn reopening_legacy_headerless_segment_stays_legacy() {
// A pre-m11p4 file (batch bytes at offset 0): appends must NOT inject a
// header mid-file — the segment keeps its legacy layout.
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let path = dir.path().join(segment_filename(ShardId::SINGLE, 1));
fs::write(&path, [0xCD; 32]).expect("write should succeed");
let writer =
SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1024).expect("open should succeed");
assert_eq!(writer.current_size(), 32);
let bytes = fs::read(&path).expect("read");
assert_eq!(&bytes[..4], &[0xCD; 4], "legacy leading bytes untouched");
}
#[test]
fn unparseable_seg_filename_is_format_unknown() {
let dir = tempfile::tempdir().expect("tempdir creation should succeed");
let _ = SegmentWriter::open(dir.path(), ShardId::SINGLE, 1, 1024).expect("open");
fs::write(dir.path().join("wal-mystery.seg"), [0u8; 8]).expect("write should succeed");
assert!(
matches!(
list_segments(dir.path()),
Err(WalError::SegmentFormatUnknown { .. })
),
"an unparseable .seg filename must refuse loudly, not read as absent"
);
assert!(
matches!(
list_segments_for_shard(dir.path(), ShardId::SINGLE),
Err(WalError::SegmentFormatUnknown { .. })
),
"the per-shard listing must refuse identically"
);
}
mod proptests { mod proptests {
use proptest::prelude::*; use proptest::prelude::*;