Three pieces addressing the cold-restart-write incident where a production deployment's first writes after every restart (33-35s each on a cold page cache) blew the op-count-scaled transact budget mid-batch: every write is itself a multi-op transaction, so one cold operation consumed the whole budget, the gate before the next operation tripped, and the write rolled back atomically - refused, retried, and refused again until the page cache warmed passively. - The budget's start-gating contract is now explicit and pinned: it gates STARTING the next operation, never rolling back completed work for elapsed time (the shipped schedule since 8.7.0, now stated in contract JSDoc, guarded by code for operation 0, and enforced by regression tests). The 30s floor is configurable via transactionBudgetFloorMs for stores whose cold operations legitimately run long. - New brain.warm() eagerly loads the vector index, metadata index, and graph adjacency so first operations after a cold restart run at steady-state cost. Returns a WarmReport with an honest per-surface outcome (warmed / probed / unavailable) - never reports a probe as a warm. warmOnOpen: true runs it during init(). New optional provider hook warm() on the vector and graph plugin contracts. - Vector-index transaction op classes renamed from the backend-specific AddToHNSWOperation / RemoveFromHNSWOperation to backend-neutral AddToVectorIndexOperation / RemoveFromVectorIndexOperation, stamping the active backend into the emitted op-name string (AddToVectorIndex(js-hnsw) vs a native provider's own identity) so journals never misdirect an operator toward an index that isn't running.
320 lines
11 KiB
TypeScript
320 lines
11 KiB
TypeScript
/**
|
||
* Transaction - Atomic Unit of Work
|
||
*
|
||
* Executes operations atomically: all succeed or all rollback.
|
||
* Prevents partial failures that leave system in inconsistent state.
|
||
*
|
||
* Usage:
|
||
* ```typescript
|
||
* const tx = new Transaction()
|
||
* tx.addOperation(operation1)
|
||
* tx.addOperation(operation2)
|
||
* await tx.execute() // Both succeed or both rollback
|
||
* ```
|
||
*/
|
||
|
||
import {
|
||
Operation,
|
||
RollbackAction,
|
||
TransactionState,
|
||
TransactionContext,
|
||
TransactionOptions
|
||
} from './types.js'
|
||
import {
|
||
InvalidTransactionStateError,
|
||
TransactionExecutionError,
|
||
TransactionRollbackError,
|
||
TransactionTimeoutError
|
||
} from './errors.js'
|
||
import { prodLog } from '../utils/logger.js'
|
||
|
||
/**
|
||
* Default transaction options
|
||
*/
|
||
const DEFAULT_OPTIONS: Required<TransactionOptions> = {
|
||
timeout: 30000, // 30 seconds
|
||
logging: false,
|
||
maxRollbackRetries: 3
|
||
}
|
||
|
||
/**
|
||
* The floor term of {@link transactTimeoutBudget}'s scaling formula when the
|
||
* caller supplies neither an explicit override nor `config.transactionBudgetFloorMs`.
|
||
*/
|
||
const DEFAULT_BUDGET_FLOOR_MS = 30_000
|
||
|
||
/**
|
||
* The apply-phase budget for a batch of `opCount` operations.
|
||
*
|
||
* An explicit `override` wins untouched (a caller-specified deadline for this
|
||
* one batch). Otherwise the budget SCALES with the batch:
|
||
* `max(floorMs, opCount × 2 000)`, where `floorMs` defaults to `30 000` and
|
||
* is overridable via `BrainyConfig.transactionBudgetFloorMs` for the whole
|
||
* brain. The per-op term is calibrated from field data — bulk imports on
|
||
* network-attached disks measure ~2 s per operation (each op pays canonical
|
||
* writes + fsync + index maintenance) — so a flat 30 s budget silently capped
|
||
* honest work at ~15 operations while looking generous for small batches. A
|
||
* 16-op batch, for example, scales to `max(30 000, 16 × 2 000) = 32 000`.
|
||
* Scaling keeps small transacts fast-failing and gives bulk ones a budget
|
||
* proportional to the work they actually asked for.
|
||
*
|
||
* This is a **mid-batch** budget only: it governs whether the transaction's
|
||
* NEXT operation may start (see {@link Transaction.execute}), never whether
|
||
* already-completed work is rolled back after the fact. A trip mid-batch
|
||
* still rolls back every operation applied so far, atomically, and throws a
|
||
* retryable, fully-labeled TransactionTimeoutError — that zero-loss guarantee
|
||
* doesn't change; only the point at which the clock stops mattering does (at
|
||
* the last operation, not one check later).
|
||
*
|
||
* @param opCount - Number of operations in the batch.
|
||
* @param override - A full override for this call; wins over everything else.
|
||
* @param floorMs - The scaling floor for this call (e.g. `config.transactionBudgetFloorMs`).
|
||
* Ignored when `override` is set. Defaults to `30 000` when omitted.
|
||
*/
|
||
export function transactTimeoutBudget(
|
||
opCount: number,
|
||
override?: number,
|
||
floorMs?: number
|
||
): number {
|
||
if (override !== undefined) return override
|
||
return Math.max(floorMs ?? DEFAULT_BUDGET_FLOOR_MS, opCount * 2_000)
|
||
}
|
||
|
||
/**
|
||
* Transaction class
|
||
*/
|
||
export class Transaction implements TransactionContext {
|
||
private operations: Operation[] = []
|
||
private rollbackActions: RollbackAction[] = []
|
||
private state: TransactionState = 'pending'
|
||
private readonly options: Required<TransactionOptions>
|
||
private startTime?: number
|
||
private endTime?: number
|
||
|
||
constructor(options: TransactionOptions = {}) {
|
||
this.options = { ...DEFAULT_OPTIONS, ...options }
|
||
}
|
||
|
||
/**
|
||
* Add an operation to the transaction
|
||
*/
|
||
addOperation(operation: Operation): void {
|
||
if (this.state !== 'pending') {
|
||
throw new InvalidTransactionStateError(
|
||
this.state,
|
||
'add operation'
|
||
)
|
||
}
|
||
this.operations.push(operation)
|
||
}
|
||
|
||
/**
|
||
* Get current transaction state
|
||
*/
|
||
getState(): TransactionState {
|
||
return this.state
|
||
}
|
||
|
||
/**
|
||
* Get number of operations in transaction
|
||
*/
|
||
getOperationCount(): number {
|
||
return this.operations.length
|
||
}
|
||
|
||
/**
|
||
* Execute all operations atomically.
|
||
*
|
||
* Budget semantics: the configured budget gates starting the next
|
||
* operation; completed work is never rolled back for elapsed time. Elapsed
|
||
* time is checked ONLY before starting operation `i` for `i ≥ 1` — never
|
||
* before operation 0 (an operation always gets to start) and never again
|
||
* after the final operation completes. Concretely: a single-op transaction
|
||
* can never time out post-hoc — it either runs (and its result stands, no
|
||
* matter how long it took) or a `TransactionTimeoutError` is thrown before
|
||
* it starts, which cannot happen since there is no operation before it. A
|
||
* multi-op transaction whose op `i` overruns the budget stops op `i+1` from
|
||
* starting, rolls back operations `0..i` (reverse order), and throws
|
||
* `TransactionTimeoutError` — that zero-loss rollback guarantee for
|
||
* mid-batch trips is unchanged; only the post-completion check is gone.
|
||
*/
|
||
async execute(): Promise<void> {
|
||
if (this.state !== 'pending') {
|
||
throw new InvalidTransactionStateError(
|
||
this.state,
|
||
'execute'
|
||
)
|
||
}
|
||
|
||
this.state = 'executing'
|
||
this.startTime = Date.now()
|
||
|
||
if (this.options.logging) {
|
||
prodLog.info(`[Transaction] Executing ${this.operations.length} operations`)
|
||
}
|
||
|
||
try {
|
||
// 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++) {
|
||
// Budget gates STARTING the next operation; completed work is never
|
||
// rolled back for elapsed time. Skipped for i === 0 (operation 0
|
||
// always gets to start — nothing has run yet to have overrun) and
|
||
// never re-checked after the final operation completes (there is no
|
||
// "next operation" left to gate). A trip here throws into the catch
|
||
// below and rolls back like any other failure — it must never bypass
|
||
// rollback.
|
||
if (i > 0 && Date.now() - this.startTime > this.options.timeout) {
|
||
throw new TransactionTimeoutError(this.options.timeout, i, {
|
||
elapsedMs: Date.now() - this.startTime,
|
||
totalOperations: this.operations.length,
|
||
operationName: this.operations[i]?.name
|
||
})
|
||
}
|
||
|
||
const operation = this.operations[i]
|
||
|
||
if (this.options.logging) {
|
||
prodLog.info(`[Transaction] Executing operation ${i}: ${operation.name || 'unnamed'}`)
|
||
}
|
||
|
||
let rollbackAction: RollbackAction | undefined
|
||
try {
|
||
rollbackAction = await operation.execute()
|
||
} catch (error) {
|
||
// 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
|
||
)
|
||
}
|
||
|
||
// Record rollback action (if the operation provided one)
|
||
if (rollbackAction) {
|
||
this.rollbackActions.push(rollbackAction)
|
||
}
|
||
}
|
||
} catch (error) {
|
||
// 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()
|
||
}
|
||
|
||
/**
|
||
* Commit the transaction
|
||
*/
|
||
private commit(): void {
|
||
this.state = 'committed'
|
||
this.endTime = Date.now()
|
||
|
||
if (this.options.logging) {
|
||
const duration = this.endTime - (this.startTime || this.endTime)
|
||
prodLog.info(`[Transaction] Committed successfully in ${duration}ms`)
|
||
}
|
||
|
||
// Clear rollback actions - no longer needed
|
||
this.rollbackActions = []
|
||
}
|
||
|
||
/**
|
||
* Rollback all executed operations in reverse order
|
||
*/
|
||
private async rollback(originalError: Error): Promise<void> {
|
||
this.state = 'rolling_back'
|
||
|
||
if (this.options.logging) {
|
||
prodLog.info(`[Transaction] Rolling back ${this.rollbackActions.length} operations`)
|
||
}
|
||
|
||
const rollbackErrors: Error[] = []
|
||
|
||
// Execute rollback actions in REVERSE order
|
||
for (let i = this.rollbackActions.length - 1; i >= 0; i--) {
|
||
const action = this.rollbackActions[i]
|
||
|
||
// Retry rollback with exponential backoff
|
||
let attempts = 0
|
||
let success = false
|
||
|
||
while (attempts < this.options.maxRollbackRetries && !success) {
|
||
try {
|
||
await action()
|
||
success = true
|
||
|
||
if (this.options.logging) {
|
||
prodLog.info(`[Transaction] Rolled back operation ${i}`)
|
||
}
|
||
|
||
} catch (error) {
|
||
attempts++
|
||
|
||
if (attempts >= this.options.maxRollbackRetries) {
|
||
// Max retries exceeded - log error and continue
|
||
const rollbackError = error as Error
|
||
rollbackErrors.push(rollbackError)
|
||
|
||
prodLog.error(
|
||
`[Transaction] Rollback failed for operation ${i} after ${attempts} attempts: ${rollbackError.message}`
|
||
)
|
||
} else {
|
||
// Retry with exponential backoff
|
||
const delayMs = Math.pow(2, attempts) * 100
|
||
await new Promise(resolve => setTimeout(resolve, delayMs))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Honest terminal state: only claim 'rolled_back' when EVERY undo applied.
|
||
// If any undo failed, the store is not cleanly rolled back — mark
|
||
// 'inconsistent' so the commit orchestration reconciles rather than trusts
|
||
// a false 'rolled_back'. (Fixes the state half of the post-commit response
|
||
// lie: the record whose undo failed is still durable.)
|
||
this.state = rollbackErrors.length > 0 ? 'inconsistent' : 'rolled_back'
|
||
this.endTime = Date.now()
|
||
|
||
if (this.options.logging) {
|
||
const duration = this.endTime - (this.startTime || this.endTime)
|
||
prodLog.info(`[Transaction] ${this.state === 'inconsistent' ? 'Rollback INCOMPLETE' : 'Rolled back'} in ${duration}ms`)
|
||
}
|
||
|
||
// If rollback encountered errors, wrap them with original error. The
|
||
// orchestration keys on `instanceof TransactionRollbackError` to know the
|
||
// undo did not fully apply and reconciliation is required.
|
||
if (rollbackErrors.length > 0) {
|
||
throw new TransactionRollbackError(
|
||
`Transaction rollback encountered ${rollbackErrors.length} errors during cleanup`,
|
||
originalError,
|
||
rollbackErrors
|
||
)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get transaction execution time in milliseconds
|
||
*/
|
||
getExecutionTimeMs(): number | undefined {
|
||
if (this.startTime && this.endTime) {
|
||
return this.endTime - this.startTime
|
||
}
|
||
return undefined
|
||
}
|
||
}
|