fix: honest response when a transaction rollback cannot complete
When a transaction failed and its rollback ALSO failed to undo a canonical write (retries exhausted), the old path logged, continued, threw TransactionRollbackError, and set state='rolled_back' — while the record it could not undo stayed durable on disk. The caller got an error implying the write was undone; a read-back showed the record. A failed rollback had no truthful response (the post-commit response lie). Give a failed rollback a two-branch honest contract, decided by observation: both commit paths already hold byte-identical before-images, so after a failed rollback the store compares current canonical state to them and classifies each touched record as reconciled, an additive orphan (present when it should be gone), or a restorative loss (gone/wrong when it should have been restored). - Adopt-forward: a single-op write whose only damage is a durably-present orphan — the record the caller asked for — is committed forward (its generation buffered) and returns success with a loud warning that the derived index may be incomplete for that id until the next rebuild/repairIndex(). No error, no double-write; the record is durable and get-able immediately. - Fail loud + quarantine: a multi-op batch, or ANY restorative loss, throws the new StoreInconsistentError naming every unreconciled record and its disposition, and puts the brain into write-quarantine (reads keep working, writes refused via assertWritable) until repairIndex() reconciles the derived indexes against canonical and lifts it. The counter is not advanced, and Transaction.rollback now reports state 'inconsistent' instead of the 'rolled_back' lie when any undo failed. New: StoreInconsistentError + UnreconciledRecord exported from the root; repairIndex() forces a rebuild and clears the quarantine. Regression tests/integration/rollback-trapdoor.test.ts injects index-add-throws + canonical-delete-undo-fails and pins adopt-forward (durable, get-able, not quarantined), fail-loud (StoreInconsistentError + quarantine + reads work + repairIndex lifts it), and the error's record naming.
This commit is contained in:
parent
036e56c9f9
commit
711d2f046a
7 changed files with 437 additions and 35 deletions
|
|
@ -174,7 +174,7 @@ import {
|
|||
type PendingChangeEvent
|
||||
} from './events/changeFeed.js'
|
||||
import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js'
|
||||
import { GenerationConflictError } from './db/errors.js'
|
||||
import { GenerationConflictError, StoreInconsistentError } from './db/errors.js'
|
||||
import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError } from './errors/brainyError.js'
|
||||
import { MemoryStorage } from './storage/adapters/memoryStorage.js'
|
||||
import type {
|
||||
|
|
@ -593,6 +593,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// set this to ReaderMode so historical instances are protected the same way.
|
||||
private operationalMode: BaseOperationalMode
|
||||
|
||||
// Write-quarantine after a failed transaction rollback left the store
|
||||
// inconsistent (see StoreInconsistentError). Set at the moment the failure is
|
||||
// surfaced; every subsequent mutation is refused via assertWritable until
|
||||
// repairIndex() reconciles canonical vs derived state and clears it. Reads are
|
||||
// never affected. null = healthy.
|
||||
private storeInconsistency: StoreInconsistentError | null = null
|
||||
|
||||
// Ready Promise state (Unified readiness API)
|
||||
// Allows consumers to await brain.ready for initialization completion
|
||||
private _readyPromise: Promise<void> | null = null
|
||||
|
|
@ -758,6 +765,17 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
`Open in writer mode to modify data.`
|
||||
)
|
||||
}
|
||||
// Write-quarantine: a prior transaction's rollback failed and left the store
|
||||
// inconsistent. Refuse further writes (which would compound the damage) until
|
||||
// repairIndex() reconciles and lifts the quarantine. Reads still work.
|
||||
if (this.storeInconsistency) {
|
||||
throw new Error(
|
||||
`Cannot call ${method}() — the store is WRITE-QUARANTINED after a failed ` +
|
||||
`transaction rollback left it inconsistent. Reads still work; run repairIndex() ` +
|
||||
`to reconcile the derived indexes against canonical storage and lift the ` +
|
||||
`quarantine. Original inconsistency: ${this.storeInconsistency.message}`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1631,14 +1649,23 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
this.emitCommitted(pendingEvents, capturedBefore, undefined, timestamp)
|
||||
return { timestamp }
|
||||
}
|
||||
const receipt = await this.generationStore.commitSingleOp({
|
||||
let receipt
|
||||
try {
|
||||
receipt = await this.generationStore.commitSingleOp({
|
||||
touched,
|
||||
precommit: captureAndCheck,
|
||||
execute: () => this.transactionManager.executeTransaction(run)
|
||||
})
|
||||
} catch (err) {
|
||||
// A failed rollback that left the store inconsistent (a remove/update
|
||||
// whose restore-undo failed) quarantines writes until repairIndex().
|
||||
if (err instanceof StoreInconsistentError) this.storeInconsistency = err
|
||||
throw err
|
||||
}
|
||||
// POST-COMMIT ONLY: an aborted commit (CAS conflict, failed apply) throws
|
||||
// above and never reaches this line — the feed cannot announce a write
|
||||
// that did not become durable.
|
||||
// that did not become durable. (A degraded adopt-forward write DID commit —
|
||||
// it carries a generation and emits normally.)
|
||||
this.emitCommitted(pendingEvents, capturedBefore, receipt.generation, receipt.timestamp)
|
||||
return receipt
|
||||
}
|
||||
|
|
@ -7407,7 +7434,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
const { generation, timestamp } = await this.generationStore.commitTransaction({
|
||||
let generation: number
|
||||
let timestamp: number
|
||||
try {
|
||||
;({ generation, timestamp } = await this.generationStore.commitTransaction({
|
||||
touched: { nouns: plan.touchedNouns, verbs: plan.touchedVerbs },
|
||||
meta: options?.meta,
|
||||
ifAtGeneration: options?.ifAtGeneration,
|
||||
|
|
@ -7419,7 +7449,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}))
|
||||
} catch (err) {
|
||||
// A batch whose rollback failed and left canonical records unreconciled
|
||||
// quarantines writes until repairIndex() reconciles and lifts it.
|
||||
if (err instanceof StoreInconsistentError) this.storeInconsistency = err
|
||||
throw err
|
||||
}
|
||||
|
||||
// Aggregation-index maintenance: derived data, applied after the commit
|
||||
// point — exactly where the single-operation methods apply it.
|
||||
|
|
@ -14750,15 +14786,36 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
|
||||
/**
|
||||
* Detect and repair corrupted metadata indexes
|
||||
* Detect and repair corrupted metadata indexes.
|
||||
*
|
||||
* Runs corruption detection and auto-rebuilds if corruption is found.
|
||||
* This is the equivalent of the old init()-time corruption check,
|
||||
* now available as an explicit operation.
|
||||
* Runs corruption detection and auto-rebuilds if corruption is found. This is
|
||||
* the equivalent of the old init()-time corruption check, now available as an
|
||||
* explicit operation.
|
||||
*
|
||||
* It is ALSO the recovery path for a write-quarantine: when a transaction's
|
||||
* rollback fails and leaves the store inconsistent ({@link StoreInconsistentError}),
|
||||
* writes are refused until this method reconciles the derived indexes against
|
||||
* canonical storage (a forced rebuild — orphaned/lost records reflected
|
||||
* consistently) and lifts the quarantine. Canonical is the source of truth: a
|
||||
* genuinely lost record cannot be resurrected here (restore from a snapshot for
|
||||
* that), but the store is made internally consistent and writes re-enabled.
|
||||
*/
|
||||
async repairIndex(): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
await this.metadataIndex.detectAndRepairCorruption()
|
||||
// Lift a failed-rollback write-quarantine: force a full rebuild so the
|
||||
// derived indexes are provably reconciled with canonical, then clear the
|
||||
// flag so writes resume.
|
||||
if (this.storeInconsistency) {
|
||||
await this.rebuildIndexesIfNeeded(true)
|
||||
const cleared = this.storeInconsistency
|
||||
this.storeInconsistency = null
|
||||
prodLog.warn(
|
||||
`[Brainy] repairIndex() reconciled the store and LIFTED the write-quarantine ` +
|
||||
`set by a failed transaction rollback (${cleared.records.length} record(s) affected). ` +
|
||||
`Writes are re-enabled.`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -159,3 +159,74 @@ export class GenerationCompactedError extends Error {
|
|||
this.horizon = horizon
|
||||
}
|
||||
}
|
||||
|
||||
/** One entity/relationship left in an unreconciled state by a failed rollback. */
|
||||
export interface UnreconciledRecord {
|
||||
/** The entity or relationship id. */
|
||||
id: string
|
||||
/** `'noun'` (entity) or `'verb'` (relationship). */
|
||||
kind: 'noun' | 'verb'
|
||||
/**
|
||||
* `'orphan'` — the record is durably PRESENT but the transaction aborted
|
||||
* (an add whose delete-undo failed); `'loss'` — the record is durably GONE
|
||||
* or wrong when it should have been restored (a remove/update whose
|
||||
* restore-undo failed — actual data loss).
|
||||
*/
|
||||
disposition: 'orphan' | 'loss'
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Thrown when a transaction's rollback could not be fully applied
|
||||
* and the resulting inconsistency cannot be safely adopted forward — i.e. a
|
||||
* multi-operation batch, or ANY case where a record was lost (a remove/update
|
||||
* whose restore-undo failed). The store is left in a known-inconsistent state:
|
||||
* the {@link records} name every entity/relationship whose canonical state no
|
||||
* longer matches what the aborted transaction should have produced.
|
||||
*
|
||||
* On throw, the brain enters **write-quarantine** — reads continue to work, but
|
||||
* further writes are refused until {@link } `repairIndex()` reconciles the
|
||||
* derived indexes against canonical storage and lifts the quarantine. This is
|
||||
* the honest, loud failure: a visible inconsistency the operator must repair,
|
||||
* never a silent partial write. The generation counter is NOT advanced.
|
||||
*
|
||||
* (A SINGLE-op add whose only damage is a durably-present orphan is NOT this
|
||||
* error: it is adopted forward as a committed generation and returned as a
|
||||
* degraded-but-successful write — the record the caller asked for exists.)
|
||||
*
|
||||
* @example
|
||||
* try {
|
||||
* await brain.transact(ops)
|
||||
* } catch (err) {
|
||||
* if (err instanceof StoreInconsistentError) {
|
||||
* console.error('store inconsistent:', err.records) // ids + dispositions
|
||||
* await brain.repairIndex() // reconcile + lift the write-quarantine
|
||||
* }
|
||||
* }
|
||||
*/
|
||||
export class StoreInconsistentError extends Error {
|
||||
/** Every record left in an unreconciled state by the failed rollback. */
|
||||
public readonly records: UnreconciledRecord[]
|
||||
/** The original error that triggered the (then-failed) rollback. */
|
||||
public override readonly cause: Error
|
||||
|
||||
/**
|
||||
* @param records - The unreconciled entities/relationships (ids + disposition).
|
||||
* @param cause - The error that triggered the rollback.
|
||||
*/
|
||||
constructor(records: UnreconciledRecord[], cause: Error) {
|
||||
const orphans = records.filter((r) => r.disposition === 'orphan').length
|
||||
const losses = records.filter((r) => r.disposition === 'loss').length
|
||||
super(
|
||||
`Store left inconsistent by a failed transaction rollback: ` +
|
||||
`${records.length} record(s) could not be reconciled ` +
|
||||
`(${orphans} orphaned, ${losses} lost). ` +
|
||||
`The brain is now WRITE-QUARANTINED (reads still work) — run repairIndex() ` +
|
||||
`to reconcile the derived indexes against canonical storage and lift the ` +
|
||||
`quarantine. Ids: ${records.map((r) => `${r.id}:${r.disposition}`).join(', ')}. ` +
|
||||
`Cause: ${cause.message}`
|
||||
)
|
||||
this.name = 'StoreInconsistentError'
|
||||
this.records = records
|
||||
this.cause = cause
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,7 +32,9 @@
|
|||
*/
|
||||
|
||||
import { prodLog } from '../utils/logger.js'
|
||||
import { GenerationCompactedError, GenerationConflictError } from './errors.js'
|
||||
import { GenerationCompactedError, GenerationConflictError, StoreInconsistentError } from './errors.js'
|
||||
import type { UnreconciledRecord } from './errors.js'
|
||||
import { TransactionRollbackError } from '../transaction/errors.js'
|
||||
import type {
|
||||
ChangedIds,
|
||||
CompactHistoryOptions,
|
||||
|
|
@ -618,18 +620,21 @@ export class GenerationStore {
|
|||
// before staging; scoped outside the try so an abort can compensate.
|
||||
let txBlobHashes: string[] = []
|
||||
|
||||
// Hoisted so the catch can reconcile canonical state against them after a
|
||||
// failed rollback (the trapdoor). Empty until populated below.
|
||||
const nounBefore = new Map<string, GenerationRecord>()
|
||||
const verbBefore = new Map<string, GenerationRecord>()
|
||||
|
||||
try {
|
||||
// -- 3. Before-images + delta (the durable undo log) ------------------
|
||||
// Read every before-image FIRST, then run the caller's CAS
|
||||
// precondition against them, and only then stage to disk — so a
|
||||
// conflicting batch aborts with zero staging I/O. The maps hold the
|
||||
// byte-identical records the staged files are written from.
|
||||
const nounBefore = new Map<string, GenerationRecord>()
|
||||
for (const id of nouns) {
|
||||
const prev = await this.storage.readNounRaw(id)
|
||||
nounBefore.set(id, { kind: 'noun', metadata: prev.metadata, vector: prev.vector })
|
||||
}
|
||||
const verbBefore = new Map<string, GenerationRecord>()
|
||||
for (const id of verbs) {
|
||||
const prev = await this.storage.readVerbRaw(id)
|
||||
verbBefore.set(id, { kind: 'verb', metadata: prev.metadata, vector: prev.vector })
|
||||
|
|
@ -737,6 +742,16 @@ export class GenerationStore {
|
|||
if (crashSimulated) {
|
||||
throw err
|
||||
}
|
||||
// The trapdoor for a batch: if rollback FAILED to fully apply, canonical
|
||||
// storage may be inconsistent. A batch is never adopted forward (its
|
||||
// other ops were rolled back — partial commit would break atomicity), so
|
||||
// any unreconciled record is a fail-loud StoreInconsistentError. Compute
|
||||
// it here (before the staging cleanup, which is always safe to run) and
|
||||
// throw it in place of the raw error at the end.
|
||||
let inconsistent: UnreconciledRecord[] = []
|
||||
if (err instanceof TransactionRollbackError) {
|
||||
inconsistent = await this.reconcileFailedRollback(nounBefore, verbBefore)
|
||||
}
|
||||
// Failed before the manifest rename: nothing is committed. Remove the
|
||||
// staging directory; the TransactionManager already restored any
|
||||
// applied operation byte-identically.
|
||||
|
|
@ -763,11 +778,68 @@ export class GenerationStore {
|
|||
// Return the reservation when no concurrent bump consumed a later
|
||||
// number, so a failed transaction leaves generation() unchanged.
|
||||
if (this.counter === gen) this.counter = gen - 1
|
||||
// A failed rollback that left canonical records unreconciled surfaces as
|
||||
// a loud StoreInconsistentError (brainy quarantines writes until repair);
|
||||
// otherwise the raw error (clean abort, or a derived-only undo failure
|
||||
// the egress guard + rebuild handle).
|
||||
if (inconsistent.length > 0) {
|
||||
throw new StoreInconsistentError(
|
||||
inconsistent,
|
||||
(err as TransactionRollbackError).originalError
|
||||
)
|
||||
}
|
||||
throw err
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* @description After a transaction's rollback FAILED to fully apply
|
||||
* (`TransactionRollbackError`), determine which touched records are now
|
||||
* unreconciled by comparing current canonical state to the before-images the
|
||||
* commit captured. This observes reality rather than trusting the opaque undo
|
||||
* closures — the authoritative signal for adopt-forward vs fail-loud.
|
||||
*
|
||||
* @param nounBefore - Byte-identical entity before-images captured pre-execute.
|
||||
* @param verbBefore - Byte-identical relationship before-images.
|
||||
* @returns Every record whose canonical state no longer matches its
|
||||
* before-image, tagged `'orphan'` (durably present when it should be gone —
|
||||
* an add whose delete-undo failed) or `'loss'` (durably gone/wrong when it
|
||||
* should have been restored — a remove/update whose restore-undo failed). An
|
||||
* empty array means canonical storage is cleanly rolled back (only a derived
|
||||
* index undo failed — the egress guard + a rebuild handle that).
|
||||
*/
|
||||
private async reconcileFailedRollback(
|
||||
nounBefore: Map<string, GenerationRecord>,
|
||||
verbBefore: Map<string, GenerationRecord>
|
||||
): Promise<UnreconciledRecord[]> {
|
||||
const out: UnreconciledRecord[] = []
|
||||
const classify = (
|
||||
before: GenerationRecord,
|
||||
current: { metadata: unknown | null; vector: unknown | null }
|
||||
): 'orphan' | 'loss' | null => {
|
||||
const beforeEmpty = before.metadata == null && before.vector == null
|
||||
const currentEmpty = current.metadata == null && current.vector == null
|
||||
if (beforeEmpty && currentEmpty) return null // add cleanly undone
|
||||
if (beforeEmpty) return 'orphan' // add's delete-undo failed → still present
|
||||
if (currentEmpty) return 'loss' // remove's restore-undo failed → gone
|
||||
// Both present: same value = reconciled; different = restore left a wrong value.
|
||||
const same =
|
||||
JSON.stringify({ m: before.metadata, v: before.vector }) ===
|
||||
JSON.stringify({ m: current.metadata, v: current.vector })
|
||||
return same ? null : 'loss'
|
||||
}
|
||||
for (const [id, before] of nounBefore) {
|
||||
const d = classify(before, await this.storage.readNounRaw(id))
|
||||
if (d) out.push({ id, kind: 'noun', disposition: d })
|
||||
}
|
||||
for (const [id, before] of verbBefore) {
|
||||
const d = classify(before, await this.storage.readVerbRaw(id))
|
||||
if (d) out.push({ id, kind: 'verb', disposition: d })
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ==========================================================================
|
||||
// Single-operation commit (Model-B per-write group-commit)
|
||||
// ==========================================================================
|
||||
|
|
@ -830,7 +902,7 @@ export class GenerationStore {
|
|||
touched: { nouns?: string[]; verbs?: string[] }
|
||||
execute: () => Promise<void>
|
||||
precommit?: (before: CommitBeforeImages) => void
|
||||
}): Promise<{ generation: number; timestamp: number }> {
|
||||
}): Promise<{ generation: number; timestamp: number; degraded?: string[] }> {
|
||||
return this.withMutex(async () => {
|
||||
const nouns = args.touched.nouns ? [...new Set(args.touched.nouns)] : []
|
||||
const verbs = args.touched.verbs ? [...new Set(args.touched.verbs)] : []
|
||||
|
|
@ -873,9 +945,42 @@ export class GenerationStore {
|
|||
await args.execute()
|
||||
} catch (err) {
|
||||
this.inTransact = false
|
||||
// Live write failed: nothing buffered, nothing on disk. Return the
|
||||
// reservation (when no concurrent bump consumed a later number) so
|
||||
// generation() is unchanged.
|
||||
// A failed rollback (TransactionRollbackError) may have left canonical
|
||||
// storage inconsistent — the trapdoor. Reconcile against the
|
||||
// before-images to decide the honest response (David's ruling:
|
||||
// adopt-forward when safe, else fail loud).
|
||||
if (err instanceof TransactionRollbackError) {
|
||||
const records = await this.reconcileFailedRollback(nounBefore, verbBefore)
|
||||
if (records.length > 0 && records.every((r) => r.disposition === 'orphan')) {
|
||||
// ADOPT FORWARD: the durably-present orphan(s) are exactly what this
|
||||
// single-op add wrote. Keep + buffer the generation so the record is
|
||||
// legitimately committed and readable; the derived index may be
|
||||
// incomplete for these ids until the next rebuild/repairIndex (the
|
||||
// egress guard prevents wrong results meanwhile). Loud, honest,
|
||||
// no double-write.
|
||||
this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp })
|
||||
this.pendingGens.push(gen)
|
||||
this.extendChains(gen, nouns, verbs)
|
||||
prodLog.warn(
|
||||
`[GenerationStore] Recovered a failed rollback FORWARD: single-op write ` +
|
||||
`committed as generation ${gen} because its canonical undo could not be ` +
|
||||
`applied (record(s) ${records.map((r) => r.id).join(', ')} are durable). ` +
|
||||
`The derived index may be incomplete for these ids — run repairIndex() to heal.`
|
||||
)
|
||||
return { generation: gen, timestamp, degraded: records.map((r) => r.id) }
|
||||
}
|
||||
if (records.length > 0) {
|
||||
// FAIL LOUD: a loss (or mixed) cannot be safely adopted. Return the
|
||||
// reservation and throw — brainy quarantines writes until repair.
|
||||
if (this.counter === gen) this.counter = gen - 1
|
||||
throw new StoreInconsistentError(records, err.originalError)
|
||||
}
|
||||
// records.length === 0: canonical is cleanly rolled back (only a
|
||||
// derived undo failed) — fall through to the clean-abort path below.
|
||||
}
|
||||
// Live write failed with canonical cleanly rolled back: nothing buffered,
|
||||
// nothing durable. Return the reservation (when no concurrent bump
|
||||
// consumed a later number) so generation() is unchanged.
|
||||
if (this.counter === gen) this.counter = gen - 1
|
||||
throw err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -174,8 +174,10 @@ export type {
|
|||
export {
|
||||
GenerationConflictError,
|
||||
SpeculativeOverlayError,
|
||||
GenerationCompactedError
|
||||
GenerationCompactedError,
|
||||
StoreInconsistentError
|
||||
} from './db/errors.js'
|
||||
export type { UnreconciledRecord } from './db/errors.js'
|
||||
export type {
|
||||
TxOperation,
|
||||
TxAddOperation,
|
||||
|
|
|
|||
|
|
@ -219,15 +219,22 @@ export class Transaction implements TransactionContext {
|
|||
}
|
||||
}
|
||||
|
||||
this.state = 'rolled_back'
|
||||
// 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] Rolled back in ${duration}ms`)
|
||||
prodLog.info(`[Transaction] ${this.state === 'inconsistent' ? 'Rollback INCOMPLETE' : 'Rolled back'} in ${duration}ms`)
|
||||
}
|
||||
|
||||
// If rollback encountered errors, wrap them with original error
|
||||
// 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`,
|
||||
|
|
|
|||
|
|
@ -14,6 +14,10 @@ export type TransactionState =
|
|||
| 'committed' // Successfully committed
|
||||
| 'rolling_back' // Rolling back due to failure
|
||||
| 'rolled_back' // Successfully rolled back
|
||||
| 'inconsistent' // Rollback ran but ≥1 undo could not be applied — the store
|
||||
// is NOT cleanly rolled back; the commit orchestration must
|
||||
// reconcile (adopt-forward or fail-loud). Never claim
|
||||
// 'rolled_back' when an undo failed.
|
||||
|
||||
/**
|
||||
* Rollback action - undoes an operation
|
||||
|
|
|
|||
156
tests/integration/rollback-trapdoor.test.ts
Normal file
156
tests/integration/rollback-trapdoor.test.ts
Normal file
|
|
@ -0,0 +1,156 @@
|
|||
/**
|
||||
* @module tests/integration/rollback-trapdoor
|
||||
* @description Regression for a consumer-reported data-integrity bug: a failed
|
||||
* CANONICAL rollback undo left the record durable while the request errored (a
|
||||
* post-commit response lie). Reproduced by a failure-injection coherence harness:
|
||||
* a late index op throws → rollback runs → an earlier op's canonical undo
|
||||
* (storage.deleteNounMetadata) fails persistently → the record stays on disk
|
||||
* while Transaction.rollback threw TransactionRollbackError AND lied with
|
||||
* state='rolled_back'.
|
||||
*
|
||||
* Fix (David's ruling — adopt-forward when safe, else fail loud):
|
||||
* - SINGLE-op add whose only damage is a durably-present orphan → adopt it
|
||||
* forward: commit the generation (the record IS what add() wanted), return
|
||||
* success + a loud warn; the derived index heals on rebuild/repairIndex.
|
||||
* - MULTI-op batch, OR any restorative LOSS → StoreInconsistentError naming the
|
||||
* unreconciled ids; the brain enters write-quarantine (reads OK, writes
|
||||
* refused) until repairIndex() reconciles and lifts it. Transaction state is
|
||||
* 'inconsistent', never the 'rolled_back' lie.
|
||||
*
|
||||
* The injection mirrors that harness: fault index.addToIndex (trigger rollback) and
|
||||
* storage.deleteNounMetadata (fail the canonical undo).
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||
import * as fs from 'node:fs'
|
||||
import * as os from 'node:os'
|
||||
import * as path from 'node:path'
|
||||
import { Brainy } from '../../src/brainy.js'
|
||||
import { NounType } from '../../src/types/graphTypes.js'
|
||||
import { StoreInconsistentError } from '../../src/db/errors.js'
|
||||
|
||||
let seq = 0
|
||||
const freshId = (): string =>
|
||||
`00000000-0000-4000-8000-${(++seq).toString(16).padStart(12, '0')}`
|
||||
|
||||
describe('rollback trapdoor — failed canonical undo (data integrity)', () => {
|
||||
let dir: string
|
||||
let brain: any
|
||||
let restore: Array<() => void>
|
||||
|
||||
beforeEach(async () => {
|
||||
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
|
||||
dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-trapdoor-'))
|
||||
brain = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'filesystem', path: dir },
|
||||
dimensions: 384,
|
||||
silent: true
|
||||
})
|
||||
await brain.init()
|
||||
restore = []
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
for (const r of restore) r()
|
||||
try { await brain.close() } catch { /* ignore */ }
|
||||
fs.rmSync(dir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
/** Fault the metadata-index add so a write's rollback is triggered. */
|
||||
const faultIndexAdd = () => {
|
||||
const idx = brain['metadataIndex']
|
||||
const orig = idx.addToIndex.bind(idx)
|
||||
idx.addToIndex = async () => { throw new Error('injected: index add failed') }
|
||||
restore.push(() => { idx.addToIndex = orig })
|
||||
}
|
||||
/** Fault the canonical additive undo (delete of a just-created record). */
|
||||
const faultCanonicalDelete = () => {
|
||||
const storage = brain['storage']
|
||||
const orig = storage.deleteNounMetadata.bind(storage)
|
||||
storage.deleteNounMetadata = async () => { throw new Error('injected: canonical undo failed') }
|
||||
restore.push(() => { storage.deleteNounMetadata = orig })
|
||||
}
|
||||
|
||||
it('ADOPT-FORWARD: a single-op add whose canonical undo fails commits as a durable, get-able record (no throw, no quarantine)', async () => {
|
||||
const id = freshId()
|
||||
const genBefore = brain.generationStore.generation()
|
||||
|
||||
faultIndexAdd() // triggers rollback
|
||||
faultCanonicalDelete() // the trapdoor: the metadata record can't be un-written
|
||||
|
||||
// add() must NOT throw — the record it wanted is durable, so it is adopted.
|
||||
const returned = await brain.add({ id, data: 'orphan adopted', type: NounType.Thing })
|
||||
expect(returned).toBe(id)
|
||||
|
||||
// Un-fault so reads/subsequent writes use the real methods.
|
||||
for (const r of restore) r()
|
||||
restore = []
|
||||
|
||||
// The record is durable and get-able, and the generation advanced.
|
||||
expect(await brain.get(id)).not.toBeNull()
|
||||
expect(brain.generationStore.generation()).toBe(genBefore + 1)
|
||||
|
||||
// The store is NOT quarantined — a subsequent write succeeds.
|
||||
const ok = await brain.add({ id: freshId(), data: 'still writable', type: NounType.Thing })
|
||||
expect(await brain.get(ok)).not.toBeNull()
|
||||
})
|
||||
|
||||
it('FAIL-LOUD: a multi-op transact whose canonical undo fails throws StoreInconsistentError and write-quarantines the brain', async () => {
|
||||
// A pre-existing record to prove reads keep working under quarantine.
|
||||
const anchor = await brain.add({ id: freshId(), data: 'anchor', type: NounType.Thing })
|
||||
await brain.flush()
|
||||
|
||||
faultIndexAdd()
|
||||
faultCanonicalDelete()
|
||||
|
||||
const a = freshId()
|
||||
const b = freshId()
|
||||
await expect(
|
||||
brain.transact([
|
||||
{ op: 'add', id: a, data: 'batch A', type: NounType.Thing },
|
||||
{ op: 'add', id: b, data: 'batch B', type: NounType.Thing }
|
||||
])
|
||||
).rejects.toBeInstanceOf(StoreInconsistentError)
|
||||
|
||||
// Writes are refused (quarantine); reads still work.
|
||||
await expect(
|
||||
brain.add({ id: freshId(), data: 'blocked', type: NounType.Thing })
|
||||
).rejects.toThrow(/WRITE-QUARANTINED/)
|
||||
expect(await brain.get(anchor)).not.toBeNull()
|
||||
|
||||
// repairIndex() reconciles and lifts the quarantine → writes resume.
|
||||
for (const r of restore) r()
|
||||
restore = []
|
||||
await brain.repairIndex()
|
||||
|
||||
const ok = await brain.add({ id: freshId(), data: 'writable again', type: NounType.Thing })
|
||||
expect(await brain.get(ok)).not.toBeNull()
|
||||
})
|
||||
|
||||
it('the StoreInconsistentError names the unreconciled records with dispositions', async () => {
|
||||
faultIndexAdd()
|
||||
faultCanonicalDelete()
|
||||
|
||||
const a = freshId()
|
||||
const b = freshId()
|
||||
let caught: unknown
|
||||
await brain
|
||||
.transact([
|
||||
{ op: 'add', id: a, data: 'A', type: NounType.Thing },
|
||||
{ op: 'add', id: b, data: 'B', type: NounType.Thing }
|
||||
])
|
||||
.catch((e: unknown) => (caught = e))
|
||||
|
||||
expect(caught).toBeInstanceOf(StoreInconsistentError)
|
||||
const err = caught as StoreInconsistentError
|
||||
expect(err.records.length).toBeGreaterThan(0)
|
||||
// Every unreconciled record here is a durably-present orphan.
|
||||
expect(err.records.every((r) => r.disposition === 'orphan')).toBe(true)
|
||||
expect(err.message).toMatch(/WRITE-QUARANTINED/)
|
||||
|
||||
// Recover so afterEach can close cleanly.
|
||||
for (const r of restore) r()
|
||||
restore = []
|
||||
await brain.repairIndex()
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue