diff --git a/CHANGELOG.md b/CHANGELOG.md index c2daed10..ef119bdd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — `segment_plane_scan` missed v0.6.0's nested-Command plane framing, risking WS/MQ/temporal data loss on upgrade (task #69) + +`WalWriterV3::recycle_aggressive`/`recycle_segments_before` gate deletion of +sealed WAL v3 segments on `segment_plane_scan`'s `blocks_recycle` verdict, +which — until this fix — classified a segment purely from its OUTER +record-type byte. v0.6.0's shard event-loop `wal_append` drain hardcoded +`WalRecordType::Command` as the outer wrapper for EVERY plane record +(`WorkspaceCreate`/`Drop`, `MqCreate`/`Push`/`Pop`/`Ack`/`Trigger`/`Drop`, +`TemporalUpsert`, `GraphTemporal`) — verified against the v0.6.0 tag's +`event_loop.rs` (lines 1452/2070). So on a v0.6.0→v0.7.0 upgrade, every +plane record in a pre-upgrade segment was invisible to the scan: it saw only +`Command` at the outer level and classified the whole segment as +recyclable pure-KV history. WS/MQ/temporal-upsert have no snapshot or +checkpoint format in ANY mode — their replay is WAL-only (task #43) — so +recycling one of these segments permanently destroyed that history, with no +way to recover it. + +`segment_plane_scan` now peeks a `Command` record's payload for a nested v3 +record frame (length-prefix + CRC32C validated, matching +`read_wal_v3_record`'s own framing) and classifies the INNER type against +the same blocking set, mirroring the unwrap +`shared_databases.rs::replay_workspace_wal`/`replay_mq_wal`/ +`replay_temporal_wal` already perform via `read_wal_v3_record(&record.payload)` +at replay time — scan and replay now agree on what counts as a nested +record. Fails closed: a structurally valid nested frame with an +unrecognized inner type byte (a future plane type this build predates) +still blocks recycling. An ordinary (non-nested) `Command` payload — the +overwhelmingly common case — is unaffected and stays recyclable. + +Byte-slice peek only, no allocation and no LZ4 decompression (`Command` +payloads are never LZ4-compressed by `write_wal_v3_record`) — off the hot +path (recycle passes only) but kept allocation-free per repo convention. + ### Fixed — `crash_recovery_disk_offload_no_aof` harness assumed eviction-throughput durability the write path no longer provides (task #44) `cold_keys_recover_after_crash_without_aof` forced LRU eviction with a diff --git a/src/persistence/wal_v3/segment.rs b/src/persistence/wal_v3/segment.rs index e609b89e..2795fc7f 100644 --- a/src/persistence/wal_v3/segment.rs +++ b/src/persistence/wal_v3/segment.rs @@ -156,9 +156,109 @@ struct SegmentPlaneScan { has_graph_temporal: bool, } +/// True for the plane record types whose ONLY durable copy is the WAL +/// itself (see [`segment_plane_scan`]'s doc-comment for why). Shared +/// between the outer-type match arm and the v0.6.0 nested-Command unwrap +/// arm so the two can never drift apart. +#[inline] +fn plane_type_blocks_recycle(t: WalRecordType) -> bool { + matches!( + t, + WalRecordType::TemporalUpsert + | WalRecordType::WorkspaceCreate + | WalRecordType::WorkspaceDrop + | WalRecordType::MqCreate + | WalRecordType::MqAck + | WalRecordType::MqPush + | WalRecordType::MqPop + | WalRecordType::MqTrigger + | WalRecordType::MqDrop + ) +} + +/// Peeks a `Command` record's raw payload bytes for a nested WAL v3 record +/// frame, without allocating or decompressing. +/// +/// **Why this exists (v0.6.0 upgrade-boundary contract, task #69):** the +/// v0.6.0 shard event-loop's `wal_append` drain hardcoded +/// `WalRecordType::Command` as the OUTER wrapper for EVERY plane record — +/// `TemporalUpsert`, `WorkspaceCreate`/`Drop`, and every `Mq*` variant — +/// verified against the v0.6.0 tag's `event_loop.rs` (lines 1452/2070). A +/// v0.6.0-written segment therefore has no plane records at the outer-type +/// level at all; they are all nested one level down, inside a `Command` +/// payload. `shared_databases.rs`'s replay path +/// (`replay_workspace_wal` / `replay_mq_wal` / `replay_temporal_wal`) +/// already unwraps this shape via `read_wal_v3_record(&record.payload)` on +/// every `Command` record it sees — this function mirrors that EXACT +/// byte layout (length prefix + 12-byte header + CRC32C trailer, see +/// `record.rs`'s module doc) so the content scan and replay never +/// disagree about what counts as a nested record. Post-v0.6.0 builds +/// pre-frame plane records the same way at the point of emission (see +/// `mq_exec.rs`, `command::temporal::apply_invalidate`), so this shape is +/// not purely historical — it is the SAME nested framing new builds still +/// produce for those planes; only bare (non-nested) records are the +/// newer, separate on-disk arrangement `segment_plane_scan`'s outer match +/// already covered. +/// +/// Returns: +/// - `Some(Some(record_type))` — `payload` contains a structurally valid +/// nested frame (length bounds AND CRC32C both check out) whose inner +/// type byte decodes to a known [`WalRecordType`]. +/// - `Some(None)` — `payload` contains a structurally valid nested frame, +/// but the inner type byte does not decode (a future plane type this +/// build predates). Callers must fail closed on this case. +/// - `None` — `payload` does not contain a validly-framed nested record at +/// all (the CRC32C did not match, or the length was out of bounds) — +/// the overwhelmingly common case: an ordinary opaque RESP command +/// payload. `write_wal_v3_record` never LZ4-compresses `Command` +/// payloads (only `FullPageImage` is compression-eligible), so no +/// decompression step is needed here — this is a pure byte-slice peek, +/// allocation-free like the rest of this scan. +fn peek_nested_plane_record_type(payload: &[u8]) -> Option> { + // Minimum inner framing: len(4) + header(12) + crc(4) = 20, matching + // `record::MIN_RECORD_SIZE` (not exported — duplicated as a literal + // here, same as this file's own outer-record framing check below). + const MIN_INNER_RECORD_SIZE: usize = 20; + if payload.len() < MIN_INNER_RECORD_SIZE { + return None; + } + let inner_len = u32::from_le_bytes([payload[0], payload[1], payload[2], payload[3]]) as usize; + if inner_len < MIN_INNER_RECORD_SIZE || inner_len > payload.len() { + return None; + } + // CRC32C covers bytes [4..inner_len-4] of the nested frame — identical + // span to `read_wal_v3_record`'s `crc32c::crc32c(&data[4..record_len - + // 4])` check in `record.rs`. + let crc_stored = u32::from_le_bytes([ + payload[inner_len - 4], + payload[inner_len - 3], + payload[inner_len - 2], + payload[inner_len - 1], + ]); + let crc_computed = crc32c::crc32c(&payload[4..inner_len - 4]); + if crc_stored != crc_computed { + return None; // not a nested frame — plain opaque Command payload + } + Some(WalRecordType::from_u8(payload[12])) +} + /// Single-pass content scan backing both the recycle-gating boolean /// (`blocks_recycle`) and the K2 `GraphTemporal` observability /// counter, so recyclers never need to read a sealed segment file twice. +/// +/// **v0.6.0 wire-history note (task #69):** every plane record a v0.6.0 +/// binary ever wrote is nested inside a `WalRecordType::Command` outer +/// record (see [`peek_nested_plane_record_type`]'s doc-comment). Without +/// unwrapping that shape here, a post-upgrade recycle pass sees only +/// `Command` at the outer level and misclassifies a pre-upgrade segment as +/// pure-KV history — deleting the sole durable copy of that segment's +/// WS/MQ/temporal records (no snapshot format exists for those planes, in +/// ANY mode). The `Some(WalRecordType::Command)` arm below exists solely +/// to close that upgrade-boundary gap; it must stay byte-layout-identical +/// to the unwrap `shared_databases.rs::replay_workspace_wal` / +/// `replay_mq_wal` / `replay_temporal_wal` perform via +/// `read_wal_v3_record(&record.payload)`, or scan and replay could +/// disagree about which segments are safe to delete. fn segment_plane_scan(path: &Path) -> SegmentPlaneScan { use super::record::WalRecordType; @@ -195,17 +295,25 @@ fn segment_plane_scan(path: &Path) -> SegmentPlaneScan { return BLOCKED; // torn/malformed sealed tail — fail closed } match WalRecordType::from_u8(data[offset + 12]) { - Some( - WalRecordType::TemporalUpsert - | WalRecordType::WorkspaceCreate - | WalRecordType::WorkspaceDrop - | WalRecordType::MqCreate - | WalRecordType::MqAck - | WalRecordType::MqPush - | WalRecordType::MqPop - | WalRecordType::MqTrigger - | WalRecordType::MqDrop, - ) => return BLOCKED, + Some(WalRecordType::Command) => { + // v0.6.0 upgrade-boundary unwrap — see doc-comment above. + let payload = &data[offset + 16..offset + record_len - 4]; + match peek_nested_plane_record_type(payload) { + Some(Some(inner_type)) if plane_type_blocks_recycle(inner_type) => { + return BLOCKED; + } + Some(Some(WalRecordType::GraphTemporal)) => has_graph_temporal = true, + Some(Some(_)) => {} + // Structurally valid nested frame, unrecognized inner + // type — fail closed (a future plane type this build + // predates might have no snapshot format either). + Some(None) => return BLOCKED, + // No valid nested frame — an ordinary opaque RESP + // command payload, never blocking. + None => {} + } + } + Some(t) if plane_type_blocks_recycle(t) => return BLOCKED, Some(WalRecordType::GraphTemporal) => has_graph_temporal = true, Some(_) => {} None => return BLOCKED, // unknown (future) type — fail closed @@ -1674,4 +1782,143 @@ mod tests { let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap(); assert_eq!(writer.append(WalRecordType::Command, b"x"), 1); } + + // ── task #69 (P1, v0.7.0 upgrade-path data loss): v0.6.0's event-loop + // `wal_append` drain hardcoded `WalRecordType::Command` as the OUTER + // wrapper for every plane record (WS/MQ/Temporal/GraphTemporal) — see + // v0.6.0 tag `event_loop.rs` lines 1452/2070. `segment_plane_scan` must + // peek inside a Command payload for that nested v3 frame shape, exactly + // like `shared_databases.rs::replay_workspace_wal` / + // `replay_mq_wal` / `replay_temporal_wal` already do via + // `read_wal_v3_record(&record.payload)` — otherwise a post-upgrade + // recycle pass misclassifies a pre-upgrade segment as pure-KV and + // deletes the sole durable copy of that plane's history. + + /// Builds the exact byte shape a v0.6.0 binary wrote for every plane + /// record: an inner v3 frame (`inner_type`/`inner_payload`) serialized + /// via `write_wal_v3_record`, with THAT whole frame used as the payload + /// of an outer `WalRecordType::Command` record. + fn v060_nested_command_payload(inner_type: WalRecordType, inner_payload: &[u8]) -> Vec { + let mut nested = Vec::new(); + write_wal_v3_record(&mut nested, 1, inner_type, inner_payload); + nested + } + + #[test] + fn test_segment_plane_scan_blocks_v060_nested_workspace_create() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap(); + writer.set_wal_bounds(0, u64::MAX); + + let nested = + v060_nested_command_payload(WalRecordType::WorkspaceCreate, b"ws-plane-payload"); + writer.append(WalRecordType::Command, &nested); + writer.flush_sync().unwrap(); + + let seg_path = WalSegment::segment_path(&wal_dir, 1); + let scan = segment_plane_scan(&seg_path); + assert!( + scan.blocks_recycle, + "a v0.6.0-style Command-nested WorkspaceCreate record must block \ + recycle — it is the sole durable copy of that plane's history" + ); + } + + #[test] + fn test_segment_plane_scan_flags_v060_nested_graph_temporal() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap(); + writer.set_wal_bounds(0, u64::MAX); + + let nested = + v060_nested_command_payload(WalRecordType::GraphTemporal, b"graph-temporal-payload"); + writer.append(WalRecordType::Command, &nested); + writer.flush_sync().unwrap(); + + let seg_path = WalSegment::segment_path(&wal_dir, 1); + let scan = segment_plane_scan(&seg_path); + assert!( + scan.has_graph_temporal, + "a v0.6.0-style Command-nested GraphTemporal record must be \ + observable via has_graph_temporal, same as a direct-typed one" + ); + assert!( + !scan.blocks_recycle, + "GraphTemporal alone must never block recycle by itself — that \ + is graph_floor_lsn's job (K2), not the content scan's" + ); + } + + /// Fail-closed contract (function doc-comment): a nested frame whose + /// length+CRC32C validate but whose inner type byte is unrecognized (a + /// future plane type this build predates) must still block recycle — + /// never silently treated as an opaque non-nested Command payload. + #[test] + fn test_segment_plane_scan_blocks_nested_frame_with_unknown_inner_type() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap(); + writer.set_wal_bounds(0, u64::MAX); + + // Hand-build a validly-framed (length + CRC32C both correct) inner + // record with a type byte no `WalRecordType` variant claims. + let inner_payload = b"future-plane-payload"; + let mut nested = Vec::new(); + let record_len = (20 + inner_payload.len()) as u32; + nested.extend_from_slice(&record_len.to_le_bytes()); + let crc_start = nested.len(); + nested.extend_from_slice(&1u64.to_le_bytes()); // lsn + nested.push(0x99); // unrecognized record-type byte + nested.push(0u8); // flags + nested.extend_from_slice(&[0u8; 2]); // padding + nested.extend_from_slice(inner_payload); + let crc = crc32c::crc32c(&nested[crc_start..]); + nested.extend_from_slice(&crc.to_le_bytes()); + assert!( + WalRecordType::from_u8(0x99).is_none(), + "0x99 must stay an unassigned discriminant for this test to be meaningful" + ); + + writer.append(WalRecordType::Command, &nested); + writer.flush_sync().unwrap(); + + let seg_path = WalSegment::segment_path(&wal_dir, 1); + let scan = segment_plane_scan(&seg_path); + assert!( + scan.blocks_recycle, + "a structurally valid nested frame with an unrecognized inner \ + type byte must fail closed and block recycle" + ); + } + + /// Pins the entire point of the content scan: an ordinary (non-nested) + /// `Command` payload — the common case, a plain RESP-encoded write — + /// must NOT block recycling. Without this the fix above could + /// over-trigger on any payload and defeat recycling altogether. + #[test] + fn test_segment_plane_scan_plain_command_payload_not_blocked() { + let tmp = tempfile::tempdir().unwrap(); + let wal_dir = tmp.path().join("wal"); + let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap(); + writer.set_wal_bounds(0, u64::MAX); + + writer.append( + WalRecordType::Command, + b"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$3\r\nval\r\n", + ); + writer.flush_sync().unwrap(); + + let seg_path = WalSegment::segment_path(&wal_dir, 1); + let scan = segment_plane_scan(&seg_path); + assert!( + !scan.blocks_recycle, + "an ordinary non-nested Command (RESP) payload must stay recyclable" + ); + assert!( + !scan.has_graph_temporal, + "an ordinary non-nested Command payload must never set has_graph_temporal" + ); + } }