Fix missing delete blocks (bluesky-social#3033)

* send relevant blocks on commit

* use relevantBlocks in pds

* test

* fix compiler errors

* send both new & relevant blocks

* build branch

* changsets

* no build branch

Author: dholms

.changeset/lovely-ghosts-beam.md

@@ -0,0 +1,5 @@
+---
+"@atproto/pds": patch
+---
+
+Ensure we emit all relevant proof blocks for commit ops

.changeset/twenty-worms-eat.md

@@ -0,0 +1,5 @@
+---
+"@atproto/repo": minor
+---
+
+Add relevant proof blocks to commit data

.github/workflows/build-and-push-pds-ghcr.yaml

@@ -3,7 +3,6 @@ on:
   push:
     branches:
       - main
-      - msieben/micro-optimizations
 env:
   REGISTRY: ghcr.io
   USERNAME: ${{ github.actor }}

packages/pds/src/actor-store/repo/transactor.ts

@@ -128,12 +128,12 @@ export class RepoTransactor extends RepoReader {
 
     // find blocks that are relevant to ops but not included in diff
     // (for instance a record that was moved but cid stayed the same)
-    const newRecordBlocks = commit.newBlocks.getMany(newRecordCids)
+    const newRecordBlocks = commit.relevantBlocks.getMany(newRecordCids)
     if (newRecordBlocks.missing.length > 0) {
       const missingBlocks = await this.storage.getBlocks(
         newRecordBlocks.missing,
       )
-      commit.newBlocks.addMap(missingBlocks.blocks)
+      commit.relevantBlocks.addMap(missingBlocks.blocks)
     }
     return commit
   }

packages/pds/src/api/com/atproto/server/activateAccount.ts

@@ -50,6 +50,7 @@ export default function (server: Server, ctx: AppContext) {
           since: null,
           prev: null,
           newBlocks: blocks.blocks,
+          relevantBlocks: blocks.blocks,
           removedCids: new CidSet(),
         }
       })

packages/pds/src/scripts/rebuild-repo.ts

@@ -73,6 +73,7 @@ export const rebuildRepo = async (ctx: AppContext, args: string[]) => {
       since: null,
       prev: null,
       newBlocks,
+      relevantBlocks: newBlocks,
       removedCids: toDelete,
     }
   })

packages/pds/src/sequencer/events.ts

@@ -22,11 +22,15 @@ export const formatSeqCommit = async (
   const blobs = new CidSet()
   let carSlice: Uint8Array
 
+  const blocksToSend = new BlockMap()
+  blocksToSend.addMap(commitData.newBlocks)
+  blocksToSend.addMap(commitData.relevantBlocks)
+
   // max 200 ops or 1MB of data
-  if (writes.length > 200 || commitData.newBlocks.byteSize > 1000000) {
+  if (writes.length > 200 || blocksToSend.byteSize > 1000000) {
     tooBig = true
     const justRoot = new BlockMap()
-    const rootBlock = commitData.newBlocks.get(commitData.cid)
+    const rootBlock = blocksToSend.get(commitData.cid)
     if (rootBlock) {
       justRoot.set(commitData.cid, rootBlock)
     }
@@ -46,7 +50,7 @@ export const formatSeqCommit = async (
       }
       ops.push({ action: w.action, path, cid })
     }
-    carSlice = await blocksToCarFile(commitData.cid, commitData.newBlocks)
+    carSlice = await blocksToCarFile(commitData.cid, blocksToSend)
   }
 
   const evt: CommitEvt = {

packages/repo/src/mst/mst.ts

@@ -756,6 +756,20 @@ export class MST {
     return cids
   }
 
+  async addBlocksForPath(key: string, blocks: BlockMap) {
+    const serialized = await this.serialize()
+    blocks.set(serialized.cid, serialized.bytes)
+    const index = await this.findGtOrEqualLeafIndex(key)
+    const found = await this.atIndex(index)
+    if (found && found.isLeaf() && found.key === key) {
+      return
+    }
+    const prev = await this.atIndex(index - 1)
+    if (prev && prev.isTree()) {
+      await prev.addBlocksForPath(key, blocks)
+    }
+  }
+
   // Matching Leaf interface
   // -------------------
 

packages/repo/src/repo.ts

@@ -1,6 +1,7 @@
 import { CID } from 'multiformats/cid'
-import { TID } from '@atproto/common'
+import { dataToCborBlock, TID } from '@atproto/common'
 import * as crypto from '@atproto/crypto'
+import { lexToIpld } from '@atproto/lexicon'
 import {
   Commit,
   CommitData,
@@ -69,6 +70,7 @@ export class Repo extends ReadableRepo {
       since: null,
       prev: null,
       newBlocks,
+      relevantBlocks: newBlocks,
       removedCids: diff.removedCids,
     }
   }
@@ -140,11 +142,22 @@ export class Repo extends ReadableRepo {
     const newBlocks = diff.newMstBlocks
     const removedCids = diff.removedCids
 
+    const relevantBlocks = new BlockMap()
+    await Promise.all(
+      writes.map((op) =>
+        data.addBlocksForPath(
+          util.formatDataKey(op.collection, op.rkey),
+          relevantBlocks,
+        ),
+      ),
+    )
+
     const addedLeaves = leaves.getMany(diff.newLeafCids.toList())
     if (addedLeaves.missing.length > 0) {
       throw new Error(`Missing leaf blocks: ${addedLeaves.missing}`)
     }
     newBlocks.addMap(addedLeaves.blocks)
+    relevantBlocks.addMap(addedLeaves.blocks)
 
     const rev = TID.nextStr(this.commit.rev)
     const commit = await util.signCommit(
@@ -157,21 +170,20 @@ export class Repo extends ReadableRepo {
       },
       keypair,
     )
-    const commitCid = await newBlocks.add(commit)
-
-    // ensure the commit cid actually changed
-    if (commitCid.equals(this.cid)) {
-      newBlocks.delete(commitCid)
-    } else {
+    const commitBlock = await dataToCborBlock(lexToIpld(commit))
+    if (!commitBlock.cid.equals(this.cid)) {
+      newBlocks.set(commitBlock.cid, commitBlock.bytes)
+      relevantBlocks.set(commitBlock.cid, commitBlock.bytes)
       removedCids.add(this.cid)
     }
 
     return {
-      cid: commitCid,
+      cid: commitBlock.cid,
       rev,
       since: this.commit.rev,
       prev: this.cid,
       newBlocks,
+      relevantBlocks,
       removedCids,
     }
   }
@@ -208,6 +220,7 @@ export class Repo extends ReadableRepo {
       since: null,
       prev: null,
       newBlocks,
+      relevantBlocks: newBlocks,
       removedCids: new CidSet([this.cid]),
     }
   }

packages/repo/src/sync/consumer.ts

@@ -93,6 +93,7 @@ export const verifyDiff = async (
       prev: repo?.cid ?? null,
       since: repo?.commit.rev ?? null,
       newBlocks,
+      relevantBlocks: newBlocks,
       removedCids,
     },
   }