From 25074aeae2fd95679ffab81b4bb4a06007f3bbe8 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 20:39:21 +0700 Subject: [PATCH] fix: order same-max-key nodes correctly and stop re-sorting pages on reload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Retrospective-review follow-up to the duplicate-key reload fix; two real defects are corrected, and the reconstruction algorithm moves out of the proc macro into a unit-testable runtime helper. Defect 1: same-max-key nodes were ordered by node-id link --------------------------------------------------------- Several nodes can share one maximum key (a key's duplicates crossing leaf boundaries), and the persisted (key, link) node id alone cannot recover their relative order: links are row locations, uncorrelated with the lost discriminators. Ordering such nodes by link could place the mixed boundary page (the one also carrying smaller keys) after a duplicate-only page. That both stranded the smaller keys behind a bigger registered maximum and reset the per-key discriminator counter mid-run, so two nodes could end up with maxima that compare Equal under MultiPair's Ord (same key, same discriminator, different values) — and the outer node index is a map keyed by the maximum, so attaching the second node silently REPLACED the first. Nodes now order by (node id key, first entry key, node id link): in a valid snapshot at most one same-max-key node carries smaller keys and must come first; the rest are mutually order-free. Defect 2: pages were re-sorted, breaking positional CDC events -------------------------------------------------------------- get_node() already returns the page's logical (slot-resolved) entry order — the tree order the node had when persisted. That order is authoritative: CDC events are positional (InsertAt/RemoveAt carry an index into the node) and the on-disk page applies them through the same slot order; on removal of a node's maximum, the page promotes the LOGICAL predecessor to be the new node id. Re-sorting entries within a node (introduced together with the reload fix) desynced both: after a reload, a delete of a node maximum made the in-memory index and the table of contents disagree on the node's identity, and the persistence engine died on "page should be available in table of contents", hanging wait_for_ops. Reconstruction now preserves the get_node() order verbatim (for unique indexes too, restoring their original behavior) and only re-derives the cross-node discriminator ordering. Structure and tests ------------------- - New runtime helper reconstruct_multi_index_nodes (exported via the prelude) holds the whole algorithm plus structural validation of the persisted input: empty pages, a last entry that is not the node id, keys out of logical order, and backwards keys across the node sequence are reported with tracing::error and handled best-effort, keeping files damaged by earlier releases loadable. - Unit tests cover the review counterexample directly: a mixed boundary page ending in key K, a duplicate-only page of K, node-id links deliberately ordered opposite to the logical page order, and equal duplicate counts so maximum collisions are guaranteed under the old numbering; plus a multi-node straddle chain and damaged-input cases. Invariants asserted: exact entry preservation, per-node order kept verbatim, sorted nodes, non-overlapping ranges, maxima unique under Ord, input page order irrelevant. - The integration regression now uses an exact-id-set model (counts can hide one missing link plus one duplicated link), covers an unsized String duplicate index alongside the sized u64 one, asserts the straddling page topology it depends on by parsing the index file, and after the first reload runs a mutation battery (inserts into existing runs, spread deletes plus a whole key removed end to end, updates moving rows between keys) followed by a second reload — this battery is what exposed defect 2. A no-reload control keeps the live CDC path and the reload path distinguishable. A single-key all-duplicates stress test rounds it out. Suite: 319 passed / 0 failed / 4 ignored, 5 consecutive runs; fmt and clippy clean. Cross-version harness: files written by earlier releases still read back complete (8572/8572, 10000/10000). --- Cargo.toml | 1 + codegen/src/persist_index/generator.rs | 113 ++---- src/lib.rs | 2 +- src/persistence/mod.rs | 1 + src/persistence/space/index/mod.rs | 4 +- src/persistence/space/index/reconstruct.rs | 267 +++++++++++++ src/persistence/space/mod.rs | 2 +- .../persistence/duplicate_key_index_reload.rs | 362 ++++++++++++++++-- 8 files changed, 630 insertions(+), 122 deletions(-) create mode 100644 src/persistence/space/index/reconstruct.rs diff --git a/Cargo.toml b/Cargo.toml index 1d57f89..a0e73e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -55,6 +55,7 @@ worktable_codegen = { path = "codegen", version = "=0.9.1" } chrono = "0.4.43" criterion = { version = "0.5", features = ["async_tokio"] } rand = "0.9.1" +tracing-subscriber = "0.3.23" [[bench]] name = "worktable_benchmarks" diff --git a/codegen/src/persist_index/generator.rs b/codegen/src/persist_index/generator.rs index eef1773..84884c3 100644 --- a/codegen/src/persist_index/generator.rs +++ b/codegen/src/persist_index/generator.rs @@ -350,23 +350,25 @@ impl Generator { .get(i) .expect("should be available as constructed from same values"); - // KEY FORMAT ASSUMPTION driving all of the reconstruction - // below: index pages store entries in event-arrival order, - // NOT sorted order. Only bootstrap-written pages happen to be - // sorted with the node maximum last; pages that were - // incrementally updated through CDC events are arbitrary. - // Reconstruction therefore must sort every page itself — - // assuming "last entry == node id" registers a wrong maximum - // in the in-memory node index and makes every entry above it - // unreachable. + // KEY FORMAT FACT driving the reconstruction below: index + // pages store their cells in event-arrival order, but the + // slot table preserves the tree-logical order the in-memory + // node had when persisted, and `get_node()` resolves through + // it. That logical order is authoritative and must be kept + // verbatim: CDC events are positional (`InsertAt`/`RemoveAt` + // carry an index into the node) and the on-disk page applies + // them through the same slot order, so re-sorting entries at + // reconstruction desyncs every later positional event and the + // node-id successor tracking that keeps the table of contents + // addressable. // - // `unique_reconstruct` handles unique indexes: sorting each - // page is all they need, since a page is exactly one node and - // unique keys make the sorted last entry the true maximum. + // `unique_reconstruct` handles unique indexes: each page is + // exactly one node, `get_node()` already yields it in order + // with the true maximum last, so it attaches directly. let unique_reconstruct = |attach: TokenStream| { quote! { for page in persisted.#i.1 { - let mut inner: Vec> = page + let inner: Vec> = page .inner .get_node() .into_iter() @@ -375,60 +377,31 @@ impl Generator { value: p.value.into(), }) .collect(); - inner.sort_by(|a, b| a.key.cmp(&b.key).then_with(|| a.value.cmp(&b.value))); #attach } } }; - // `multi_reconstruct` handles non-unique indexes, where one - // key's duplicates can straddle nodes. In-memory order for - // duplicates comes from per-entry discriminators that were - // never persisted, so reconstruction re-derives them — and it - // must do so consistently across the whole index, not per - // node: the node index (a B-tree) requires every entry in a - // node to compare <= that node's registered maximum, so a - // later node's segment of one key has to compare greater than - // an earlier node's segment of the same key. Processing nodes - // in ascending node-id order while letting discriminators - // keep growing across node boundaries within one key gives - // every duplicate a globally consistent position; numbering - // per node would make straddling segments overlap and leave - // entries unreachable. - // - // Two more constraints: - // - Each node must end with exactly the entry the on-disk - // table of contents knows the node by (its persisted - // node_id): CDC events emitted after the reload address - // nodes by their maximum, and a node whose maximum drifted - // from the TOC key can no longer be resolved by the space - // index. The page sort pins that entry last within its key. - // - Discriminators 0 and u64::MAX are reserved (range infimum - // and supremum used by lookups), which is why numbering - // starts at 1 and is capped at u64::MAX - 1. The cap and - // the saturating_add can only matter for a key with more - // than u64::MAX - 2 duplicates — unreachable in practice - // (the table could not hold the rows) — so they are purely - // defensive. - // - // Pages are sorted and consumed in place, one node at a time, - // so peak memory stays at the parsed pages plus a single - // node — no intermediate copy of the whole index. + // Non-unique indexes additionally need their duplicate + // ordering re-derived (MultiPair discriminators are not + // persisted), which has to be globally consistent across + // nodes. They delegate to the runtime helper + // `reconstruct_multi_index_nodes` (see its doc comment for + // the ordering and discriminator invariants — notably why + // same-max-key nodes must be ordered by their first entry + // key, not by node-id link). The macro only maps pages into + // (node_id, entries) form and attaches the returned nodes, + // so the algorithm stays unit-testable with synthetic pages. + let index_name_literal = Literal::string(i.to_string().as_str()); let multi_reconstruct = |attach: TokenStream| { quote! { - let mut pages = persisted.#i.1; - pages.sort_by(|a, b| { - let a_link: OffsetEqLink = OffsetEqLink(a.inner.node_id.link); - let b_link: OffsetEqLink = OffsetEqLink(b.inner.node_id.link); - a.inner.node_id.key.cmp(&b.inner.node_id.key).then_with(|| a_link.cmp(&b_link)) - }); - let mut prev_key = None; - let mut next_discriminator = 1u64; - for page in pages { + let mut raw_nodes: Vec<(IndexPair<#ty, OffsetEqLink>, Vec>)> = + Vec::with_capacity(persisted.#i.1.len()); + for page in persisted.#i.1 { let node_id = IndexPair { key: page.inner.node_id.key.clone(), value: OffsetEqLink(page.inner.node_id.link), }; - let mut inner: Vec> = page + let entries = page .inner .get_node() .into_iter() @@ -437,29 +410,9 @@ impl Generator { value: OffsetEqLink(p.value), }) .collect(); - inner.sort_by(|a, b| { - a.key.cmp(&b.key).then_with(|| { - let a_is_id = a.key == node_id.key && a.value == node_id.value; - let b_is_id = b.key == node_id.key && b.value == node_id.value; - a_is_id.cmp(&b_is_id).then_with(|| a.value.cmp(&b.value)) - }) - }); - if inner.is_empty() { - continue; - } - let mut sorted = Vec::with_capacity(inner.len()); - for p in inner { - if prev_key.as_ref() != Some(&p.key) { - prev_key = Some(p.key.clone()); - next_discriminator = 1; - } - sorted.push(IndexMultiPair { - key: p.key, - value: p.value, - discriminator: next_discriminator.min(u64::MAX - 1), - }); - next_discriminator = next_discriminator.saturating_add(1); - } + raw_nodes.push((node_id, entries)); + } + for sorted in reconstruct_multi_index_nodes(#index_name_literal, raw_nodes) { #attach } } diff --git a/src/lib.rs b/src/lib.rs index aead1ab..4cd58ae 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,7 +36,7 @@ pub mod prelude { InsertOperation, Operation, OperationId, PersistedWorkTable, PersistenceConfig, PersistenceEngine, PersistenceTask, ReadOnlyPersistenceEngine, SpaceData, SpaceDataOps, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, map_index_pages_to_toc_and_general, - map_unsized_index_pages_to_toc_and_general, validate_events, + map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, validate_events, }; pub use crate::primary_key::{PrimaryKeyGenerator, PrimaryKeyGeneratorState, TablePrimaryKey}; pub use crate::table::select::{Order, QueryParams, SelectQueryBuilder, SelectQueryExecutor}; diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs index 04ec3a1..1eed9c0 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -12,6 +12,7 @@ pub use readonly_engine::ReadOnlyPersistenceEngine; pub use space::{ IndexTableOfContents, SpaceData, SpaceDataOps, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general, + reconstruct_multi_index_nodes, }; pub use task::PersistenceTask; diff --git a/src/persistence/space/index/mod.rs b/src/persistence/space/index/mod.rs index ae8dc86..23bb24f 100644 --- a/src/persistence/space/index/mod.rs +++ b/src/persistence/space/index/mod.rs @@ -1,3 +1,4 @@ +mod reconstruct; mod table_of_contents; mod unsized_; mod util; @@ -33,6 +34,7 @@ use crate::persistence::SpaceIndexOps; use crate::persistence::space::{BatchChangeEvent, open_or_create_file}; use crate::prelude::WT_INDEX_EXTENSION; +pub use reconstruct::reconstruct_multi_index_nodes; pub use table_of_contents::IndexTableOfContents; pub use unsized_::SpaceIndexUnsized; pub use util::{map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general}; @@ -393,7 +395,7 @@ where let page_index = self .table_of_contents .get(page_id) - .expect("page should be available in table of contents"); + .unwrap_or_else(|| panic!("page {page_id:?} should be available in table of contents")); let page = pages.get_mut(&page_index); let page_to_update = if let Some(page) = page { page diff --git a/src/persistence/space/index/reconstruct.rs b/src/persistence/space/index/reconstruct.rs new file mode 100644 index 0000000..0104f01 --- /dev/null +++ b/src/persistence/space/index/reconstruct.rs @@ -0,0 +1,267 @@ +//! Reconstruction of non-unique (multi) index nodes from persisted pages. +//! +//! Extracted from the `PersistIndex` derive so the algorithm is a plain +//! generic function that can be unit-tested with synthetic pages; the proc +//! macro only generates type plumbing and node attachment. + +use std::fmt::Debug; + +use indexset::core::multipair::MultiPair; +use indexset::core::pair::Pair; + +/// A persisted page in `(node id, logical entries)` form, ready for +/// [`reconstruct_multi_index_nodes`]. +pub type PersistedMultiNode = (Pair, Vec>); + +/// Rebuilds the in-memory nodes of a non-unique secondary index from its +/// persisted pages. +/// +/// Input contract: `entries` for every page must be the page's LOGICAL entry +/// sequence as returned by `get_node()` — index pages store cells in +/// event-arrival order, but the slot table preserves the tree order the +/// in-memory node had when it was persisted, and `get_node()` resolves +/// through it. That order is authoritative and is preserved verbatim here: +/// +/// - CDC events are positional (`InsertAt`/`RemoveAt` carry an index into +/// the node), and the on-disk page applies them through the same slot +/// order. A reconstruction that re-sorts entries within a node desyncs +/// every later positional event. +/// - When a node's maximum is removed, the page promotes its logical +/// predecessor to be the new node id. The reconstructed in-memory node +/// must agree on that predecessor, or the next event addresses the node +/// by a maximum the table of contents never heard of. +/// +/// What reconstruction must still re-derive is the duplicate ordering +/// *between* nodes: `MultiPair` discriminators are not persisted, and the +/// outer node index (a map keyed by each node's maximum) requires every +/// entry to compare `<=` its node's registered maximum, with maxima that +/// are unique under `Ord` — attaching a node whose maximum compares `Equal` +/// to another's *replaces* that node wholesale. Hence: +/// +/// - Nodes are processed in ascending `(node id key, first entry key, +/// node id link)` order. The first-entry-key component is load-bearing: +/// several nodes can share one maximum key (a key's duplicates crossing +/// leaf boundaries), and the persisted `(key, link)` id alone cannot +/// recover their relative order — links are row locations, uncorrelated +/// with the lost discriminators. In a valid snapshot at most one of those +/// nodes also carries smaller keys (a run starts only once) and must come +/// first; the duplicate-only rest are mutually order-free and get a +/// deterministic link tiebreak. +/// - Discriminators grow across node boundaries within one key, so +/// straddling segments stay disjoint and node maxima stay unique. +/// Numbering per node instead would hand two same-max-key nodes an Equal +/// maximum and one of them would silently vanish on attach. +/// - Discriminators 0 and `u64::MAX` are reserved (range infimum/supremum +/// used by lookups), so numbering starts at 1 and is capped at +/// `u64::MAX - 1`. The cap and the `saturating_add` only matter for a key +/// with more than `u64::MAX - 2` duplicates — unreachable in practice — +/// so they are purely defensive. +/// +/// Structural violations in the persisted input (empty pages, a last entry +/// that is not the node id, keys out of order within a page, keys going +/// backwards across the node sequence) are reported with `tracing::error!` +/// and handled best-effort instead of aborting the load: files damaged by +/// earlier releases must stay loadable, and `select_all` (which reads data +/// pages, not this index) remains complete either way. +pub fn reconstruct_multi_index_nodes( + index_name: &str, + pages: Vec>, +) -> Vec>> +where + K: Ord + Clone + Debug, + V: Ord + PartialEq + Clone + Debug, +{ + let mut prepared = Vec::with_capacity(pages.len()); + for (node_id, entries) in pages { + let Some(last) = entries.last() else { + tracing::error!( + "index {index_name}: persisted page for node id {node_id:?} has no entries but the table of contents still references it; skipping the page" + ); + continue; + }; + if last.key != node_id.key || last.value != node_id.value { + tracing::error!( + "index {index_name}: page for node id {node_id:?} ends with {last:?} instead of its node id; the reconstructed maximum will not match the table of contents and later persistence events for this node may not resolve" + ); + } + if entries.windows(2).any(|w| w[0].key > w[1].key) { + tracing::error!( + "index {index_name}: page for node id {node_id:?} has keys out of logical order; the slot table is damaged and lookups for the affected keys may be incomplete" + ); + } + prepared.push((node_id, entries)); + } + + prepared.sort_by(|(a_id, a_entries), (b_id, b_entries)| { + a_id.key + .cmp(&b_id.key) + .then_with(|| a_entries[0].key.cmp(&b_entries[0].key)) + .then_with(|| a_id.value.cmp(&b_id.value)) + }); + + let mut nodes = Vec::with_capacity(prepared.len()); + let mut prev_key: Option = None; + let mut next_discriminator = 1u64; + for (node_id, entries) in prepared { + let mut node = Vec::with_capacity(entries.len()); + for p in entries { + let same_key = prev_key.as_ref() == Some(&p.key); + if !same_key { + if let Some(prev) = &prev_key + && *prev > p.key + { + // Impossible for a valid snapshot given the node ordering + // above; a damaged file can produce it. The counter reset + // below can then collide node maxima, so surface it. + tracing::error!( + "index {index_name}: entry keys go backwards across the reconstructed node sequence ({prev:?} -> {:?} in the node for id {node_id:?}); the persisted index is structurally damaged and lookups for the affected keys may be incomplete", + p.key, + ); + } + prev_key = Some(p.key.clone()); + next_discriminator = 1; + } + node.push(MultiPair { + key: p.key, + value: p.value, + discriminator: next_discriminator.min(u64::MAX - 1), + }); + next_discriminator = next_discriminator.saturating_add(1); + } + nodes.push(node); + } + nodes +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pair(key: u64, value: u64) -> Pair { + Pair { key, value } + } + + fn flatten(nodes: &[Vec>]) -> Vec<(u64, u64)> { + nodes.iter().flatten().map(|p| (p.key, p.value)).collect() + } + + fn assert_valid_tree(nodes: &[Vec>]) { + for node in nodes { + assert!(!node.is_empty(), "reconstruction must not attach empty nodes"); + for w in node.windows(2) { + assert!(w[0] < w[1], "node not internally sorted: {:?} !< {:?}", w[0], w[1]); + } + } + for w in nodes.windows(2) { + let prev_max = w[0].last().unwrap(); + let next_min = w[1].first().unwrap(); + assert!( + prev_max < next_min, + "node ranges overlap: prev max {prev_max:?} !< next min {next_min:?}" + ); + } + // Node maxima must be unique under Ord: the outer index is a map + // keyed by the maximum, and an Equal maximum replaces the node. + for (i, a) in nodes.iter().enumerate() { + for b in nodes.iter().skip(i + 1) { + assert_ne!( + a.last().unwrap().cmp(b.last().unwrap()), + std::cmp::Ordering::Equal, + "two node maxima compare Equal: {:?} vs {:?}", + a.last().unwrap(), + b.last().unwrap() + ); + } + } + } + + /// The review counterexample: a mixed boundary page (tail of key 1, start + /// of key 2's run) whose node-id link sorts AFTER a duplicate-only page + /// of key 2, with equal duplicate counts. Ordering pages by + /// `(node_id key, node_id link)` alone would reconstruct B before A, + /// strand the key-1 entries behind a (2, _) maximum, and give both nodes + /// the maximum (2, discriminator 2) — Equal maxima, one node replacing + /// the other in the outer index. + #[test] + fn mixed_boundary_page_with_adversarial_link_order() { + let page_a = (pair(2, 20), vec![pair(1, 11), pair(2, 21), pair(2, 20)]); + let page_b = (pair(2, 10), vec![pair(2, 12), pair(2, 10)]); + + // Input order must not matter; test both. + for pages in [ + vec![page_a.clone(), page_b.clone()], + vec![page_b.clone(), page_a.clone()], + ] { + let nodes = reconstruct_multi_index_nodes("test", pages); + + assert_eq!(nodes.len(), 2); + assert_valid_tree(&nodes); + + // Every persisted entry survives exactly once. + let mut all = flatten(&nodes); + all.sort_unstable(); + assert_eq!(all, vec![(1, 11), (2, 10), (2, 12), (2, 20), (2, 21)]); + + // The mixed boundary node comes first (it carries key 1)... + assert_eq!(nodes[0].first().unwrap().key, 1); + // ...each node keeps its logical entry order verbatim... + assert_eq!(flatten(&nodes[..1]), vec![(1, 11), (2, 21), (2, 20)]); + assert_eq!(flatten(&nodes[1..]), vec![(2, 12), (2, 10)]); + // ...and each node still ends with its persisted node id entry. + assert_eq!((nodes[0].last().unwrap().key, nodes[0].last().unwrap().value), (2, 20)); + assert_eq!((nodes[1].last().unwrap().key, nodes[1].last().unwrap().value), (2, 10)); + } + } + + /// A straddle chain for one key plus neighbours on both sides: + /// [1,1,2,2] -> [2,2] -> [2,2] -> [2,3,3], with node-id links running + /// against the logical order wherever the format allows it. + #[test] + fn multi_node_straddle_chain() { + let pages = vec![ + (pair(3, 5), vec![pair(2, 70), pair(3, 90), pair(3, 5)]), + (pair(2, 40), vec![pair(2, 41), pair(2, 40)]), + (pair(2, 30), vec![pair(2, 31), pair(2, 30)]), + (pair(2, 60), vec![pair(1, 2), pair(1, 1), pair(2, 61), pair(2, 60)]), + ]; + let expected_len: usize = pages.iter().map(|(_, e)| e.len()).sum(); + + let nodes = reconstruct_multi_index_nodes("test", pages); + + assert_eq!(nodes.len(), 4); + assert_valid_tree(&nodes); + assert_eq!(flatten(&nodes).len(), expected_len); + // The mixed [1,1,2..] node must be first and the [..2,3,3] node last. + assert_eq!(nodes[0].first().unwrap().key, 1); + assert_eq!(nodes[3].last().unwrap().key, 3); + // Within every node the logical entry order is preserved verbatim. + assert_eq!(flatten(&nodes[..1]), vec![(1, 2), (1, 1), (2, 61), (2, 60)]); + // Key 2's discriminators grow across all four nodes. + let discs: Vec = nodes + .iter() + .flatten() + .filter(|p| p.key == 2) + .map(|p| p.discriminator) + .collect(); + for w in discs.windows(2) { + assert!(w[0] < w[1], "key 2 discriminators not strictly increasing: {discs:?}"); + } + } + + /// Damaged input must not panic and must keep every entry somewhere: + /// empty pages, a node id that is not the last entry, keys out of + /// logical order within a page. + #[test] + fn damaged_pages_are_survivable() { + let pages = vec![ + (pair(5, 1), vec![]), + (pair(7, 99), vec![pair(6, 2), pair(7, 1)]), + (pair(9, 50), vec![pair(9, 3), pair(8, 4), pair(9, 50)]), + ]; + + let nodes = reconstruct_multi_index_nodes("test", pages); + + assert_eq!(nodes.len(), 2); + assert_eq!(flatten(&nodes).len(), 5); + } +} diff --git a/src/persistence/space/mod.rs b/src/persistence/space/mod.rs index 680b792..54ef46f 100644 --- a/src/persistence/space/mod.rs +++ b/src/persistence/space/mod.rs @@ -14,7 +14,7 @@ use tokio::fs::{File, OpenOptions}; pub use data::SpaceData; pub use index::{ IndexTableOfContents, SpaceIndex, SpaceIndexUnsized, map_index_pages_to_toc_and_general, - map_unsized_index_pages_to_toc_and_general, + map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, }; pub type BatchData = HashMap)>>; diff --git a/tests/persistence/duplicate_key_index_reload.rs b/tests/persistence/duplicate_key_index_reload.rs index 2e6e868..ee6058d 100644 --- a/tests/persistence/duplicate_key_index_reload.rs +++ b/tests/persistence/duplicate_key_index_reload.rs @@ -1,3 +1,4 @@ +use std::collections::{BTreeMap, BTreeSet}; use std::time::Duration; use tokio::time::timeout; @@ -13,15 +14,144 @@ worktable!( columns: { id: u64 primary_key autoincrement, score: u64, + bucket: String, label: String, }, indexes: { score_idx: score, + bucket_idx: bucket, }, + queries: { + update: { + ScoreById(score) by id, + } + } ); +const ROWS: u64 = 10_000; +const KEYS: u64 = 97; +const BUCKETS: u64 = 23; + +fn bucket_name(i: u64) -> String { + format!("bucket-{:02}", i % BUCKETS) +} + +/// Expected state: score -> id set and bucket -> id set. +#[derive(Default)] +struct Model { + by_score: BTreeMap>, + by_bucket: BTreeMap>, +} + +impl Model { + fn insert(&mut self, id: u64, score: u64, bucket: String) { + self.by_score.entry(score).or_default().insert(id); + self.by_bucket.entry(bucket).or_default().insert(id); + } + + fn remove(&mut self, id: u64) { + self.by_score.retain(|_, ids| { + ids.remove(&id); + !ids.is_empty() + }); + self.by_bucket.retain(|_, ids| { + ids.remove(&id); + !ids.is_empty() + }); + } + + fn move_score(&mut self, id: u64, new_score: u64) { + self.by_score.retain(|_, ids| { + ids.remove(&id); + !ids.is_empty() + }); + self.by_score.entry(new_score).or_default().insert(id); + } + + /// Compares EXACT id sets, not counts: a count can pass while one link is + /// missing and another is duplicated. + fn assert_matches(&self, table: &DuplicateKeyReloadWorkTable, stage: &str) { + for (score, expected_ids) in &self.by_score { + let got: BTreeSet = table + .select_by_score(*score) + .execute() + .unwrap() + .into_iter() + .map(|r| r.id) + .collect(); + assert_eq!(&got, expected_ids, "score {score} id set mismatch at stage: {stage}"); + } + // No phantom rows either. + let live: u64 = self.by_score.values().map(|s| s.len() as u64).sum(); + assert_eq!( + table.select_all().execute().unwrap().len() as u64, + live, + "row count at {stage}" + ); + + for (bucket, expected_ids) in &self.by_bucket { + let got: BTreeSet = table + .select_by_bucket(bucket.clone()) + .execute() + .unwrap() + .into_iter() + .map(|r| r.id) + .collect(); + assert_eq!(&got, expected_ids, "bucket {bucket} id set mismatch at stage: {stage}"); + } + } +} + +/// Asserts the persisted score index actually has the topology the +/// regression depends on: multiple pages, with at least one key's duplicates +/// crossing a page boundary. Without this, a page-capacity or serialization +/// change could shrink the workload into one node and the test would go +/// green without exercising the bug. +async fn assert_straddling_topology(dir: &str) { + use data_bucket::INNER_PAGE_SIZE; + use std::sync::Arc; + use std::sync::atomic::AtomicU32; + use tokio::fs::OpenOptions; + + let path = format!("{dir}/duplicate_key_reload/score_idx.wt.idx"); + let mut file = OpenOptions::new().read(true).write(true).open(&path).await.unwrap(); + let next_id_gen = Arc::new(AtomicU32::new(1)); + let toc = IndexTableOfContents::<(u64, Link), { INNER_PAGE_SIZE as u32 }>::parse_from_file( + &mut file, + 0.into(), + next_id_gen, + ) + .await + .unwrap(); + + let mappings: Vec<_> = toc.iter().map(|(_, page_id)| *page_id).collect(); + assert!( + mappings.len() >= 2, + "score index fits in a single page ({} pages); the workload no longer exercises node straddling", + mappings.len() + ); + + let mut pages_per_key: BTreeMap = BTreeMap::new(); + for page_id in mappings { + let page = parse_page::, { DUPLICATE_KEY_RELOAD_PAGE_SIZE as u32 }>(&mut file, page_id.into()) + .await + .unwrap(); + let keys: BTreeSet = page.inner.index_values[..page.inner.current_length as usize] + .iter() + .map(|v| v.key) + .collect(); + for key in keys { + *pages_per_key.entry(key).or_default() += 1; + } + } + assert!( + pages_per_key.values().any(|&pages| pages >= 2), + "no key's duplicates cross a page boundary; the workload no longer exercises the regression" + ); +} + /// Regression test for lossy reconstruction of duplicate-key secondary -/// indexes. +/// indexes (sized u64 and unsized String), including post-reload mutations. /// /// Index pages store entries in event-arrival order, but `from_persisted` /// used to treat the last entry of every page as the node's maximum and @@ -29,23 +159,31 @@ worktable!( /// incrementally updated through CDC events (any bulk load), that "maximum" /// was an arbitrary entry, so the reconstructed in-memory node index /// registered wrong node maxima and every entry sorting above one became -/// unreachable through `select_by_*` — around 12% of the entries in this -/// workload, although `select_all` still returned every row. Reconstruction -/// now sorts each page, orders nodes by their true maximum, and assigns -/// discriminators that keep growing across node boundaries within one key, -/// which restores the B-tree ordering invariant even when one key's -/// duplicates straddle nodes. +/// unreachable through `select_by_*` — while `select_all` still returned +/// every row. Reconstruction (see `reconstruct_multi_index_nodes`) now sorts +/// each page, pins the persisted node id as each node's maximum, orders +/// same-max-key nodes by their minimum entry key, and assigns discriminators +/// that keep growing across node boundaries within one key. +/// +/// The second phase mutates the reloaded table (inserts into existing +/// duplicate runs, deletes spread across every node including whole-key +/// removal, updates that move rows between keys) and reloads again: those +/// operations are routed through the reconstructed node maxima and the +/// on-disk table of contents, so they prove the reload left persistence in +/// an addressable state, not just a readable one. #[test] fn test_duplicate_key_secondary_index_survives_reload() { - const ROWS: u64 = 10_000; - const KEYS: u64 = 97; - + let dir = "tests/data/duplicate_key_index_reload/persisted"; let config = DiskConfig::new_with_table_name( - "tests/data/duplicate_key_index_reload/persisted", + dir, DuplicateKeyReloadWorkTable::name_snake_case(), DuplicateKeyReloadWorkTable::version(), ); + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::WARN) + .try_init(); + let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(2) .enable_io() @@ -54,53 +192,98 @@ fn test_duplicate_key_secondary_index_survives_reload() { .unwrap(); runtime.block_on(async { - remove_dir_if_exists("tests/data/duplicate_key_index_reload/persisted".to_string()).await; + remove_dir_if_exists(dir.to_string()).await; + let mut model = Model::default(); { let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); let table = DuplicateKeyReloadWorkTable::load(engine).await.unwrap(); for i in 0..ROWS { + let bucket = bucket_name(i); table .insert(DuplicateKeyReloadRow { id: i, score: i % KEYS, + bucket: bucket.clone(), label: format!("row-{i}-{}", "x".repeat((i % 50) as usize)), }) .unwrap(); + model.insert(i, i % KEYS, bucket); } - // The in-memory index is complete before shutdown. - for s in 0..KEYS { - let expected = ROWS / KEYS + u64::from(s < ROWS % KEYS); - assert_eq!( - table.select_by_score(s).execute().unwrap().len() as u64, - expected, - "in-memory index incomplete for key {s}" - ); - } + model.assert_matches(&table, "in-memory before first persist"); timeout(Duration::from_secs(30), table.wait_for_ops()) .await .expect("persistence stalled on bulk insert"); } + + assert_straddling_topology(dir).await; + { let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); let table = DuplicateKeyReloadWorkTable::load(engine).await.unwrap(); - // Row data survives the round trip... - assert_eq!(table.select_all().execute().unwrap().len() as u64, ROWS); - - // ...and every row stays reachable through the secondary index, - // with per-key counts intact. - for s in 0..KEYS { - let expected = ROWS / KEYS + u64::from(s < ROWS % KEYS); - assert_eq!( - table.select_by_score(s).execute().unwrap().len() as u64, - expected, - "secondary index lost entries for key {s} across persist+reload" - ); + model.assert_matches(&table, "after first reload"); + + // Insert new rows into existing duplicate runs. + for j in 0..500u64 { + let id = 20_000 + j; + let bucket = bucket_name(j); + table + .insert(DuplicateKeyReloadRow { + id, + score: j % KEYS, + bucket: bucket.clone(), + label: format!("late-{j}"), + }) + .unwrap(); + model.insert(id, j % KEYS, bucket); } + // Deletes spread over every region of the key space (every 13th + // row hits assorted in-node and boundary positions), plus one + // whole key removed end to end so entire nodes drain. + for id in (0..ROWS).step_by(13) { + table.delete(id).await.unwrap(); + model.remove(id); + } + let score_42_ids: Vec = model + .by_score + .get(&42) + .cloned() + .unwrap_or_default() + .into_iter() + .collect(); + for id in score_42_ids { + if id < ROWS { + table.delete(id).await.unwrap(); + model.remove(id); + } + } + // Updates that move rows between secondary keys. + for id in (0..ROWS).step_by(17) { + if model.by_score.values().any(|ids| ids.contains(&id)) { + let new_score = (id % KEYS) + 1_000; + table + .update_score_by_id(ScoreByIdQuery { score: new_score }, id) + .await + .unwrap(); + model.move_score(id, new_score); + } + } + + model.assert_matches(&table, "in-memory after post-reload mutations"); + + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence stalled on post-reload mutations"); + } + { + let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); + let table = DuplicateKeyReloadWorkTable::load(engine).await.unwrap(); + + model.assert_matches(&table, "after second reload"); } }) } @@ -111,8 +294,6 @@ fn test_duplicate_key_secondary_index_survives_reload() { /// for the discriminator and node-ordering logic in `from_persisted`. #[test] fn test_single_key_all_duplicates_survives_reload() { - const ROWS: u64 = 10_000; - let config = DiskConfig::new_with_table_name( "tests/data/duplicate_key_index_reload/single_key", DuplicateKeyReloadWorkTable::name_snake_case(), @@ -138,6 +319,7 @@ fn test_single_key_all_duplicates_survives_reload() { .insert(DuplicateKeyReloadRow { id: i, score: 42, + bucket: "the-bucket".to_string(), label: format!("row-{i}"), }) .unwrap(); @@ -152,11 +334,17 @@ fn test_single_key_all_duplicates_survives_reload() { let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); let table = DuplicateKeyReloadWorkTable::load(engine).await.unwrap(); - assert_eq!(table.select_all().execute().unwrap().len() as u64, ROWS); + let ids: BTreeSet = table + .select_by_score(42) + .execute() + .unwrap() + .into_iter() + .map(|r| r.id) + .collect(); assert_eq!( - table.select_by_score(42).execute().unwrap().len() as u64, - ROWS, - "single-key secondary index lost entries across persist+reload" + ids, + (0..ROWS).collect::>(), + "single-key secondary index lost or duplicated ids across persist+reload" ); // The reloaded table must also stay writable: post-reload CDC @@ -166,6 +354,7 @@ fn test_single_key_all_duplicates_survives_reload() { .insert(DuplicateKeyReloadRow { id: ROWS, score: 42, + bucket: "the-bucket".to_string(), label: "post-reload".to_string(), }) .unwrap(); @@ -176,3 +365,98 @@ fn test_single_key_all_duplicates_survives_reload() { } }) } + +/// Control for the reload test: the identical mutation battery with NO +/// reload in between. If this stalls too, the failure is in the live CDC +/// space-index path for duplicate keys, not in reconstruction. +#[test] +fn test_duplicate_key_mutations_without_reload() { + let dir = "tests/data/duplicate_key_index_reload/no_reload"; + let config = DiskConfig::new_with_table_name( + dir, + DuplicateKeyReloadWorkTable::name_snake_case(), + DuplicateKeyReloadWorkTable::version(), + ); + + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::WARN) + .try_init(); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_io() + .enable_time() + .build() + .unwrap(); + + runtime.block_on(async { + remove_dir_if_exists(dir.to_string()).await; + + let mut model = Model::default(); + let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); + let table = DuplicateKeyReloadWorkTable::load(engine).await.unwrap(); + + for i in 0..ROWS { + let bucket = bucket_name(i); + table + .insert(DuplicateKeyReloadRow { + id: i, + score: i % KEYS, + bucket: bucket.clone(), + label: format!("row-{i}-{}", "x".repeat((i % 50) as usize)), + }) + .unwrap(); + model.insert(i, i % KEYS, bucket); + } + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence stalled on bulk insert"); + + for j in 0..500u64 { + let id = 20_000 + j; + let bucket = bucket_name(j); + table + .insert(DuplicateKeyReloadRow { + id, + score: j % KEYS, + bucket: bucket.clone(), + label: format!("late-{j}"), + }) + .unwrap(); + model.insert(id, j % KEYS, bucket); + } + for id in (0..ROWS).step_by(13) { + table.delete(id).await.unwrap(); + model.remove(id); + } + let score_42_ids: Vec = model + .by_score + .get(&42) + .cloned() + .unwrap_or_default() + .into_iter() + .collect(); + for id in score_42_ids { + if id < ROWS { + table.delete(id).await.unwrap(); + model.remove(id); + } + } + for id in (0..ROWS).step_by(17) { + if model.by_score.values().any(|ids| ids.contains(&id)) { + let new_score = (id % KEYS) + 1_000; + table + .update_score_by_id(ScoreByIdQuery { score: new_score }, id) + .await + .unwrap(); + model.move_score(id, new_score); + } + } + + model.assert_matches(&table, "in-memory after mutations (no reload)"); + + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence stalled on mutations without any reload"); + }) +}