mirror of
https://github.com/garrytan/gbrain.git
synced 2026-08-14 00:48:18 +00:00
fix(schema): apply mutation batches atomically so a mid-batch failure leaves the pack untouched (#2581) (#3446)
Adversarial review: survived a hostile reviewer plus two independent refuters, each told to assume the PR was broken and to default to refuting when uncertain. 19 of 62 PRs cleared that bar. Applying a schema mutation batch was not atomic: a failure partway through left earlier mutations permanently written. Reproduced on disk — a failure at index 2 left mutation 0 applied with no way to tell from the pack's state that it was half-done. The fix validates the whole batch first and writes once, which makes partial application impossible by construction rather than by careful ordering. Verified before merge: the failure was reproduced by injecting one rather than reasoning about it; the PR's own tests fail when the fix is reverted; typecheck clean; MERGEABLE/CLEAN at 22/22 on the current base after batches 1-4 landed. Sequenced last deliberately — it collides with #3531 on docs/architecture/KEY_FILES.md and with #3667 on src/core/operations.ts, both of which landed earlier today. Known gap, recorded rather than hidden: lock contention under concurrent writers was reasoned about, not stress-tested.
This commit is contained in:
Masa
@@ -492,11 +492,11 @@ Key files (v0.40.7.0 additions):
|
||||
- `src/core/schema-pack/lint-rules.ts` — 12 pure rule functions. `withMutation`'s pre-write validation gate composes the 10 file-plane rules; the 2 DB-aware rules (`extractable_empty_corpus`, `mutation_count_anomaly`) need an engine. Single source of truth consumed by CLI lint + MCP `schema_lint` + the pre-write validation gate. New file-plane rule `link_regex_catastrophic_backtrack` — advisory ReDoS pre-screen flagging the classic nested-quantifier shapes (`(a+)+`, `(a*)*`, `(a+)*`, `(\w+)+`) in a link_type's `inference.regex` via `NESTED_QUANTIFIER_RE`. WARNING not error: a hard reject would disable the whole pack on upgrade (pages fall back to legacy typing). The runtime input-length cap in `redos-guard.ts` is the actual safety net; this rule tells the pack author to fix the pattern.
|
||||
- `src/core/schema-pack/redos-guard.ts` + `src/core/schema-pack/link-inference.ts` — ReDoS hardening for pack inference regexes. `redos-guard.ts` adds `MAX_REGEX_INPUT_CHARS` (default 64_000, env `GBRAIN_MAX_REGEX_INPUT_CHARS`) — a hard input-length cap, the real runtime safety net (catastrophic backtracking needs a long input; a link-extraction `context` is normally a sentence or short paragraph). Over the cap, `runRegexBounded` throws the tagged `RegexInputTooLargeError` and the regex is skipped (degrade-to-mentions) without entering the `node:vm`. `link-inference.ts:inferLinkTypeFromPack` no-budget branch (test contexts) now routes through `runRegexBounded` so the input-length cap + per-regex vm timeout (`PER_REGEX_TIMEOUT_MS = 50`) apply on every path (previously this branch ran `new RegExp(pattern).test(context)` unbounded — the one ReDoS hole with no timeout). Defensive hardening + diagnostics; the deterministic ~3100-file sync-wedge root cause remains open. Pinned by `test/redos-hardening.test.ts` + `test/schema-pack-lint-rules.test.ts`.
|
||||
- `src/core/schema-pack/query-cache-invalidator.ts` — `invalidateQueryCache(engine, sourceId?)` DELETEs query_cache rows so cached search results bound to old page types don't survive a schema mutation.
|
||||
- `src/core/schema-pack/mutate.ts` — 8-step `withMutation` skeleton (bundled-guard → lock → read → mutator → validate → atomic write → audit → invalidate). 11 mutation primitives: `addTypeToPack`, `removeTypeFromPack` (with reference check), `updateTypeOnPack`, `addAliasToType`, `removeAliasFromType`, `addPrefixToType`, `removePrefixFromType`, `addLinkTypeToPack`, `removeLinkTypeFromPack`, `setExtractableOnType`, `setExpertRoutingOnType`. Atomic write via `.tmp + fsync + rename` — the pack file on disk is NEVER partial. Inline minimal JSON→YAML emitter so YAML packs stay YAML (does NOT preserve comments — pin pack.json if you care about layout).
|
||||
- `src/core/schema-pack/mutate.ts` — 8-step `withMutation` skeleton (bundled-guard → lock → read → mutator → validate → atomic write → audit → invalidate) backs the 11 single-mutation primitives: `addTypeToPack`, `removeTypeFromPack` (with reference check), `updateTypeOnPack`, `addAliasToType`, `removeAliasFromType`, `addPrefixToType`, `removePrefixFromType`, `addLinkTypeToPack`, `removeLinkTypeFromPack`, `setExtractableOnType`, `setExpertRoutingOnType`. Each primitive's business-rule validation + transform is factored into a `build*Mutator(...)` pure `(manifest) => manifest` function shared with `applyMutationsAtomic` (the `schema_apply_mutations` batch entry point) so single-call and batched mutations can never validate differently. `applyMutationsAtomic` locks + reads the pack file ONCE, applies + lint-validates every mutation in the batch against an in-memory manifest, and calls `writePackManifest` at MOST ONCE — only after the whole batch checks out — so a batch that fails partway leaves the pack file byte-identical to its pre-batch state. Atomic single write via `.tmp + fsync + rename` — the pack file on disk is NEVER partial, for either a single mutation or a batch. Inline minimal JSON→YAML emitter so YAML packs stay YAML (does NOT preserve comments — pin pack.json if you care about layout).
|
||||
- `src/core/schema-pack/stats.ts` — `runStatsCore(engine, opts)` returns per-source + aggregate page counts + coverage % + `dead_prefixes` (declared prefixes with zero matching pages — agent drilldown signal). Multi-source aware (`sourceIds[]` federated, `sourceId` single, or whole-brain). PGLite + Postgres parity via `executeRaw`. Empty brain → coverage:1.0 (vacuous truth).
|
||||
- `src/core/schema-pack/sync.ts` — `runSyncCore(engine, opts)` chunked UPDATE in 1000-row batches per declared prefix. Concurrent writers never block on a single row >100ms. Write-side scoping via `ctx.sourceId` directly (NOT `sourceScopeOpts`, which inherits OAuth read federation). Idempotent on `--apply` re-run.
|
||||
- `src/commands/schema.ts` extension — 14 CLI verbs in the dispatch table: `add-type`, `remove-type`, `update-type`, `add-alias`, `remove-alias`, `add-prefix`, `remove-prefix`, `add-link-type`, `remove-link-type`, `set-extractable`, `set-expert-routing`, `stats`, `sync`, `reload`. `withConnectedEngine` routes `loadConfig()` through the canonical `toEngineConfig()` helper and passes the complete result (`database_url` and `database_path`) to factory construction and connect, so PGLite schema commands open the configured brain. Lifecycle-grouped help text (Inspection / Activation / Authoring / Discovery+repair). Pinned by `test/schema-cli-database-path.serial.test.ts`.
|
||||
- `src/core/operations.ts` extension — 9 MCP ops: `get_active_schema_pack`, `list_schema_packs`, `schema_stats`, `schema_lint`, `schema_graph`, `schema_explain_type`, `schema_review_orphans` (all read-scope, NOT localOnly), plus `schema_apply_mutations` (admin scope, NOT localOnly so remote agents can author packs over HTTPS MCP — batched, one MCP tool taking a `mutations[]` array atomically inside ONE `withPackLock`, audit log captures `actor: mcp:<clientId8>`) and `reload_schema_pack` (admin, NOT localOnly). Trust posture: per-call `schema_pack` opt STAYS rejected for remote callers via `op-trust-gate.ts`.
|
||||
- `src/core/operations.ts` extension — 9 MCP ops: `get_active_schema_pack`, `list_schema_packs`, `schema_stats`, `schema_lint`, `schema_graph`, `schema_explain_type`, `schema_review_orphans` (all read-scope, NOT localOnly), plus `schema_apply_mutations` (admin scope, NOT localOnly so remote agents can author packs over HTTPS MCP — batched, one MCP tool taking a `mutations[]` array, delegating to `applyMutationsAtomic` for a single lock + single read + single write across the whole batch; a mid-batch failure reports `mutations_applied: 0` + `pack_unchanged: true` (never a `partial_results` list — nothing is written until every mutation validates), audit log captures `actor: mcp:<clientId8>`) and `reload_schema_pack` (admin, NOT localOnly). Trust posture: per-call `schema_pack` opt STAYS rejected for remote callers via `op-trust-gate.ts`.
|
||||
- `src/commands/whoknows.ts` + `src/core/operations.ts:find_experts` — T1.5 wiring sites. Pack-aware via `expertTypesFromPack(pack.manifest)` from `best-effort.ts`. Pack-load failure → EMPTY filter (NOT hardcoded `['person', 'company']` defaults). A `researcher` type declared `--expert` now surfaces in `whoknows` results.
|
||||
- `skills/schema-author/SKILL.md` — Agent dispatcher for "evolve the schema pack." Triggers: 15+ phrasings incl. "add a page type", "my brain has untyped pages", "propose new types from my corpus", "backfill page types". Explicit Non-goals callout to `brain-taxonomist` (files one page) and `eiirp` (schema-check during iteration) so agents pick the right surface. 7-phase workflow: brain → assess → propose → apply → sync → verify → commit. Lists every gbrain schema CLI verb + every MCP op the skill uses. `brain_first: exempt` frontmatter. Required conformance sections: Contract, Anti-Patterns, Output Format.
|
||||
- `skills/conventions/schema-evolution.md` — Canonical convention: "when to add a type vs alias vs prefix." Decision tree: <20 pages → don't pack-codify; 20-100 → alias or narrow prefix on existing type; 100+ → first-class type. Don'ts section + "when to remove a type" + "when to commit the pack" all answered in one place.
|
||||
|
||||
+24
-92
@@ -5043,7 +5043,7 @@ const schema_review_orphans: Operation = {
|
||||
|
||||
const schema_apply_mutations: Operation = {
|
||||
name: 'schema_apply_mutations',
|
||||
description: 'v0.40.7.0: batched schema pack mutation. ATOMIC: all mutations succeed or all roll back. Audit log records one batch_id. Admin scope; NOT localOnly so remote agents (your OpenClaw, etc.) can author packs over normal MCP. Mutation shape per ApplyMutationsRequest type — supports add_type / remove_type / update_type / add_alias / remove_alias / add_prefix / remove_prefix / add_link_type / remove_link_type / set_extractable / set_expert_routing.',
|
||||
description: 'v0.40.7.0: batched schema pack mutation. ATOMIC: every mutation is validated against an in-memory manifest first, and the pack file is written to disk at most once, after the FULL batch has proven valid — so a failure at any point leaves the pack file byte-identical to its pre-batch state (never a partial write). Audit log records one batch_id. Admin scope; NOT localOnly so remote agents (your OpenClaw, etc.) can author packs over normal MCP. Mutation shape per ApplyMutationsRequest type — supports add_type / remove_type / update_type / add_alias / remove_alias / add_prefix / remove_prefix / add_link_type / remove_link_type / set_extractable / set_expert_routing.',
|
||||
params: {
|
||||
pack: { type: 'string', required: true, description: 'Pack to mutate (must not be bundled)' },
|
||||
mutations: {
|
||||
@@ -5066,92 +5066,20 @@ const schema_apply_mutations: Operation = {
|
||||
const batchId = `batch-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
|
||||
const actor = ctx.auth?.clientId ? `mcp:${ctx.auth.clientId.slice(0, 8)}` : 'cli';
|
||||
const sourceId = ctx.sourceId; // codex C5: write-side scoping
|
||||
// Compose every mutation inside ONE withPackLock so the batch is
|
||||
// truly atomic. The withMutation skeleton handles audit / cache
|
||||
// invalidation per operation; we orchestrate the lock + iteration.
|
||||
const { withPackLock } = await import('./schema-pack/pack-lock.ts');
|
||||
const {
|
||||
addTypeToPack, removeTypeFromPack, updateTypeOnPack,
|
||||
addAliasToType, removeAliasFromType, addPrefixToType, removePrefixFromType,
|
||||
addLinkTypeToPack, removeLinkTypeFromPack,
|
||||
setExtractableOnType, setExpertRoutingOnType,
|
||||
SchemaPackMutationError,
|
||||
} = await import('./schema-pack/mutate.ts');
|
||||
const baseMutateOpts = {
|
||||
actor: actor as 'cli' | `mcp:${string}`,
|
||||
batchId,
|
||||
engine: ctx.engine,
|
||||
...(sourceId ? { sourceId } : {}),
|
||||
...(force ? { force: true } : {}),
|
||||
};
|
||||
const results: unknown[] = [];
|
||||
// `applyMutationsAtomic` (issue #2581) owns the lock + single read +
|
||||
// single write for the whole batch: every mutation is validated
|
||||
// in-memory first, and the pack file is written at most once, only
|
||||
// after the FULL batch checks out. That is what makes this actually
|
||||
// atomic (a failure at any index can never leave earlier mutations on
|
||||
// disk), vs. the old per-mutation-writes-as-it-goes shape.
|
||||
const { applyMutationsAtomic } = await import('./schema-pack/mutate.ts');
|
||||
try {
|
||||
// Outer lock: hold the pack for the whole batch so other writers
|
||||
// can't slip in between mutations.
|
||||
await withPackLock(pack, { force, lockDir: undefined }, async () => {
|
||||
for (let i = 0; i < mutations.length; i++) {
|
||||
const m = mutations[i]!;
|
||||
// Each primitive acquires the lock internally; the outer
|
||||
// withPackLock makes that re-entrant via fast-stale-detect
|
||||
// (--force option for the inner call). To keep semantics
|
||||
// simple, we pass {force:true} to the inner calls because
|
||||
// they're nested inside our outer lock — we already own it.
|
||||
const innerOpts = { ...baseMutateOpts, force: true };
|
||||
let r: unknown;
|
||||
switch (m.op) {
|
||||
case 'add_type':
|
||||
r = await addTypeToPack(pack, {
|
||||
name: m.name as string,
|
||||
primitive: m.primitive as never,
|
||||
prefix: m.prefix as string,
|
||||
extractable: m.extractable as boolean | undefined,
|
||||
expertRouting: m.expert_routing as boolean | undefined,
|
||||
aliases: m.aliases as string[] | undefined,
|
||||
}, innerOpts);
|
||||
break;
|
||||
case 'remove_type':
|
||||
r = await removeTypeFromPack(pack, m.name as string, innerOpts);
|
||||
break;
|
||||
case 'update_type':
|
||||
r = await updateTypeOnPack(pack, { name: m.name as string, patch: (m.patch as object) ?? {} }, innerOpts);
|
||||
break;
|
||||
case 'add_alias':
|
||||
r = await addAliasToType(pack, m.type as string, m.alias as string, innerOpts);
|
||||
break;
|
||||
case 'remove_alias':
|
||||
r = await removeAliasFromType(pack, m.type as string, m.alias as string, innerOpts);
|
||||
break;
|
||||
case 'add_prefix':
|
||||
r = await addPrefixToType(pack, m.type as string, m.prefix as string, innerOpts);
|
||||
break;
|
||||
case 'remove_prefix':
|
||||
r = await removePrefixFromType(pack, m.type as string, m.prefix as string, innerOpts);
|
||||
break;
|
||||
case 'add_link_type':
|
||||
r = await addLinkTypeToPack(pack, {
|
||||
name: m.name as string,
|
||||
inverse: m.inverse as string | undefined,
|
||||
inference: m.inference as { regex?: string; page_type?: string; target_type?: string } | undefined,
|
||||
}, innerOpts);
|
||||
break;
|
||||
case 'remove_link_type':
|
||||
r = await removeLinkTypeFromPack(pack, m.name as string, innerOpts);
|
||||
break;
|
||||
case 'set_extractable':
|
||||
r = await setExtractableOnType(pack, m.type as string, m.value as boolean, innerOpts);
|
||||
break;
|
||||
case 'set_expert_routing':
|
||||
r = await setExpertRoutingOnType(pack, m.type as string, m.value as boolean, innerOpts);
|
||||
break;
|
||||
default:
|
||||
throw new SchemaPackMutationError(
|
||||
'INVALID_RESULT',
|
||||
`unknown mutation op: '${m.op}' at index ${i}`,
|
||||
{ index: i, op: m.op },
|
||||
);
|
||||
}
|
||||
results.push({ index: i, op: m.op, ...(r as object) });
|
||||
}
|
||||
const results = await applyMutationsAtomic(pack, mutations, {
|
||||
actor: actor as 'cli' | `mcp:${string}`,
|
||||
batchId,
|
||||
engine: ctx.engine,
|
||||
...(sourceId ? { sourceId } : {}),
|
||||
...(force ? { force: true } : {}),
|
||||
});
|
||||
return {
|
||||
schema_version: 1,
|
||||
@@ -5162,17 +5090,21 @@ const schema_apply_mutations: Operation = {
|
||||
};
|
||||
} catch (e) {
|
||||
const code = (e as { code?: string }).code ?? 'UNKNOWN';
|
||||
const failedAtIndex = (e as { details?: { index?: number } }).details?.index;
|
||||
return {
|
||||
error: 'mutation_failed',
|
||||
code,
|
||||
message: (e as Error).message,
|
||||
batch_id: batchId,
|
||||
// Partial results recorded so the agent can inspect which
|
||||
// mutations landed before the failure (the atomic guarantee
|
||||
// is at the LOCK level — individual mutations are sequential
|
||||
// and each is atomic; pack state reflects everything up to the
|
||||
// failed mutation).
|
||||
partial_results: results,
|
||||
// Nothing was written to disk — applyMutationsAtomic only writes
|
||||
// once, after every mutation in the batch has validated cleanly.
|
||||
// (Pre-fix, this field was `partial_results` and listed mutations
|
||||
// that HAD already landed on disk, because the old implementation
|
||||
// wrote as it went — that shape is gone; a failed batch can no
|
||||
// longer imply partial application.)
|
||||
mutations_applied: 0,
|
||||
pack_unchanged: true,
|
||||
...(failedAtIndex !== undefined ? { failed_at_index: failedAtIndex } : {}),
|
||||
};
|
||||
}
|
||||
},
|
||||
|
||||
@@ -186,6 +186,9 @@ export {
|
||||
removeLinkTypeFromPack,
|
||||
setExtractableOnType,
|
||||
setExpertRoutingOnType,
|
||||
type BatchMutationRequest,
|
||||
type BatchMutationResult,
|
||||
applyMutationsAtomic,
|
||||
} from './mutate.ts';
|
||||
|
||||
export { invalidateQueryCache } from './query-cache-invalidator.ts';
|
||||
|
||||
+286
-27
@@ -497,11 +497,18 @@ export interface AddTypeOpts {
|
||||
aliases?: string[];
|
||||
}
|
||||
|
||||
export async function addTypeToPack(packName: string, opts: AddTypeOpts, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
// Each `build*Mutator` below does the primitive's up-front (file-free,
|
||||
// lock-free) shape validation and returns the pure `(current) => next`
|
||||
// transform. The public async functions wrap the builder with
|
||||
// `withMutation` for the single-mutation (CLI) path; `applyMutationsAtomic`
|
||||
// (batch path, below) reuses the SAME builders so single-call and batched
|
||||
// mutations can never drift in what they accept or reject.
|
||||
|
||||
function buildAddTypeMutator(opts: AddTypeOpts): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
validateTypeName(opts.name);
|
||||
validatePrimitive(opts.primitive);
|
||||
validatePrefix(opts.prefix);
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
if (m.page_types.some((pt) => pt.name === opts.name)) {
|
||||
throw new SchemaPackMutationError(
|
||||
'TYPE_EXISTS',
|
||||
@@ -518,16 +525,24 @@ export async function addTypeToPack(packName: string, opts: AddTypeOpts, mutateO
|
||||
expert_routing: opts.expertRouting ?? false,
|
||||
};
|
||||
return { ...m, page_types: [...m.page_types, newType] };
|
||||
}, 'add_type', { type: opts.name, prefix: opts.prefix });
|
||||
};
|
||||
}
|
||||
|
||||
export async function removeTypeFromPack(packName: string, name: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
export async function addTypeToPack(packName: string, opts: AddTypeOpts, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildAddTypeMutator(opts), 'add_type', { type: opts.name, prefix: opts.prefix });
|
||||
}
|
||||
|
||||
function buildRemoveTypeMutator(name: string): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
validateTypeName(name);
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
findType(m, name); // throws TYPE_NOT_FOUND if missing
|
||||
checkNoReferences(m, name); // codex C14
|
||||
return { ...m, page_types: m.page_types.filter((t) => t.name !== name) };
|
||||
}, 'remove_type', { type: name });
|
||||
};
|
||||
}
|
||||
|
||||
export async function removeTypeFromPack(packName: string, name: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildRemoveTypeMutator(name), 'remove_type', { type: name });
|
||||
}
|
||||
|
||||
export interface UpdateTypeOpts {
|
||||
@@ -535,56 +550,76 @@ export interface UpdateTypeOpts {
|
||||
patch: Partial<Omit<PackPageType, 'name'>>;
|
||||
}
|
||||
|
||||
export async function updateTypeOnPack(packName: string, opts: UpdateTypeOpts, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
function buildUpdateTypeMutator(opts: UpdateTypeOpts): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
validateTypeName(opts.name);
|
||||
if (opts.patch.primitive !== undefined) validatePrimitive(opts.patch.primitive);
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
const existing = findType(m, opts.name);
|
||||
const updated: PackPageType = { ...existing, ...opts.patch, name: existing.name };
|
||||
return { ...m, page_types: m.page_types.map((t) => (t.name === opts.name ? updated : t)) };
|
||||
}, 'update_type', { type: opts.name });
|
||||
};
|
||||
}
|
||||
|
||||
export async function addAliasToType(packName: string, typeName: string, alias: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
export async function updateTypeOnPack(packName: string, opts: UpdateTypeOpts, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildUpdateTypeMutator(opts), 'update_type', { type: opts.name });
|
||||
}
|
||||
|
||||
function buildAddAliasMutator(typeName: string, alias: string): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
validateTypeName(typeName);
|
||||
validateTypeName(alias);
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
const t = findType(m, typeName);
|
||||
if (t.aliases.includes(alias)) return m; // idempotent
|
||||
const next: PackPageType = { ...t, aliases: [...t.aliases, alias] };
|
||||
return { ...m, page_types: m.page_types.map((pt) => (pt.name === typeName ? next : pt)) };
|
||||
}, 'add_alias', { type: typeName });
|
||||
};
|
||||
}
|
||||
|
||||
export async function removeAliasFromType(packName: string, typeName: string, alias: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
export async function addAliasToType(packName: string, typeName: string, alias: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildAddAliasMutator(typeName, alias), 'add_alias', { type: typeName });
|
||||
}
|
||||
|
||||
function buildRemoveAliasMutator(typeName: string, alias: string): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
validateTypeName(typeName);
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
const t = findType(m, typeName);
|
||||
if (!t.aliases.includes(alias)) return m; // idempotent
|
||||
const next: PackPageType = { ...t, aliases: t.aliases.filter((a) => a !== alias) };
|
||||
return { ...m, page_types: m.page_types.map((pt) => (pt.name === typeName ? next : pt)) };
|
||||
}, 'remove_alias', { type: typeName });
|
||||
};
|
||||
}
|
||||
|
||||
export async function addPrefixToType(packName: string, typeName: string, prefix: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
export async function removeAliasFromType(packName: string, typeName: string, alias: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildRemoveAliasMutator(typeName, alias), 'remove_alias', { type: typeName });
|
||||
}
|
||||
|
||||
function buildAddPrefixMutator(typeName: string, prefix: string): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
validateTypeName(typeName);
|
||||
validatePrefix(prefix);
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
const t = findType(m, typeName);
|
||||
if (t.path_prefixes.includes(prefix)) return m;
|
||||
const next: PackPageType = { ...t, path_prefixes: [...t.path_prefixes, prefix] };
|
||||
return { ...m, page_types: m.page_types.map((pt) => (pt.name === typeName ? next : pt)) };
|
||||
}, 'add_prefix', { type: typeName, prefix });
|
||||
};
|
||||
}
|
||||
|
||||
export async function removePrefixFromType(packName: string, typeName: string, prefix: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
export async function addPrefixToType(packName: string, typeName: string, prefix: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildAddPrefixMutator(typeName, prefix), 'add_prefix', { type: typeName, prefix });
|
||||
}
|
||||
|
||||
function buildRemovePrefixMutator(typeName: string, prefix: string): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
validateTypeName(typeName);
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
const t = findType(m, typeName);
|
||||
if (!t.path_prefixes.includes(prefix)) return m;
|
||||
const next: PackPageType = { ...t, path_prefixes: t.path_prefixes.filter((p) => p !== prefix) };
|
||||
return { ...m, page_types: m.page_types.map((pt) => (pt.name === typeName ? next : pt)) };
|
||||
}, 'remove_prefix', { type: typeName, prefix });
|
||||
};
|
||||
}
|
||||
|
||||
export async function removePrefixFromType(packName: string, typeName: string, prefix: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildRemovePrefixMutator(typeName, prefix), 'remove_prefix', { type: typeName, prefix });
|
||||
}
|
||||
|
||||
export interface AddLinkTypeOpts {
|
||||
@@ -593,11 +628,11 @@ export interface AddLinkTypeOpts {
|
||||
inference?: { regex?: string; page_type?: string; target_type?: string };
|
||||
}
|
||||
|
||||
export async function addLinkTypeToPack(packName: string, opts: AddLinkTypeOpts, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
function buildAddLinkTypeMutator(opts: AddLinkTypeOpts): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
if (typeof opts.name !== 'string' || opts.name.length === 0) {
|
||||
throw new SchemaPackMutationError('INVALID_RESULT', `link_type.name is required`);
|
||||
}
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
return (m) => {
|
||||
if (m.link_types.some((lt) => lt.name === opts.name)) {
|
||||
throw new SchemaPackMutationError(
|
||||
'TYPE_EXISTS',
|
||||
@@ -611,11 +646,15 @@ export async function addLinkTypeToPack(packName: string, opts: AddLinkTypeOpts,
|
||||
...(opts.inference ? { inference: opts.inference } : {}),
|
||||
} as PackLinkType;
|
||||
return { ...m, link_types: [...m.link_types, newLink] };
|
||||
}, 'add_link_type', { type: opts.name });
|
||||
};
|
||||
}
|
||||
|
||||
export async function removeLinkTypeFromPack(packName: string, linkName: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, (m) => {
|
||||
export async function addLinkTypeToPack(packName: string, opts: AddLinkTypeOpts, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildAddLinkTypeMutator(opts), 'add_link_type', { type: opts.name });
|
||||
}
|
||||
|
||||
function buildRemoveLinkTypeMutator(linkName: string): (m: SchemaPackManifest) => SchemaPackManifest {
|
||||
return (m) => {
|
||||
if (!m.link_types.some((lt) => lt.name === linkName)) {
|
||||
throw new SchemaPackMutationError(
|
||||
'TYPE_NOT_FOUND',
|
||||
@@ -633,7 +672,11 @@ export async function removeLinkTypeFromPack(packName: string, linkName: string,
|
||||
);
|
||||
}
|
||||
return { ...m, link_types: m.link_types.filter((lt) => lt.name !== linkName) };
|
||||
}, 'remove_link_type', { type: linkName });
|
||||
};
|
||||
}
|
||||
|
||||
export async function removeLinkTypeFromPack(packName: string, linkName: string, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return withMutation(packName, mutateOpts, buildRemoveLinkTypeMutator(linkName), 'remove_link_type', { type: linkName });
|
||||
}
|
||||
|
||||
export async function setExtractableOnType(packName: string, typeName: string, value: boolean, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
@@ -643,3 +686,219 @@ export async function setExtractableOnType(packName: string, typeName: string, v
|
||||
export async function setExpertRoutingOnType(packName: string, typeName: string, value: boolean, mutateOpts: MutateOpts = {}): Promise<MutateResult> {
|
||||
return updateTypeOnPack(packName, { name: typeName, patch: { expert_routing: value } }, { ...mutateOpts });
|
||||
}
|
||||
|
||||
// ────────────────────────────────────────────────────────────────────────
|
||||
// Atomic batch application (issue #2581) — one lock, one file read, one
|
||||
// write. `schema_apply_mutations` used to loop over these same primitives
|
||||
// and let each one independently read/validate/WRITE the pack file, so a
|
||||
// batch that failed partway left every earlier mutation permanently on
|
||||
// disk even though the op is documented as all-or-nothing. Here every
|
||||
// mutation in the batch is applied + lint-validated against an IN-MEMORY
|
||||
// manifest only; `writePackManifest` is called at most once, after every
|
||||
// mutation in the batch has been proven valid. A failure at any index
|
||||
// therefore leaves the pack file byte-identical to its pre-batch state —
|
||||
// partial application is structurally impossible, not just cleaned up
|
||||
// after the fact.
|
||||
// ────────────────────────────────────────────────────────────────────────
|
||||
|
||||
export interface BatchMutationRequest {
|
||||
op: string;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
export interface BatchMutationResult {
|
||||
index: number;
|
||||
op: string;
|
||||
pack: string;
|
||||
path: string;
|
||||
format: PackFileFormat;
|
||||
/** sha8 of the manifest immediately before this mutation (chained). */
|
||||
prev_sha8: string;
|
||||
/** sha8 of the manifest immediately after this mutation (chained). */
|
||||
new_sha8: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve one batch entry to its pure mutator + audit context, reusing the
|
||||
* exact same `build*Mutator` a single-mutation call would use. Throws
|
||||
* `SchemaPackMutationError('INVALID_RESULT', ...)` for an unrecognized
|
||||
* `op`, matching the pre-existing single-mutation shape-validation
|
||||
* contract: this runs before the file is touched, so it is deliberately
|
||||
* NOT audit-logged here (mirrors `addTypeToPack` etc. throwing from their
|
||||
* own up-front `validate*` calls, before `withMutation` ever starts).
|
||||
*/
|
||||
function buildBatchMutator(
|
||||
m: BatchMutationRequest,
|
||||
index: number,
|
||||
): { mutate: (current: SchemaPackManifest) => SchemaPackManifest; auditContext: { type?: string; prefix?: string } } {
|
||||
switch (m.op) {
|
||||
case 'add_type':
|
||||
return {
|
||||
mutate: buildAddTypeMutator({
|
||||
name: m.name as string,
|
||||
primitive: m.primitive as never,
|
||||
prefix: m.prefix as string,
|
||||
extractable: m.extractable as boolean | undefined,
|
||||
expertRouting: m.expert_routing as boolean | undefined,
|
||||
aliases: m.aliases as string[] | undefined,
|
||||
}),
|
||||
auditContext: { type: m.name as string, prefix: m.prefix as string },
|
||||
};
|
||||
case 'remove_type':
|
||||
return { mutate: buildRemoveTypeMutator(m.name as string), auditContext: { type: m.name as string } };
|
||||
case 'update_type':
|
||||
return {
|
||||
mutate: buildUpdateTypeMutator({ name: m.name as string, patch: (m.patch as object) ?? {} }),
|
||||
auditContext: { type: m.name as string },
|
||||
};
|
||||
case 'add_alias':
|
||||
return { mutate: buildAddAliasMutator(m.type as string, m.alias as string), auditContext: { type: m.type as string } };
|
||||
case 'remove_alias':
|
||||
return { mutate: buildRemoveAliasMutator(m.type as string, m.alias as string), auditContext: { type: m.type as string } };
|
||||
case 'add_prefix':
|
||||
return {
|
||||
mutate: buildAddPrefixMutator(m.type as string, m.prefix as string),
|
||||
auditContext: { type: m.type as string, prefix: m.prefix as string },
|
||||
};
|
||||
case 'remove_prefix':
|
||||
return {
|
||||
mutate: buildRemovePrefixMutator(m.type as string, m.prefix as string),
|
||||
auditContext: { type: m.type as string, prefix: m.prefix as string },
|
||||
};
|
||||
case 'add_link_type':
|
||||
return {
|
||||
mutate: buildAddLinkTypeMutator({
|
||||
name: m.name as string,
|
||||
inverse: m.inverse as string | undefined,
|
||||
inference: m.inference as { regex?: string; page_type?: string; target_type?: string } | undefined,
|
||||
}),
|
||||
auditContext: { type: m.name as string },
|
||||
};
|
||||
case 'remove_link_type':
|
||||
return { mutate: buildRemoveLinkTypeMutator(m.name as string), auditContext: { type: m.name as string } };
|
||||
case 'set_extractable':
|
||||
return {
|
||||
mutate: buildUpdateTypeMutator({ name: m.type as string, patch: { extractable: m.value as boolean } }),
|
||||
auditContext: { type: m.type as string },
|
||||
};
|
||||
case 'set_expert_routing':
|
||||
return {
|
||||
mutate: buildUpdateTypeMutator({ name: m.type as string, patch: { expert_routing: m.value as boolean } }),
|
||||
auditContext: { type: m.type as string },
|
||||
};
|
||||
default:
|
||||
throw new SchemaPackMutationError('INVALID_RESULT', `unknown mutation op: '${m.op}' at index ${index}`, { index, op: m.op });
|
||||
}
|
||||
}
|
||||
|
||||
export async function applyMutationsAtomic(
|
||||
packName: string,
|
||||
mutations: BatchMutationRequest[],
|
||||
opts: MutateOpts,
|
||||
): Promise<BatchMutationResult[]> {
|
||||
const actor: MutationActor = opts.actor ?? 'cli';
|
||||
const firstOp = (mutations[0]?.op as MutationOp) ?? 'add_type';
|
||||
|
||||
// Bundled-pack guard, same as withMutation step 1 — happens once for
|
||||
// the whole batch since `pack` is constant across mutations.
|
||||
let path: string;
|
||||
let format: PackFileFormat;
|
||||
try {
|
||||
({ path, format } = locateMutablePackFile(packName));
|
||||
} catch (e) {
|
||||
if (e instanceof SchemaPackMutationError) {
|
||||
await logMutationFailure({ op: firstOp, pack: packName, actor, reason: e.code, batch_id: opts.batchId });
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
|
||||
return withPackLock(packName, opts, async () => {
|
||||
let current: SchemaPackManifest;
|
||||
let batchPrevSha8: string;
|
||||
try {
|
||||
current = loadPackFromFile(path);
|
||||
batchPrevSha8 = await computeManifestSha8(current);
|
||||
} catch (e) {
|
||||
const err = new SchemaPackMutationError(
|
||||
'PACK_CORRUPT',
|
||||
`cannot read or parse pack file at ${path}: ${(e as Error).message}`,
|
||||
{ path },
|
||||
);
|
||||
await logMutationFailure({ op: firstOp, pack: packName, actor, reason: err.code, batch_id: opts.batchId });
|
||||
throw err;
|
||||
}
|
||||
|
||||
// Phase 1: apply + lint-validate every mutation against the IN-MEMORY
|
||||
// manifest only. Nothing here touches disk — a throw at any index
|
||||
// propagates straight out (lock released by withPackLock's finally)
|
||||
// and `path` is left completely untouched.
|
||||
const pending: Array<{ index: number; op: string; auditContext: { type?: string; prefix?: string }; prevSha8: string; newSha8: string }> = [];
|
||||
let runningPrevSha8 = batchPrevSha8;
|
||||
for (let i = 0; i < mutations.length; i++) {
|
||||
const m = mutations[i]!;
|
||||
const opForAudit = (m.op as MutationOp) ?? firstOp;
|
||||
const built = buildBatchMutator(m, i); // shape validation — unaudited, matches single-mutation contract
|
||||
let next: SchemaPackManifest;
|
||||
try {
|
||||
next = built.mutate(current);
|
||||
} catch (e) {
|
||||
const base = e instanceof SchemaPackMutationError ? e : new SchemaPackMutationError('INVALID_RESULT', (e as Error).message);
|
||||
// Re-wrap so `details.index` is always present for the batch
|
||||
// caller (operations.ts) to report which mutation failed,
|
||||
// without losing the primitive's own code/message/details.
|
||||
const wrapped = new SchemaPackMutationError(base.code, base.message, { ...base.details, index: i });
|
||||
await logMutationFailure({
|
||||
op: opForAudit, pack: packName, actor, ...built.auditContext,
|
||||
reason: wrapped.code, prev_sha8: runningPrevSha8, batch_id: opts.batchId,
|
||||
});
|
||||
throw wrapped;
|
||||
}
|
||||
const lintReport = await runFilePlaneLintRules(next);
|
||||
if (!lintReport.ok) {
|
||||
const msg = lintReport.errors.map((iss) => `${iss.rule}: ${iss.message}`).join('; ');
|
||||
const err = new SchemaPackMutationError('INVALID_RESULT', `mutation would produce invalid pack: ${msg}`, { index: i, errors: lintReport.errors });
|
||||
await logMutationFailure({
|
||||
op: opForAudit, pack: packName, actor, ...built.auditContext,
|
||||
reason: err.code, prev_sha8: runningPrevSha8, batch_id: opts.batchId,
|
||||
});
|
||||
throw err;
|
||||
}
|
||||
const newSha8 = await computeManifestSha8(next);
|
||||
pending.push({ index: i, op: m.op, auditContext: built.auditContext, prevSha8: runningPrevSha8, newSha8 });
|
||||
current = next;
|
||||
runningPrevSha8 = newSha8;
|
||||
}
|
||||
|
||||
// Phase 2: every mutation validated clean — write ONCE.
|
||||
try {
|
||||
writePackManifest(path, current, format);
|
||||
} catch (e) {
|
||||
const err = e instanceof SchemaPackMutationError ? e : new SchemaPackMutationError('IO_ERROR', (e as Error).message, { path });
|
||||
const last = pending[pending.length - 1];
|
||||
await logMutationFailure({
|
||||
op: (last?.op as MutationOp) ?? firstOp, pack: packName, actor, ...(last?.auditContext ?? {}),
|
||||
reason: err.code, prev_sha8: batchPrevSha8, batch_id: opts.batchId,
|
||||
});
|
||||
throw err;
|
||||
}
|
||||
|
||||
// Step 7 equivalent: best-effort post-hooks, once for the whole batch.
|
||||
try { invalidatePackCache(packName); } catch { /* swallow — cache invalidation must not block mutation success */ }
|
||||
if (opts.engine) {
|
||||
try { await invalidateQueryCache(opts.engine, opts.sourceId); } catch { /* swallow */ }
|
||||
}
|
||||
|
||||
// Only now — after the single write has actually landed on disk — do
|
||||
// we log success and report results. Nothing above this point may
|
||||
// ever be reported as applied.
|
||||
const results: BatchMutationResult[] = [];
|
||||
for (const p of pending) {
|
||||
await logMutationSuccess({
|
||||
op: p.op as MutationOp, pack: packName, actor, ...p.auditContext,
|
||||
prev_sha8: p.prevSha8, new_sha8: p.newSha8, batch_id: opts.batchId,
|
||||
});
|
||||
results.push({ index: p.index, op: p.op, pack: packName, path, format, prev_sha8: p.prevSha8, new_sha8: p.newSha8 });
|
||||
}
|
||||
return results;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -283,19 +283,79 @@ describe('schema_apply_mutations', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('returns partial_results on mid-batch failure with a single batch_id', async () => {
|
||||
it('mid-batch failure reports nothing applied — no partial_results implying a landed write (#2581)', async () => {
|
||||
await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_AUDIT_DIR: auditDir }, async () => {
|
||||
seedPack('mine');
|
||||
const packPath = seedPack('mine');
|
||||
const before = readFileSync(packPath, 'utf-8');
|
||||
const result = await operationsByName.schema_apply_mutations!.handler(ctxOf(), {
|
||||
pack: 'mine',
|
||||
mutations: [
|
||||
{ op: 'add_type', name: 'company', primitive: 'entity', prefix: 'companies/' },
|
||||
{ op: 'add_type', name: 'person', primitive: 'entity', prefix: 'people/' }, // collides with seed
|
||||
{ op: 'add_type', name: 'person', primitive: 'entity', prefix: 'people/' }, // name collision with seed
|
||||
],
|
||||
}) as Record<string, unknown>;
|
||||
expect(result.error).toBe('mutation_failed');
|
||||
const partial = result.partial_results as Array<unknown>;
|
||||
expect(partial.length).toBe(1); // first mutation succeeded
|
||||
expect(result.code).toBe('TYPE_EXISTS');
|
||||
// Nothing was written: mutations_applied is 0, the response says so
|
||||
// explicitly, and there is no `partial_results` field implying the
|
||||
// first mutation landed on disk (it never did — see the byte-identical
|
||||
// assertion in the dedicated regression test below).
|
||||
expect(result.mutations_applied).toBe(0);
|
||||
expect(result.pack_unchanged).toBe(true);
|
||||
expect(result.failed_at_index).toBe(1);
|
||||
expect('partial_results' in result).toBe(false);
|
||||
expect(readFileSync(packPath, 'utf-8')).toBe(before);
|
||||
});
|
||||
});
|
||||
|
||||
// Regression test for #2581: schema_apply_mutations documented itself as
|
||||
// ATOMIC ("all mutations succeed or all roll back"), but each mutation
|
||||
// independently read/validated/WROTE the pack file as the batch loop ran.
|
||||
// A batch that failed partway therefore left every earlier mutation
|
||||
// permanently applied to disk — the exact repro from the issue (7
|
||||
// add_type mutations, a later one fails prefix_collision, and the type
|
||||
// from index 0 is found already written to pack.yaml). This test fails
|
||||
// on pre-fix code (the sha8/content changes) and passes once the batch
|
||||
// validates entirely in-memory before a single write.
|
||||
it('#2581: a batch that fails partway leaves the pack file byte-identical to its pre-batch state', async () => {
|
||||
await withEnv({ GBRAIN_HOME: tmpDir, GBRAIN_AUDIT_DIR: auditDir }, async () => {
|
||||
const packPath = seedPack('mine');
|
||||
const beforeContent = readFileSync(packPath, 'utf-8');
|
||||
|
||||
const result = await operationsByName.schema_apply_mutations!.handler(ctxOf(), {
|
||||
pack: 'mine',
|
||||
mutations: [
|
||||
{ op: 'add_type', name: 'alpha', primitive: 'entity', prefix: 'alpha/' },
|
||||
{ op: 'add_type', name: 'beta', primitive: 'entity', prefix: 'beta/' },
|
||||
// Same path_prefix as `alpha` — fails schema_apply_mutations'
|
||||
// prefix_collision lint rule, matching the issue's repro.
|
||||
{ op: 'add_type', name: 'gamma', primitive: 'entity', prefix: 'alpha/' },
|
||||
],
|
||||
}) as Record<string, unknown>;
|
||||
|
||||
expect(result.error).toBe('mutation_failed');
|
||||
expect(result.code).toBe('INVALID_RESULT');
|
||||
expect(String(result.message)).toContain('prefix_collision');
|
||||
expect(result.mutations_applied).toBe(0);
|
||||
expect(result.pack_unchanged).toBe(true);
|
||||
expect(result.failed_at_index).toBe(2);
|
||||
|
||||
const afterContent = readFileSync(packPath, 'utf-8');
|
||||
expect(afterContent).toBe(beforeContent);
|
||||
|
||||
// A corrected re-submission (without the colliding prefix) must
|
||||
// succeed cleanly — pre-fix, this failed with TYPE_EXISTS for
|
||||
// `alpha` because it was already stuck on disk from the failed
|
||||
// batch, wedging the user until they restored from a backup.
|
||||
const retry = await operationsByName.schema_apply_mutations!.handler(ctxOf(), {
|
||||
pack: 'mine',
|
||||
mutations: [
|
||||
{ op: 'add_type', name: 'alpha', primitive: 'entity', prefix: 'alpha/' },
|
||||
{ op: 'add_type', name: 'beta', primitive: 'entity', prefix: 'beta/' },
|
||||
],
|
||||
}) as Record<string, unknown>;
|
||||
expect(retry.error).toBeUndefined();
|
||||
expect(retry.mutations_applied).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
Reference in New Issue
Block a user