David Snelling
003e2a74ea
A production incident: a native-provider op ground 38-40s inside a transaction, blew the apply-phase budget, rolled back, and a downstream pipeline hot-retried the identical operation into a 6-minute CPU storm. Brainy itself never auto-retried the timeout; the gap was that TransactionTimeoutError only said "retryable" in prose, with nothing machine-readable for a caller to branch on. - TransactionTimeoutError gains two typed, always-true fields: retryable (a later attempt may succeed once the slowness resolves or the budget is raised) and hotRetryUnsafe (an immediate identical retry re-pays the full cost that just timed out and can cascade into a CPU storm -- callers must latch and back off, never loop). context's existing telemetry fields (timeoutMs, operationIndex, elapsedMs, totalOperations, operationName) are now documented as the caller's backoff inputs. - Updated the "retryable" doc-prose sites (transact()'s timeoutMs option, transactionBudgetFloorMs, Transaction.execute()'s contract) to point at the new fields instead of bare prose. - Regression pin (tests/unit/transaction/timeout-never-internally-retried.test.ts): an execution counter proves the engine never re-drives a timed-out operation, through both the single-op engine TransactionManager/Transaction drives for every single-record write, and add()'s upsert-race retry loop (which must exit on the first TransactionTimeoutError, never treat it like the lost-insert-race signal it retries on). - Removed TransactionManager.executeTransactionWithResult -- zero callers anywhere in the codebase.
336 lines
11 KiB
TypeScript
336 lines
11 KiB
TypeScript
/**
|
||
* TransactionManager Tests
|
||
*
|
||
* Tests the TransactionManager for:
|
||
* - High-level transaction API
|
||
* - Statistics tracking
|
||
* - Error handling
|
||
*/
|
||
|
||
import { describe, it, expect, beforeEach } from 'vitest'
|
||
import { TransactionManager } from '../../src/transaction/TransactionManager.js'
|
||
import { transactTimeoutBudget } from '../../src/transaction/Transaction.js'
|
||
import { TransactionError } from '../../src/transaction/errors.js'
|
||
|
||
describe('TransactionManager', () => {
|
||
let manager: TransactionManager
|
||
|
||
beforeEach(() => {
|
||
manager = new TransactionManager()
|
||
})
|
||
|
||
describe('executeTransaction', () => {
|
||
it('should execute transaction and return user value', async () => {
|
||
const result = await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
return 'success'
|
||
})
|
||
|
||
expect(result).toBe('success')
|
||
})
|
||
|
||
it('should execute multiple operations', async () => {
|
||
const executionOrder: number[] = []
|
||
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
executionOrder.push(1)
|
||
return async () => {}
|
||
}
|
||
})
|
||
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
executionOrder.push(2)
|
||
return async () => {}
|
||
}
|
||
})
|
||
})
|
||
|
||
expect(executionOrder).toEqual([1, 2])
|
||
})
|
||
|
||
it('should throw TransactionError on failure', async () => {
|
||
await expect(
|
||
manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
throw new Error('Operation failed')
|
||
}
|
||
})
|
||
})
|
||
).rejects.toThrow(TransactionError)
|
||
})
|
||
|
||
it('should pass through custom transaction options', async () => {
|
||
// Should timeout with many operations
|
||
await expect(
|
||
manager.executeTransaction(
|
||
async (tx) => {
|
||
for (let i = 0; i < 10; i++) {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
await new Promise(resolve => setTimeout(resolve, 10))
|
||
return async () => {}
|
||
}
|
||
})
|
||
}
|
||
},
|
||
{ timeout: 50 }
|
||
)
|
||
).rejects.toThrow()
|
||
})
|
||
})
|
||
|
||
describe('Statistics Tracking', () => {
|
||
it('should track total transactions', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.totalTransactions).toBe(2)
|
||
})
|
||
|
||
it('should track successful transactions', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.successfulTransactions).toBe(1)
|
||
expect(stats.failedTransactions).toBe(0)
|
||
})
|
||
|
||
it('should track failed transactions', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
}).catch(() => {})
|
||
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
throw new Error('Failed')
|
||
}
|
||
})
|
||
}).catch(() => {})
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.successfulTransactions).toBe(1)
|
||
expect(stats.failedTransactions).toBe(1)
|
||
})
|
||
|
||
it('should track rolled back transactions', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
return async () => { /* rollback */ }
|
||
}
|
||
})
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
throw new Error('Failed')
|
||
}
|
||
})
|
||
}).catch(() => {})
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.rolledBackTransactions).toBe(1)
|
||
})
|
||
|
||
it('should calculate average execution time', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
await new Promise(resolve => setTimeout(resolve, 10))
|
||
return async () => {}
|
||
}
|
||
})
|
||
})
|
||
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
await new Promise(resolve => setTimeout(resolve, 10))
|
||
return async () => {}
|
||
}
|
||
})
|
||
})
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.averageExecutionTimeMs).toBeGreaterThan(0)
|
||
expect(stats.totalTransactions).toBe(2)
|
||
})
|
||
|
||
it('should calculate average operations per transaction', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
tx.addOperation({ execute: async () => undefined })
|
||
tx.addOperation({ execute: async () => undefined })
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.averageOperationsPerTransaction).toBe(3) // (2 + 4) / 2
|
||
})
|
||
|
||
it('should allow resetting statistics', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
|
||
let stats = manager.getStats()
|
||
expect(stats.totalTransactions).toBe(1)
|
||
|
||
manager.resetStats()
|
||
|
||
stats = manager.getStats()
|
||
expect(stats.totalTransactions).toBe(0)
|
||
expect(stats.successfulTransactions).toBe(0)
|
||
expect(stats.failedTransactions).toBe(0)
|
||
expect(stats.rolledBackTransactions).toBe(0)
|
||
expect(stats.averageExecutionTimeMs).toBe(0)
|
||
expect(stats.averageOperationsPerTransaction).toBe(0)
|
||
})
|
||
})
|
||
|
||
describe('Concurrent Transactions', () => {
|
||
it('should handle concurrent transactions', async () => {
|
||
const results = await Promise.all([
|
||
manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
return 1
|
||
}),
|
||
manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
return 2
|
||
}),
|
||
manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
return 3
|
||
})
|
||
])
|
||
|
||
expect(results).toEqual([1, 2, 3])
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.totalTransactions).toBe(3)
|
||
expect(stats.successfulTransactions).toBe(3)
|
||
})
|
||
|
||
it('should track mixed success/failure in concurrent transactions', async () => {
|
||
const results = await Promise.allSettled([
|
||
manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
}),
|
||
manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
throw new Error('Failed')
|
||
}
|
||
})
|
||
}),
|
||
manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
])
|
||
|
||
expect(results[0].status).toBe('fulfilled')
|
||
expect(results[1].status).toBe('rejected')
|
||
expect(results[2].status).toBe('fulfilled')
|
||
|
||
const stats = manager.getStats()
|
||
expect(stats.totalTransactions).toBe(3)
|
||
expect(stats.successfulTransactions).toBe(2)
|
||
expect(stats.failedTransactions).toBe(1)
|
||
})
|
||
})
|
||
|
||
describe('Error Context', () => {
|
||
it('should wrap non-TransactionError errors', async () => {
|
||
const error = await manager.executeTransaction(async () => {
|
||
throw new Error('Custom error')
|
||
}).catch(e => e)
|
||
|
||
expect(error).toBeInstanceOf(TransactionError)
|
||
expect(error.message).toContain('Custom error')
|
||
})
|
||
|
||
it('should preserve TransactionError instances', async () => {
|
||
const error = await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({
|
||
execute: async () => {
|
||
throw new Error('Operation failed')
|
||
}
|
||
})
|
||
}).catch(e => e)
|
||
|
||
expect(error).toBeInstanceOf(TransactionError)
|
||
})
|
||
})
|
||
|
||
describe('Stats Immutability', () => {
|
||
it('should return immutable stats copy', async () => {
|
||
await manager.executeTransaction(async (tx) => {
|
||
tx.addOperation({ execute: async () => undefined })
|
||
})
|
||
|
||
const stats1 = manager.getStats()
|
||
const stats2 = manager.getStats()
|
||
|
||
expect(stats1).not.toBe(stats2) // Different objects
|
||
expect(stats1).toEqual(stats2) // Same values
|
||
})
|
||
})
|
||
|
||
describe('Timeout budget + telemetry', () => {
|
||
it('transactTimeoutBudget: explicit override wins; default scales with batch size', () => {
|
||
expect(transactTimeoutBudget(1)).toBe(30_000) // small batches keep the 30s floor
|
||
expect(transactTimeoutBudget(15)).toBe(30_000) // the old flat cap's break-even point
|
||
expect(transactTimeoutBudget(100)).toBe(200_000) // 100 ops × 2s — bulk gets an honest budget
|
||
expect(transactTimeoutBudget(1000, 5_000)).toBe(5_000) // caller override is untouched
|
||
})
|
||
|
||
it('a tripped budget rolls back and names the operation, progress, and budget', async () => {
|
||
const rolledBack: string[] = []
|
||
|
||
const failing = manager.executeTransaction(
|
||
async (tx) => {
|
||
tx.addOperation({
|
||
name: 'slow-first-op',
|
||
execute: async () => {
|
||
await new Promise((r) => setTimeout(r, 30))
|
||
return async () => {
|
||
rolledBack.push('slow-first-op')
|
||
}
|
||
}
|
||
})
|
||
tx.addOperation({
|
||
name: 'never-reached',
|
||
execute: async () => undefined
|
||
})
|
||
},
|
||
{ timeout: 5 } // the first op's 30ms sleep guarantees the pre-op-2 check trips
|
||
)
|
||
|
||
await expect(failing).rejects.toThrow(TransactionError)
|
||
const err = await failing.catch((e) => e)
|
||
expect(err.name).toBe('TransactionTimeoutError')
|
||
expect(err.message).toContain('operation 1/2') // which op, of how many
|
||
expect(err.message).toContain("('never-reached')") // its name
|
||
expect(err.message).toContain('budget 5ms') // the budget that tripped
|
||
expect(err.message).toContain('rolled back') // the retryability statement
|
||
expect(rolledBack).toEqual(['slow-first-op']) // the applied op was undone
|
||
})
|
||
})
|
||
})
|