feat(8.0): #18 coordinated migration LOCK — block-and-queue the 7.x→8.0 auto-upgrade
Replaces rc.8's no-freeze deference with an automatic, observable, coordinated
migration LOCK (David's reversal: "unknown/halfway states are more dangerous
than blocking"). While a native provider runs its one-time 7.x→8.0
rebuild-from-canonical (`isMigrating()===true`), brainy blocks/queues data-plane
reads AND writes so no operation ever touches a half-built index — closing the
three seam-map gaps (lost mid-rebuild writes, degraded reads, read-time rebuild).
Mechanism (one choke point):
- awaitMigrationLock() at ensureInitialized() covers all ~40 data-plane methods;
a non-migrating brain pays one boolean check. Polls isMigrating() at 250ms;
after migrationWaitTimeoutMs (default 30s) throws a retryable, exported
MigrationInProgressError. The timeout bounds the CALLER'S WAIT, not the
migration — the rebuild is unbounded and never interrupted.
- init() awaits the lock before VFS bootstrap + before serving ("not ready until
upgraded"); a rebuild past the budget surfaces MigrationInProgressError at init
(raise the budget, or run the offline migrator).
Observability (readiness-probe correct — never gated):
- getIndexStatus() gains `migrating` + `migration` (MigrationProgress); a probe
maps migrating→HTTP 503+Retry-After, not 500. health()/checkHealth() lock-exempt
(health() reports a `warn` migration check). stampBrainFormat()/close() are
ungated so cor can clear the lock — no deadlock.
- Optional provider migrationStatus() is relayed verbatim for a live %.
verifyGraphAdjacencyLive() honors the lock (no self-rebuild mid-migration); the
stamp is still withheld while any provider migrates (cor stamps after verify).
Adversarially verified: fixed a critical init/VFS-bootstrap deadlock, a mixed-
provider busy-spin (dropped the event-driven signal → pure poll), a not-yet-
initialized getIndexStatus crash, and once-per-window log/clock resets. 10 lock
tests (incl. the init-during-migration regression). Gates: typecheck 0, build 0,
test:unit 1753/1753.
This commit is contained in:
parent
ca9129a924
commit
67bbf69a5c
5 changed files with 534 additions and 31 deletions
299
src/brainy.ts
299
src/brainy.ts
|
|
@ -151,6 +151,7 @@ import type { MigrationPreview, MigrationResult, MigrateOptions } from './migrat
|
|||
import { AggregationIndex } from './aggregation/AggregationIndex.js'
|
||||
import { AggregateMaterializer } from './aggregation/materializer.js'
|
||||
import type { AggregateDefinition, AggregateQueryParams, AggregateResult } from './types/brainy.types.js'
|
||||
import type { MigrationProgress } from './types/brainy.types.js'
|
||||
import { resolveJsHnswConfig, DEFAULT_RECALL } from './utils/recallPreset.js'
|
||||
import * as fs from 'node:fs'
|
||||
import * as os from 'node:os'
|
||||
|
|
@ -168,7 +169,7 @@ import {
|
|||
import { GenerationStore } from './db/generationStore.js'
|
||||
import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js'
|
||||
import { GenerationConflictError } from './db/errors.js'
|
||||
import { BrainyError, GraphIndexNotReadyError } from './errors/brainyError.js'
|
||||
import { BrainyError, GraphIndexNotReadyError, MigrationInProgressError } from './errors/brainyError.js'
|
||||
import { MemoryStorage } from './storage/adapters/memoryStorage.js'
|
||||
import type {
|
||||
CompactHistoryOptions,
|
||||
|
|
@ -245,6 +246,7 @@ type ResolvedBrainyConfig = Required<
|
|||
| 'integrations'
|
||||
| 'retention'
|
||||
| 'eagerEmbeddings'
|
||||
| 'migrationWaitTimeoutMs'
|
||||
>
|
||||
> &
|
||||
Pick<
|
||||
|
|
@ -256,6 +258,7 @@ type ResolvedBrainyConfig = Required<
|
|||
| 'integrations'
|
||||
| 'retention'
|
||||
| 'eagerEmbeddings'
|
||||
| 'migrationWaitTimeoutMs'
|
||||
>
|
||||
|
||||
/**
|
||||
|
|
@ -324,6 +327,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
private static shutdownHooksRegisteredGlobally = false
|
||||
private static instances: Brainy[] = []
|
||||
|
||||
/** Poll cadence (ms) for the migration LOCK when a provider exposes no
|
||||
* event-driven `whenMigrationComplete()` signal. See {@link awaitMigrationLock}. */
|
||||
private static readonly MIGRATION_POLL_INTERVAL_MS = 250
|
||||
|
||||
// Core components
|
||||
private index!: JsHnswVectorIndex
|
||||
private storage!: BaseStorage
|
||||
|
|
@ -414,6 +421,17 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
private _graphAdjacencyVerified = false
|
||||
/** Re-entrancy guard: a verify (rebuild → reads) is in flight. */
|
||||
private _graphAdjacencyVerifying = false
|
||||
/**
|
||||
* Coordinated migration LOCK (#18): dedup guards so the "upgrading, blocking"
|
||||
* and "upgrade complete, resumed" lines each log once per migration window,
|
||||
* not once per blocked operation. See {@link awaitMigrationLock}.
|
||||
*/
|
||||
private _migrationBlockLogged = false
|
||||
private _migrationReleaseLogged = false
|
||||
/** Epoch ms when brainy first observed the migration LOCK held, or `null` when
|
||||
* not migrating — the fallback `elapsedMs` a provider that omits `migrationStatus()`
|
||||
* timing still gets in `getIndexStatus().migration`. See {@link migrationSnapshot}. */
|
||||
private _migrationObservedAt: number | null = null
|
||||
/**
|
||||
* 8.0 ⇄ native-provider version handshake — the on-disk {@link BrainFormat}
|
||||
* marker (`_system/brain-format.json`) as read during the store-open phase,
|
||||
|
|
@ -1011,6 +1029,17 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// VFS.init() needs brain to be marked initialized to call brain methods
|
||||
this.initialized = true
|
||||
|
||||
// Migration LOCK (#18): if a native provider is running the one-time
|
||||
// 7.x → 8.0 rebuild, WAIT for it to finish HERE — before VFS bootstrap and
|
||||
// before the brain serves — so init-internal reads/writes (the VFS root
|
||||
// get/add/find below) run against the rebuilt indexes, never a half-built
|
||||
// one. This is the "not ready until upgraded" contract: a blocking upgrade
|
||||
// to a known-good state. Bounded by `migrationWaitTimeoutMs`: a brain whose
|
||||
// upgrade outlives the budget surfaces MigrationInProgressError here (raise
|
||||
// the budget, or run the offline migrator) instead of bootstrapping the VFS
|
||||
// against a partial index. A non-migrating brain returns instantly.
|
||||
await this.awaitMigrationLock()
|
||||
|
||||
// Initialize VFS: Ensure VFS is ready when accessed as property
|
||||
// This eliminates need for separate vfs.init() calls - zero additional complexity
|
||||
this._vfs = new VirtualFileSystem(this)
|
||||
|
|
@ -1207,13 +1236,23 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
* re-initializing — using a closed Brainy is a consumer bug, not a lazy-init
|
||||
* opportunity.
|
||||
*/
|
||||
private async ensureInitialized(): Promise<void> {
|
||||
private async ensureInitialized(opts?: { bypassMigrationLock?: boolean }): Promise<void> {
|
||||
if (this.closed) {
|
||||
throw new Error('Brainy instance is not initialized: it was closed via close(). Create a new instance.')
|
||||
}
|
||||
if (!this.initialized) {
|
||||
await this.init()
|
||||
}
|
||||
// Coordinated migration LOCK (#18): every data-plane read and write funnels
|
||||
// through here, so this is the single choke point that holds operations while
|
||||
// a native provider runs its one-time 7.x → 8.0 rebuild-from-canonical — no
|
||||
// op touches a half-built index. Observability (`health`/`checkHealth`) and
|
||||
// the lock-clearing path (`stampBrainFormat`, which does not route through
|
||||
// here) opt out so an operator can always watch progress and cor can stamp.
|
||||
// A brain that never migrates pays one boolean check (see awaitMigrationLock).
|
||||
if (!opts?.bypassMigrationLock) {
|
||||
await this.awaitMigrationLock()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -2765,6 +2804,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
*/
|
||||
private async verifyGraphAdjacencyLive(): Promise<'live' | 'rebuilt'> {
|
||||
if (this._graphAdjacencyVerified) return 'live'
|
||||
// Coordinated migration LOCK (#18): while the graph provider owns a locked
|
||||
// rebuild-from-canonical, brainy must NOT fire its own graphIndex.rebuild()
|
||||
// on a read — that would race the provider's in-place rebuild. The data-plane
|
||||
// lock (awaitMigrationLock in ensureInitialized) already makes callers wait,
|
||||
// so this is normally unreachable mid-migration; the guard is defensive. It
|
||||
// deliberately does NOT set `_graphAdjacencyVerified`, so the real verify runs
|
||||
// once the migration clears.
|
||||
if (this.providerIsMigrating(this.graphIndex)) return 'live'
|
||||
// Re-entrancy: rebuild() can trigger reads (neighbors/related) that call back into this
|
||||
// guard. While a verify is in flight, short-circuit so we cannot recurse into rebuild().
|
||||
if (this._graphAdjacencyVerifying) return 'live'
|
||||
|
|
@ -9131,6 +9178,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
initialized: boolean
|
||||
lazyRebuildCompleted: boolean
|
||||
disableAutoRebuild: boolean
|
||||
/** `true` while a native provider runs the one-time 7.x → 8.0 rebuild LOCK.
|
||||
* A readiness probe should map this to HTTP 503 + Retry-After (transiently
|
||||
* not-ready), NOT 200-ready and NOT 500-broken. Never gated by the lock. */
|
||||
migrating: boolean
|
||||
/** Structured progress while {@link migrating}; absent otherwise. */
|
||||
migration?: MigrationProgress
|
||||
hnswIndex: {
|
||||
size: number
|
||||
populated: boolean
|
||||
|
|
@ -9147,6 +9200,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
totalEntities: number
|
||||
}
|
||||
}> {
|
||||
// Callable before init() — a readiness/liveness probe may fire during
|
||||
// construction or a fired-not-awaited init(). Report a safe not-initialized
|
||||
// snapshot rather than dereferencing providers that init() has not assigned.
|
||||
if (!this.initialized || this.metadataIndex == null || this.index == null || this.graphIndex == null) {
|
||||
return {
|
||||
initialized: false,
|
||||
lazyRebuildCompleted: this.lazyRebuildCompleted,
|
||||
disableAutoRebuild: this.config.disableAutoRebuild || false,
|
||||
migrating: false,
|
||||
hnswIndex: { size: 0, populated: false },
|
||||
metadataIndex: { entries: 0, populated: false },
|
||||
graphIndex: { relationships: 0, populated: false },
|
||||
storage: { totalEntities: 0 }
|
||||
}
|
||||
}
|
||||
|
||||
const metadataStats = await this.metadataIndex.getStats()
|
||||
const hnswSize = this.index.size()
|
||||
const graphSize = await this.graphIndex.size()
|
||||
|
|
@ -9164,6 +9233,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
initialized: this.initialized,
|
||||
lazyRebuildCompleted: this.lazyRebuildCompleted,
|
||||
disableAutoRebuild: this.config.disableAutoRebuild || false,
|
||||
...this.migrationSnapshot(),
|
||||
hnswIndex: {
|
||||
size: hnswSize,
|
||||
populated: hnswSize > 0
|
||||
|
|
@ -9206,7 +9276,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
details?: Record<string, unknown>
|
||||
}>
|
||||
}> {
|
||||
await this.ensureInitialized()
|
||||
// Lock-exempt: an operator must be able to probe health WHILE the brain
|
||||
// upgrades — that is the whole point of an observable migration.
|
||||
await this.ensureInitialized({ bypassMigrationLock: true })
|
||||
const checks: Array<{
|
||||
name: string
|
||||
status: 'pass' | 'warn' | 'fail'
|
||||
|
|
@ -9214,6 +9286,28 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
details?: Record<string, unknown>
|
||||
}> = []
|
||||
|
||||
// Migration LOCK (#18): while a native provider rebuilds the derived indexes,
|
||||
// the parity/field checks below would report transient mismatches that read
|
||||
// as failures but are not — surface the upgrade as the single honest signal.
|
||||
// `warn` (not `fail`) so a readiness probe treats it as "not ready yet, retry".
|
||||
const migSnap = this.migrationSnapshot()
|
||||
if (migSnap.migrating) {
|
||||
const pct = migSnap.migration?.percent
|
||||
return {
|
||||
overall: 'warn',
|
||||
checks: [
|
||||
{
|
||||
name: 'migration',
|
||||
status: 'warn',
|
||||
message:
|
||||
`Brain is upgrading (7.x → 8.0)${pct !== undefined ? `, ${Math.round(pct)}% complete` : ''}. ` +
|
||||
'Reads and writes are blocked until it completes; retry shortly.',
|
||||
details: { ...migSnap.migration }
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
const hnswSize = this.index.size()
|
||||
const metadataStats = await this.metadataIndex.getStats()
|
||||
const graphSize = await this.graphIndex.size()
|
||||
|
|
@ -12894,6 +12988,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// Memory management escape hatches over the RAM-derived auto limits
|
||||
maxQueryLimit: config?.maxQueryLimit ?? undefined,
|
||||
reservedQueryMemory: config?.reservedQueryMemory ?? undefined,
|
||||
// Migration LOCK wait budget — left undefined when omitted so
|
||||
// awaitMigrationLock() applies its 30 s default at the read site.
|
||||
migrationWaitTimeoutMs: config?.migrationWaitTimeoutMs ?? undefined,
|
||||
// Vector index configuration (8.0) — algorithm-neutral surface with
|
||||
// `recall` preset + `persistMode` (folded in from 7.x's hnswPersistMode).
|
||||
// See BRAINY-8.0-RENAME-COORDINATION § A.2 + § G.2.
|
||||
|
|
@ -12961,12 +13058,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
return
|
||||
}
|
||||
|
||||
// No-freeze deference (rc.8): when the vector provider is running its own
|
||||
// non-blocking background build-new→verify→swap migration, a first query must
|
||||
// NOT trigger brainy's blocking force-rebuild — cor owns the index and serves
|
||||
// correct reads from canonical until it verifies-and-swaps. Defer (no
|
||||
// `lazyRebuildCompleted` latch) so the check re-runs each query: once the swap
|
||||
// lands, `index.size() > 0` above ends the lazy path normally.
|
||||
// Migration LOCK (#18) deference: while the vector provider runs its one-time
|
||||
// 7.x → 8.0 rebuild-from-canonical, a first query must NOT trigger brainy's
|
||||
// force-rebuild — the provider owns that index. Normally unreachable here: the
|
||||
// data-plane lock (awaitMigrationLock) makes the caller wait upstream, so a
|
||||
// query only reaches this point once the migration has cleared. Defensive
|
||||
// (no `lazyRebuildCompleted` latch) so the check re-runs: once the provider
|
||||
// clears the lock, `index.size() > 0` above ends the lazy path normally.
|
||||
if (this.providerIsMigrating(this.index)) {
|
||||
return
|
||||
}
|
||||
|
|
@ -13152,10 +13250,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// three from the canonical records (not just the empty ones below).
|
||||
const epochStale = this._indexEpochStale
|
||||
|
||||
// No-freeze deference (rc.8): when a native provider is running its OWN
|
||||
// non-blocking background build-new→verify→swap migration, brainy must NOT
|
||||
// rebuild that index — the provider owns it and serves correct reads from
|
||||
// canonical until it verifies-and-swaps, then calls brain.stampBrainFormat().
|
||||
// Migration LOCK (#18) deference: when a native provider holds the lock for
|
||||
// its one-time 7.x → 8.0 rebuild-from-canonical, brainy must NOT rebuild
|
||||
// that index — the provider owns it in place and clears the lock only after
|
||||
// it verifies and calls brain.stampBrainFormat(). Reads and writes are held
|
||||
// by awaitMigrationLock meanwhile (nothing serves from a half-built index).
|
||||
// Gated per-index, so a non-migrating sibling still rebuilds when it needs
|
||||
// to; a migrating provider is skipped even under epoch-drift or size()===0.
|
||||
const metadataMigrating = this.providerIsMigrating(this.metadataIndex)
|
||||
|
|
@ -13238,9 +13337,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
|
||||
// Consistency verification: metadata index must match storage entity count.
|
||||
// If mismatch, the rebuild missed entities — force a second attempt. Skipped
|
||||
// when the metadata provider is mid-migration: a 0 count there reflects the
|
||||
// background swap in progress (it serves from canonical), not a missed
|
||||
// rebuild, so forcing a second rebuild would break deference.
|
||||
// when the metadata provider holds the migration lock: a 0 count there
|
||||
// reflects its in-place rebuild in progress, not a missed rebuild, so
|
||||
// forcing a second rebuild would collide with the provider's own.
|
||||
if (metadataCountAfter === 0 && totalCount > 0 && !metadataMigrating) {
|
||||
console.error(
|
||||
`[Brainy] CRITICAL: Metadata index has 0 entries but storage has ${totalCount} entities. ` +
|
||||
|
|
@ -13258,11 +13357,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// the (idempotent) rebuild; the marker is never advanced ahead of the
|
||||
// indexes it certifies. A no-op when the epoch was already current.
|
||||
//
|
||||
// EXCEPT while a provider is mid-migration (the no-freeze path): brainy
|
||||
// deferred that index's rebuild, so it is NOT yet at this epoch. The marker
|
||||
// (which certifies ALL derived indexes) must not advance ahead of the index
|
||||
// cor is still building — cor calls brain.stampBrainFormat() once its
|
||||
// background build-new→verify→swap has verified-and-swapped.
|
||||
// EXCEPT while a provider holds the migration lock: brainy deferred that
|
||||
// index's rebuild, so it is NOT yet at this epoch. The marker (which
|
||||
// certifies ALL derived indexes) must not advance ahead of the index the
|
||||
// provider is still rebuilding in place — the provider calls
|
||||
// brain.stampBrainFormat() itself once it has verified parity.
|
||||
if (!anyMigrating) {
|
||||
await this.stampBrainFormatIfNeeded()
|
||||
}
|
||||
|
|
@ -13304,26 +13403,163 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
|
||||
/**
|
||||
* @description Whether an index provider is running its own non-blocking
|
||||
* background migration (the no-freeze build-new→verify→swap path). A native
|
||||
* provider's `init()` flips an OPTIONAL `isMigrating(): boolean` to true the
|
||||
* moment it detects a large epoch-drift and keeps it true until it has
|
||||
* verified-and-swapped its new derived index. While true, brainy DEFERS its
|
||||
* own rebuild of that index and lets the provider own it (the provider serves
|
||||
* correct reads from canonical meanwhile) — never a minutes-long blocking
|
||||
* rebuild-on-open. Feature-detected: a provider that omits `isMigrating`
|
||||
* (every JS index) is treated as not migrating, so behavior is unchanged.
|
||||
* @description Whether an index provider holds the coordinated migration LOCK
|
||||
* (#18). A native provider's `init()` flips an OPTIONAL `isMigrating(): boolean`
|
||||
* to `true` the moment it detects a large 7.x epoch-drift and keeps it `true`
|
||||
* while it rebuilds all derived indexes IN PLACE from the canonical records,
|
||||
* clearing it only after it has verified parity and stamped the marker. While
|
||||
* `true`, brainy BLOCKS/QUEUES data-plane reads AND writes on the lock (see
|
||||
* {@link awaitMigrationLock}) so no operation touches a half-built index —
|
||||
* the "unknown/halfway state" is closed by waiting for a known-good one.
|
||||
* Feature-detected: a provider that omits `isMigrating` (every JS index) is
|
||||
* treated as not migrating, so behavior is unchanged.
|
||||
* @param provider - A registered index provider (metadata / vector / graph).
|
||||
* @returns `true` only when the provider exposes `isMigrating()` and it reports
|
||||
* an in-flight migration.
|
||||
*/
|
||||
private providerIsMigrating(provider: unknown): boolean {
|
||||
return (
|
||||
provider != null &&
|
||||
typeof (provider as { isMigrating?: () => boolean }).isMigrating === 'function' &&
|
||||
(provider as { isMigrating: () => boolean }).isMigrating() === true
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* @description `true` when ANY index provider (metadata / vector / graph) holds
|
||||
* the migration LOCK. The single predicate the data-plane gate and the status
|
||||
* surface consult; a brain with no migrating provider evaluates three cheap
|
||||
* feature-detects and returns `false`.
|
||||
*/
|
||||
private anyProviderMigrating(): boolean {
|
||||
return (
|
||||
this.providerIsMigrating(this.metadataIndex) ||
|
||||
this.providerIsMigrating(this.index) ||
|
||||
this.providerIsMigrating(this.graphIndex)
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Rich migration progress relayed verbatim from whichever provider
|
||||
* exposes the OPTIONAL `migrationStatus(): { phase?, index?, percent?, … } | null`.
|
||||
* The provider owns the rebuild, so it owns the truth of the percentage; brainy
|
||||
* surfaces it through {@link getIndexStatus} and {@link health}. Returns `null`
|
||||
* when no provider reports structured progress (brainy then shows only
|
||||
* `migrating: true` + `elapsedMs`). Feature-detected and best-effort.
|
||||
*/
|
||||
private providerMigrationStatus(): MigrationProgress | null {
|
||||
for (const provider of [this.metadataIndex, this.index, this.graphIndex]) {
|
||||
const fn = (provider as { migrationStatus?: () => unknown }).migrationStatus
|
||||
if (typeof fn === 'function') {
|
||||
try {
|
||||
const status = fn.call(provider)
|
||||
if (status && typeof status === 'object') {
|
||||
return status as MigrationProgress
|
||||
}
|
||||
} catch {
|
||||
// best-effort: a provider status hiccup never breaks the gate/probe
|
||||
}
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Block-and-queue on the coordinated migration LOCK (#18). Called
|
||||
* at the {@link ensureInitialized} choke point (and once inside `init()` before
|
||||
* VFS bootstrap), so every data-plane read and write — and startup itself —
|
||||
* waits here while a native provider runs its one-time 7.x → 8.0
|
||||
* rebuild-from-canonical. The caller gets the correct answer, never a partial
|
||||
* read of a half-built index and never a lost write. A brain that is not
|
||||
* migrating pays a single boolean check and returns immediately.
|
||||
*
|
||||
* IMPORTANT: this timeout bounds THE CALLER'S WAIT, not the migration. The
|
||||
* migration itself is unbounded — the native provider rebuilds a billion-scale
|
||||
* brain for as long as it needs; brainy cannot and does not abort it. The lock
|
||||
* is polled every {@link Brainy.MIGRATION_POLL_INTERVAL_MS}; a small/medium
|
||||
* upgrade clears well within `migrationWaitTimeoutMs` (default 30 s) and the
|
||||
* caller resumes transparently. If the wait outlives the budget, a retryable
|
||||
* {@link MigrationInProgressError} is thrown — the upgrade keeps running in the
|
||||
* background, so the caller retries, watches `getIndexStatus().migration`
|
||||
* (never gated — the primary large-upgrade signal for a readiness probe), or,
|
||||
* for a very large brain, runs the offline migrator. The block/release lines
|
||||
* log once per window; the flags reset on release so a later upgrade re-logs.
|
||||
*/
|
||||
private async awaitMigrationLock(): Promise<void> {
|
||||
if (!this.anyProviderMigrating()) return // fast path — the overwhelming common case
|
||||
|
||||
const startedAt = Date.now()
|
||||
const timeoutMs = this.config.migrationWaitTimeoutMs ?? 30_000
|
||||
|
||||
if (!this._migrationBlockLogged) {
|
||||
this._migrationBlockLogged = true
|
||||
prodLog.info(
|
||||
'[Brainy] Auto-upgrade in progress (7.x → 8.0): reads and writes are blocked ' +
|
||||
'until the rebuild completes. This is automatic and one-time; data is safe.'
|
||||
)
|
||||
}
|
||||
|
||||
while (this.anyProviderMigrating()) {
|
||||
const remaining = timeoutMs - (Date.now() - startedAt)
|
||||
if (remaining <= 0) {
|
||||
const status = this.providerMigrationStatus()
|
||||
throw new MigrationInProgressError(
|
||||
`Brain is still upgrading (7.x → 8.0)` +
|
||||
(status?.percent !== undefined ? `, ${Math.round(status.percent)}% complete` : '') +
|
||||
` after ${Math.round((Date.now() - startedAt) / 1000)}s. The wait timed out; the upgrade ` +
|
||||
`continues in the background. Retry shortly, watch brain.getIndexStatus().migration for ` +
|
||||
`progress, raise config.migrationWaitTimeoutMs to wait longer, or for a very large brain ` +
|
||||
`run the offline migrator. Nothing is lost.`,
|
||||
Date.now() - startedAt,
|
||||
status?.percent
|
||||
)
|
||||
}
|
||||
await new Promise((r) => setTimeout(r, Math.min(Brainy.MIGRATION_POLL_INTERVAL_MS, remaining)))
|
||||
}
|
||||
|
||||
if (this._migrationBlockLogged && !this._migrationReleaseLogged) {
|
||||
this._migrationReleaseLogged = true
|
||||
prodLog.info(
|
||||
`[Brainy] Auto-upgrade complete in ${Math.round((Date.now() - startedAt) / 1000)}s — ` +
|
||||
'reads and writes resumed.'
|
||||
)
|
||||
}
|
||||
// Reset the once-per-window guards so a subsequent upgrade in this same
|
||||
// instance re-logs cleanly and re-clocks its elapsed (a fresh observed-at).
|
||||
this._migrationBlockLogged = false
|
||||
this._migrationReleaseLogged = false
|
||||
this._migrationObservedAt = null
|
||||
}
|
||||
|
||||
/**
|
||||
* @description The lock-exempt observability snapshot consumed by
|
||||
* {@link getIndexStatus} and {@link health}: whether the brain is upgrading and,
|
||||
* if so, the provider's structured {@link MigrationProgress} enriched with a
|
||||
* `startedAt` / `elapsedMs` that brainy tracks itself (so a provider that omits
|
||||
* timing still yields a live elapsed). Reads the migration flag without waiting,
|
||||
* so a readiness probe can always report `migrating` → HTTP 503 + Retry-After
|
||||
* rather than routing traffic into a half-built index. Lazily stamps the
|
||||
* observed-at clock on first sight and clears it once the lock releases.
|
||||
*/
|
||||
private migrationSnapshot(): { migrating: boolean; migration?: MigrationProgress } {
|
||||
if (!this.anyProviderMigrating()) {
|
||||
this._migrationObservedAt = null
|
||||
return { migrating: false }
|
||||
}
|
||||
if (this._migrationObservedAt === null) {
|
||||
this._migrationObservedAt = Date.now()
|
||||
}
|
||||
const provided = this.providerMigrationStatus()
|
||||
return {
|
||||
migrating: true,
|
||||
migration: {
|
||||
...provided,
|
||||
startedAt: provided?.startedAt ?? this._migrationObservedAt,
|
||||
elapsedMs: provided?.elapsedMs ?? Date.now() - this._migrationObservedAt
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check health of metadata indexes
|
||||
*
|
||||
|
|
@ -13341,7 +13577,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
indexEntryCount: number
|
||||
recommendation: string | null
|
||||
}> {
|
||||
await this.ensureInitialized()
|
||||
// Lock-exempt: diagnostics must answer while the brain upgrades.
|
||||
await this.ensureInitialized({ bypassMigrationLock: true })
|
||||
const result = await this.metadataIndex.validateConsistency()
|
||||
// Fold in a non-fatal index-rebuild failure recorded at init() so the degraded
|
||||
// state is observable through the same health surface (not just the console).
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ export type BrainyErrorType =
|
|||
| 'VALIDATION'
|
||||
| 'FIELD_NOT_INDEXED'
|
||||
| 'GRAPH_INDEX_NOT_READY'
|
||||
| 'MIGRATION_IN_PROGRESS'
|
||||
|
||||
/**
|
||||
* Custom error class for Brainy operations
|
||||
|
|
@ -253,3 +254,46 @@ export class GraphIndexNotReadyError extends BrainyError {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Thrown when a data-plane read or write is issued against a brain that is
|
||||
* running its one-time, automatic 7.x → 8.0 on-disk upgrade — the coordinated
|
||||
* migration LOCK. While a native provider rebuilds all derived indexes from the
|
||||
* canonical records, brainy blocks reads and writes so no operation touches a
|
||||
* half-built index; the caller waits for the correct answer rather than getting
|
||||
* a partial one. This error is raised ONLY when the wait exceeds the configured
|
||||
* window ({@link BrainyConfig.migrationWaitTimeoutMs}, default 30 s) — never
|
||||
* instead of a partial/incorrect result.
|
||||
*
|
||||
* It is `retryable`: the upgrade continues in the background. Retry shortly,
|
||||
* watch `brain.getIndexStatus().migration` for progress, or for a very large
|
||||
* brain run the offline migrator. Data is safe; nothing is lost. Consumers can
|
||||
* catch this (e.g. request middleware) and answer HTTP 503 + `Retry-After`.
|
||||
*
|
||||
* @example
|
||||
* try {
|
||||
* await brain.find({ query })
|
||||
* } catch (e) {
|
||||
* if (e instanceof MigrationInProgressError) {
|
||||
* res.set('Retry-After', '5').status(503).json({ upgrading: true, percent: e.percent })
|
||||
* return
|
||||
* }
|
||||
* throw e
|
||||
* }
|
||||
*/
|
||||
export class MigrationInProgressError extends BrainyError {
|
||||
/** Milliseconds the operation waited on the migration lock before timing out. */
|
||||
public readonly elapsedMs: number
|
||||
/** Latest observed migration progress (0–100), when the provider reports it. */
|
||||
public readonly percent?: number
|
||||
|
||||
constructor(message: string, elapsedMs: number, percent?: number, originalError?: Error) {
|
||||
super(message, 'MIGRATION_IN_PROGRESS', true, originalError)
|
||||
this.name = 'MigrationInProgressError'
|
||||
this.elapsedMs = elapsedMs
|
||||
this.percent = percent
|
||||
if (Error.captureStackTrace) {
|
||||
Error.captureStackTrace(this, MigrationInProgressError)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -137,6 +137,11 @@ export { RevisionConflictError } from './transaction/RevisionConflictError.js'
|
|||
// transact/with() when a referenced entity or relation does not exist.
|
||||
export { EntityNotFoundError, RelationNotFoundError } from './errors/notFound.js'
|
||||
|
||||
// Base error + typed migration-lock error — thrown by any data-plane call while a
|
||||
// brain runs its one-time 7.x→8.0 upgrade; catch to answer HTTP 503 + Retry-After.
|
||||
export { BrainyError, MigrationInProgressError, GraphIndexNotReadyError } from './errors/brainyError.js'
|
||||
export type { BrainyErrorType } from './errors/brainyError.js'
|
||||
|
||||
// ============= 8.0 Db API — generational MVCC =============
|
||||
// Immutable database values: brain.now() / brain.transact() / brain.asOf() /
|
||||
// db.with() / db.persist() / Brainy.load(). See src/db/ for the record layer.
|
||||
|
|
|
|||
|
|
@ -1575,6 +1575,31 @@ export interface BrainyStats {
|
|||
/**
|
||||
* Brainy configuration
|
||||
*/
|
||||
/**
|
||||
* Structured progress of the one-time, automatic 7.x → 8.0 migration (the
|
||||
* coordinated LOCK). Relayed verbatim from the native provider's optional
|
||||
* `migrationStatus()` and surfaced on `getIndexStatus().migration` while the
|
||||
* brain is upgrading. Every field is optional — a provider may report only a
|
||||
* phase, or nothing (in which case `getIndexStatus()` shows `migrating: true`
|
||||
* plus `elapsedMs` alone).
|
||||
*/
|
||||
export interface MigrationProgress {
|
||||
/** Coarse stage, e.g. `'rebuilding'` | `'verifying'`. */
|
||||
phase?: string
|
||||
/** Which derived index is currently rebuilding. */
|
||||
index?: 'metadata' | 'vector' | 'graph'
|
||||
/** Overall progress, 0–100. */
|
||||
percent?: number
|
||||
/** Canonical entities processed so far. */
|
||||
entitiesDone?: number
|
||||
/** Total canonical entities to process. */
|
||||
entitiesTotal?: number
|
||||
/** When the migration was first observed (epoch ms). */
|
||||
startedAt?: number
|
||||
/** Milliseconds elapsed since the migration was first observed. */
|
||||
elapsedMs?: number
|
||||
}
|
||||
|
||||
export interface BrainyConfig {
|
||||
/**
|
||||
* Storage configuration: either a config object resolved through the
|
||||
|
|
@ -1626,6 +1651,28 @@ export interface BrainyConfig {
|
|||
*/
|
||||
disableAutoRebuild?: boolean
|
||||
|
||||
/**
|
||||
* How long (ms) an operation **waits** on the coordinated 7.x → 8.0 migration
|
||||
* before throwing a retryable `MigrationInProgressError`.
|
||||
*
|
||||
* IMPORTANT: this bounds the *caller's wait*, NOT the migration. The migration
|
||||
* itself is **unbounded** — a native provider rebuilds a billion-scale brain
|
||||
* for as long as it needs, and this timeout never interrupts it. While the
|
||||
* rebuild runs, reads and writes (and `init()` before it serves) block so no
|
||||
* operation touches a half-built index:
|
||||
* - **Small/medium upgrade (< this window):** the caller waits, then resumes
|
||||
* transparently — no error.
|
||||
* - **Large upgrade (> this window):** the caller gets a retryable
|
||||
* `MigrationInProgressError` (the rebuild continues in the background);
|
||||
* `getIndexStatus().migrating` — never gated — is the readiness-probe signal
|
||||
* that maps to HTTP 503 + Retry-After so an orchestrator waits rather than
|
||||
* routing traffic in.
|
||||
*
|
||||
* Raise it for a latency-tolerant batch job (wait longer / effectively wait
|
||||
* through); lower it to fail fast on a request path. Default: `30000` (30 s).
|
||||
*/
|
||||
migrationWaitTimeoutMs?: number
|
||||
|
||||
/**
|
||||
* Vector index configuration (Brainy 8.0).
|
||||
*
|
||||
|
|
|
|||
170
tests/unit/migration-lock.test.ts
Normal file
170
tests/unit/migration-lock.test.ts
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
/**
|
||||
* Migration LOCK (#18) — coordinated, automatic 7.x → 8.0 auto-upgrade.
|
||||
*
|
||||
* Exercises the real block-and-queue behavior of `awaitMigrationLock` at the
|
||||
* `ensureInitialized` choke point: while ANY index provider reports
|
||||
* `isMigrating() === true`, data-plane reads and writes WAIT (no operation
|
||||
* touches a half-built index); observability (`getIndexStatus`, `health`) and
|
||||
* the lock-clearing `stampBrainFormat` stay exempt; and a lock that outlives the
|
||||
* configured window surfaces a retryable `MigrationInProgressError`.
|
||||
*
|
||||
* A native (cortex) provider owns the real migration; here we simulate the lock
|
||||
* by feature-detect-injecting `isMigrating()` onto a live provider, exactly as
|
||||
* the production feature-detection reads it.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach } from 'vitest'
|
||||
import { Brainy, NounType, MigrationInProgressError } from '../../src/index.js'
|
||||
import { GraphAdjacencyIndex } from '../../src/graph/graphAdjacencyIndex.js'
|
||||
|
||||
/** Force the graph provider to hold (or release) the migration lock. */
|
||||
function setMigrating(brain: any, migrating: boolean): void {
|
||||
brain.graphIndex.isMigrating = () => migrating
|
||||
}
|
||||
|
||||
describe('Migration LOCK (#18) — coordinated 7.x→8.0 auto-upgrade', () => {
|
||||
let brain: any
|
||||
|
||||
beforeEach(async () => {
|
||||
brain = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'memory' },
|
||||
migrationWaitTimeoutMs: 5000 // generous; timing tests clear well within it
|
||||
})
|
||||
await brain.init()
|
||||
})
|
||||
|
||||
it('does not gate operations when no provider is migrating (fast path)', async () => {
|
||||
const id = await brain.add({ data: 'hello', type: NounType.Concept })
|
||||
expect(id).toBeTruthy()
|
||||
const status = await brain.getIndexStatus()
|
||||
expect(status.migrating).toBe(false)
|
||||
expect(status.migration).toBeUndefined()
|
||||
})
|
||||
|
||||
it('getIndexStatus() reports migrating WITHOUT blocking (lock-exempt observability)', async () => {
|
||||
setMigrating(brain, true)
|
||||
const status = await brain.getIndexStatus() // must resolve, never hang
|
||||
expect(status.migrating).toBe(true)
|
||||
expect(status.migration).toBeDefined()
|
||||
expect(typeof status.migration.elapsedMs).toBe('number')
|
||||
})
|
||||
|
||||
it('health() surfaces the migration as a warn check without blocking', async () => {
|
||||
setMigrating(brain, true)
|
||||
const h = await brain.health() // lock-exempt
|
||||
expect(h.overall).toBe('warn')
|
||||
expect(h.checks.some((c: any) => c.name === 'migration')).toBe(true)
|
||||
})
|
||||
|
||||
it('checkHealth() answers during a migration (lock-exempt diagnostics)', async () => {
|
||||
setMigrating(brain, true)
|
||||
const r = await brain.checkHealth()
|
||||
expect(r).toHaveProperty('healthy')
|
||||
})
|
||||
|
||||
it('blocks a write while migrating (poll path), then completes when the lock clears', async () => {
|
||||
setMigrating(brain, true)
|
||||
let resolved = false
|
||||
const p = brain
|
||||
.add({ data: 'queued write', type: NounType.Concept })
|
||||
.then((id: string) => {
|
||||
resolved = true
|
||||
return id
|
||||
})
|
||||
|
||||
// Still blocked shortly after issue.
|
||||
await new Promise((r) => setTimeout(r, 40))
|
||||
expect(resolved).toBe(false)
|
||||
|
||||
// Clearing the lock releases the queued write against the good indexes.
|
||||
setMigrating(brain, false)
|
||||
const id = await p
|
||||
expect(resolved).toBe(true)
|
||||
expect(id).toBeTruthy()
|
||||
})
|
||||
|
||||
it('blocks a read while migrating, then releases when the flag clears (poll path)', async () => {
|
||||
let migrating = true
|
||||
brain.graphIndex.isMigrating = () => migrating
|
||||
|
||||
const p = brain.find({ type: NounType.Concept }) // a read → gated
|
||||
let resolved = false
|
||||
p.then(() => {
|
||||
resolved = true
|
||||
})
|
||||
|
||||
await new Promise((r) => setTimeout(r, 40))
|
||||
expect(resolved).toBe(false)
|
||||
|
||||
// Clearing the flag lets the next poll release the read against good indexes.
|
||||
migrating = false
|
||||
await expect(p).resolves.toEqual([])
|
||||
})
|
||||
|
||||
it('throws a retryable MigrationInProgressError when the lock outlives the timeout', async () => {
|
||||
const shortBrain: any = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'memory' },
|
||||
migrationWaitTimeoutMs: 120
|
||||
})
|
||||
await shortBrain.init()
|
||||
setMigrating(shortBrain, true) // never clears
|
||||
|
||||
await expect(shortBrain.find({ type: NounType.Concept })).rejects.toBeInstanceOf(
|
||||
MigrationInProgressError
|
||||
)
|
||||
try {
|
||||
await shortBrain.add({ data: 'x', type: NounType.Concept })
|
||||
throw new Error('expected MigrationInProgressError')
|
||||
} catch (e: any) {
|
||||
expect(e).toBeInstanceOf(MigrationInProgressError)
|
||||
expect(e.retryable).toBe(true)
|
||||
expect(typeof e.elapsedMs).toBe('number')
|
||||
}
|
||||
})
|
||||
|
||||
it('stampBrainFormat() is NOT gated — cor clears the lock through it (no deadlock)', async () => {
|
||||
setMigrating(brain, true)
|
||||
// Must resolve, not hang, even while a migration is "in progress".
|
||||
await expect(brain.stampBrainFormat()).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('close() is NOT gated — an instance can shut down mid-migration', async () => {
|
||||
setMigrating(brain, true)
|
||||
await expect(brain.close()).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('C1: init() waits out a migration BEFORE VFS bootstrap (no deadlock, no failure)', async () => {
|
||||
// The critical regression: init()'s own VFS bootstrap (get/add/find on the
|
||||
// root) routes through the gated ensureInitialized. If a provider migrates
|
||||
// during init, init must WAIT for the lock to clear BEFORE bootstrapping VFS —
|
||||
// otherwise it deadlocks (< timeout) or throws MigrationInProgressError past
|
||||
// the timeout. We make the graph provider report migrating for a short window
|
||||
// spanning init, then clear it, and assert init completes and the brain works.
|
||||
const orig = (GraphAdjacencyIndex.prototype as any).isMigrating
|
||||
let migrating = true
|
||||
;(GraphAdjacencyIndex.prototype as any).isMigrating = () => migrating
|
||||
const clearMigration = setTimeout(() => {
|
||||
migrating = false
|
||||
}, 60)
|
||||
try {
|
||||
const b: any = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'memory' },
|
||||
migrationWaitTimeoutMs: 5000
|
||||
})
|
||||
await b.init() // must resolve once the migration clears — not hang, not throw
|
||||
expect(b.isInitialized).toBe(true)
|
||||
// VFS bootstrapped post-migration → normal write/read work end-to-end.
|
||||
const id = await b.add({ data: 'after upgrade', type: NounType.Concept })
|
||||
expect(id).toBeTruthy()
|
||||
expect(await b.get(id)).not.toBeNull()
|
||||
await b.close()
|
||||
} finally {
|
||||
clearTimeout(clearMigration)
|
||||
if (orig) (GraphAdjacencyIndex.prototype as any).isMigrating = orig
|
||||
else delete (GraphAdjacencyIndex.prototype as any).isMigrating
|
||||
}
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue