Remove old user blocks

.sqlx/query-03faaf7b8676e0af1bf620759425632560dabfd5748d0383971c10f9b2847d7d.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            DELETE FROM user_blocks\n            WHERE user_id = $1\n            AND block_cid = ANY($2)\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "ByteaArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "03faaf7b8676e0af1bf620759425632560dabfd5748d0383971c10f9b2847d7d"
+}

.sqlx/query-d71881b1dd8111b2afff6a7af8829651379afbe050dcc8a93e0b91eced31ca89.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                    INSERT INTO user_blocks (user_id, block_cid)\n                    SELECT $1, block_cid FROM UNNEST($2::bytea[]) AS t(block_cid)\n                    ON CONFLICT (user_id, block_cid) DO NOTHING\n                    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "ByteaArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d71881b1dd8111b2afff6a7af8829651379afbe050dcc8a93e0b91eced31ca89"
+}

.sqlx/query-e70fc3dced4eb7dc220ca2a18cdfcbd5f2d66dff2262bb083fd4118b032ff978.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT block_cid FROM user_blocks WHERE user_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "block_cid",
+        "type_info": "Bytea"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "e70fc3dced4eb7dc220ca2a18cdfcbd5f2d66dff2262bb083fd4118b032ff978"
+}

migrations/20260106_clear_user_blocks.sql

@@ -0,0 +1 @@
+TRUNCATE TABLE user_blocks;

src/api/backup.rs

@@ -220,14 +220,15 @@ pub async fn create_backup(State(state): State<AppState>, auth: BearerAuth) -> R
         }
     };
 
-    let car_bytes = match generate_full_backup(&state.block_store, &head_cid).await {
-        Ok(bytes) => bytes,
-        Err(e) => {
-            error!("Failed to generate CAR: {:?}", e);
-            return ApiError::InternalError(Some("Failed to generate backup".into()))
-                .into_response();
-        }
-    };
+    let car_bytes =
+        match generate_full_backup(&state.db, &state.block_store, user.id, &head_cid).await {
+            Ok(bytes) => bytes,
+            Err(e) => {
+                error!("Failed to generate CAR: {:?}", e);
+                return ApiError::InternalError(Some("Failed to generate backup".into()))
+                    .into_response();
+            }
+        };
 
     let block_count = crate::scheduled::count_car_blocks(&car_bytes);
     let size_bytes = car_bytes.len() as i64;

src/api/repo/record/batch.rs

@@ -388,18 +388,15 @@ pub async fn apply_writes(
             return ApiError::InternalError(Some("Failed to persist MST".into())).into_response();
         }
     };
-    let mut relevant_blocks = std::collections::BTreeMap::new();
+    let mut new_mst_blocks = std::collections::BTreeMap::new();
+    let mut old_mst_blocks = std::collections::BTreeMap::new();
     for key in &modified_keys {
-        if mst
-            .blocks_for_path(key, &mut relevant_blocks)
-            .await
-            .is_err()
-        {
+        if mst.blocks_for_path(key, &mut new_mst_blocks).await.is_err() {
             return ApiError::InternalError(Some("Failed to get new MST blocks for path".into()))
                 .into_response();
         }
         if original_mst
-            .blocks_for_path(key, &mut relevant_blocks)
+            .blocks_for_path(key, &mut old_mst_blocks)
             .await
             .is_err()
         {
@@ -407,16 +404,36 @@ pub async fn apply_writes(
                 .into_response();
         }
     }
-    let mut written_cids = tracking_store.get_all_relevant_cids();
-    for cid in relevant_blocks.keys() {
-        if !written_cids.contains(cid) {
-            written_cids.push(*cid);
+    let mut relevant_blocks = new_mst_blocks.clone();
+    relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone())));
+    let written_cids: Vec<Cid> = tracking_store
+        .get_all_relevant_cids()
+        .into_iter()
+        .chain(relevant_blocks.keys().copied())
+        .collect::<std::collections::HashSet<_>>()
+        .into_iter()
+        .collect();
+    let written_cids_str: Vec<String> = written_cids.iter().map(|c| c.to_string()).collect();
+    let prev_record_cids = ops.iter().filter_map(|op| match op {
+        RecordOp::Update {
+            prev: Some(cid), ..
         }
-    }
-    let written_cids_str = written_cids
-        .iter()
-        .map(|c| c.to_string())
-        .collect::<Vec<_>>();
+        | RecordOp::Delete {
+            prev: Some(cid), ..
+        } => Some(*cid),
+        _ => None,
+    });
+    let obsolete_cids: Vec<Cid> = std::iter::once(current_root_cid)
+        .chain(
+            old_mst_blocks
+                .keys()
+                .filter(|cid| !new_mst_blocks.contains_key(*cid))
+                .copied(),
+        )
+        .chain(prev_record_cids)
+        .collect::<std::collections::HashSet<_>>()
+        .into_iter()
+        .collect();
     let commit_res = match commit_and_log(
         &state,
         CommitParams {
@@ -428,6 +445,7 @@ pub async fn apply_writes(
             ops,
             blocks_cids: &written_cids_str,
             blobs: &all_blob_cids,
+            obsolete_cids,
         },
     )
     .await

src/api/repo/record/delete.rs

@@ -129,9 +129,10 @@ pub async fn delete_record(
         rkey: rkey_for_audit.clone(),
         prev: prev_record_cid,
     };
-    let mut relevant_blocks = std::collections::BTreeMap::new();
+    let mut new_mst_blocks = std::collections::BTreeMap::new();
+    let mut old_mst_blocks = std::collections::BTreeMap::new();
     if new_mst
-        .blocks_for_path(&key, &mut relevant_blocks)
+        .blocks_for_path(&key, &mut new_mst_blocks)
         .await
         .is_err()
     {
@@ -139,23 +140,32 @@ pub async fn delete_record(
             .into_response();
     }
     if mst
-        .blocks_for_path(&key, &mut relevant_blocks)
+        .blocks_for_path(&key, &mut old_mst_blocks)
         .await
         .is_err()
     {
         return ApiError::InternalError(Some("Failed to get old MST blocks for path".into()))
             .into_response();
     }
-    let mut written_cids = tracking_store.get_all_relevant_cids();
-    for cid in relevant_blocks.keys() {
-        if !written_cids.contains(cid) {
-            written_cids.push(*cid);
-        }
-    }
-    let written_cids_str = written_cids
-        .iter()
-        .map(|c| c.to_string())
-        .collect::<Vec<_>>();
+    let mut relevant_blocks = new_mst_blocks.clone();
+    relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone())));
+    let written_cids: Vec<Cid> = tracking_store
+        .get_all_relevant_cids()
+        .into_iter()
+        .chain(relevant_blocks.keys().copied())
+        .collect::<std::collections::HashSet<_>>()
+        .into_iter()
+        .collect();
+    let written_cids_str: Vec<String> = written_cids.iter().map(|c| c.to_string()).collect();
+    let obsolete_cids: Vec<Cid> = std::iter::once(current_root_cid)
+        .chain(
+            old_mst_blocks
+                .keys()
+                .filter(|cid| !new_mst_blocks.contains_key(*cid))
+                .copied(),
+        )
+        .chain(prev_record_cid)
+        .collect();
     let commit_result = match commit_and_log(
         &state,
         CommitParams {
@@ -167,6 +177,7 @@ pub async fn delete_record(
             ops: vec![op],
             blocks_cids: &written_cids_str,
             blobs: &[],
+            obsolete_cids,
         },
     )
     .await

src/api/repo/record/utils.rs

@@ -92,6 +92,7 @@ pub struct CommitParams<'a> {
     pub ops: Vec<RecordOp>,
     pub blocks_cids: &'a [String],
     pub blobs: &'a [String],
+    pub obsolete_cids: Vec<Cid>,
 }
 
 pub async fn commit_and_log(
@@ -107,6 +108,7 @@ pub async fn commit_and_log(
         ops,
         blocks_cids,
         blobs,
+        obsolete_cids,
     } = params;
     let key_row = sqlx::query!(
         "SELECT key_bytes, encryption_version FROM user_keys WHERE user_id = $1",
@@ -201,6 +203,21 @@ pub async fn commit_and_log(
         .await
         .map_err(|e| format!("DB Error (user_blocks): {}", e))?;
     }
+    if !obsolete_cids.is_empty() {
+        let obsolete_bytes: Vec<Vec<u8>> = obsolete_cids.iter().map(|c| c.to_bytes()).collect();
+        sqlx::query!(
+            r#"
+            DELETE FROM user_blocks
+            WHERE user_id = $1
+            AND block_cid = ANY($2)
+            "#,
+            user_id,
+            &obsolete_bytes as &[Vec<u8>]
+        )
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| format!("DB Error (user_blocks delete obsolete): {}", e))?;
+    }
     let mut upsert_collections: Vec<String> = Vec::new();
     let mut upsert_rkeys: Vec<String> = Vec::new();
     let mut upsert_cids: Vec<String> = Vec::new();
@@ -404,21 +421,33 @@ pub async fn create_record_internal(
         rkey: rkey.to_string(),
         cid: record_cid,
     };
-    let mut relevant_blocks = std::collections::BTreeMap::new();
+    let mut new_mst_blocks = std::collections::BTreeMap::new();
+    let mut old_mst_blocks = std::collections::BTreeMap::new();
     new_mst
-        .blocks_for_path(&key, &mut relevant_blocks)
+        .blocks_for_path(&key, &mut new_mst_blocks)
         .await
         .map_err(|e| format!("Failed to get new MST blocks for path: {:?}", e))?;
-    mst.blocks_for_path(&key, &mut relevant_blocks)
+    mst.blocks_for_path(&key, &mut old_mst_blocks)
         .await
         .map_err(|e| format!("Failed to get old MST blocks for path: {:?}", e))?;
+    let obsolete_cids: Vec<Cid> = std::iter::once(current_root_cid)
+        .chain(
+            old_mst_blocks
+                .keys()
+                .filter(|cid| !new_mst_blocks.contains_key(*cid))
+                .copied(),
+        )
+        .collect();
+    let mut relevant_blocks = new_mst_blocks;
+    relevant_blocks.extend(old_mst_blocks);
     relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes));
-    let mut written_cids = tracking_store.get_all_relevant_cids();
-    for cid in relevant_blocks.keys() {
-        if !written_cids.contains(cid) {
-            written_cids.push(*cid);
-        }
-    }
+    let written_cids: Vec<Cid> = tracking_store
+        .get_all_relevant_cids()
+        .into_iter()
+        .chain(relevant_blocks.keys().copied())
+        .collect::<std::collections::HashSet<_>>()
+        .into_iter()
+        .collect();
     let written_cids_str: Vec<String> = written_cids.iter().map(|c| c.to_string()).collect();
     let blob_cids = extract_blob_cids(record);
     let result = commit_and_log(
@@ -432,6 +461,7 @@ pub async fn create_record_internal(
             ops: vec![op],
             blocks_cids: &written_cids_str,
             blobs: &blob_cids,
+            obsolete_cids,
         },
     )
     .await?;

src/api/repo/record/write.rs

@@ -266,9 +266,10 @@ pub async fn create_record(
         rkey: rkey.to_string(),
         cid: record_cid,
     };
-    let mut relevant_blocks = std::collections::BTreeMap::new();
+    let mut new_mst_blocks = std::collections::BTreeMap::new();
+    let mut old_mst_blocks = std::collections::BTreeMap::new();
     if new_mst
-        .blocks_for_path(&key, &mut relevant_blocks)
+        .blocks_for_path(&key, &mut new_mst_blocks)
         .await
         .is_err()
     {
@@ -276,25 +277,33 @@ pub async fn create_record(
             .into_response();
     }
     if mst
-        .blocks_for_path(&key, &mut relevant_blocks)
+        .blocks_for_path(&key, &mut old_mst_blocks)
         .await
         .is_err()
     {
         return ApiError::InternalError(Some("Failed to get old MST blocks for path".into()))
             .into_response();
     }
+    let mut relevant_blocks = new_mst_blocks.clone();
+    relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone())));
     relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes));
-    let mut written_cids = tracking_store.get_all_relevant_cids();
-    for cid in relevant_blocks.keys() {
-        if !written_cids.contains(cid) {
-            written_cids.push(*cid);
-        }
-    }
-    let written_cids_str = written_cids
-        .iter()
-        .map(|c| c.to_string())
-        .collect::<Vec<_>>();
+    let written_cids: Vec<Cid> = tracking_store
+        .get_all_relevant_cids()
+        .into_iter()
+        .chain(relevant_blocks.keys().copied())
+        .collect::<std::collections::HashSet<_>>()
+        .into_iter()
+        .collect();
+    let written_cids_str: Vec<String> = written_cids.iter().map(|c| c.to_string()).collect();
     let blob_cids = extract_blob_cids(&input.record);
+    let obsolete_cids: Vec<Cid> = std::iter::once(current_root_cid)
+        .chain(
+            old_mst_blocks
+                .keys()
+                .filter(|cid| !new_mst_blocks.contains_key(*cid))
+                .copied(),
+        )
+        .collect();
     let commit_result = match commit_and_log(
         &state,
         CommitParams {
@@ -306,6 +315,7 @@ pub async fn create_record(
             ops: vec![op],
             blocks_cids: &written_cids_str,
             blobs: &blob_cids,
+            obsolete_cids,
         },
     )
     .await
@@ -512,9 +522,10 @@ pub async fn put_record(
             cid: record_cid,
         }
     };
-    let mut relevant_blocks = std::collections::BTreeMap::new();
+    let mut new_mst_blocks = std::collections::BTreeMap::new();
+    let mut old_mst_blocks = std::collections::BTreeMap::new();
     if new_mst
-        .blocks_for_path(&key, &mut relevant_blocks)
+        .blocks_for_path(&key, &mut new_mst_blocks)
         .await
         .is_err()
     {
@@ -522,26 +533,35 @@ pub async fn put_record(
             .into_response();
     }
     if mst
-        .blocks_for_path(&key, &mut relevant_blocks)
+        .blocks_for_path(&key, &mut old_mst_blocks)
         .await
         .is_err()
     {
         return ApiError::InternalError(Some("Failed to get old MST blocks for path".into()))
             .into_response();
     }
+    let mut relevant_blocks = new_mst_blocks.clone();
+    relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone())));
     relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes));
-    let mut written_cids = tracking_store.get_all_relevant_cids();
-    for cid in relevant_blocks.keys() {
-        if !written_cids.contains(cid) {
-            written_cids.push(*cid);
-        }
-    }
-    let written_cids_str = written_cids
-        .iter()
-        .map(|c| c.to_string())
-        .collect::<Vec<_>>();
+    let written_cids: Vec<Cid> = tracking_store
+        .get_all_relevant_cids()
+        .into_iter()
+        .chain(relevant_blocks.keys().copied())
+        .collect::<std::collections::HashSet<_>>()
+        .into_iter()
+        .collect();
+    let written_cids_str: Vec<String> = written_cids.iter().map(|c| c.to_string()).collect();
     let is_update = existing_cid.is_some();
     let blob_cids = extract_blob_cids(&input.record);
+    let obsolete_cids: Vec<Cid> = std::iter::once(current_root_cid)
+        .chain(
+            old_mst_blocks
+                .keys()
+                .filter(|cid| !new_mst_blocks.contains_key(*cid))
+                .copied(),
+        )
+        .chain(existing_cid)
+        .collect();
     let commit_result = match commit_and_log(
         &state,
         CommitParams {
@@ -553,6 +573,7 @@ pub async fn put_record(
             ops: vec![op],
             blocks_cids: &written_cids_str,
             blobs: &blob_cids,
+            obsolete_cids,
         },
     )
     .await

src/scheduled.rs

@@ -226,74 +226,94 @@ pub async fn backfill_user_blocks(db: &PgPool, block_store: PostgresBlockStore)
             }
         };
 
-        let mut block_cids: Vec<Vec<u8>> = Vec::new();
-        let mut to_visit = vec![root_cid];
-        let mut visited = std::collections::HashSet::new();
-
-        while let Some(cid) = to_visit.pop() {
-            if visited.contains(&cid) {
-                continue;
-            }
-            visited.insert(cid);
-            block_cids.push(cid.to_bytes());
-
-            let block = match block_store.get(&cid).await {
-                Ok(Some(b)) => b,
-                _ => continue,
-            };
-
-            if let Ok(commit) = Commit::from_cbor(&block) {
-                to_visit.push(commit.data);
-                if let Some(prev) = commit.prev {
-                    to_visit.push(prev);
+        match collect_current_repo_blocks(&block_store, &root_cid).await {
+            Ok(block_cids) => {
+                if block_cids.is_empty() {
+                    failed += 1;
+                    continue;
                 }
-            } else if let Ok(Ipld::Map(ref obj)) = serde_ipld_dagcbor::from_slice::<Ipld>(&block) {
-                if let Some(Ipld::Link(left_cid)) = obj.get("l") {
-                    to_visit.push(*left_cid);
-                }
-                if let Some(Ipld::List(entries)) = obj.get("e") {
-                    for entry in entries {
-                        if let Ipld::Map(entry_obj) = entry {
-                            if let Some(Ipld::Link(tree_cid)) = entry_obj.get("t") {
-                                to_visit.push(*tree_cid);
-                            }
-                            if let Some(Ipld::Link(val_cid)) = entry_obj.get("v") {
-                                to_visit.push(*val_cid);
-                            }
-                        }
-                    }
+
+                if let Err(e) = sqlx::query!(
+                    r#"
+                    INSERT INTO user_blocks (user_id, block_cid)
+                    SELECT $1, block_cid FROM UNNEST($2::bytea[]) AS t(block_cid)
+                    ON CONFLICT (user_id, block_cid) DO NOTHING
+                    "#,
+                    user.user_id,
+                    &block_cids
+                )
+                .execute(db)
+                .await
+                {
+                    warn!(user_id = %user.user_id, error = %e, "Failed to backfill user_blocks");
+                    failed += 1;
+                } else {
+                    info!(user_id = %user.user_id, block_count = block_cids.len(), "Backfilled user_blocks");
+                    success += 1;
                 }
             }
-        }
-
-        if block_cids.is_empty() {
-            failed += 1;
-            continue;
-        }
-
-        if let Err(e) = sqlx::query!(
-            r#"
-            INSERT INTO user_blocks (user_id, block_cid)
-            SELECT $1, block_cid FROM UNNEST($2::bytea[]) AS t(block_cid)
-            ON CONFLICT (user_id, block_cid) DO NOTHING
-            "#,
-            user.user_id,
-            &block_cids
-        )
-        .execute(db)
-        .await
-        {
-            warn!(user_id = %user.user_id, error = %e, "Failed to backfill user_blocks");
-            failed += 1;
-        } else {
-            info!(user_id = %user.user_id, block_count = block_cids.len(), "Backfilled user_blocks");
-            success += 1;
+            Err(e) => {
+                warn!(user_id = %user.user_id, error = %e, "Failed to collect repo blocks for backfill");
+                failed += 1;
+            }
         }
     }
 
     info!(success, failed, "Completed user_blocks backfill");
 }
 
+pub async fn collect_current_repo_blocks(
+    block_store: &PostgresBlockStore,
+    head_cid: &Cid,
+) -> Result<Vec<Vec<u8>>, String> {
+    let mut block_cids: Vec<Vec<u8>> = Vec::new();
+    let mut to_visit = vec![*head_cid];
+    let mut visited = std::collections::HashSet::new();
+
+    while let Some(cid) = to_visit.pop() {
+        if visited.contains(&cid) {
+            continue;
+        }
+        visited.insert(cid);
+        block_cids.push(cid.to_bytes());
+
+        let block = match block_store.get(&cid).await {
+            Ok(Some(b)) => b,
+            Ok(None) => continue,
+            Err(e) => return Err(format!("Failed to get block {}: {:?}", cid, e)),
+        };
+
+        if let Ok(commit) = Commit::from_cbor(&block) {
+            to_visit.push(commit.data);
+        } else if let Ok(Ipld::Map(ref obj)) = serde_ipld_dagcbor::from_slice::<Ipld>(&block) {
+            if let Some(Ipld::Link(left_cid)) = obj.get("l") {
+                to_visit.push(*left_cid);
+            }
+            if let Some(Ipld::List(entries)) = obj.get("e") {
+                to_visit.extend(
+                    entries
+                        .iter()
+                        .filter_map(|entry| match entry {
+                            Ipld::Map(entry_obj) => Some(entry_obj),
+                            _ => None,
+                        })
+                        .flat_map(|entry_obj| {
+                            [entry_obj.get("t"), entry_obj.get("v")]
+                                .into_iter()
+                                .flatten()
+                                .filter_map(|v| match v {
+                                    Ipld::Link(cid) => Some(*cid),
+                                    _ => None,
+                                })
+                        }),
+                );
+            }
+        }
+    }
+
+    Ok(block_cids)
+}
+
 pub async fn backfill_record_blobs(db: &PgPool, block_store: PostgresBlockStore) {
     let users_needing_backfill = match sqlx::query!(
         r#"
@@ -664,7 +684,7 @@ async fn process_scheduled_backups(
             }
         };
 
-        let car_result = generate_full_backup(block_store, &head_cid).await;
+        let car_result = generate_full_backup(db, block_store, user.user_id, &head_cid).await;
         let car_bytes = match car_result {
             Ok(bytes) => bytes,
             Err(e) => {
@@ -736,67 +756,105 @@ pub async fn generate_repo_car(
     head_cid: &Cid,
 ) -> Result<Vec<u8>, String> {
     use jacquard_repo::storage::BlockStore;
-    use std::io::Write;
 
-    let mut car_bytes =
+    let block_cids_bytes = collect_current_repo_blocks(block_store, head_cid).await?;
+    let block_cids: Vec<Cid> = block_cids_bytes
+        .iter()
+        .filter_map(|b| Cid::try_from(b.as_slice()).ok())
+        .collect();
+
+    let car_bytes =
         encode_car_header(head_cid).map_err(|e| format!("Failed to encode CAR header: {}", e))?;
 
-    let mut stack = vec![*head_cid];
-    let mut visited = std::collections::HashSet::new();
+    let blocks = block_store
+        .get_many(&block_cids)
+        .await
+        .map_err(|e| format!("Failed to fetch blocks: {:?}", e))?;
 
-    while let Some(cid) = stack.pop() {
-        if visited.contains(&cid) {
-            continue;
-        }
-        visited.insert(cid);
+    let car_bytes = block_cids
+        .iter()
+        .zip(blocks.iter())
+        .filter_map(|(cid, block_opt)| block_opt.as_ref().map(|block| (cid, block)))
+        .fold(car_bytes, |mut acc, (cid, block)| {
+            acc.extend(encode_car_block(cid, block));
+            acc
+        });
 
-        if let Ok(Some(block)) = block_store.get(&cid).await {
-            let cid_bytes = cid.to_bytes();
-            let total_len = cid_bytes.len() + block.len();
-            let mut writer = Vec::new();
-            crate::sync::car::write_varint(&mut writer, total_len as u64)
-                .expect("Writing to Vec<u8> should never fail");
-            writer
-                .write_all(&cid_bytes)
-                .expect("Writing to Vec<u8> should never fail");
-            writer
-                .write_all(&block)
-                .expect("Writing to Vec<u8> should never fail");
-            car_bytes.extend_from_slice(&writer);
+    Ok(car_bytes)
+}
 
-            if let Ok(value) = serde_ipld_dagcbor::from_slice::<Ipld>(&block) {
-                extract_links(&value, &mut stack);
-            }
+fn encode_car_block(cid: &Cid, block: &[u8]) -> Vec<u8> {
+    use std::io::Write;
+    let cid_bytes = cid.to_bytes();
+    let total_len = cid_bytes.len() + block.len();
+    let mut writer = Vec::new();
+    crate::sync::car::write_varint(&mut writer, total_len as u64)
+        .expect("Writing to Vec<u8> should never fail");
+    writer
+        .write_all(&cid_bytes)
+        .expect("Writing to Vec<u8> should never fail");
+    writer
+        .write_all(block)
+        .expect("Writing to Vec<u8> should never fail");
+    writer
+}
+
+pub async fn generate_repo_car_from_user_blocks(
+    db: &PgPool,
+    block_store: &PostgresBlockStore,
+    user_id: uuid::Uuid,
+    head_cid: &Cid,
+) -> Result<Vec<u8>, String> {
+    use jacquard_repo::storage::BlockStore;
+
+    let block_cid_bytes: Vec<Vec<u8>> = sqlx::query_scalar!(
+        "SELECT block_cid FROM user_blocks WHERE user_id = $1",
+        user_id
+    )
+    .fetch_all(db)
+    .await
+    .map_err(|e| format!("Failed to fetch user_blocks: {}", e))?;
+
+    if block_cid_bytes.is_empty() {
+        let cids = collect_current_repo_blocks(block_store, head_cid).await?;
+        if cids.is_empty() {
+            return Err("No blocks found for repo".to_string());
         }
+        return generate_repo_car(block_store, head_cid).await;
     }
 
+    let block_cids: Vec<Cid> = block_cid_bytes
+        .iter()
+        .filter_map(|bytes| Cid::try_from(bytes.as_slice()).ok())
+        .collect();
+
+    let car_bytes =
+        encode_car_header(head_cid).map_err(|e| format!("Failed to encode CAR header: {}", e))?;
+
+    let blocks = block_store
+        .get_many(&block_cids)
+        .await
+        .map_err(|e| format!("Failed to fetch blocks: {:?}", e))?;
+
+    let car_bytes = block_cids
+        .iter()
+        .zip(blocks.iter())
+        .filter_map(|(cid, block_opt)| block_opt.as_ref().map(|block| (cid, block)))
+        .fold(car_bytes, |mut acc, (cid, block)| {
+            acc.extend(encode_car_block(cid, block));
+            acc
+        });
+
     Ok(car_bytes)
 }
 
 pub async fn generate_full_backup(
+    db: &PgPool,
     block_store: &PostgresBlockStore,
+    user_id: uuid::Uuid,
     head_cid: &Cid,
 ) -> Result<Vec<u8>, String> {
-    generate_repo_car(block_store, head_cid).await
-}
-
-fn extract_links(value: &Ipld, stack: &mut Vec<Cid>) {
-    match value {
-        Ipld::Link(cid) => {
-            stack.push(*cid);
-        }
-        Ipld::Map(map) => {
-            for v in map.values() {
-                extract_links(v, stack);
-            }
-        }
-        Ipld::List(arr) => {
-            for v in arr {
-                extract_links(v, stack);
-            }
-        }
-        _ => {}
-    }
+    generate_repo_car_from_user_blocks(db, block_store, user_id, head_cid).await
 }
 
 pub fn count_car_blocks(car_bytes: &[u8]) -> i32 {

src/sync/repo.rs

@@ -1,4 +1,5 @@
 use crate::api::error::ApiError;
+use crate::scheduled::generate_repo_car_from_user_blocks;
 use crate::state::AppState;
 use crate::sync::car::encode_car_header;
 use crate::sync::util::assert_repo_availability;
@@ -8,15 +9,12 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use cid::Cid;
-use ipld_core::ipld::Ipld;
 use jacquard_repo::storage::BlockStore;
 use serde::Deserialize;
 use std::io::Write;
 use std::str::FromStr;
 use tracing::error;
 
-const MAX_REPO_BLOCKS_TRAVERSAL: usize = 20_000;
-
 fn parse_get_blocks_query(query_string: &str) -> Result<(String, Vec<String>), String> {
     let did = crate::util::parse_repeated_query_param(Some(query_string), "did")
         .into_iter()
@@ -138,43 +136,21 @@ pub async fn get_repo(
         return get_repo_since(&state, &query.did, &head_cid, since).await;
     }
 
-    let mut car_bytes = match encode_car_header(&head_cid) {
-        Ok(h) => h,
+    let car_bytes = match generate_repo_car_from_user_blocks(
+        &state.db,
+        &state.block_store,
+        account.user_id,
+        &head_cid,
+    )
+    .await
+    {
+        Ok(bytes) => bytes,
         Err(e) => {
-            error!("Failed to encode CAR header: {}", e);
+            error!("Failed to generate repo CAR: {}", e);
             return ApiError::InternalError(None).into_response();
         }
     };
-    let mut stack = vec![head_cid];
-    let mut visited = std::collections::HashSet::new();
-    let mut remaining = MAX_REPO_BLOCKS_TRAVERSAL;
-    while let Some(cid) = stack.pop() {
-        if visited.contains(&cid) {
-            continue;
-        }
-        visited.insert(cid);
-        if remaining == 0 {
-            break;
-        }
-        remaining -= 1;
-        if let Ok(Some(block)) = state.block_store.get(&cid).await {
-            let cid_bytes = cid.to_bytes();
-            let total_len = cid_bytes.len() + block.len();
-            let mut writer = Vec::new();
-            crate::sync::car::write_varint(&mut writer, total_len as u64)
-                .expect("Writing to Vec<u8> should never fail");
-            writer
-                .write_all(&cid_bytes)
-                .expect("Writing to Vec<u8> should never fail");
-            writer
-                .write_all(&block)
-                .expect("Writing to Vec<u8> should never fail");
-            car_bytes.extend_from_slice(&writer);
-            if let Ok(value) = serde_ipld_dagcbor::from_slice::<Ipld>(&block) {
-                extract_links_ipld(&value, &mut stack);
-            }
-        }
-    }
+
     (
         StatusCode::OK,
         [(axum::http::header::CONTENT_TYPE, "application/vnd.ipld.car")],
@@ -275,25 +251,6 @@ async fn get_repo_since(state: &AppState, did: &str, head_cid: &Cid, since: &str
         .into_response()
 }
 
-fn extract_links_ipld(value: &Ipld, stack: &mut Vec<Cid>) {
-    match value {
-        Ipld::Link(cid) => {
-            stack.push(*cid);
-        }
-        Ipld::Map(map) => {
-            for v in map.values() {
-                extract_links_ipld(v, stack);
-            }
-        }
-        Ipld::List(arr) => {
-            for v in arr {
-                extract_links_ipld(v, stack);
-            }
-        }
-        _ => {}
-    }
-}
-
 #[derive(Deserialize)]
 pub struct GetRecordQuery {
     pub did: String,

tests/account_lifecycle.rs

@@ -93,11 +93,17 @@ async fn test_check_account_status_returns_correct_block_count() {
     let body3: Value = status3.json().await.unwrap();
     let after_delete_blocks = body3["repoBlocks"].as_i64().unwrap();
     assert!(
-        after_delete_blocks >= after_create_blocks,
-        "Block count should not decrease after deleting a record (was {}, now {})",
+        after_delete_blocks <= after_create_blocks,
+        "Block count should decrease or stay same after deleting a record (was {}, now {})",
         after_create_blocks,
         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
+    );
 }
 
 #[tokio::test]

tests/delete_account.rs

@@ -174,7 +174,7 @@ async fn test_delete_account_invalid_token() {
         .send()
         .await
         .expect("Failed to send delete request");
-    assert_eq!(delete_res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(delete_res.status(), StatusCode::UNAUTHORIZED);
     let body: Value = delete_res.json().await.unwrap();
     assert_eq!(body["error"], "InvalidToken");
 }
@@ -228,7 +228,7 @@ async fn test_delete_account_expired_token() {
         .send()
         .await
         .expect("Failed to send delete request");
-    assert_eq!(delete_res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(delete_res.status(), StatusCode::UNAUTHORIZED);
     let body: Value = delete_res.json().await.unwrap();
     assert_eq!(body["error"], "ExpiredToken");
 }
@@ -280,7 +280,7 @@ async fn test_delete_account_token_mismatch() {
         .send()
         .await
         .expect("Failed to send delete request");
-    assert_eq!(delete_res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(delete_res.status(), StatusCode::UNAUTHORIZED);
     let body: Value = delete_res.json().await.unwrap();
     assert_eq!(body["error"], "InvalidToken");
 }

tests/email_update.rs

@@ -193,7 +193,7 @@ async fn test_update_email_invalid_token() {
         .send()
         .await
         .expect("Failed to attempt email update");
-    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
     let body: Value = res.json().await.expect("Invalid JSON");
     assert_eq!(body["error"], "InvalidToken");
 }
@@ -390,7 +390,7 @@ async fn test_confirm_email_invalid_token() {
         .send()
         .await
         .expect("Failed to confirm email");
-    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
     let body: Value = res.json().await.expect("Invalid JSON");
     assert_eq!(body["error"], "InvalidToken");
 }

tests/oauth.rs

@@ -261,14 +261,15 @@ async fn test_full_oauth_flow() {
             .to_string();
     }
     assert!(
-        location.starts_with(redirect_uri),
-        "Redirect to wrong URI: {}",
+        location.contains("code="),
+        "No code in redirect URI: {}",
         location
     );
-    assert!(location.contains("code="), "No code in redirect");
     assert!(
-        location.contains(&format!("state={}", state)),
-        "Wrong state"
+        location.contains(&format!("state={}", state))
+            || location.contains(&format!("state%3D{}", state)),
+        "Wrong state in redirect: {}",
+        location
     );
     let code = location
         .split("code=")
@@ -527,7 +528,11 @@ async fn test_oauth_2fa_flow() {
     );
     let twofa_body: Value = twofa_res.json().await.unwrap();
     let final_location = twofa_body["redirect_uri"].as_str().unwrap();
-    assert!(final_location.starts_with(redirect_uri) && final_location.contains("code="));
+    assert!(
+        final_location.contains("code="),
+        "No code in redirect URI: {}",
+        final_location
+    );
     let auth_code = final_location
         .split("code=")
         .nth(1)
@@ -805,7 +810,11 @@ async fn test_account_selector_with_2fa() {
     );
     let twofa_body: Value = twofa_res.json().await.unwrap();
     let final_location = twofa_body["redirect_uri"].as_str().unwrap();
-    assert!(final_location.starts_with(redirect_uri) && final_location.contains("code="));
+    assert!(
+        final_location.contains("code="),
+        "No code in redirect URI: {}",
+        final_location
+    );
     let final_code = final_location
         .split("code=")
         .nth(1)

tests/password_reset.rs

@@ -177,7 +177,7 @@ async fn test_reset_password_with_invalid_token() {
         .send()
         .await
         .expect("Failed to reset password");
-    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
     let body: Value = res.json().await.expect("Invalid JSON");
     assert_eq!(body["error"], "InvalidToken");
 }
@@ -241,7 +241,7 @@ async fn test_reset_password_with_expired_token() {
         .send()
         .await
         .expect("Failed to reset password");
-    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
     let body: Value = res.json().await.expect("Invalid JSON");
     assert_eq!(body["error"], "ExpiredToken");
 }

tests/plc_operations.rs

@@ -76,7 +76,7 @@ async fn test_sign_plc_operation_validation() {
         .send()
         .await
         .unwrap();
-    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
     let body: serde_json::Value = res.json().await.unwrap();
     assert!(body["error"] == "InvalidToken" || body["error"] == "ExpiredToken");
 }