feat: scaled transact budgets + labeled timeout diagnostics + envelope docs
- The transact apply budget scales with the batch — max(30s, opCount x 2s) — or is exactly the caller's new TransactOptions.timeoutMs. A flat 30s cap silently limited honest bulk work to ~15 operations on network-attached storage (~2s/op measured in a production import) while looking generous for small batches. Internal batch paths (removeMany chunks) get the same scaling via the shared transactTimeoutBudget helper. - TransactionTimeoutError now reports the operation it stopped at as i/N with the operation's name, elapsed vs budgeted time, and states the batch rolled back atomically and is retryable; context carries the fields programmatically. - The transact operational envelope is documented (optimistic-concurrency guide): budget math, chunking with ifAbsent idempotency, when to reach for addMany vs transact, and the precompute pattern (embedBatch + per-op vector) that keeps inference out of the commit path. - Embedding claims made honest per an end-to-end probe of the built dist: batch and single embedding paths produce bit-identical vectors and a vector-supplied add is fully searchable, but batch throughput on the default WASM engine measures comparable to sequential (~160ms/text) — the 5-10x speedup claim in the addMany JSDoc is replaced with the measured reality and the actual win (inference outside the budgeted write path).
This commit is contained in:
parent
e0f6e7722f
commit
6ef9fcb7a2
7 changed files with 175 additions and 15 deletions
22
RELEASES.md
22
RELEASES.md
|
|
@ -10,6 +10,28 @@ Full auto-generated changelog: `CHANGELOG.md` · Releases: https://github.com/so
|
|||
|
||||
---
|
||||
|
||||
## v8.7.0 — 2026-07-17 (bulk-transact ergonomics: scaled budgets + timeout telemetry)
|
||||
|
||||
The bulk-import ergonomics release, from a consumer's measured production incident (a serial
|
||||
import on network-attached storage at ~2 s/op met a flat 30 s transaction budget):
|
||||
|
||||
- **The transact apply budget now scales with the batch** — `max(30 s, opCount × 2 s)` — or
|
||||
is exactly what you pass as the new `TransactOptions.timeoutMs`. A flat 30 s cap silently
|
||||
limited honest bulk work to ~15 operations on slow disks while looking generous for small
|
||||
batches. Internal batch paths (e.g. `removeMany` chunks) get the same scaling.
|
||||
- **`TransactionTimeoutError` is a diagnosis, not just a failure**: it now reports the
|
||||
operation it stopped at as `i/N` with the operation's name, elapsed vs budgeted time, and
|
||||
states the batch rolled back atomically and is retryable. Its `context` carries the same
|
||||
fields programmatically.
|
||||
- **The transact envelope is documented** — batch sizing, budget math, chunking with
|
||||
`ifAbsent` idempotency, and the precompute pattern (`embedBatch` + per-op `vector`) that
|
||||
keeps model inference out of the commit path. Guide: `docs/guides/optimistic-concurrency.md`.
|
||||
- Note: `brain.embed()` / `brain.embedBatch()` (the precompute APIs) already ship — public,
|
||||
with native-provider passthrough, verified end-to-end (batch and single paths produce
|
||||
bit-identical vectors; a vector-supplied `add` is fully searchable). Honest measurement:
|
||||
on the default WASM engine, batch throughput ≈ sequential (~160 ms/text) — the precompute
|
||||
win is keeping inference out of the budgeted commit path, not raw embedding speed.
|
||||
|
||||
## v8.6.0 — 2026-07-17 (brain.auditGraph — the graph-truth verification instrument)
|
||||
|
||||
A minor release adding one new public API, from the fleet's graph-trust program: a read-only
|
||||
|
|
|
|||
|
|
@ -180,3 +180,35 @@ Brainy 8.0 has exactly two write-coordination counters, at two granularities:
|
|||
They compose: a `transact()` batch can carry per-entity `ifRev` checks *and* a whole-store `ifAtGeneration`; any failed check rejects the entire batch before anything is staged. Generations also power snapshots and time travel (`brain.now()`, `brain.asOf()`, `db.persist()`) — see the [consistency model](../concepts/consistency-model.md) and [Snapshots & Time Travel](./snapshots-and-time-travel.md).
|
||||
|
||||
A snapshot or historical view captures each entity *including* its `_rev` at that moment, so reading the past and writing back with `ifRev` against the live state works exactly as you'd hope: the write fails if the entity moved since the state you copied from.
|
||||
|
||||
## The transact envelope: batch size, budget, and bulk imports
|
||||
|
||||
`transact()` applies its batch atomically under one commit — which means the whole batch
|
||||
shares one **apply budget**. Since 8.7.0 the budget scales with the batch:
|
||||
`max(30 s, opCount × 2 s)`, or exactly what you pass as `timeoutMs`. A tripped budget rolls
|
||||
the entire batch back (nothing partial survives) and throws a retryable
|
||||
`TransactionTimeoutError` that names the operation it stopped at, the batch size, and the
|
||||
elapsed vs budgeted time — a diagnosis, not just a failure:
|
||||
|
||||
```
|
||||
Transaction timed out at operation 41/120 ('add') — 246012ms elapsed, budget 240000ms.
|
||||
The batch rolled back atomically; retry with a higher timeoutMs or a smaller batch.
|
||||
```
|
||||
|
||||
Practical envelope guidance for bulk work:
|
||||
|
||||
1. **Precompute embeddings outside the commit path.** Embedding inside `transact()` spends
|
||||
the budget on model inference. Use `brain.embedBatch(texts)` and pass each vector via
|
||||
the op's `vector` field — the commit then pays only storage costs, and a retried batch
|
||||
never re-pays inference. (The win is *where* the inference happens, not raw embedding
|
||||
throughput: on the default WASM engine, batch and sequential embedding measure
|
||||
comparably, ~160 ms/text; native embedding providers may batch faster.)
|
||||
2. **Chunk very large imports** into batches of a few hundred ops with one `transact()`
|
||||
each. You lose whole-import atomicity but keep per-chunk atomicity, bounded memory, and
|
||||
resumability — pair with `ifAbsent` upserts so a retried chunk is idempotent.
|
||||
3. **Slow disks change the math, not the contract.** On network-attached storage a single
|
||||
op can cost ~2 s (canonical write + fsync + index maintenance). The scaled default
|
||||
absorbs that; pass an explicit `timeoutMs` only when you know better than the scale.
|
||||
4. **`addMany`/`relateMany` are the convenience tier** — they chunk and batch-embed for
|
||||
you, with per-item error reporting instead of batch atomicity. Choose by what you need:
|
||||
atomic-all-or-nothing → `transact()`; resilient bulk load → `addMany`.
|
||||
|
|
|
|||
|
|
@ -71,6 +71,7 @@ import type {
|
|||
} from './plugin.js'
|
||||
import { ConnectionsCodec } from './hnsw/connectionsCodec.js'
|
||||
import { TransactionManager } from './transaction/TransactionManager.js'
|
||||
import { transactTimeoutBudget } from './transaction/Transaction.js'
|
||||
import { RevisionConflictError } from './transaction/RevisionConflictError.js'
|
||||
import { EntityNotFoundError, RelationNotFoundError } from './errors/notFound.js'
|
||||
import {
|
||||
|
|
@ -1752,7 +1753,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
captureAndCheck({ nouns, verbs } as CommitBeforeImages)
|
||||
}
|
||||
await this.generationStore.runWithoutGeneration(() =>
|
||||
this.transactionManager.executeTransaction(run)
|
||||
this.transactionManager.executeTransaction(run, {
|
||||
timeout: transactTimeoutBudget(
|
||||
(touched.nouns?.length ?? 0) + (touched.verbs?.length ?? 0)
|
||||
)
|
||||
})
|
||||
)
|
||||
const timestamp = Date.now()
|
||||
// Bootstrap writes are not generation-stamped; emit without one.
|
||||
|
|
@ -1764,7 +1769,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
receipt = await this.generationStore.commitSingleOp({
|
||||
touched,
|
||||
precommit: captureAndCheck,
|
||||
execute: () => this.transactionManager.executeTransaction(run)
|
||||
execute: () =>
|
||||
this.transactionManager.executeTransaction(run, {
|
||||
timeout: transactTimeoutBudget(
|
||||
(touched.nouns?.length ?? 0) + (touched.verbs?.length ?? 0)
|
||||
)
|
||||
})
|
||||
})
|
||||
} catch (err) {
|
||||
// A failed rollback that left the store inconsistent (a remove/update
|
||||
|
|
@ -6655,10 +6665,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
/**
|
||||
* Add multiple entities in a single batch operation
|
||||
*
|
||||
* Uses batch embedding (embedBatch) to pre-compute all vectors in a single
|
||||
* WASM forward pass instead of N individual embed() calls, providing 5-10x
|
||||
* speedup on bulk inserts. Automatically adapts batch size and parallelism
|
||||
* to the storage adapter (e.g., smaller batches for cloud storage).
|
||||
* Uses batch embedding (embedBatch) to pre-compute all vectors before any
|
||||
* storage write, keeping model inference out of the per-item write path.
|
||||
* (On the default WASM engine, batch throughput is comparable to sequential
|
||||
* embed() calls — measured ~160 ms/text either way; a native embedding
|
||||
* provider may batch faster.) Automatically adapts batch size and
|
||||
* parallelism to the storage adapter (e.g., smaller batches for cloud
|
||||
* storage).
|
||||
*
|
||||
* @param params - Batch add parameters
|
||||
* @param params.items - Array of AddParams (same shape as brain.add())
|
||||
|
|
@ -7797,11 +7810,17 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
ifAtGeneration: options?.ifAtGeneration,
|
||||
precommit: casPrecommit,
|
||||
execute: async () => {
|
||||
await this.transactionManager.executeTransaction(async (tx) => {
|
||||
for (const operation of plan.operations) {
|
||||
tx.addOperation(operation)
|
||||
}
|
||||
})
|
||||
await this.transactionManager.executeTransaction(
|
||||
async (tx) => {
|
||||
for (const operation of plan.operations) {
|
||||
tx.addOperation(operation)
|
||||
}
|
||||
},
|
||||
// Budget scales with the batch (or the caller's explicit
|
||||
// timeoutMs): a flat 30s cap silently limited honest bulk work
|
||||
// to ~15 ops on network disks (~2s/op measured in the field).
|
||||
{ timeout: transactTimeoutBudget(plan.operations.length, options?.timeoutMs) }
|
||||
)
|
||||
}
|
||||
}))
|
||||
} catch (err) {
|
||||
|
|
|
|||
|
|
@ -116,6 +116,16 @@ export interface TransactOptions {
|
|||
* record is staged.
|
||||
*/
|
||||
ifAtGeneration?: number
|
||||
/**
|
||||
* Budget (ms) for the atomic apply phase. When omitted, the budget SCALES
|
||||
* with the batch: `max(30 000, opCount × 2 000)` — production imports on
|
||||
* network-attached disks measure ~2 s per operation, so a flat 30 s budget
|
||||
* silently capped honest bulk work at ~15 operations. A tripped budget
|
||||
* rolls the whole batch back and throws a retryable
|
||||
* `TransactionTimeoutError` naming the operation it stopped at, the batch
|
||||
* size, and the elapsed/budget times.
|
||||
*/
|
||||
timeoutMs?: number
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -37,6 +37,24 @@ const DEFAULT_OPTIONS: Required<TransactionOptions> = {
|
|||
maxRollbackRetries: 3
|
||||
}
|
||||
|
||||
/**
|
||||
* The apply-phase budget for a batch of `opCount` operations.
|
||||
*
|
||||
* An explicit override wins untouched. Otherwise the budget SCALES with the
|
||||
* batch: `max(30 000 ms, opCount × 2 000 ms)`. 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. Scaling keeps small transacts
|
||||
* fast-failing and gives bulk ones a budget proportional to the work they
|
||||
* actually asked for; a trip still rolls back atomically and throws a
|
||||
* retryable, fully-labeled TransactionTimeoutError.
|
||||
*/
|
||||
export function transactTimeoutBudget(opCount: number, override?: number): number {
|
||||
if (override !== undefined) return override
|
||||
return Math.max(30_000, opCount * 2_000)
|
||||
}
|
||||
|
||||
/**
|
||||
* Transaction class
|
||||
*/
|
||||
|
|
@ -114,7 +132,11 @@ export class Transaction implements TransactionContext {
|
|||
// into the catch below and rolls back like any other failure — it must
|
||||
// never bypass rollback.
|
||||
if (Date.now() - this.startTime > this.options.timeout) {
|
||||
throw new TransactionTimeoutError(this.options.timeout, i)
|
||||
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]
|
||||
|
|
|
|||
|
|
@ -77,11 +77,24 @@ export class InvalidTransactionStateError extends TransactionError {
|
|||
export class TransactionTimeoutError extends TransactionError {
|
||||
constructor(
|
||||
timeoutMs: number,
|
||||
operationIndex: number
|
||||
operationIndex: number,
|
||||
telemetry?: {
|
||||
elapsedMs?: number
|
||||
totalOperations?: number
|
||||
operationName?: string
|
||||
}
|
||||
) {
|
||||
const progress =
|
||||
telemetry?.totalOperations !== undefined
|
||||
? `${operationIndex}/${telemetry.totalOperations}`
|
||||
: String(operationIndex)
|
||||
const name = telemetry?.operationName ? ` ('${telemetry.operationName}')` : ''
|
||||
const elapsed =
|
||||
telemetry?.elapsedMs !== undefined ? `${telemetry.elapsedMs}ms elapsed, ` : ''
|
||||
super(
|
||||
`Transaction timed out after ${timeoutMs}ms at operation ${operationIndex}`,
|
||||
{ timeoutMs, operationIndex }
|
||||
`Transaction timed out at operation ${progress}${name} — ${elapsed}budget ${timeoutMs}ms. ` +
|
||||
`The batch rolled back atomically; retry with a higher timeoutMs or a smaller batch.`,
|
||||
{ timeoutMs, operationIndex, ...telemetry }
|
||||
)
|
||||
this.name = 'TransactionTimeoutError'
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@
|
|||
|
||||
import { describe, it, expect, beforeEach } from 'vitest'
|
||||
import { TransactionManager } from '../../src/transaction/TransactionManager.js'
|
||||
import { transactTimeoutBudget } from '../../src/transaction/Transaction.js'
|
||||
import { TransactionError } from '../../src/transaction/errors.js'
|
||||
|
||||
describe('TransactionManager', () => {
|
||||
|
|
@ -328,4 +329,45 @@ describe('TransactionManager', () => {
|
|||
expect(stats1).toEqual(stats2) // Same values
|
||||
})
|
||||
})
|
||||
|
||||
describe('Timeout budget + telemetry', () => {
|
||||
it('transactTimeoutBudget: explicit override wins; default scales with batch size', () => {
|
||||
expect(transactTimeoutBudget(1)).toBe(30_000) // small batches keep the 30s floor
|
||||
expect(transactTimeoutBudget(15)).toBe(30_000) // the old flat cap's break-even point
|
||||
expect(transactTimeoutBudget(100)).toBe(200_000) // 100 ops × 2s — bulk gets an honest budget
|
||||
expect(transactTimeoutBudget(1000, 5_000)).toBe(5_000) // caller override is untouched
|
||||
})
|
||||
|
||||
it('a tripped budget rolls back and names the operation, progress, and budget', async () => {
|
||||
const rolledBack: string[] = []
|
||||
|
||||
const failing = manager.executeTransaction(
|
||||
async (tx) => {
|
||||
tx.addOperation({
|
||||
name: 'slow-first-op',
|
||||
execute: async () => {
|
||||
await new Promise((r) => setTimeout(r, 30))
|
||||
return async () => {
|
||||
rolledBack.push('slow-first-op')
|
||||
}
|
||||
}
|
||||
})
|
||||
tx.addOperation({
|
||||
name: 'never-reached',
|
||||
execute: async () => undefined
|
||||
})
|
||||
},
|
||||
{ timeout: 5 } // the first op's 30ms sleep guarantees the pre-op-2 check trips
|
||||
)
|
||||
|
||||
await expect(failing).rejects.toThrow(TransactionError)
|
||||
const err = await failing.catch((e) => e)
|
||||
expect(err.name).toBe('TransactionTimeoutError')
|
||||
expect(err.message).toContain('operation 1/2') // which op, of how many
|
||||
expect(err.message).toContain("('never-reached')") // its name
|
||||
expect(err.message).toContain('budget 5ms') // the budget that tripped
|
||||
expect(err.message).toContain('rolled back') // the retryability statement
|
||||
expect(rolledBack).toEqual(['slow-first-op']) // the applied op was undone
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue