fix(core): mutations are non-transactional, leaving orphaned/partial data

.changeset/atomic-mutations.md

@@ -0,0 +1,9 @@
+---
+"@keystone-6/core": patch
+---
+
+`create`, `update` and `delete` mutations now run their write inside a database transaction, so each item is written atomically. Previously a mutation with nested creates could leave orphaned related rows behind (or a partially-applied item) if a nested write, a `beforeOperation`/`validate` hook, or a database constraint failed partway through — now the whole write rolls back cleanly.
+
+For the `*Many` mutations (`createMany`, `updateMany`, `deleteMany`) each item is written in its own transaction and the items are processed one at a time, so a single failing item rolls back on its own while the rest of the batch still succeeds — the existing per-item partial-success behaviour is unchanged.
+
+`afterOperation` hooks are unaffected: they still run after the write has committed, and an error thrown from an `afterOperation` hook still does not roll the write back.

packages/core/src/lib/context/createContext.ts

@@ -51,13 +51,16 @@ export function createContext({
     session,
     internal,
     sudo,
+    transactional = false,
   }: {
     prisma: any
     req?: IncomingMessage
     res?: ServerResponse
     session?: unknown
     internal: boolean
     sudo: boolean
+    // true when this context is already bound to an open interactive transaction
+    transactional?: boolean
   }) => {
     const schema = internal ? graphQLSchemas.internal : graphQLSchemas.public
     const rawGraphQL: KeystoneGraphQLAPI['raw'] = async ({ query, variables }) => {
@@ -83,6 +86,12 @@ export function createContext({
       graphql: { raw: rawGraphQL, run: runGraphQL, schema },
 
       transaction: async (f, opts) => {
+        // already inside an interactive transaction: join it rather than opening
+        // a nested one (Prisma does not support nested interactive transactions).
+        // This keeps mutations - which now open a transaction of their own -
+        // working when they are called from within `context.transaction(...)`.
+        if (transactional) return await f(context)
+
         return await prisma.$transaction(async (prisma_: any) => {
           const newContext = construct({
             prisma: prisma_,
@@ -91,6 +100,7 @@ export function createContext({
             session,
             internal,
             sudo,
+            transactional: true,
           })
 
           return await f(newContext)
@@ -110,19 +120,21 @@ export function createContext({
           session,
           internal,
           sudo,
+          transactional,
         })
         return newContext.withSession(
           (await config.session?.get({ context: newContext })) ?? undefined
         )
       },
 
       withSession: session => {
-        return construct({ prisma, req, res, session, internal, sudo })
+        return construct({ prisma, req, res, session, internal, sudo, transactional })
       },
 
       // privilege escalation
-      internal: () => construct({ prisma, req, res, session, internal: true, sudo }),
-      sudo: () => construct({ prisma, req, res, session, internal: true, sudo: true }),
+      internal: () => construct({ prisma, req, res, session, internal: true, sudo, transactional }),
+      sudo: () =>
+        construct({ prisma, req, res, session, internal: true, sudo: true, transactional }),
 
       __internal: {
         sudo,

packages/core/src/lib/core/mutations/index.ts

@@ -109,7 +109,11 @@ async function createSingle__(
 }
 
 export class NestedMutationState {
-  #afterOperations: (() => void | Promise<void>)[] = []
+  // afterOperation hooks run once the enclosing transaction has committed, so
+  // they are deferred here and later invoked with a post-commit context (see
+  // `afterOperation` below) rather than the transaction-bound context they were
+  // created with.
+  #afterOperations: ((context: KeystoneContext) => void | Promise<void>)[] = []
   #context: KeystoneContext
   constructor(context: KeystoneContext) {
     this.#context = context
@@ -124,12 +128,12 @@ export class NestedMutationState {
     const { item, afterOperation } = await createSingle__(data, list, context)
 
     // after operation
-    this.#afterOperations.push(() => afterOperation(item))
+    this.#afterOperations.push(afterContext => afterOperation(afterContext, item))
     return { id: item.id as IdType }
   }
 
-  async afterOperation() {
-    await promiseAllRejectWithAllErrors(this.#afterOperations.map(async x => x()))
+  async afterOperation(context: KeystoneContext) {
+    await promiseAllRejectWithAllErrors(this.#afterOperations.map(async x => x(context)))
   }
 }
 
@@ -152,34 +156,43 @@ async function updateSingle__(
   return await withSpan(
     `update ${list.graphql.names.outputTypeNameLower}`,
     async span => {
-      // validate and resolve the input filter
-      const uniqueWhere = await resolveUniqueWhereInput(where, list, context)
-
-      // filter and item access control - throws an AccessDeniedError if not allowed
-      const item = await getFilteredItem(list, context, uniqueWhere!, accessFilters, 'update')
-
-      // throw an accessDeniedError if not allowed
-      await enforceListLevelAccessControl(context, 'update', list, inputData ?? {}, item)
-      await enforceFieldLevelAccessControl(context, 'update', list, inputData ?? {}, item)
-      const { beforeOperation, afterOperation, data } = await resolveInputForCreateOrUpdate(
-        list,
-        context,
-        inputData ?? {},
-        item
-      )
-
-      // before operation
-      await beforeOperation()
-
-      // operation
-      const result = await context.prisma[list.listKey].update({
-        where: { id: item.id },
-        data,
+      // run the access-filtered read, any nested writes, and the update itself
+      // atomically, so a failure at any step rolls back cleanly instead of
+      // leaving a partially-written item behind. The transaction-bound context
+      // is shadowed as `context` so every call inside runs against it.
+      const { result, afterOperation } = await context.transaction(async context => {
+        // validate and resolve the input filter
+        const uniqueWhere = await resolveUniqueWhereInput(where, list, context)
+
+        // filter and item access control - throws an AccessDeniedError if not allowed
+        const item = await getFilteredItem(list, context, uniqueWhere!, accessFilters, 'update')
+
+        // throw an accessDeniedError if not allowed
+        await enforceListLevelAccessControl(context, 'update', list, inputData ?? {}, item)
+        await enforceFieldLevelAccessControl(context, 'update', list, inputData ?? {}, item)
+        const { beforeOperation, afterOperation, data } = await resolveInputForCreateOrUpdate(
+          list,
+          context,
+          inputData ?? {},
+          item
+        )
+
+        // before operation
+        await beforeOperation()
+
+        // operation
+        const result = await context.prisma[list.listKey].update({
+          where: { id: item.id },
+          data,
+        })
+        span.setAttribute('keystone.result.id', result?.id ?? '')
+
+        return { result, afterOperation }
       })
-      span.setAttribute('keystone.result.id', result?.id ?? '')
 
-      // after operation
-      await afterOperation(result)
+      // after operation - runs once the transaction has committed, using the
+      // original (post-commit) context; an error here does not roll the write back
+      await afterOperation(context, result)
 
       return result
     },
@@ -196,39 +209,51 @@ async function deleteSingle__(
   return await withSpan(
     `delete ${list.graphql.names.outputTypeNameLower}`,
     async span => {
-      // validate and resolve the input filter
-      const uniqueWhere = await resolveUniqueWhereInput(where, list, context)
+      // run the access-filtered read, the validate/beforeOperation hooks, and the
+      // delete itself atomically. The transaction-bound context is shadowed as
+      // `context` so every call inside runs against it.
+      const { result, item } = await context.transaction(async context => {
+        // validate and resolve the input filter
+        const uniqueWhere = await resolveUniqueWhereInput(where, list, context)
+
+        // filter and item access control throw an AccessDeniedError if not allowed
+        // apply access.filter.* controls
+        const item = await getFilteredItem(list, context, uniqueWhere!, accessFilters, 'delete')
+
+        await enforceListLevelAccessControl(context, 'delete', list, {}, item)
+        // WARNING: no field level access control for delete operations
+
+        const hookArgs = {
+          operation: 'delete' as const,
+          listKey: list.listKey,
+          context,
+          item,
+          resolvedData: undefined,
+          inputData: undefined,
+        }
 
-      // filter and item access control throw an AccessDeniedError if not allowed
-      // apply access.filter.* controls
-      const item = await getFilteredItem(list, context, uniqueWhere!, accessFilters, 'delete')
+        // hooks
+        await validate({ list, hookArgs })
 
-      await enforceListLevelAccessControl(context, 'delete', list, {}, item)
-      // WARNING: no field level access control for delete operations
+        // before operation
+        await runSideEffectOnlyHook(list, 'beforeOperation', hookArgs)
+
+        // operation
+        const result = await context.prisma[list.listKey].delete({ where: { id: item.id } })
+        span.setAttribute('keystone.result.id', result?.id ?? '')
+
+        return { result, item }
+      })
 
-      const hookArgs = {
+      // after operation - runs once the transaction has committed, using the
+      // original (post-commit) context; an error here does not roll the delete back
+      await runSideEffectOnlyHook(list, 'afterOperation', {
         operation: 'delete' as const,
         listKey: list.listKey,
         context,
-        item,
+        item: undefined,
         resolvedData: undefined,
         inputData: undefined,
-      }
-
-      // hooks
-      await validate({ list, hookArgs })
-
-      // before operation
-      await runSideEffectOnlyHook(list, 'beforeOperation', hookArgs)
-
-      // operation
-      const result = await context.prisma[list.listKey].delete({ where: { id: item.id } })
-      span.setAttribute('keystone.result.id', result?.id ?? '')
-
-      // after operation
-      await runSideEffectOnlyHook(list, 'afterOperation', {
-        ...hookArgs,
-        item: undefined,
         originalItem: item,
       })
 
@@ -270,6 +295,28 @@ async function actionSingle__(
 
 //
 
+// create a single item and its nested writes atomically: the item and every
+// nested create it triggers are written inside one transaction, so a failure at
+// any point rolls all of them back instead of leaving orphaned related rows
+// behind. `createSingle__` is also used for nested creates (where it must run
+// inside the parent's transaction), so the transaction is opened here at the top
+// level rather than inside `createSingle__` to avoid nested interactive
+// transactions. afterOperation hooks run after the transaction has committed,
+// using the original (post-commit) `context` - consistent with the existing
+// behaviour where an error thrown from afterOperation does not roll back the
+// write.
+async function createSingleInTransaction(
+  inputData: InputData,
+  list: InitialisedList,
+  context: KeystoneContext
+) {
+  const { item, afterOperation } = await context.transaction(async context =>
+    createSingle__(inputData ?? {}, list, context)
+  )
+  await afterOperation(context, item)
+  return item
+}
+
 async function createOne(inputData: InputData, list: InitialisedList, context: KeystoneContext) {
   const operationAccess = await getOperationAccess(list, context, 'create')
   if (!operationAccess) throw accessDeniedError(cannotForItem('create', list))
@@ -278,12 +325,7 @@ async function createOne(inputData: InputData, list: InitialisedList, context: K
   //   NOTHING - no filters for create operations
 
   // operation
-  const { item, afterOperation } = await createSingle__(inputData ?? {}, list, context)
-
-  // after operation // TODO: move to createSingle__
-  await afterOperation(item)
-
-  return item
+  return createSingleInTransaction(inputData, list, context)
 }
 
 async function createMany(
@@ -297,17 +339,16 @@ async function createMany(
   // get list-level access control filters
   //   NOTHING - no filters for create operations
 
-  return inputDatas.map(async inputData => {
+  // each item is created in its own transaction and is independent of the
+  // others (partial success is preserved), but items are run one at a time so we
+  // never hold multiple interactive transactions open at once - SQLite is a
+  // single-writer database and cannot service concurrent write transactions.
+  return runSequentially(inputDatas, async inputData => {
     // throw for each attempt
     if (!operationAccess) throw accessDeniedError(cannotForItem('create', list))
 
     // operation
-    const { item, afterOperation } = await createSingle__(inputData ?? {}, list, context)
-
-    // after operation // TODO: move to createSingle__
-    await afterOperation(item)
-
-    return item
+    return createSingleInTransaction(inputData, list, context)
   })
 }
 
@@ -336,7 +377,9 @@ async function updateMany(
   // get list-level access control filters
   const accessFilters = await getAccessFilters(list, context, 'update')
 
-  return updateManyInput.map(async updateInput => {
+  // each item is updated in its own transaction (see createMany for why they run
+  // one at a time); partial success across the batch is preserved.
+  return runSequentially(updateManyInput, async updateInput => {
     // throw for each attempt
     if (!operationAccess) throw accessDeniedError(cannotForItem('update', list))
 
@@ -369,7 +412,9 @@ async function deleteMany(
   // get list-level access control filters
   const accessFilters = await getAccessFilters(list, context, 'delete')
 
-  return wheres.map(async where => {
+  // each item is deleted in its own transaction (see createMany for why they run
+  // one at a time); partial success across the batch is preserved.
+  return runSequentially(wheres, async where => {
     // throw for each attempt
     if (!operationAccess) throw accessDeniedError(cannotForItem('delete', list))
 
@@ -609,12 +654,17 @@ async function resolveInputForCreateOrUpdate(
       // before operation
       await runSideEffectOnlyHook(list, 'beforeOperation', hookArgs)
     },
-    afterOperation: async (updatedItem: BaseItem) => {
-      await nestedMutationState.afterOperation()
+    // `afterContext` is the context afterOperation hooks run against. The write
+    // happens inside a transaction, but afterOperation hooks run once that
+    // transaction has committed, so they are handed a post-commit context rather
+    // than the (now closed) transaction-bound context used for the write.
+    afterOperation: async (afterContext: KeystoneContext, updatedItem: BaseItem) => {
+      await nestedMutationState.afterOperation(afterContext)
 
       // after operation
       await runSideEffectOnlyHook(list, 'afterOperation', {
         ...hookArgs,
+        context: afterContext,
         item: updatedItem,
       })
     },
@@ -665,6 +715,30 @@ function transformForPrismaClient(
   ])
 }
 
+// Run each item through `fn` one at a time (never concurrently), returning an
+// array of per-item promises. Partial-success semantics are preserved: a rejected
+// item settles independently and never prevents later items from running. Running
+// sequentially means we only ever hold one interactive transaction open at a time,
+// which is required for SQLite (a single-writer database) and harmless elsewhere.
+function runSequentially<T, R>(
+  items: T[],
+  fn: (item: T, index: number) => Promise<R>
+): Promise<R>[] {
+  const results: Promise<R>[] = []
+  // a promise that always resolves (never rejects), used to gate the next item on
+  // the previous one having settled regardless of its outcome
+  let gate: Promise<unknown> = Promise.resolve()
+  for (const [index, item] of items.entries()) {
+    const result = gate.then(() => fn(item, index))
+    gate = result.then(
+      () => {},
+      () => {}
+    )
+    results.push(result)
+  }
+  return results
+}
+
 // This is not a thing that I really agree with but it's to make the behaviour consistent with old keystone.
 // Basically, old keystone uses Promise.allSettled and then after that maps that into promises that resolve and reject,
 // whereas the new stuff is just like "here are some promises" with no guarantees about the order they will be settled in.