Skip to content

Commit f729a92

Browse files
committed
fix(pager): refresh the nonce anchor mid-operation instead of per-commit
The main-db nonce counter could only run anchor_budget values ahead of the anchor durable in the A/B header, and only a header write advanced it. Ordinary commits write a header every time, but a compaction rebuild, a rekey re-seal, and an incremental apply all seal many pages before publishing anything, so each drew its entire run from a single window and silently stopped working above a fixed store size. Add a HeaderCursor/anchor-refresh path (src/pager/anchor.rs) that rewrites the live header with only the anchor advanced, publishing nothing else, and wire every writer path (commit, catalog write, apply-journal clear, snapshot apply, rekey) through the pager's shared cursor instead of carrying its own copy of the A/B slot and sequence. Rebuild the dense-repack and rekey paths on top of this: replace the one-shot BTree::bulk_load with a streaming, push-style BulkLoader (src/btree/tree/bulk/) so a rebuild holds one leaf and one internal node per level rather than the whole dataset, and stream the source tree into it in bounded batches with periodic flushes bounded by the buffer-pool budget. Also drop the legacy 13-byte rekey-state row and its conservative same-KEK migration: every durable intent row still on disk is now the fixed-width form, so decode_rekey_state returns a RekeyIntent directly and an off-width row is rejected outright.
1 parent 478d5bf commit f729a92

26 files changed

Lines changed: 1747 additions & 498 deletions

src/btree/tests/basic.rs

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -733,6 +733,99 @@ async fn bulk_load_rejects_non_strict_key_order_without_poisoning_tree() {
733733
}
734734
}
735735

736+
/// The streaming loader carries the strictly-increasing check across `push`
737+
/// calls, so a caller feeding it batch by batch is held to exactly the contract
738+
/// a caller handing over the whole input is held to.
739+
#[tokio::test(flavor = "current_thread")]
740+
async fn bulk_loader_rejects_non_increasing_keys_across_pushes() {
741+
for second_key in [b"a".as_slice(), b"b".as_slice()] {
742+
let pager = fresh_pager().await;
743+
let mut tree = fresh_tree(pager);
744+
{
745+
let mut loader = tree.bulk_loader().unwrap();
746+
loader.push(b"b".to_vec(), b"one".to_vec()).await.unwrap();
747+
let error = loader
748+
.push(second_key.to_vec(), b"two".to_vec())
749+
.await
750+
.expect_err("descending and duplicate keys must both be rejected");
751+
assert!(
752+
matches!(error, PagedbError::BulkLoadNotMonotonic),
753+
"expected BulkLoadNotMonotonic, got {error:?}"
754+
);
755+
}
756+
assert_eq!(tree.root_page_id(), 0);
757+
assert_eq!(tree.next_page_id(), 4);
758+
}
759+
}
760+
761+
/// The record preflight runs before the record allocates anything, so a rejected
762+
/// key leaves the allocator exactly where it was.
763+
#[tokio::test(flavor = "current_thread")]
764+
async fn bulk_loader_rejects_oversized_key_before_allocating() {
765+
let pager = fresh_pager().await;
766+
let mut tree = fresh_tree(pager);
767+
// Fits a leaf but cannot fit an internal separator.
768+
let key_len = body_capacity(PAGE) - 32;
769+
{
770+
let mut loader = tree.bulk_loader().unwrap();
771+
let error = loader
772+
.push(vec![b'a'; key_len], Vec::new())
773+
.await
774+
.expect_err("a key too large to separate must be rejected");
775+
assert!(matches!(error, PagedbError::PayloadTooLarge));
776+
}
777+
assert_eq!(tree.root_page_id(), 0);
778+
assert_eq!(
779+
tree.next_page_id(),
780+
4,
781+
"the preflight must reject before any page is allocated"
782+
);
783+
}
784+
785+
/// A tree built by pushing records through the loader must be a well-formed
786+
/// multi-level tree: every record present, in order, overflow values resolved,
787+
/// and no dangling child or overflow pointer.
788+
#[tokio::test(flavor = "current_thread")]
789+
async fn bulk_loader_builds_a_multi_level_tree_that_scans_in_order() {
790+
let pager = fresh_pager().await;
791+
let mut tree = fresh_tree(pager);
792+
// Inline values wide enough that a few records fill a leaf, so the record
793+
// count below produces enough leaves to need more than one internal level.
794+
let inline = vec![0x2Bu8; 900];
795+
let spilled = vec![0xB4u8; PAGE]; // > PAGE/4 → overflow chain
796+
let n = 1200usize;
797+
798+
{
799+
let mut loader = tree.bulk_loader().unwrap();
800+
for i in 0..n {
801+
let value = if i % 11 == 0 { &spilled } else { &inline };
802+
loader
803+
.push(format!("k-{i:05}").into_bytes(), value.clone())
804+
.await
805+
.unwrap();
806+
}
807+
loader.finish().await.unwrap();
808+
}
809+
810+
assert_ne!(tree.root_page_id(), 0);
811+
assert!(
812+
tree.find_dangling().await.is_none(),
813+
"streamed tree must be structurally intact"
814+
);
815+
816+
let all = tree.collect_all().await.unwrap();
817+
assert_eq!(all.len(), n);
818+
for (i, (key, value)) in all.iter().enumerate() {
819+
assert_eq!(key.as_slice(), format!("k-{i:05}").as_bytes());
820+
let want = if i % 11 == 0 { &spilled } else { &inline };
821+
assert_eq!(value, want, "value mismatch at record {i}");
822+
}
823+
assert_eq!(
824+
tree.get(b"k-00777").await.unwrap().as_deref(),
825+
Some(inline.as_slice())
826+
);
827+
}
828+
736829
#[tokio::test(flavor = "current_thread")]
737830
async fn put_rejects_oversized_key_without_poisoning_tree() {
738831
let pager = fresh_pager().await;

src/btree/tree/bulk.rs

Lines changed: 0 additions & 184 deletions
This file was deleted.

0 commit comments

Comments
 (0)