From cd8d8c63d3be4f0494ce0336ffbffe3e1a9cf17b Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 18:19:56 +0700 Subject: [PATCH 1/2] tests: add ignored reproduction for lossy duplicate-key secondary index persistence Problem ------- A plain bulk insert of 10_000 rows spread over 97 distinct values of a non-unique secondary index (~103 duplicates per key) is fully visible through select_by_* while the table is in memory, but after wait_for_ops and a reopen via load() roughly 12% of the entries are no longer reachable through the index. select_all still returns every row, so the data pages are intact; only the persisted index is incomplete. The loss is baked into the index file at write time: re-reading the same files gives the same wrong counts, while separate write runs lose different amounts. An extra wait_for_ops plus a sleep before shutdown does not help, so this is not the known wait_for_ops race. Unique-valued secondary indexes survive the identical round trip intact. The existing assertions that would have caught this are commented out in tests/persistence/sync/string_secondary_index.rs (lines 449-452). Reproduction ------------ The test is #[ignore]d so CI stays green until the fix lands. Run with: cargo test --test mod test_duplicate_key_secondary_index_survives_reload -- --ignored Current failure on master: secondary index lost 968 of 10000 entries across persist+reload --- .../persistence/duplicate_key_index_reload.rs | 101 ++++++++++++++++++ tests/persistence/mod.rs | 1 + 2 files changed, 102 insertions(+) create mode 100644 tests/persistence/duplicate_key_index_reload.rs diff --git a/tests/persistence/duplicate_key_index_reload.rs b/tests/persistence/duplicate_key_index_reload.rs new file mode 100644 index 0000000..15bb78e --- /dev/null +++ b/tests/persistence/duplicate_key_index_reload.rs @@ -0,0 +1,101 @@ +use std::time::Duration; + +use tokio::time::timeout; + +use crate::remove_dir_if_exists; +use worktable::prelude::PersistedWorkTable; +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: DuplicateKeyReload, + persist: true, + columns: { + id: u64 primary_key autoincrement, + score: u64, + label: String, + }, + indexes: { + score_idx: score, + }, +); + +/// Reproduction for lossy persistence of duplicate-key secondary indexes. +/// +/// A plain bulk insert of 10_000 rows spread over 97 distinct `score` values +/// (~103 duplicates per key) is fully visible through `select_by_score` while +/// the table is in memory, but after wait_for_ops + reopen a chunk of the +/// entries (~12% as of 0.9.0) is no longer reachable through the index, even +/// though `select_all` still returns every row. The loss is baked into the +/// index file at write time: re-reading the same files gives the same wrong +/// counts. Unique-valued secondary indexes survive the same round trip intact. +#[test] +#[ignore = "exposes lossy persistence of duplicate-key secondary indexes (~12% of entries unreachable after reload)"] +fn test_duplicate_key_secondary_index_survives_reload() { + const ROWS: u64 = 10_000; + const KEYS: u64 = 97; + + let config = DiskConfig::new_with_table_name( + "tests/data/duplicate_key_index_reload/persisted", + DuplicateKeyReloadWorkTable::name_snake_case(), + DuplicateKeyReloadWorkTable::version(), + ); + + 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("tests/data/duplicate_key_index_reload/persisted".to_string()).await; + + { + let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); + let table = DuplicateKeyReloadWorkTable::load(engine).await.unwrap(); + + for i in 0..ROWS { + table + .insert(DuplicateKeyReloadRow { + id: i, + score: i % KEYS, + label: format!("row-{i}-{}", "x".repeat((i % 50) as usize)), + }) + .unwrap(); + } + + // 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}" + ); + } + + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence stalled on bulk insert"); + } + { + 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 must stay reachable through the secondary index. + let mut reachable = 0u64; + for s in 0..KEYS { + reachable += table.select_by_score(s).execute().unwrap().len() as u64; + } + assert_eq!( + reachable, ROWS, + "secondary index lost {} of {ROWS} entries across persist+reload", + ROWS - reachable + ); + } + }) +} diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index bdcfa80..67e293b 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -4,6 +4,7 @@ use worktable::worktable; mod bulk_load_stall; mod concurrent; +mod duplicate_key_index_reload; mod failure; mod index_page; mod read; From 0b97128ef605ce67fd20acf86c7cf8d86c52e276 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 19:08:19 +0700 Subject: [PATCH 2/2] fix: reconstruct duplicate-key secondary indexes correctly on reload Problem ------- Index pages store entries in event-arrival order: only bootstrap-written pages happen to be sorted with the node maximum last, while pages that were incrementally updated through CDC events (any bulk load) are arbitrary. from_persisted assumed "last entry == node maximum" for non-unique indexes: it popped the last entry of every page and re-appended it with discriminator u64::MAX - 1 as the node id. For CDC-updated pages that promoted an arbitrary entry to node maximum, so the reconstructed in-memory node index registered wrong maxima and every entry sorting above one became unreachable through select_by_* (~12% of entries in the regression workload; ~62% under 0.9.0-beta), even though select_all still returned every row. The wrong in-memory maximum also desynced reloaded tables from the on-disk table of contents, which addresses nodes by their maximum: CDC events emitted after a reload could reference node ids the TOC never had. Fix --- Reconstruction now: - sorts every page's entries (a no-op for bootstrap-written pages), - forces the page's persisted node_id entry to sort last within its key, so the reconstructed node maximum is exactly the entry the TOC knows the node by, - attaches nodes in ascending order of their node id, and - hands out discriminators that keep growing across node boundaries within one key, so a key's duplicates straddling nodes keep the B-tree ordering invariant (every entry <= its node's registered maximum). Discriminator 0 stays reserved for the range infimum and u64::MAX for the supremum. Unique indexes get the page sort as well: their reconstruction assumed sorted pages the same way, it just never had duplicate keys to lose. The reload fix is retroactive for 0.9.0-written files: the entries were always all present in the files (verified by parsing pages directly), only reconstruction lost them, so existing tables read back complete without a rewrite. Files written by 0.9.0-beta had additional write-side damage and recover only partially. The repro test is un-ignored, asserts exact per-key counts after reload, and passes 5/5 full-suite runs (317 passed / 4 ignored). --- codegen/src/persist_index/generator.rs | 187 ++++++++++-------- .../persistence/duplicate_key_index_reload.rs | 39 ++-- 2 files changed, 127 insertions(+), 99 deletions(-) diff --git a/codegen/src/persist_index/generator.rs b/codegen/src/persist_index/generator.rs index 7c7a864..e8419f8 100644 --- a/codegen/src/persist_index/generator.rs +++ b/codegen/src/persist_index/generator.rs @@ -356,24 +356,43 @@ impl Generator { .get(i) .expect("should be available as constructed from same values"); - if is_unsized(&ty.to_string()) { - let node = if is_unique { - quote! { - let node = page - .inner - .get_node() - .into_iter() - .map(|p| IndexPair { - key: p.key, - value: p.value.into(), - }) - .collect(); - let node = UnsizedNode::from_inner(node, #const_name); - #i.attach_node(node); - } - } else { - quote! { - let mut inner: Vec<_> = page + // Index pages carry entries in event-arrival order, not sorted + // order: only bootstrap-written pages happen to be sorted with + // the node maximum last, while pages that were incrementally + // updated through CDC events are arbitrary. Reconstruction + // therefore must sort every page and derive the node maximum + // from the sorted entries; assuming "last entry == node id" + // registers a wrong maximum in the in-memory node index and + // makes every entry above it unreachable. + let sort_page_entries = quote! { + inner.sort_by(|a, b| a.key.cmp(&b.key).then_with(|| a.value.cmp(&b.value))); + }; + // For non-unique indexes one key's duplicates can straddle + // nodes, and the B-tree invariant (every entry in a node + // compares <= that node's registered maximum) only survives + // reconstruction if a later node's segment of the same key + // compares greater than the previous node's maximum. So nodes + // are attached in ascending order of their true maximum and + // discriminators keep growing across node boundaries within + // one key. Discriminator 0 is reserved for the range infimum + // and u64::MAX for the supremum, so numbering starts at 1 and + // is capped below the supremum. + let multi_reconstruct = |attach: TokenStream| { + quote! { + let mut nodes: Vec<(IndexPair<#ty, OffsetEqLink>, Vec>)> = Vec::new(); + for page in persisted.#i.1 { + // The persisted node id is the node maximum the + // on-disk table of contents knows this node by. + // The reconstructed node must end with exactly + // this entry: 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. + let node_id = IndexPair { + key: page.inner.node_id.key.clone(), + value: OffsetEqLink(page.inner.node_id.link), + }; + let mut inner: Vec> = page .inner .get_node() .into_iter() @@ -382,41 +401,75 @@ impl Generator { value: OffsetEqLink(p.value), }) .collect(); - - let node_id = inner.pop(); - - let mut sorted: Vec<_> = inner.into_iter() - .map(|p| IndexMultiPair { - key: p.key, - value: p.value, - discriminator: 0, + 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)) }) - .collect(); - sorted.sort(); - - let mut current_discriminator = 1u64; - for entry in sorted.iter_mut() { - entry.discriminator = current_discriminator.min(u64::MAX - 1); - current_discriminator += 1; + }); + if !inner.is_empty() { + nodes.push((node_id, inner)); } - if let Some(node_id) = node_id { - sorted.push(IndexMultiPair { key: node_id.key, value: node_id.value, discriminator: u64::MAX - 1 }); + } + nodes.sort_by(|(a, _), (b, _)| a.key.cmp(&b.key).then_with(|| a.value.cmp(&b.value))); + let mut prev_key = None; + let mut next_discriminator = 1u64; + for (_, inner) in nodes { + 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); } + #attach + } + } + }; + if is_unsized(&ty.to_string()) { + if is_unique { + quote! { + let #i: #t<_, OffsetEqLink, UnsizedNode<_>> = #t::with_maximum_node_size(#const_name); + for page in persisted.#i.1 { + let mut inner: Vec> = page + .inner + .get_node() + .into_iter() + .map(|p| IndexPair { + key: p.key, + value: p.value.into(), + }) + .collect(); + #sort_page_entries + let node = UnsizedNode::from_inner(inner, #const_name); + #i.attach_node(node); + } + } + } else { + let attach = quote! { let node = UnsizedNode::from_inner(sorted, #const_name); #i.attach_multi_node(node); + }; + let body = multi_reconstruct(attach); + quote! { + let #i: #t<_, OffsetEqLink, UnsizedNode<_>> = #t::with_maximum_node_size(#const_name); + #body } - }; + } + } else if is_unique { quote! { - let #i: #t<_, OffsetEqLink, UnsizedNode<_>> = #t::with_maximum_node_size(#const_name); + let size = get_index_page_size_from_data_length::<#ty>(#const_name); + let #i: #t<_, OffsetEqLink> = #t::with_maximum_node_size(size); for page in persisted.#i.1 { - #node - } - } - } else { - let node = if is_unique { - quote! { - let node = page + let mut inner: Vec> = page .inner .get_node() .into_iter() @@ -425,49 +478,19 @@ impl Generator { value: p.value.into(), }) .collect(); - #i.attach_node(node); - } - } else { - quote! { - let mut inner: Vec<_> = page - .inner - .get_node() - .into_iter() - .map(|p| IndexPair { - key: p.key, - value: OffsetEqLink(p.value), - }) - .collect(); - - let node_id = inner.pop(); - - let mut sorted: Vec<_> = inner.into_iter() - .map(|p| IndexMultiPair { - key: p.key, - value: p.value, - discriminator: 0, - }) - .collect(); - sorted.sort(); - - let mut current_discriminator = 1u64; - for entry in sorted.iter_mut() { - entry.discriminator = current_discriminator.min(u64::MAX - 1); - current_discriminator += 1; - } - if let Some(node_id) = node_id { - sorted.push(IndexMultiPair { key: node_id.key, value: node_id.value, discriminator: u64::MAX - 1 }); - } - - #i.attach_multi_node(sorted); + #sort_page_entries + #i.attach_node(inner); } + } + } else { + let attach = quote! { + #i.attach_multi_node(sorted); }; + let body = multi_reconstruct(attach); quote! { let size = get_index_page_size_from_data_length::<#ty>(#const_name); let #i: #t<_, OffsetEqLink> = #t::with_maximum_node_size(size); - for page in persisted.#i.1 { - #node - } + #body } } }) diff --git a/tests/persistence/duplicate_key_index_reload.rs b/tests/persistence/duplicate_key_index_reload.rs index 15bb78e..c25ff0c 100644 --- a/tests/persistence/duplicate_key_index_reload.rs +++ b/tests/persistence/duplicate_key_index_reload.rs @@ -20,17 +20,22 @@ worktable!( }, ); -/// Reproduction for lossy persistence of duplicate-key secondary indexes. +/// Regression test for lossy reconstruction of duplicate-key secondary +/// indexes. /// -/// A plain bulk insert of 10_000 rows spread over 97 distinct `score` values -/// (~103 duplicates per key) is fully visible through `select_by_score` while -/// the table is in memory, but after wait_for_ops + reopen a chunk of the -/// entries (~12% as of 0.9.0) is no longer reachable through the index, even -/// though `select_all` still returns every row. The loss is baked into the -/// index file at write time: re-reading the same files gives the same wrong -/// counts. Unique-valued secondary indexes survive the same round trip intact. +/// 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 +/// re-append it with discriminator `u64::MAX - 1`. For pages that were +/// 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. #[test] -#[ignore = "exposes lossy persistence of duplicate-key secondary indexes (~12% of entries unreachable after reload)"] fn test_duplicate_key_secondary_index_survives_reload() { const ROWS: u64 = 10_000; const KEYS: u64 = 97; @@ -86,16 +91,16 @@ fn test_duplicate_key_secondary_index_survives_reload() { // Row data survives the round trip... assert_eq!(table.select_all().execute().unwrap().len() as u64, ROWS); - // ...and every row must stay reachable through the secondary index. - let mut reachable = 0u64; + // ...and every row stays reachable through the secondary index, + // with per-key counts intact. for s in 0..KEYS { - reachable += table.select_by_score(s).execute().unwrap().len() as u64; + 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" + ); } - assert_eq!( - reachable, ROWS, - "secondary index lost {} of {ROWS} entries across persist+reload", - ROWS - reachable - ); } }) }