fix(repo): use mst diff instead of full tree walk for obsolete blocks

Lewis: May this revision serve well! <lu5a@proton.me>

Cargo.lock

@@ -7405,7 +7405,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-api"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "anyhow",
  "axum",
@@ -7456,7 +7456,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-auth"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "anyhow",
  "base32",
@@ -7479,7 +7479,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-cache"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7493,7 +7493,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-comms"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7511,7 +7511,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-config"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "confique",
  "serde",
@@ -7519,7 +7519,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-crypto"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "aes-gcm",
  "base64 0.22.1",
@@ -7535,7 +7535,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-db"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "chrono",
@@ -7552,7 +7552,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-db-traits"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7568,7 +7568,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-infra"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "bytes",
@@ -7579,7 +7579,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-lexicon"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "chrono",
  "hickory-resolver",
@@ -7597,7 +7597,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-oauth"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "anyhow",
  "axum",
@@ -7620,7 +7620,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-oauth-server"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "axum",
  "base64 0.22.1",
@@ -7653,7 +7653,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-pds"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -7745,7 +7745,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-repo"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "bytes",
  "cid",
@@ -7757,7 +7757,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-ripple"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "backon",
@@ -7782,7 +7782,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-scopes"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "axum",
  "futures",
@@ -7798,7 +7798,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-server"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "axum",
  "clap",
@@ -7819,7 +7819,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-signal"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "chrono",
@@ -7842,7 +7842,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-storage"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "aws-config",
@@ -7859,7 +7859,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-store"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "async-trait",
  "bytes",
@@ -7872,6 +7872,7 @@ dependencies = [
  "jacquard-common",
  "jacquard-repo",
  "k256",
+ "libc",
  "lsm-tree",
  "memmap2",
  "multihash",
@@ -7905,7 +7906,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-sync"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "anyhow",
  "axum",
@@ -7927,7 +7928,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-types"
-version = "0.5.2"
+version = "0.5.3"
 dependencies = [
  "chrono",
  "cid",

Cargo.toml

@@ -26,7 +26,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.5.2"
+version = "0.5.3"
 edition = "2024"
 license = "AGPL-3.0-or-later"
 

Dockerfile

@@ -1,4 +1,4 @@
-FROM node:24-alpine AS builder
+FROM node:24-alpine AS frontend
 RUN corepack enable && corepack prepare pnpm@latest --activate
 WORKDIR /app
 COPY frontend/package.json frontend/pnpm-lock.yaml ./

crates/tranquil-pds/src/repo_ops.rs

@@ -13,7 +13,6 @@ use jacquard_repo::mst::util::compute_cid;
 use jacquard_repo::storage::BlockStore;
 use k256::ecdsa::SigningKey;
 use serde_json::{Value, json};
-use std::collections::BTreeSet;
 use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::OwnedMutexGuard;
@@ -226,30 +225,6 @@ pub async fn begin_repo_write(
     Ok((ctx, mst))
 }
 
-pub async fn compute_obsolete_cids(
-    original_mst: &Mst<TrackingBlockStore>,
-    new_mst: &Mst<TrackingBlockStore>,
-    original_root_cid: CommitCid,
-) -> Result<Vec<Cid>, jacquard_repo::error::RepoError> {
-    let (old_nodes, new_nodes, old_leaves, new_leaves) = tokio::try_join!(
-        original_mst.collect_node_cids(),
-        new_mst.collect_node_cids(),
-        original_mst.leaves(),
-        new_mst.leaves(),
-    )?;
-    let old_nodes_set: BTreeSet<Cid> = old_nodes.into_iter().collect();
-    let new_nodes_set: BTreeSet<Cid> = new_nodes.into_iter().collect();
-    let old_leaf_set: BTreeSet<Cid> = old_leaves.iter().map(|(_, cid)| *cid).collect();
-    let new_leaf_set: BTreeSet<Cid> = new_leaves.iter().map(|(_, cid)| *cid).collect();
-    let removed_nodes = old_nodes_set.difference(&new_nodes_set).copied();
-    let removed_leaves = old_leaf_set.difference(&new_leaf_set).copied();
-    let obsolete: BTreeSet<Cid> = std::iter::once(original_root_cid.into_cid())
-        .chain(removed_nodes)
-        .chain(removed_leaves)
-        .collect();
-    Ok(obsolete.into_iter().collect())
-}
-
 pub async fn finalize_repo_write(
     state: &AppState,
     ctx: RepoWriteContext,
@@ -267,42 +242,22 @@ pub async fn finalize_repo_write(
     let original_settled = Mst::load(storage_for_diff.clone(), ctx.prev_data_cid, None);
     let new_settled = Mst::load(storage_for_diff, new_mst_root, None);
 
-    let (obsolete_result, new_tree_result) = tokio::join!(
-        compute_obsolete_cids(
-            &original_settled,
-            &new_settled,
-            CommitCid::from(ctx.current_root_cid),
-        ),
-        async {
-            let (nodes, leaves) =
-                tokio::try_join!(new_settled.collect_node_cids(), new_settled.leaves(),)?;
-            Ok::<Vec<Cid>, jacquard_repo::error::RepoError>(
-                nodes
-                    .into_iter()
-                    .chain(leaves.iter().map(|(_, cid)| *cid))
-                    .collect(),
-            )
-        },
-    );
+    let new_tree_cids: Vec<Cid> = block_bytes.keys().copied().collect();
 
-    let new_tree_cids = match new_tree_result {
-        Ok(cids) => cids,
-        Err(e) => {
-            error!(
-                "new tree walk failed: {e}. \
-                 Falling back to written-block CIDs only; \
-                 shared subtree ownership already tracked by prior commits."
+    let obsolete_cids = match original_settled.diff(&new_settled).await {
+        Ok(diff) => {
+            let mut obsolete: Vec<Cid> = Vec::with_capacity(
+                1 + diff.removed_mst_blocks.len() + diff.removed_cids.len(),
             );
-            block_bytes.keys().copied().collect()
+            obsolete.push(ctx.current_root_cid);
+            obsolete.extend(diff.removed_mst_blocks);
+            obsolete.extend(diff.removed_cids);
+            obsolete
         }
-    };
-
-    let obsolete_cids = match obsolete_result {
-        Ok(cids) => cids,
         Err(e) => {
             error!(
                 "MST diff failed during finalize_repo_write: {e}. \
-                 Proceeding with empty obsolete set; leaked blocks \
+                 Proceeding with commit CID only; leaked blocks \
                  will be reclaimed by reachability GC."
             );
             vec![ctx.current_root_cid]

crates/tranquil-pds/tests/account_lifecycle.rs

@@ -99,9 +99,8 @@ async fn test_check_account_status_returns_correct_block_count() {
         after_delete_blocks
     );
     assert!(
-        after_delete_blocks >= initial_blocks,
-        "Block count after delete should be at least initial count (initial {}, now {})",
-        initial_blocks,
+        after_delete_blocks >= 2,
+        "Block count after delete should have at least commit + MST root (got {})",
         after_delete_blocks
     );
 }

crates/tranquil-pds/tests/mst_diff_equivalence.rs

@@ -0,0 +1,543 @@
+use std::collections::BTreeSet;
+use std::sync::Arc;
+
+use cid::Cid;
+use jacquard_repo::mst::Mst;
+use jacquard_repo::storage::MemoryBlockStore;
+
+fn test_cid(n: u32) -> Cid {
+    let data = n.to_be_bytes();
+    let mut buf = [0u8; 32];
+    buf[..4].copy_from_slice(&data);
+    buf[4] = (n >> 8) as u8 ^ 0xAB;
+    buf[5] = (n & 0xFF) as u8 ^ 0xCD;
+    let mh = multihash::Multihash::wrap(0x12, &buf).unwrap();
+    Cid::new_v1(0x71, mh)
+}
+
+async fn compute_obsolete_full_walk<S: jacquard_repo::storage::BlockStore + Sync + Send + 'static>(
+    old: &Mst<S>,
+    new: &Mst<S>,
+) -> BTreeSet<Cid> {
+    let old_nodes = old.collect_node_cids().await.unwrap();
+    let new_nodes = new.collect_node_cids().await.unwrap();
+    let old_leaves = old.leaves().await.unwrap();
+    let new_leaves = new.leaves().await.unwrap();
+    let old_nodes_set: BTreeSet<Cid> = old_nodes.into_iter().collect();
+    let new_nodes_set: BTreeSet<Cid> = new_nodes.into_iter().collect();
+    let old_leaf_set: BTreeSet<Cid> = old_leaves.iter().map(|(_, cid)| *cid).collect();
+    let new_leaf_set: BTreeSet<Cid> = new_leaves.iter().map(|(_, cid)| *cid).collect();
+    old_nodes_set
+        .difference(&new_nodes_set)
+        .copied()
+        .chain(old_leaf_set.difference(&new_leaf_set).copied())
+        .collect()
+}
+
+fn compute_obsolete_from_diff(
+    diff: &jacquard_repo::mst::diff::MstDiff,
+) -> BTreeSet<Cid> {
+    diff.removed_mst_blocks
+        .iter()
+        .copied()
+        .chain(diff.removed_cids.iter().copied())
+        .collect()
+}
+
+async fn assert_equivalence(
+    old_records: &[(String, u32)],
+    new_records: &[(String, u32)],
+    scenario: &str,
+) {
+    let storage = Arc::new(MemoryBlockStore::new());
+
+    let mut old_tree = Mst::new(storage.clone());
+    for (key, val) in old_records {
+        old_tree = old_tree.add(key, test_cid(*val)).await.unwrap();
+    }
+    let old_root = old_tree.persist().await.unwrap();
+
+    let mut new_tree = Mst::new(storage.clone());
+    for (key, val) in new_records {
+        new_tree = new_tree.add(key, test_cid(*val)).await.unwrap();
+    }
+    let new_root = new_tree.persist().await.unwrap();
+
+    let old_settled = Mst::load(storage.clone(), old_root, None);
+    let new_settled = Mst::load(storage.clone(), new_root, None);
+
+    let full_walk_obsolete = compute_obsolete_full_walk(&old_settled, &new_settled).await;
+
+    let old_for_diff = Mst::load(storage.clone(), old_root, None);
+    let new_for_diff = Mst::load(storage, new_root, None);
+    let diff = old_for_diff.diff(&new_for_diff).await.unwrap();
+    let diff_obsolete = compute_obsolete_from_diff(&diff);
+
+    assert_eq!(
+        full_walk_obsolete, diff_obsolete,
+        "MISMATCH in scenario: {scenario}\n  full_walk count: {}\n  diff count: {}\n  in full_walk but not diff: {:?}\n  in diff but not full_walk: {:?}",
+        full_walk_obsolete.len(),
+        diff_obsolete.len(),
+        full_walk_obsolete.difference(&diff_obsolete).collect::<Vec<_>>(),
+        diff_obsolete.difference(&full_walk_obsolete).collect::<Vec<_>>(),
+    );
+}
+
+fn make_key(collection: &str, i: u32) -> String {
+    format!("{collection}/{i:06}")
+}
+
+fn generate_records(collection: &str, range: std::ops::Range<u32>) -> Vec<(String, u32)> {
+    range.map(|i| (make_key(collection, i), i)).collect()
+}
+
+fn generate_multi_collection_records(
+    collections: &[&str],
+    per_collection: u32,
+) -> Vec<(String, u32)> {
+    collections
+        .iter()
+        .enumerate()
+        .flat_map(|(ci, coll)| {
+            let base = ci as u32 * per_collection;
+            (0..per_collection).map(move |i| (make_key(coll, i), base + i))
+        })
+        .collect()
+}
+
+fn apply_scattered_updates(
+    records: &[(String, u32)],
+    stride: usize,
+    cid_offset: u32,
+) -> Vec<(String, u32)> {
+    records
+        .iter()
+        .enumerate()
+        .map(|(idx, (key, val))| {
+            if idx % stride == 0 {
+                (key.clone(), val + cid_offset)
+            } else {
+                (key.clone(), *val)
+            }
+        })
+        .collect()
+}
+
+fn remove_every_nth(records: &[(String, u32)], n: usize) -> Vec<(String, u32)> {
+    records
+        .iter()
+        .enumerate()
+        .filter(|(idx, _)| idx % n != 0)
+        .map(|(_, r)| r.clone())
+        .collect()
+}
+
+fn remove_range(records: &[(String, u32)], start: usize, count: usize) -> Vec<(String, u32)> {
+    records
+        .iter()
+        .enumerate()
+        .filter(|(idx, _)| *idx < start || *idx >= start + count)
+        .map(|(_, r)| r.clone())
+        .collect()
+}
+
+fn keep_only_collection(records: &[(String, u32)], collection: &str) -> Vec<(String, u32)> {
+    records
+        .iter()
+        .filter(|(key, _)| key.starts_with(collection))
+        .cloned()
+        .collect()
+}
+
+fn append_records(
+    base: &[(String, u32)],
+    collection: &str,
+    range: std::ops::Range<u32>,
+    cid_base: u32,
+) -> Vec<(String, u32)> {
+    let mut result = base.to_vec();
+    result.extend(range.map(|i| (make_key(collection, i), cid_base + i)));
+    result.sort_by(|(a, _), (b, _)| a.cmp(b));
+    result
+}
+
+#[tokio::test]
+async fn massive_tree_single_create() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec = append_records(&old, "app.bsky.feed.post", 2000..2001, 2000);
+    assert_equivalence(&old, &new_rec, "2000 records + 1 create").await;
+}
+
+#[tokio::test]
+async fn massive_tree_single_delete() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec = remove_range(&old, 1000, 1);
+    assert_equivalence(&old, &new_rec, "2000 records - 1 delete from middle").await;
+}
+
+#[tokio::test]
+async fn massive_tree_single_update() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec: Vec<_> = old
+        .iter()
+        .map(|(k, v)| {
+            if k == "app.bsky.feed.post/001000" {
+                (k.clone(), v + 50000)
+            } else {
+                (k.clone(), *v)
+            }
+        })
+        .collect();
+    assert_equivalence(&old, &new_rec, "2000 records - 1 update in middle").await;
+}
+
+#[tokio::test]
+async fn massive_tree_scattered_updates_every_3rd() {
+    let old = generate_records("app.bsky.feed.post", 0..1500);
+    let new_rec = apply_scattered_updates(&old, 3, 10000);
+    assert_equivalence(&old, &new_rec, "1500 records - update every 3rd").await;
+}
+
+#[tokio::test]
+async fn massive_tree_scattered_updates_every_7th() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec = apply_scattered_updates(&old, 7, 20000);
+    assert_equivalence(&old, &new_rec, "2000 records - update every 7th").await;
+}
+
+#[tokio::test]
+async fn massive_tree_delete_every_2nd() {
+    let old = generate_records("app.bsky.feed.post", 0..1000);
+    let new_rec = remove_every_nth(&old, 2);
+    assert_equivalence(&old, &new_rec, "1000 records - delete every 2nd").await;
+}
+
+#[tokio::test]
+async fn massive_tree_delete_every_5th() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec = remove_every_nth(&old, 5);
+    assert_equivalence(&old, &new_rec, "2000 records - delete every 5th").await;
+}
+
+#[tokio::test]
+async fn massive_tree_delete_first_half() {
+    let old = generate_records("app.bsky.feed.post", 0..1500);
+    let new_rec = remove_range(&old, 0, 750);
+    assert_equivalence(&old, &new_rec, "1500 records - delete first 750").await;
+}
+
+#[tokio::test]
+async fn massive_tree_delete_last_half() {
+    let old = generate_records("app.bsky.feed.post", 0..1500);
+    let new_rec = remove_range(&old, 750, 750);
+    assert_equivalence(&old, &new_rec, "1500 records - delete last 750").await;
+}
+
+#[tokio::test]
+async fn massive_tree_delete_middle_chunk() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec = remove_range(&old, 800, 400);
+    assert_equivalence(&old, &new_rec, "2000 records - delete 400 from middle").await;
+}
+
+#[tokio::test]
+async fn empty_to_massive() {
+    let new_rec = generate_records("app.bsky.feed.post", 0..1500);
+    assert_equivalence(&[], &new_rec, "empty to 1500 records").await;
+}
+
+#[tokio::test]
+async fn massive_to_empty() {
+    let old = generate_records("app.bsky.feed.post", 0..1500);
+    assert_equivalence(&old, &[], "1500 records to empty").await;
+}
+
+#[tokio::test]
+async fn massive_complete_replacement() {
+    let old = generate_records("app.bsky.feed.post", 0..1000);
+    let new_rec = generate_records("app.bsky.feed.post", 1000..2000);
+    assert_equivalence(&old, &new_rec, "1000 records fully replaced with 1000 different").await;
+}
+
+#[tokio::test]
+async fn massive_no_change() {
+    let records = generate_records("app.bsky.feed.post", 0..1500);
+    assert_equivalence(&records, &records, "1500 records unchanged").await;
+}
+
+#[tokio::test]
+async fn multi_collection_5_collections_500_each() {
+    let collections = [
+        "app.bsky.feed.like",
+        "app.bsky.feed.post",
+        "app.bsky.feed.repost",
+        "app.bsky.graph.follow",
+        "app.bsky.graph.block",
+    ];
+    let old = generate_multi_collection_records(&collections, 500);
+    let new_rec = apply_scattered_updates(&old, 4, 30000);
+    assert_equivalence(&old, &new_rec, "5 collections x 500 records - update every 4th").await;
+}
+
+#[tokio::test]
+async fn multi_collection_wipe_one_collection() {
+    let collections = [
+        "app.bsky.feed.like",
+        "app.bsky.feed.post",
+        "app.bsky.feed.repost",
+        "app.bsky.graph.follow",
+    ];
+    let old = generate_multi_collection_records(&collections, 400);
+
+    let new_rec: Vec<_> = old
+        .iter()
+        .filter(|(key, _)| !key.starts_with("app.bsky.feed.repost"))
+        .cloned()
+        .collect();
+    assert_equivalence(&old, &new_rec, "4 collections x 400 - wipe repost collection").await;
+}
+
+#[tokio::test]
+async fn multi_collection_keep_only_one() {
+    let collections = [
+        "app.bsky.feed.like",
+        "app.bsky.feed.post",
+        "app.bsky.feed.repost",
+        "app.bsky.graph.follow",
+        "app.bsky.graph.block",
+    ];
+    let old = generate_multi_collection_records(&collections, 300);
+    let new_rec = keep_only_collection(&old, "app.bsky.feed.post");
+    assert_equivalence(&old, &new_rec, "5 collections x 300 - keep only posts").await;
+}
+
+#[tokio::test]
+async fn multi_collection_add_new_collection() {
+    let old_collections = [
+        "app.bsky.feed.like",
+        "app.bsky.feed.post",
+    ];
+    let old = generate_multi_collection_records(&old_collections, 500);
+    let new_rec = append_records(&old, "app.bsky.graph.follow", 0..500, 40000);
+    assert_equivalence(&old, &new_rec, "2 collections x 500 + add 500 follows").await;
+}
+
+#[tokio::test]
+async fn mixed_ops_massive_tree() {
+    let collections = [
+        "app.bsky.feed.like",
+        "app.bsky.feed.post",
+        "app.bsky.feed.repost",
+        "app.bsky.graph.follow",
+    ];
+    let old = generate_multi_collection_records(&collections, 400);
+
+    let mut new_rec: Vec<_> = old
+        .iter()
+        .filter(|(key, _)| !key.starts_with("app.bsky.feed.repost"))
+        .enumerate()
+        .map(|(idx, (key, val))| {
+            if key.starts_with("app.bsky.feed.like") && idx % 3 == 0 {
+                (key.clone(), val + 50000)
+            } else {
+                (key.clone(), *val)
+            }
+        })
+        .collect();
+
+    new_rec.extend((0..200u32).map(|i| (make_key("app.bsky.graph.block", i), 60000 + i)));
+    new_rec.sort_by(|(a, _), (b, _)| a.cmp(b));
+
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "4 collections x 400: wipe reposts, update every 3rd like, add 200 blocks",
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn grow_tree_by_double() {
+    let old = generate_records("app.bsky.feed.post", 0..1000);
+    let new_rec = generate_records("app.bsky.feed.post", 0..2000);
+    assert_equivalence(&old, &new_rec, "grow from 1000 to 2000").await;
+}
+
+#[tokio::test]
+async fn shrink_tree_by_half() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec = generate_records("app.bsky.feed.post", 0..1000);
+    assert_equivalence(&old, &new_rec, "shrink from 2000 to 1000").await;
+}
+
+#[tokio::test]
+async fn interleaved_keys_disjoint_ranges() {
+    let old: Vec<_> = (0..1000u32)
+        .map(|i| (make_key("app.bsky.feed.post", i * 2), i))
+        .collect();
+    let new_rec: Vec<_> = (0..1000u32)
+        .map(|i| (make_key("app.bsky.feed.post", i * 2 + 1), i + 10000))
+        .collect();
+    assert_equivalence(&old, &new_rec, "1000 even-keyed records replaced by 1000 odd-keyed").await;
+}
+
+#[tokio::test]
+async fn sparse_keys_wide_gaps() {
+    let old: Vec<_> = (0..500u32)
+        .map(|i| (make_key("app.bsky.feed.post", i * 100), i))
+        .collect();
+    let new_rec: Vec<_> = (0..500u32)
+        .map(|i| {
+            if i % 10 == 0 {
+                (make_key("app.bsky.feed.post", i * 100), i + 70000)
+            } else {
+                (make_key("app.bsky.feed.post", i * 100), i)
+            }
+        })
+        .collect();
+    assert_equivalence(&old, &new_rec, "500 sparse keys - update every 10th").await;
+}
+
+#[tokio::test]
+async fn many_collections_few_records_each() {
+    let collections: Vec<String> = (0..50u32)
+        .map(|i| format!("com.example.lexicon{i:02}.record"))
+        .collect();
+    let old: Vec<_> = collections
+        .iter()
+        .enumerate()
+        .flat_map(|(ci, coll)| {
+            let base = ci as u32 * 20;
+            (0..20u32).map(move |i| (make_key(coll, i), base + i))
+        })
+        .collect();
+
+    let new_rec: Vec<_> = old
+        .iter()
+        .enumerate()
+        .filter_map(|(idx, (key, val))| {
+            if idx % 15 == 0 {
+                None
+            } else if idx % 7 == 0 {
+                Some((key.clone(), val + 80000))
+            } else {
+                Some((key.clone(), *val))
+            }
+        })
+        .collect();
+
+    assert_equivalence(&old, &new_rec, "50 collections x 20 records - delete every 15th, update every 7th").await;
+}
+
+#[tokio::test]
+async fn update_all_records() {
+    let old = generate_records("app.bsky.feed.post", 0..1000);
+    let new_rec: Vec<_> = old
+        .iter()
+        .map(|(key, val)| (key.clone(), val + 90000))
+        .collect();
+    assert_equivalence(&old, &new_rec, "1000 records - update every single one").await;
+}
+
+#[tokio::test]
+async fn delete_all_but_one() {
+    let old = generate_records("app.bsky.feed.post", 0..1500);
+    let new_rec = vec![old[750].clone()];
+    assert_equivalence(&old, &new_rec, "1500 records - delete all but middle one").await;
+}
+
+#[tokio::test]
+async fn one_to_massive() {
+    let old = vec![(make_key("app.bsky.feed.post", 500), 500u32)];
+    let new_rec = generate_records("app.bsky.feed.post", 0..1500);
+    assert_equivalence(&old, &new_rec, "1 record to 1500 records").await;
+}
+
+#[tokio::test]
+async fn delete_head_and_tail() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let new_rec: Vec<_> = old[200..1800].to_vec();
+    assert_equivalence(&old, &new_rec, "2000 records - delete first 200 and last 200").await;
+}
+
+#[tokio::test]
+async fn keep_head_and_tail_only() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let mut new_rec: Vec<_> = old[..100].to_vec();
+    new_rec.extend_from_slice(&old[1900..]);
+    assert_equivalence(&old, &new_rec, "2000 records - keep only first 100 and last 100").await;
+}
+
+#[tokio::test]
+async fn massive_tree_update_first_and_last() {
+    let old = generate_records("app.bsky.feed.post", 0..2000);
+    let mut new_rec = old.clone();
+    new_rec[0].1 += 99000;
+    new_rec[1999].1 += 99000;
+    assert_equivalence(&old, &new_rec, "2000 records - update only first and last").await;
+}
+
+#[tokio::test]
+async fn overlapping_collection_swap() {
+    let old_collections = [
+        "app.bsky.feed.like",
+        "app.bsky.feed.post",
+        "app.bsky.feed.repost",
+    ];
+    let old = generate_multi_collection_records(&old_collections, 500);
+
+    let mut new_rec: Vec<_> = old
+        .iter()
+        .filter(|(key, _)| key.starts_with("app.bsky.feed.post"))
+        .cloned()
+        .collect();
+    new_rec.extend((0..500u32).map(|i| (make_key("app.bsky.graph.follow", i), 70000 + i)));
+    new_rec.extend((0..500u32).map(|i| (make_key("app.bsky.graph.block", i), 71000 + i)));
+    new_rec.sort_by(|(a, _), (b, _)| a.cmp(b));
+
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "swap 2 of 3 collections, keep 1 (posts), 500 each",
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn swiss_cheese_deletions() {
+    let old = generate_records("app.bsky.feed.post", 0..1500);
+    let new_rec: Vec<_> = old
+        .iter()
+        .enumerate()
+        .filter(|(idx, _)| {
+            let bucket = idx / 50;
+            bucket % 3 != 0
+        })
+        .map(|(_, r)| r.clone())
+        .collect();
+    assert_equivalence(&old, &new_rec, "1500 records - delete every 3rd chunk of 50").await;
+}
+
+#[tokio::test]
+async fn mixed_ops_with_key_density_change() {
+    let old: Vec<_> = (0..1000u32)
+        .map(|i| (make_key("app.bsky.feed.post", i * 3), i))
+        .collect();
+
+    let mut new_rec: Vec<_> = old
+        .iter()
+        .filter(|(_, val)| val % 4 != 0)
+        .cloned()
+        .collect();
+    new_rec.extend((0..500u32).map(|i| {
+        (make_key("app.bsky.feed.post", i * 3 + 1), i + 100000)
+    }));
+    new_rec.sort_by(|(a, _), (b, _)| a.cmp(b));
+
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "1000 sparse records: delete every 4th, insert 500 in gaps",
+    )
+    .await;
+}

crates/tranquil-store/Cargo.toml

@@ -52,6 +52,7 @@ k256 = { workspace = true }
 rand = { workspace = true }
 tikv-jemallocator = "0.6"
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
+libc = "0.2"
 
 [[bench]]
 name = "blockstore"

crates/tranquil-store/src/blockstore/compaction.rs

@@ -93,8 +93,7 @@ pub(super) fn compact_on_writer_thread<S: StorageIO>(
             Err(e)
         }
         Ok((new_size, live_count, dead_count)) => {
-            let positions = hint_positions.snapshot();
-            if let Err(e) = index.write_checkpoint(epoch.current(), &positions) {
+            if let Err(e) = index.write_checkpoint(epoch.current(), hint_positions) {
                 tracing::warn!(error = %e, "pre-delete checkpoint failed during compaction");
             }
 

crates/tranquil-store/src/blockstore/group_commit.rs

@@ -813,8 +813,7 @@ fn maybe_checkpoint(
     if !elapsed && !threshold {
         return;
     }
-    let positions = hint_positions.snapshot();
-    match index.write_checkpoint(epoch.current(), &positions) {
+    match index.write_checkpoint(epoch.current(), hint_positions) {
         Ok(()) => {
             *last_checkpoint = std::time::Instant::now();
             *writes_since_checkpoint = 0;
@@ -831,8 +830,7 @@ fn shutdown_checkpoint(
     epoch: &EpochCounter,
     hint_positions: &ShardHintPositions,
 ) {
-    let positions = hint_positions.snapshot();
-    match index.write_checkpoint(epoch.current(), &positions) {
+    match index.write_checkpoint(epoch.current(), hint_positions) {
         Ok(()) => tracing::debug!("shutdown checkpoint written"),
         Err(e) => tracing::warn!(error = %e, "shutdown checkpoint failed"),
     }
@@ -924,8 +922,6 @@ fn commit_loop<S: StorageIO>(
             if let Ok((ref dedup, _)) = result {
                 writes_since_checkpoint =
                     writes_since_checkpoint.saturating_add(dedup.len() as u64);
-                ctx.hint_positions
-                    .update(ctx.shard_id, state.file_id, state.hint_position);
             }
 
             dispatch_responses(drain.entries, result.map(|(dedup, _proof)| dedup));
@@ -1024,8 +1020,6 @@ fn drain_and_process_remaining<S: StorageIO>(
 
         if let Ok((ref _dedup, ref proof)) = result {
             run_post_sync_hook(post_sync_hook, proof);
-            ctx.hint_positions
-                .update(ctx.shard_id, state.file_id, state.hint_position);
         }
 
         dispatch_responses(entries, result.map(|(dedup, _proof)| dedup));
@@ -1102,6 +1096,7 @@ fn process_batch<S: StorageIO>(
             let location = match dedup.get(cid_bytes) {
                 Some(&loc) => {
                     dedup_hits = dedup_hits.saturating_add(1);
+                    hint_writer.append_hint(cid_bytes, &loc)?;
                     loc
                 }
                 None => {
@@ -1194,7 +1189,19 @@ fn process_batch<S: StorageIO>(
     };
     let t = std::time::Instant::now();
     index
-        .batch_put(&index_entries, &all_decrements, cursor, current_epoch, now)
+        .batch_put_and_advance_position(
+            &index_entries,
+            &all_decrements,
+            cursor,
+            current_epoch,
+            now,
+            super::hash_index::PositionUpdate {
+                hint_positions: &ctx.hint_positions,
+                shard_id: ctx.shard_id,
+                file_id: state.file_id,
+                offset: state.hint_position,
+            },
+        )
         .map_err(CommitError::from)?;
     let index_nanos = t.elapsed().as_nanos() as u64;
 

crates/tranquil-store/src/blockstore/hash_index.rs

@@ -5,11 +5,19 @@ use std::path::{Path, PathBuf};
 use parking_lot::RwLock;
 
 use super::data_file::CID_SIZE;
+use super::group_commit::ShardHintPositions;
 use super::types::{
     BlockLength, BlockLocation, BlockOffset, CidBytes, CollectionResult, CommitEpoch, DataFileId,
-    HintOffset, IndexEntry, LivenessInfo, RefCount, WallClockMs, WriteCursor,
+    HintOffset, IndexEntry, LivenessInfo, RefCount, ShardId, WallClockMs, WriteCursor,
 };
 
+pub struct PositionUpdate<'a> {
+    pub hint_positions: &'a ShardHintPositions,
+    pub shard_id: ShardId,
+    pub file_id: DataFileId,
+    pub offset: HintOffset,
+}
+
 const EMPTY_CID: [u8; CID_SIZE] = [0u8; CID_SIZE];
 
 fn is_empty(cid: &[u8; CID_SIZE]) -> bool {
@@ -1197,6 +1205,30 @@ impl BlockIndex {
         cursor: WriteCursor,
         epoch: CommitEpoch,
         now: WallClockMs,
+    ) -> Result<(), BlockIndexError> {
+        self.batch_put_inner(entries, decrements, cursor, epoch, now, None)
+    }
+
+    pub fn batch_put_and_advance_position(
+        &self,
+        entries: &[([u8; CID_SIZE], BlockLocation)],
+        decrements: &[[u8; CID_SIZE]],
+        cursor: WriteCursor,
+        epoch: CommitEpoch,
+        now: WallClockMs,
+        position_update: PositionUpdate<'_>,
+    ) -> Result<(), BlockIndexError> {
+        self.batch_put_inner(entries, decrements, cursor, epoch, now, Some(position_update))
+    }
+
+    fn batch_put_inner(
+        &self,
+        entries: &[([u8; CID_SIZE], BlockLocation)],
+        decrements: &[[u8; CID_SIZE]],
+        cursor: WriteCursor,
+        epoch: CommitEpoch,
+        now: WallClockMs,
+        position_update: Option<PositionUpdate<'_>>,
     ) -> Result<(), BlockIndexError> {
         let mut table = self.table.write();
 
@@ -1217,6 +1249,12 @@ impl BlockIndex {
         });
 
         table.set_write_cursor(cursor);
+
+        if let Some(pos) = position_update {
+            pos.hint_positions
+                .update(pos.shard_id, pos.file_id, pos.offset);
+        }
+
         Ok(())
     }
 
@@ -1418,6 +1456,17 @@ impl BlockIndex {
     }
 
     pub fn write_checkpoint(
+        &self,
+        epoch: CommitEpoch,
+        hint_positions: &ShardHintPositions,
+    ) -> io::Result<()> {
+        let _guard = self.checkpoint_lock.lock();
+        let table = self.table.read();
+        let positions = hint_positions.snapshot();
+        write_checkpoint_ab(&table, &self.index_dir, epoch, &positions)
+    }
+
+    pub fn write_checkpoint_with_positions(
         &self,
         epoch: CommitEpoch,
         positions: &CheckpointPositions,

crates/tranquil-store/tests/checkpoint_race.rs

@@ -0,0 +1,223 @@
+mod common;
+
+use std::io;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+
+use tranquil_store::blockstore::{
+    BlockStoreConfig, BlocksSynced, CidBytes, GroupCommitConfig, TranquilBlockStore,
+};
+use tranquil_store::PostBlockstoreHook;
+
+struct SlowHook;
+
+impl PostBlockstoreHook for SlowHook {
+    fn on_blocks_synced(&self, _proof: &BlocksSynced) -> io::Result<()> {
+        std::thread::sleep(std::time::Duration::from_millis(1));
+        Ok(())
+    }
+}
+
+fn refcount(store: &TranquilBlockStore, cid: &CidBytes) -> Option<u32> {
+    store.block_index().get(cid).map(|e| e.refcount.raw())
+}
+
+fn race_config(dir: &std::path::Path) -> BlockStoreConfig {
+    BlockStoreConfig {
+        data_dir: dir.join("data"),
+        index_dir: dir.join("index"),
+        max_file_size: 256 * 1024,
+        group_commit: GroupCommitConfig {
+            checkpoint_interval_ms: 10,
+            checkpoint_write_threshold: 20,
+            ..GroupCommitConfig::default()
+        },
+        shard_count: 4,
+    }
+}
+
+fn cid_for(shard: u8, seq: u32) -> CidBytes {
+    let mut cid = [0u8; 36];
+    cid[0] = 0x01;
+    cid[1] = 0x71;
+    cid[2] = 0x12;
+    cid[3] = 0x20;
+    cid[4] = shard;
+    cid[8..12].copy_from_slice(&seq.to_le_bytes());
+    (12..36).for_each(|i| cid[i] = (seq as u8).wrapping_add(i as u8));
+    cid
+}
+
+fn write_phase(base: &std::path::Path, use_hook: bool) -> Vec<CidBytes> {
+    let config = race_config(base);
+    let hook: Option<Arc<dyn PostBlockstoreHook>> = use_hook.then(|| Arc::new(SlowHook) as _);
+    let store = Arc::new(TranquilBlockStore::open_with_hook(config, hook).unwrap());
+
+    let running = Arc::new(AtomicBool::new(true));
+    let total_cycles = Arc::new(AtomicU64::new(0));
+
+    let writers: Vec<_> = (0..4u8)
+        .map(|shard| {
+            let store = Arc::clone(&store);
+            let running = Arc::clone(&running);
+            let total_cycles = Arc::clone(&total_cycles);
+            std::thread::spawn(move || {
+                let mut targets = Vec::new();
+                let mut seq = 0u32;
+                while running.load(Ordering::Relaxed) {
+                    let cid = cid_for(shard, seq);
+                    store
+                        .put_blocks_blocking(vec![(cid, vec![shard; 60])])
+                        .unwrap();
+                    store
+                        .put_blocks_blocking(vec![(cid, vec![shard; 60])])
+                        .unwrap();
+                    store
+                        .apply_commit_blocking(vec![], vec![cid])
+                        .unwrap();
+                    targets.push(cid);
+                    seq += 1;
+                    total_cycles.fetch_add(1, Ordering::Relaxed);
+                }
+                targets
+            })
+        })
+        .collect();
+
+    while total_cycles.load(Ordering::Relaxed) < 500 {
+        std::thread::yield_now();
+    }
+
+    running.store(false, Ordering::Relaxed);
+
+    let all_targets: Vec<CidBytes> = writers
+        .into_iter()
+        .flat_map(|w| w.join().unwrap())
+        .collect();
+
+    all_targets.iter().for_each(|cid| {
+        assert_eq!(refcount(&store, cid), Some(1), "pre-crash sanity");
+    });
+
+    let store = Arc::try_unwrap(store).ok().unwrap();
+    std::mem::forget(store);
+
+    all_targets
+}
+
+fn verify_phase(base: &std::path::Path, targets: &[CidBytes]) -> usize {
+    let config = race_config(base);
+    let store = TranquilBlockStore::open(config).unwrap();
+    let bad = targets
+        .iter()
+        .filter(|cid| refcount(&store, cid) != Some(1))
+        .count();
+    drop(store);
+    bad
+}
+
+#[test]
+fn crash_recovery_preserves_refcounts() {
+    common::with_runtime(|| {
+        let mut corrupted = 0u32;
+        let total = 20u32;
+
+        (0..total).for_each(|_| {
+            let dir = tempfile::TempDir::new().unwrap();
+            let exe = std::env::current_exe().unwrap();
+            let dir_str = dir.path().to_str().unwrap();
+
+            let output = std::process::Command::new(&exe)
+                .arg("--exact")
+                .arg("__crash_write_phase")
+                .env("CRASH_TEST_DIR", dir_str)
+                .env("CRASH_TEST_HOOK", "0")
+                .output()
+                .unwrap();
+
+            assert!(output.status.success() || output.status.code() == Some(0));
+
+            let target_bytes = std::fs::read(dir.path().join("targets.bin")).unwrap();
+            let targets: Vec<CidBytes> = target_bytes
+                .chunks_exact(36)
+                .map(|chunk| {
+                    let mut cid = [0u8; 36];
+                    cid.copy_from_slice(chunk);
+                    cid
+                })
+                .collect();
+
+            if verify_phase(dir.path(), &targets) > 0 {
+                corrupted += 1;
+            }
+        });
+
+        assert_eq!(
+            corrupted, 0,
+            "{corrupted}/{total} iterations had refcount corruption after crash recovery"
+        );
+    });
+}
+
+#[test]
+fn crash_with_slow_hook_preserves_refcounts() {
+    common::with_runtime(|| {
+        let mut corrupted = 0u32;
+        let total = 20u32;
+
+        (0..total).for_each(|_| {
+            let dir = tempfile::TempDir::new().unwrap();
+            let exe = std::env::current_exe().unwrap();
+            let dir_str = dir.path().to_str().unwrap();
+
+            let output = std::process::Command::new(&exe)
+                .arg("--exact")
+                .arg("__crash_write_phase")
+                .env("CRASH_TEST_DIR", dir_str)
+                .env("CRASH_TEST_HOOK", "1")
+                .output()
+                .unwrap();
+
+            assert!(output.status.success() || output.status.code() == Some(0));
+
+            let target_bytes = std::fs::read(dir.path().join("targets.bin")).unwrap();
+            let targets: Vec<CidBytes> = target_bytes
+                .chunks_exact(36)
+                .map(|chunk| {
+                    let mut cid = [0u8; 36];
+                    cid.copy_from_slice(chunk);
+                    cid
+                })
+                .collect();
+
+            if verify_phase(dir.path(), &targets) > 0 {
+                corrupted += 1;
+            }
+        });
+
+        assert_eq!(
+            corrupted, 0,
+            "{corrupted}/{total} iterations had refcount corruption after crash with slow hook"
+        );
+    });
+}
+
+#[test]
+fn __crash_write_phase() {
+    let dir = match std::env::var("CRASH_TEST_DIR") {
+        Ok(d) => d,
+        Err(_) => return,
+    };
+    let use_hook = std::env::var("CRASH_TEST_HOOK").map(|v| v == "1").unwrap_or(false);
+    let base = std::path::Path::new(&dir);
+
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let _guard = rt.enter();
+
+    let targets = write_phase(base, use_hook);
+
+    let target_bytes: Vec<u8> = targets.iter().flat_map(|cid| cid.iter().copied()).collect();
+    std::fs::write(base.join("targets.bin"), &target_bytes).unwrap();
+
+    unsafe { libc::_exit(0) }
+}

crates/tranquil-store/tests/sim_blockstore.rs

@@ -112,7 +112,7 @@ impl SimHarness {
             HintOffset::new(entries.len() as u64 * HINT_RECORD_SIZE as u64),
         );
         index
-            .write_checkpoint(CommitEpoch::zero(), &positions)
+            .write_checkpoint_with_positions(CommitEpoch::zero(), &positions)
             .unwrap();
     }
 

flake.nix

@@ -32,7 +32,7 @@
           { lib, pkgs, ... }:
           {
             _file = "${self.outPath}/flake.nix#nixosModules.tranquil-pds";
-            imports = [ (import ./module.nix self) ];
+            imports = [ ./module.nix ];
             config.services.tranquil-pds = {
               package = self.packages.${pkgs.stdenv.hostPlatform.system}.tranquil-pds;
               settings.frontend.package = self.packages.${pkgs.stdenv.hostPlatform.system}.tranquil-frontend;

frontend.nix

@@ -1,6 +1,7 @@
 {
   lib,
   stdenvNoCC,
+  nodejs,
   pnpm,
   pnpmConfigHook,
   fetchPnpmDeps,
@@ -23,6 +24,7 @@ stdenvNoCC.mkDerivation (finalAttrs: {
 
   nativeBuildInputs = [
     pnpm
+    nodejs
     pnpmConfigHook
   ];
 

justfile

@@ -56,7 +56,6 @@ test-misc:
     ./scripts/run-tests.sh --test actor --test commit_signing --test image_processing --test lifecycle_social --test notifications --test server --test signing_key --test verify_live_commit
 
 test *args:
-    @just test-store
     @just test-unit
     ./scripts/run-tests.sh {{args}}
 

scripts/run-tests.sh

@@ -23,7 +23,7 @@ cargo test --no-run 2>&1 | tail -1
 
 echo "Running tests..."
 echo ""
-cargo nextest run "$@"
+cargo nextest run -E 'not package(tranquil-store)' "$@"
 
 echo ""
 echo "All tests passed."