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"); + }) +}