From 13d95ba0ab1b666f708703ba439a280468a16c46 Mon Sep 17 00:00:00 2001 From: Masa <98894508+Masashi-Ono0611@users.noreply.github.com> Date: Sat, 1 Aug 2026 05:09:12 +0900 Subject: [PATCH] fix(schema): apply mutation batches atomically so a mid-batch failure leaves the pack untouched (#2581) (#3446) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- docs/architecture/KEY_FILES.md | 4 +- src/core/operations.ts | 116 +++-------- src/core/schema-pack/index.ts | 3 + src/core/schema-pack/mutate.ts | 313 +++++++++++++++++++++++++--- test/operations-schema-pack.test.ts | 70 ++++++- 5 files changed, 380 insertions(+), 126 deletions(-) diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 33d8d974d..6a465c00e 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -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:`) 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:`) 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. diff --git a/src/core/operations.ts b/src/core/operations.ts index bbcb0f1a0..2d1e61459 100644 --- a/src/core/operations.ts +++ b/src/core/operations.ts @@ -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 } : {}), }; } }, diff --git a/src/core/schema-pack/index.ts b/src/core/schema-pack/index.ts index 1b1a72422..02c4ca335 100644 --- a/src/core/schema-pack/index.ts +++ b/src/core/schema-pack/index.ts @@ -186,6 +186,9 @@ export { removeLinkTypeFromPack, setExtractableOnType, setExpertRoutingOnType, + type BatchMutationRequest, + type BatchMutationResult, + applyMutationsAtomic, } from './mutate.ts'; export { invalidateQueryCache } from './query-cache-invalidator.ts'; diff --git a/src/core/schema-pack/mutate.ts b/src/core/schema-pack/mutate.ts index eaf375e47..2f5ff5a92 100644 --- a/src/core/schema-pack/mutate.ts +++ b/src/core/schema-pack/mutate.ts @@ -497,11 +497,18 @@ export interface AddTypeOpts { aliases?: string[]; } -export async function addTypeToPack(packName: string, opts: AddTypeOpts, mutateOpts: MutateOpts = {}): Promise { +// 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 { +export async function addTypeToPack(packName: string, opts: AddTypeOpts, mutateOpts: MutateOpts = {}): Promise { + 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 { + return withMutation(packName, mutateOpts, buildRemoveTypeMutator(name), 'remove_type', { type: name }); } export interface UpdateTypeOpts { @@ -535,56 +550,76 @@ export interface UpdateTypeOpts { patch: Partial>; } -export async function updateTypeOnPack(packName: string, opts: UpdateTypeOpts, mutateOpts: MutateOpts = {}): Promise { +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 { +export async function updateTypeOnPack(packName: string, opts: UpdateTypeOpts, mutateOpts: MutateOpts = {}): Promise { + 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 { +export async function addAliasToType(packName: string, typeName: string, alias: string, mutateOpts: MutateOpts = {}): Promise { + 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 { +export async function removeAliasFromType(packName: string, typeName: string, alias: string, mutateOpts: MutateOpts = {}): Promise { + 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 { +export async function addPrefixToType(packName: string, typeName: string, prefix: string, mutateOpts: MutateOpts = {}): Promise { + 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 { + 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 { +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 { - return withMutation(packName, mutateOpts, (m) => { +export async function addLinkTypeToPack(packName: string, opts: AddLinkTypeOpts, mutateOpts: MutateOpts = {}): Promise { + 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 { + return withMutation(packName, mutateOpts, buildRemoveLinkTypeMutator(linkName), 'remove_link_type', { type: linkName }); } export async function setExtractableOnType(packName: string, typeName: string, value: boolean, mutateOpts: MutateOpts = {}): Promise { @@ -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 { 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 { + 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; + }); +} diff --git a/test/operations-schema-pack.test.ts b/test/operations-schema-pack.test.ts index 0142a29d9..752401974 100644 --- a/test/operations-schema-pack.test.ts +++ b/test/operations-schema-pack.test.ts @@ -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; expect(result.error).toBe('mutation_failed'); - const partial = result.partial_results as Array; - 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; + + 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; + expect(retry.error).toBeUndefined(); + expect(retry.mutations_applied).toBe(2); }); });