fix: transaction timeout rolls back applied operations (no torn state)

Transaction.execute() checked its time budget at the top of the operation
loop and threw TransactionTimeoutError from OUTSIDE the per-operation
try/catch, so a mid-flight timeout bypassed rollback entirely — only
per-operation failures rolled back. A bulk transact that crossed its 30s
budget mid-flight left the operations already applied to canonical storage
in place while the generation was never stamped: torn, generation-less
state. The generation-store commit path's abort cleanup explicitly assumes a
throw from execute() already restored the applied operations byte-identically
(it only discards the uncommitted staging directory), so the missing
rollback broke that invariant.

Give execute() a single rollback point: the operation loop is the sole
rollback-guarded region, and any error escaping it — an operation failure OR
a mid-flight timeout — rolls back every applied operation in reverse order,
then surfaces the original error (a rollback failure still supersedes it via
TransactionRollbackError). The per-exit-path rollback that let the timeout
throw slip past is gone; atomicity now holds by construction for every error
type. An aborted transaction leaves generation() unchanged and storage
byte-identical to its pre-transaction state.

Regression (tests/unit/transaction/timeout-rollback.test.ts): a mid-flight
timeout leaves an in-memory canonical store byte-identical with the tx in the
rolled_back terminal state; the operation-failure and rollback-failure paths
through the same single rollback point; and a clean transaction still commits.

src/transaction/Transaction.ts

@@ -98,49 +98,60 @@ export class Transaction implements TransactionContext {
     }
 
     try {
-      // Execute each operation in order
+      // Execute each operation in order. This loop is the sole rollback-guarded
+      // region: ANY error that escapes it — an operation failure OR a
+      // mid-flight timeout — is caught below and rolls back every operation
+      // applied so far, in reverse order. That single guarantee is the
+      // transaction's atomicity contract, and the generation-store commit path
+      // depends on it: its abort cleanup assumes a throw from here already
+      // restored the applied operations byte-identically (it only discards the
+      // uncommitted staging directory). A rollback per exit path — the previous
+      // design — let the timeout throw slip past rollback and strand the
+      // already-applied writes as torn, generation-less state in canonical
+      // storage.
       for (let i = 0; i < this.operations.length; i++) {
-        // Check timeout
+        // Budget check BEFORE starting the next operation. A trip here throws
+        // into the catch below and rolls back like any other failure — it must
+        // never bypass rollback.
         if (Date.now() - this.startTime > this.options.timeout) {
           throw new TransactionTimeoutError(this.options.timeout, i)
         }
 
         const operation = this.operations[i]
 
+        if (this.options.logging) {
+          prodLog.info(`[Transaction] Executing operation ${i}: ${operation.name || 'unnamed'}`)
+        }
+
+        let rollbackAction: RollbackAction | undefined
         try {
-          if (this.options.logging) {
-            prodLog.info(`[Transaction] Executing operation ${i}: ${operation.name || 'unnamed'}`)
-          }
-
-          // Execute operation
-          const rollbackAction = await operation.execute()
-
-          // Record rollback action (if provided)
-          if (rollbackAction) {
-            this.rollbackActions.push(rollbackAction)
-          }
-
+          rollbackAction = await operation.execute()
         } catch (error) {
-          // Operation failed - rollback and re-throw
-          const executionError = new TransactionExecutionError(
+          // Normalize an operation failure to a TransactionExecutionError; the
+          // outer catch performs the (single) rollback and surfaces it.
+          throw new TransactionExecutionError(
             `Operation ${i} failed: ${(error as Error).message}`,
             i,
             operation.name,
             error as Error
           )
+        }
 
-          await this.rollback(executionError)
-          throw executionError
+        // Record rollback action (if the operation provided one)
+        if (rollbackAction) {
+          this.rollbackActions.push(rollbackAction)
         }
       }
-
-      // All operations succeeded - commit
-      this.commit()
-
     } catch (error) {
-      // Error already handled in rollback
+      // Single rollback point: undo everything applied so far, in reverse
+      // order, then surface the original error. A rollback failure supersedes
+      // it (rollback() throws TransactionRollbackError wrapping this error).
+      await this.rollback(error as Error)
       throw error
     }
+
+    // All operations succeeded — commit.
+    this.commit()
   }
 
   /**

tests/unit/transaction/timeout-rollback.test.ts

@@ -0,0 +1,177 @@
+/**
+ * @module tests/unit/transaction/timeout-rollback
+ * @description Regression for a consumer-reported P0 data-integrity bug:
+ * `Transaction.execute()` checked its time budget at the TOP of the operation
+ * loop and threw `TransactionTimeoutError` from OUTSIDE the per-operation
+ * try/catch — so the timeout bypassed rollback entirely (only per-operation
+ * failures rolled back). On a bulk transact that tripped the 30s budget
+ * mid-flight, the operations already applied to canonical storage were left in
+ * place while the generation was never stamped: torn, generation-less state.
+ * The generation-store commit path's abort cleanup explicitly ASSUMES a throw
+ * from `execute()` already restored the applied operations byte-identically
+ * (it only discards the uncommitted staging directory), so the missing
+ * rollback broke that invariant.
+ *
+ * Fix: `execute()` now has a SINGLE rollback point — any error escaping the
+ * operation loop (operation failure OR mid-flight timeout) rolls back every
+ * applied operation in reverse order, then surfaces the original error.
+ *
+ * These tests model canonical storage as an in-memory map and give each
+ * operation a real rollback action that restores the map byte-identically —
+ * exercising the exact fixed code path and asserting the reported requirement:
+ * a mid-flight timeout leaves storage byte-identical to its pre-transaction
+ * state.
+ */
+import { describe, it, expect } from 'vitest'
+import { Transaction } from '../../../src/transaction/Transaction.js'
+import type { Operation, RollbackAction } from '../../../src/transaction/types.js'
+import {
+  TransactionTimeoutError,
+  TransactionExecutionError,
+  TransactionRollbackError
+} from '../../../src/transaction/errors.js'
+
+/** A tiny stand-in for canonical storage. */
+type Store = Map<string, string>
+
+const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms))
+
+/**
+ * An operation that writes `value` at `key` in `store` and returns a rollback
+ * action restoring the key's PRIOR state byte-identically (delete if it was
+ * absent, else restore the previous value) — exactly how the real graph/metadata
+ * operations undo themselves. `delayMs` lets an op consume wall-clock so a small
+ * transaction budget trips on the NEXT loop iteration; `fail` makes it throw
+ * after doing its write-and-rollback capture (to exercise the shared path via an
+ * operation failure rather than a timeout).
+ */
+function writeOp(
+  store: Store,
+  key: string,
+  value: string,
+  opts: { delayMs?: number; fail?: boolean; name?: string } = {}
+): Operation & { executed: boolean } {
+  const op = {
+    name: opts.name ?? `write:${key}`,
+    executed: false,
+    async execute(): Promise<RollbackAction | undefined> {
+      op.executed = true
+      if (opts.delayMs) await sleep(opts.delayMs)
+      // Operations are individually atomic: a failing op throws having left NO
+      // net change (it never gets far enough to register an undo), so only the
+      // PRIOR operations need rolling back.
+      if (opts.fail) throw new Error(`operation ${key} failed`)
+      const had = store.has(key)
+      const prev = store.get(key)
+      store.set(key, value)
+      return async () => {
+        if (had) store.set(key, prev as string)
+        else store.delete(key)
+      }
+    }
+  }
+  return op
+}
+
+describe('Transaction — mid-flight timeout rolls back (P0 data integrity)', () => {
+  it('a mid-flight timeout leaves storage BYTE-IDENTICAL to its pre-transaction state', async () => {
+    const store: Store = new Map([['seed', 'unchanged']])
+    const snapshot = new Map(store)
+
+    // Budget 1ms; op0 sleeps 60ms so the budget is blown when the loop checks
+    // it before op1 — a deterministic mid-flight timeout at operation index 1.
+    const op0 = writeOp(store, 'A', 'applied-then-rolled-back', { delayMs: 60 })
+    const op1 = writeOp(store, 'B', 'never-applied')
+
+    const tx = new Transaction({ timeout: 1 })
+    tx.addOperation(op0)
+    tx.addOperation(op1)
+
+    await expect(tx.execute()).rejects.toBeInstanceOf(TransactionTimeoutError)
+
+    // op0 ran (and was rolled back); op1 never started.
+    expect(op0.executed).toBe(true)
+    expect(op1.executed).toBe(false)
+    // The whole point: canonical storage is byte-identical — no torn state.
+    expect(store).toEqual(snapshot)
+    expect(store.has('A')).toBe(false)
+    expect(store.has('B')).toBe(false)
+    // And the transaction ended in the rolled-back terminal state (so the
+    // manager counts it correctly), not stranded in 'executing'.
+    expect(tx.getState()).toBe('rolled_back')
+  })
+
+  it('an operation failure mid-flight rolls back byte-identically (the shared single-rollback path)', async () => {
+    const store: Store = new Map([['seed', 'unchanged']])
+    const snapshot = new Map(store)
+
+    const op0 = writeOp(store, 'A', 'applied-then-rolled-back')
+    const op1 = writeOp(store, 'B', 'partially-applied', { fail: true })
+    const op2 = writeOp(store, 'C', 'never-applied')
+
+    const tx = new Transaction()
+    tx.addOperation(op0)
+    tx.addOperation(op1)
+    tx.addOperation(op2)
+
+    await expect(tx.execute()).rejects.toBeInstanceOf(TransactionExecutionError)
+
+    expect(op0.executed).toBe(true)
+    expect(op1.executed).toBe(true)
+    expect(op2.executed).toBe(false)
+    // op1's partial write is undone by its own rollback; op0's write is undone;
+    // op2 never wrote. Byte-identical.
+    expect(store).toEqual(snapshot)
+    expect(tx.getState()).toBe('rolled_back')
+  })
+
+  it('a rollback failure during a timeout surfaces TransactionRollbackError wrapping the timeout (loud, not silent)', async () => {
+    const store: Store = new Map()
+
+    // op0 applies, but its rollback throws every time (maxRollbackRetries
+    // exhausted) — the manager must surface a TransactionRollbackError whose
+    // originalError is the timeout, never swallow it.
+    const op0: Operation & { executed: boolean } = {
+      name: 'unrollbackable',
+      executed: false,
+      async execute() {
+        op0.executed = true
+        await sleep(60)
+        store.set('A', 'applied')
+        return async () => {
+          throw new Error('rollback is impossible for this op')
+        }
+      }
+    }
+    const op1 = writeOp(store, 'B', 'never-applied')
+
+    const tx = new Transaction({ timeout: 1, maxRollbackRetries: 1 })
+    tx.addOperation(op0)
+    tx.addOperation(op1)
+
+    let caught: unknown
+    await tx.execute().catch((e) => (caught = e))
+
+    expect(caught).toBeInstanceOf(TransactionRollbackError)
+    expect((caught as TransactionRollbackError).originalError).toBeInstanceOf(
+      TransactionTimeoutError
+    )
+    expect(op1.executed).toBe(false)
+  })
+
+  it('a clean (non-timeout) transaction still commits and applies all writes', async () => {
+    const store: Store = new Map()
+    const op0 = writeOp(store, 'A', 'a')
+    const op1 = writeOp(store, 'B', 'b')
+
+    const tx = new Transaction()
+    tx.addOperation(op0)
+    tx.addOperation(op1)
+
+    await tx.execute()
+
+    expect(tx.getState()).toBe('committed')
+    expect(store.get('A')).toBe('a')
+    expect(store.get('B')).toBe('b')
+  })
+})