Skip to content

fix(persistence): replayed DEL/FLUSH tombstone the cold plane; spills no longer suppress AOF recovery #257

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pilotspacex-byte merged 1 commit into from
Jul 10, 2026
Merged

fix(persistence): replayed DEL/FLUSH tombstone the cold plane; spills no longer suppress AOF recovery #257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
wall-clock timeout with zero actual test failures. Bumped to 30m to match
the Windows Test step's headroom; removes recurring false-negative reds on
rebased branches.
### Fixed — Disk-offload: crash recovery no longer resurrects deleted cold keys, nor drops the AOF after a spill (PR #TBD)

- **Deleted/flushed cold keys stayed deleted only until the next crash.** A
DEL/UNLINK/FLUSHALL/FLUSHDB of a spilled key tombstones the in-memory
ColdIndex, but the manifest entry stays `Active` until the orphan sweep. A
kill-9 inside that window lost the tombstone, and boot-time replay could not
re-apply it: the replayed DEL ran against databases whose `cold_index` was
`None` — a silent cold-plane no-op — so the key resurrected via cold
read-through (measured 85–97/200 probes under `--appendonly yes` +
`--disk-offload`). Two detach points fixed: (a) `recover_shard_v3_pitr` now
attaches the Phase-3-rebuilt ColdIndex to the databases BEFORE Phase 4 WAL
replay (was: stashed on `RecoveryResult`, merged only after recovery
returned); (b) the per-shard/multi-part AOF replay now keeps cold wiring
live during incr replay — main.rs re-attaches it after the pre-replay hot
wipe, and `replay_per_shard`/`replay_multi_part` bridge it across
`rdb::load`'s wholesale database swap (which silently dropped it). New
crash suite `tests/crash_recovery_cold_del_resurrection.rs` (DEL + FLUSHALL
scenarios, kill-9 inside the pre-sweep window).
- **Any disk-offload spill silently discarded the AOF on the next restart.**
The "WAL replayed 0 commands → fall back to the AOF" gate counted vector +
file-lifecycle records: after a spill wrote `FileCreate` records into the
shard WAL, the gate saw a "non-empty" WAL, skipped `appendonly.aof` (the
only complete KV history — `--wal-kv-log` is auto-off when the AOF is the
authority), and every KV write since the last snapshot was lost. The gate
now keys on KV `Command` records only.

### Fixed — Conn-plane: monoio central listener joins the SO_REUSEPORT group (PR #250)

Expand Down
47 changes: 32 additions & 15 deletions src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1174,12 +1174,13 @@ fn main() -> anyhow::Result<()> {
// (eviction.rs::next_spill_file_id_seed): the seed keeps recovered cold
// files immutable so these preserved entries stay valid until the
// steady-state cascade refreshes them.
let preserved_cold_wiring: Vec<
type PreservedColdWiring = Vec<
Vec<(
Option<std::path::PathBuf>,
Option<moon::storage::tiered::cold_index::ColdIndex>,
)>,
> = if disk_offload_base.is_some() {
>;
let mut preserved_cold_wiring: PreservedColdWiring = if disk_offload_base.is_some() {
shards
.iter_mut()
.map(|s| {
Expand All @@ -1192,6 +1193,23 @@ fn main() -> anyhow::Result<()> {
} else {
Vec::new()
};
// Consuming re-attach: first call restores the wiring, later calls no-op.
// Called (a) after each replay branch's hot wipe and BEFORE its replay —
// replayed DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane,
// and with `cold_index == None` those paths are silent cold no-ops, so a
// key deleted in the AOF tail resurrects via cold read-through after
// restart — and (b) unconditionally after the whole block for the
// branches that never replay (fresh boot, tokio single-shard).
// `rdb::load`'s wholesale `*live = temp` swap inside the replay is
// bridged separately (take/restore around the load in shard_replay.rs).
fn reattach_cold_wiring(shards: &mut [moon::shard::Shard], wiring: &mut PreservedColdWiring) {
for (shard, dbs) in shards.iter_mut().zip(std::mem::take(wiring)) {
for (db, (cold_shard_dir, cold_index)) in shard.databases.iter_mut().zip(dbs) {
db.cold_shard_dir = cold_shard_dir;
db.cold_index = cold_index;
}
}
}
if config.appendonly == "yes"
&& let Some(ref dir) = persistence_dir
{
Expand Down Expand Up @@ -1220,6 +1238,8 @@ fn main() -> anyhow::Result<()> {
for db in shards[0].databases.iter_mut() {
db.clear();
}
// Cold wiring live during replay — see reattach_cold_wiring.
reattach_cold_wiring(&mut shards, &mut preserved_cold_wiring);
let loaded = moon::persistence::aof_manifest::replay_multi_part(
&mut shards[0].databases,
manifest,
Expand Down Expand Up @@ -1280,6 +1300,8 @@ fn main() -> anyhow::Result<()> {
db.clear();
}
}
// Cold wiring live during replay — see reattach_cold_wiring.
reattach_cold_wiring(&mut shards, &mut preserved_cold_wiring);

// Borrow each shard's `databases` mutably and route through
// `replay_per_shard`. The split_at_mut walk constructs a
Expand Down Expand Up @@ -1470,19 +1492,14 @@ fn main() -> anyhow::Result<()> {
// removed: multi-shard PerShard AOF is now loaded on tokio too, and the
// single-shard tokio warn is emitted inline above.)

// ── B-2: re-attach cold-tier wiring dropped by the rdb::load swap ───────
// Restore the `cold_shard_dir` + rebuilt `cold_index` captured before the
// AOF replay block. Without this, the handler reads a database with
// `cold_shard_dir = None`, so every read-through of a re-evicted key misses
// and disk-offload silently loses data across a restart.
if !preserved_cold_wiring.is_empty() {
for (shard, dbs) in shards.iter_mut().zip(preserved_cold_wiring) {
for (db, (cold_shard_dir, cold_index)) in shard.databases.iter_mut().zip(dbs) {
db.cold_shard_dir = cold_shard_dir;
db.cold_index = cold_index;
}
}
}
// ── B-2: re-attach cold-tier wiring for the branches that never replayed ─
// The replay branches above already consumed the wiring (it must be live
// DURING replay so DEL/FLUSH tombstone the cold plane); this covers fresh
// boots and the tokio single-shard path. Without it, the handler reads a
// database with `cold_shard_dir = None`, so every read-through of a
// re-evicted key misses and disk-offload silently loses data across a
// restart. No-op when already consumed.
reattach_cold_wiring(&mut shards, &mut preserved_cold_wiring);

// Extract databases from all shards and wrap in ShardDatabases
let all_dbs: Vec<Vec<moon::storage::Database>> = shards
Expand Down
53 changes: 49 additions & 4 deletions src/persistence/aof_manifest/shard_replay.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,40 @@

use super::*;

/// Cold-tier wiring (`cold_shard_dir` + `cold_index`) captured per database so
/// it survives the `rdb::load` swap. See [`take_cold_wiring`].
type ColdWiring = Vec<(
Option<std::path::PathBuf>,
Option<crate::storage::tiered::cold_index::ColdIndex>,
)>;

/// Detach cold-tier wiring from every database before an `rdb::load`.
///
/// `rdb::load` loads the base RDB into fresh `Database::new()` temporaries and
/// swaps them wholesale into the live databases (`*live = temp`). That swap
/// silently drops `cold_shard_dir` and the recovery-rebuilt `cold_index` —
/// live-tier topology, not part of the RDB hot snapshot. Capture both here and
/// re-attach with [`restore_cold_wiring`] BEFORE the incr replay: replayed
/// DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane, and with
/// `cold_index == None` those paths (`remove_counting_cold()`/`clear()`) are
/// silent cold no-ops, so a key deleted in the incr tail resurrects via cold
/// read-through after restart (its manifest entry stays Active until the
/// orphan sweep).
fn take_cold_wiring(databases: &mut [crate::storage::Database]) -> ColdWiring {
databases
.iter_mut()
.map(|db| (db.cold_shard_dir.take(), db.cold_index.take()))
.collect()
}

/// Re-attach wiring captured by [`take_cold_wiring`].
fn restore_cold_wiring(databases: &mut [crate::storage::Database], wiring: ColdWiring) {
for (db, (cold_shard_dir, cold_index)) in databases.iter_mut().zip(wiring) {
db.cold_shard_dir = cold_shard_dir;
db.cold_index = cold_index;
}
}

/// Replay multi-part AOF: load base RDB then replay incremental RESP.
///
/// Returns total keys/commands loaded.
Expand All @@ -16,10 +50,15 @@ pub fn replay_multi_part(
) -> Result<usize, crate::error::MoonError> {
let mut total = 0usize;

// Load base RDB
// Load base RDB. Cold wiring must survive the swap AND be live for the
// incr replay below (replayed DEL/FLUSH must tombstone the cold plane) —
// see take_cold_wiring.
let base_path = manifest.base_path();
if base_path.exists() {
match crate::persistence::rdb::load(databases, &base_path) {
let cold_wiring = take_cold_wiring(databases);
let load_result = crate::persistence::rdb::load(databases, &base_path);
restore_cold_wiring(databases, cold_wiring);
match load_result {
Ok(n) => {
info!(
"AOF base RDB loaded: {} keys from {}",
Expand Down Expand Up @@ -439,9 +478,15 @@ pub fn replay_per_shard(
let mut shard_max_lsn: u64 = 0;
let mut shard_ordered: Vec<OrderedEntry> = Vec::new();

// Load this shard's base RDB.
// Load this shard's base RDB. Cold wiring must survive the
// swap AND be live for the incr replay below (replayed
// DEL/FLUSH must tombstone the cold plane) — see
// take_cold_wiring.
if base_path.exists() {
match crate::persistence::rdb::load(*databases, &base_path) {
let cold_wiring = take_cold_wiring(databases);
let load_result = crate::persistence::rdb::load(*databases, &base_path);
restore_cold_wiring(databases, cold_wiring);
match load_result {
Ok(n) => {
info!(
"AOF shard-{} base RDB loaded: {} keys from {}",
Expand Down
107 changes: 89 additions & 18 deletions src/persistence/recovery.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,9 @@ pub struct RecoveryResult {
pub cold_segments: Vec<(u64, std::path::PathBuf)>,
/// Number of cold segments discovered.
pub cold_segments_loaded: usize,
/// Cold index rebuilt from heap DataFiles (None if no KvLeaf entries).
pub cold_index: Option<crate::storage::tiered::cold_index::ColdIndex>,
// NOTE: the ColdIndex rebuilt in Phase 3 is attached directly to
// `databases[0]` BEFORE Phase 4 replay (never returned here) so that
// replayed deletes tombstone the cold plane — see the Phase 3 comment.
}

/// 6-phase recovery protocol for disk-offload mode.
Expand Down Expand Up @@ -329,7 +330,20 @@ pub fn recover_shard_v3_pitr(
shard_id,
cold_idx.len()
);
result.cold_index = Some(cold_idx);
// Attach to the database BEFORE Phase 4 replay. Replayed
// DEL/UNLINK/FLUSH*/EXPIRE-past must tombstone the cold plane:
// with `cold_index == None`, `remove_counting_cold()`/`clear()`
// are silent no-ops (`as_mut()` on None), so a key deleted in
// the WAL tail resurrects via cold read-through after restart
// whenever the crash lands inside the pre-orphan-sweep window
// (the manifest entry is still Active until the sweep).
if let Some(db0) = databases.first_mut() {
db0.cold_shard_dir = Some(shard_dir.to_path_buf());
match db0.cold_index.as_mut() {
Some(existing) => existing.merge(cold_idx),
None => db0.cold_index = Some(cold_idx),
}
}
}
// Crash-orphan sweep: heap files written but never registered in
// the manifest (crash between spill write and manifest commit)
Expand Down Expand Up @@ -369,6 +383,14 @@ pub fn recover_shard_v3_pitr(
}

// ── Phase 4: WAL REPLAY ───────────────────────────────────────────
// KV `Command` records counted separately from vector/file-lifecycle
// records: the Phase 4b "WAL gave us no KV history → fall back to the
// AOF" gate must key on THIS count. Gating on the aggregate
// `commands_replayed` silently skipped the AOF (the only complete KV
// history — `wal_kv_log` is auto-off when the AOF is the authority)
// whenever disk-offload spills had written FileCreate records into the
// WAL: every KV write since the last snapshot was lost on restart.
let mut kv_commands_replayed = 0usize;
let wal_dir = shard_dir.join("wal-v3");
if wal_dir.exists() {
let mut selected_db = 0usize;
Expand All @@ -395,6 +417,7 @@ pub fn recover_shard_v3_pitr(
&mut selected_db,
);
result.commands_replayed += 1;
kv_commands_replayed += 1;
}
}
}
Expand Down Expand Up @@ -584,28 +607,38 @@ pub fn recover_shard_v3_pitr(
}

// ── Phase 4b: legacy-dir AOF / WAL v3 FALLBACK ─────────────────────
// When the disk-offload WAL v3 replay (Phase 4, above) produced 0
// commands and a legacy persistence directory is available, fall back to
// it. `appendonly.aof` is the AUTHORITY there (post-#211, WAL v3's KV
// coverage is intentionally partial: `--wal-kv-log` default-off, and
// connection-local writes bypass it even when on — a non-empty WAL v3
// must never shadow the AOF). Only when NO AOF exists at all does the
// legacy-mode WAL v3 directory (`v2_dir/shard-N/wal-v3/`, distinct from
// `shard_dir`'s Phase-4 WAL despite the shared name) serve as a
// last-resort partial recovery source. The WAL v2 rung (per-shard
// `shard-N.wal`) was removed in the pre-1.0 WAL-v3-only format freeze
// and is never replayed by this build.
if result.commands_replayed == 0 {
// When the disk-offload WAL v3 replay (Phase 4, above) produced 0 KV
// `Command` records and a legacy persistence directory is available, fall
// back to it. The gate is `kv_commands_replayed` — NOT the aggregate
// `commands_replayed`, which also counts vector + file-lifecycle records:
// disk-offload spills write FileCreate records into the same WAL, and
// gating on the aggregate skipped the AOF (the only complete KV history)
// on every post-spill restart. `appendonly.aof` is the AUTHORITY there
// (post-#211, WAL v3's KV coverage is intentionally partial:
// `--wal-kv-log` default-off, and connection-local writes bypass it even
// when on — a non-empty WAL v3 must never shadow the AOF). Only when NO
// AOF exists at all does the legacy-mode WAL v3 directory
// (`v2_dir/shard-N/wal-v3/`, distinct from `shard_dir`'s Phase-4 WAL
// despite the shared name) serve as a last-resort partial recovery
// source. The WAL v2 rung (per-shard `shard-N.wal`) was removed in the
// pre-1.0 WAL-v3-only format freeze and is never replayed by this build.
//
// Known follow-up: when the Phase-4 WAL DID carry some KV records
// (`--wal-kv-log on` / CDC active) alongside an existing AOF, this gate
// keeps the WAL's partial KV view and still skips the AOF (replaying
// both would double-apply non-idempotent commands). Resolving that needs
// an AOF-first redesign of Phase 4, tracked in the roadmap.
if kv_commands_replayed == 0 {
if let Some(v2_dir) = v2_persistence_dir {
let aof_path = v2_dir.join("appendonly.aof");
if aof_path.exists() {
info!(
"Shard {}: WAL empty, falling back to AOF replay from {:?}",
"Shard {}: WAL carried no KV commands, falling back to AOF replay from {:?}",
shard_id, aof_path
);
match crate::persistence::aof::replay_aof(databases, &aof_path, engine) {
Ok(n) => {
result.commands_replayed = n;
result.commands_replayed += n;
info!("Shard {}: AOF fallback replayed {} commands", shard_id, n);
}
Err(e) => {
Expand All @@ -626,7 +659,7 @@ pub fn recover_shard_v3_pitr(
) {
Ok(0) => {}
Ok(n) => {
result.commands_replayed = n;
result.commands_replayed += n;
tracing::warn!(
"Shard {}: no appendonly.aof found — replayed {} records \
from legacy-mode WAL v3 at {:?} as a LAST-RESORT \
Expand Down Expand Up @@ -1236,6 +1269,44 @@ mod tests {
);
}

#[test]
fn test_spill_lifecycle_records_do_not_suppress_aof_fallback() {
// Disk-offload spills write FileCreate/FileDelete/FileTierChange
// records into the shard's own wal-v3 — but NO KV Command records
// (`--wal-kv-log` is auto-off when the AOF is the authority). The
// Phase 4b gate must key on KV Command records only: gating on the
// aggregate count made any post-spill restart skip the AOF entirely,
// losing every KV write since the last snapshot (and resurrecting
// deleted cold keys, since the AOF'd DELs never replayed).
let tmp = tempfile::tempdir().unwrap();
let shard_dir = tmp.path().join("shard-0");
let offload_wal_dir = shard_dir.join("wal-v3");
std::fs::create_dir_all(&offload_wal_dir).unwrap();
let mut wal_data = make_v3_header(0);
// One spill lifecycle record: 16-byte payload (file_id + page_offset).
write_wal_v3_record(&mut wal_data, 1, WalRecordType::FileCreate, &[0u8; 16]);
std::fs::write(offload_wal_dir.join("000000000001.wal"), &wal_data).unwrap();

let v2_dir = tmp.path().join("legacy");
std::fs::create_dir_all(&v2_dir).unwrap();
let aof_payload = b"*1\r\n$4\r\nPING\r\n*1\r\n$4\r\nPING\r\n"; // 2 KV records
std::fs::write(v2_dir.join("appendonly.aof"), aof_payload).unwrap();

let mut databases = vec![Database::new()];
let engine = crate::persistence::replay::DispatchReplayEngine::new();
let result =
recover_shard_v3_with_fallback(&mut databases, 0, &shard_dir, &engine, Some(&v2_dir))
.unwrap();

assert_eq!(
result.commands_replayed, 3,
"1 FileCreate lifecycle record + 2 AOF commands must all be \
replayed -- got {} (1 means the lifecycle record suppressed the \
AOF fallback and every KV write was dropped)",
result.commands_replayed
);
}

#[test]
fn test_legacy_wal_v3_last_resort_when_no_aof() {
let tmp = tempfile::tempdir().unwrap();
Expand Down
Loading
Loading