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
105
src/brainy.ts
105
src/brainy.ts
|
|
@ -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({
|
||||
touched,
|
||||
precommit: captureAndCheck,
|
||||
execute: () => this.transactionManager.executeTransaction(run)
|
||||
})
|
||||
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,19 +7434,28 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
const { generation, timestamp } = await this.generationStore.commitTransaction({
|
||||
touched: { nouns: plan.touchedNouns, verbs: plan.touchedVerbs },
|
||||
meta: options?.meta,
|
||||
ifAtGeneration: options?.ifAtGeneration,
|
||||
precommit: casPrecommit,
|
||||
execute: async () => {
|
||||
await this.transactionManager.executeTransaction(async (tx) => {
|
||||
for (const operation of plan.operations) {
|
||||
tx.addOperation(operation)
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
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,
|
||||
precommit: casPrecommit,
|
||||
execute: async () => {
|
||||
await this.transactionManager.executeTransaction(async (tx) => {
|
||||
for (const operation of plan.operations) {
|
||||
tx.addOperation(operation)
|
||||
}
|
||||
})
|
||||
}
|
||||
}))
|
||||
} 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.`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue