Move sequenceCommit into actor store transaction (#3580)

* serialize sequencing by moving sequenceCommit into actor store transaction

* do not throw on updateRepoRoot failure

* build branch

* changset

* dont build branch

.changeset/good-dryers-poke.md

@@ -0,0 +1,5 @@
+---
+"@atproto/pds": patch
+---
+
+Fix bug where racing writes to the same repository can get sequenced out-of-order.

packages/pds/src/api/com/atproto/repo/applyWrites.ts

@@ -12,6 +12,7 @@ import {
   isDelete,
   isUpdate,
 } from '../../../../lexicon/types/com/atproto/repo/applyWrites'
+import { dbLogger } from '../../../../logger'
 import {
   BadCommitSwapError,
   InvalidRecordError,
@@ -119,19 +120,28 @@ export default function (server: Server, ctx: AppContext) {
       const swapCommitCid = swapCommit ? CID.parse(swapCommit) : undefined
 
       const commit = await ctx.actorStore.transact(did, async (actorTxn) => {
-        try {
-          return await actorTxn.repo.processWrites(writes, swapCommitCid)
-        } catch (err) {
-          if (err instanceof BadCommitSwapError) {
-            throw new InvalidRequestError(err.message, 'InvalidSwap')
-          } else {
-            throw err
-          }
-        }
+        const commit = await actorTxn.repo
+          .processWrites(writes, swapCommitCid)
+          .catch((err) => {
+            if (err instanceof BadCommitSwapError) {
+              throw new InvalidRequestError(err.message, 'InvalidSwap')
+            } else {
+              throw err
+            }
+          })
+
+        await ctx.sequencer.sequenceCommit(did, commit)
+        return commit
       })
 
-      await ctx.sequencer.sequenceCommit(did, commit)
-      await ctx.accountManager.updateRepoRoot(did, commit.cid, commit.rev)
+      await ctx.accountManager
+        .updateRepoRoot(did, commit.cid, commit.rev)
+        .catch((err) => {
+          dbLogger.error(
+            { err, did, cid: commit.cid, rev: commit.rev },
+            'failed to update account root',
+          )
+        })
 
       return {
         encoding: 'application/json',

packages/pds/src/api/com/atproto/repo/createRecord.ts

@@ -3,6 +3,7 @@ import { InvalidRecordKeyError } from '@atproto/syntax'
 import { AuthRequiredError, InvalidRequestError } from '@atproto/xrpc-server'
 import { AppContext } from '../../../../context'
 import { Server } from '../../../../lexicon'
+import { dbLogger } from '../../../../logger'
 import {
   BadCommitSwapError,
   InvalidRecordError,
@@ -82,22 +83,26 @@ export default function (server: Server, ctx: AppContext) {
           }),
         )
         const writes = [...backlinkDeletions, write]
-        try {
-          const commit = await actorTxn.repo.processWrites(
-            writes,
-            swapCommitCid,
-          )
-          return commit
-        } catch (err) {
-          if (err instanceof BadCommitSwapError) {
-            throw new InvalidRequestError(err.message, 'InvalidSwap')
-          }
-          throw err
-        }
+        const commit = await actorTxn.repo
+          .processWrites(writes, swapCommitCid)
+          .catch((err) => {
+            if (err instanceof BadCommitSwapError) {
+              throw new InvalidRequestError(err.message, 'InvalidSwap')
+            }
+            throw err
+          })
+        await ctx.sequencer.sequenceCommit(did, commit)
+        return commit
       })
 
-      await ctx.sequencer.sequenceCommit(did, commit)
-      await ctx.accountManager.updateRepoRoot(did, commit.cid, commit.rev)
+      await ctx.accountManager
+        .updateRepoRoot(did, commit.cid, commit.rev)
+        .catch((err) => {
+          dbLogger.error(
+            { err, did, cid: commit.cid, rev: commit.rev },
+            'failed to update account root',
+          )
+        })
 
       return {
         encoding: 'application/json',

packages/pds/src/api/com/atproto/repo/deleteRecord.ts

@@ -2,6 +2,7 @@ import { CID } from 'multiformats/cid'
 import { AuthRequiredError, InvalidRequestError } from '@atproto/xrpc-server'
 import { AppContext } from '../../../../context'
 import { Server } from '../../../../lexicon'
+import { dbLogger } from '../../../../logger'
 import {
   BadCommitSwapError,
   BadRecordSwapError,
@@ -56,24 +57,35 @@ export default function (server: Server, ctx: AppContext) {
         if (!record) {
           return null // No-op if record already doesn't exist
         }
-        try {
-          return await actorTxn.repo.processWrites([write], swapCommitCid)
-        } catch (err) {
-          if (
-            err instanceof BadCommitSwapError ||
-            err instanceof BadRecordSwapError
-          ) {
-            throw new InvalidRequestError(err.message, 'InvalidSwap')
-          } else {
-            throw err
-          }
-        }
+
+        const commit = await actorTxn.repo
+          .processWrites([write], swapCommitCid)
+          .catch((err) => {
+            if (
+              err instanceof BadCommitSwapError ||
+              err instanceof BadRecordSwapError
+            ) {
+              throw new InvalidRequestError(err.message, 'InvalidSwap')
+            } else {
+              throw err
+            }
+          })
+
+        await ctx.sequencer.sequenceCommit(did, commit)
+        return commit
       })
 
       if (commit !== null) {
-        await ctx.sequencer.sequenceCommit(did, commit)
-        await ctx.accountManager.updateRepoRoot(did, commit.cid, commit.rev)
+        await ctx.accountManager
+          .updateRepoRoot(did, commit.cid, commit.rev)
+          .catch((err) => {
+            dbLogger.error(
+              { err, did, cid: commit.cid, rev: commit.rev },
+              'failed to update account root',
+            )
+          })
       }
+
       return {
         encoding: 'application/json',
         body: {

packages/pds/src/api/com/atproto/repo/putRecord.ts

@@ -7,10 +7,10 @@ import { AppContext } from '../../../../context'
 import { Server } from '../../../../lexicon'
 import { ids } from '../../../../lexicon/lexicons'
 import { Record as ProfileRecord } from '../../../../lexicon/types/app/bsky/actor/profile'
+import { dbLogger } from '../../../../logger'
 import {
   BadCommitSwapError,
   BadRecordSwapError,
-  CommitDataWithOps,
   InvalidRecordError,
   PreparedCreate,
   PreparedUpdate,
@@ -104,26 +104,34 @@ export default function (server: Server, ctx: AppContext) {
             }
           }
 
-          let commit: CommitDataWithOps
-          try {
-            commit = await actorTxn.repo.processWrites([write], swapCommitCid)
-          } catch (err) {
-            if (
-              err instanceof BadCommitSwapError ||
-              err instanceof BadRecordSwapError
-            ) {
-              throw new InvalidRequestError(err.message, 'InvalidSwap')
-            } else {
-              throw err
-            }
-          }
+          const commit = await actorTxn.repo
+            .processWrites([write], swapCommitCid)
+            .catch((err) => {
+              if (
+                err instanceof BadCommitSwapError ||
+                err instanceof BadRecordSwapError
+              ) {
+                throw new InvalidRequestError(err.message, 'InvalidSwap')
+              } else {
+                throw err
+              }
+            })
+
+          await ctx.sequencer.sequenceCommit(did, commit)
+
           return { commit, write }
         },
       )
 
       if (commit !== null) {
-        await ctx.sequencer.sequenceCommit(did, commit)
-        await ctx.accountManager.updateRepoRoot(did, commit.cid, commit.rev)
+        await ctx.accountManager
+          .updateRepoRoot(did, commit.cid, commit.rev)
+          .catch((err) => {
+            dbLogger.error(
+              { err, did, cid: commit.cid, rev: commit.rev },
+              'failed to update account root',
+            )
+          })
       }
 
       return {