diff --git a/server/src/db/repositories/PluginEntityMetadata.repository.ts b/server/src/db/repositories/PluginEntityMetadata.repository.ts index 3bce51fba..bdfc9aa0d 100644 --- a/server/src/db/repositories/PluginEntityMetadata.repository.ts +++ b/server/src/db/repositories/PluginEntityMetadata.repository.ts @@ -68,4 +68,34 @@ export class PluginEntityMetadataRepository extends TrekRepository 0; } + + /** + * Plan 3j Task 7 fix wave, must-land 2 (task-7-review.md): MR2 (the + * pre-write existence check), MR3 (the `META_KEYS_MAX` count gate) and MR4 + * (the upsert) made atomic. `meta.rpc.ts#set` used to run these as three + * separately-awaited calls — two concurrent `meta.set` calls for two + * DIFFERENT new keys on the same (plugin, entity) could both read + * `count < max` before either's write landed, so a 100-key cap let a live + * burst land 130 rows. Wrapped in `this.getEntityManager().transactional` + * — the same MikroORM API `UnitOfWork.transactional` itself calls (see + * that class's own docstring on why its default NESTED/savepoint + * propagation is safe even when a caller is already inside a transaction) + * — so `MetaRpc` needs no new constructor dependency: this fix wave found + * `tests/helpers/plugin-host.ts` (which hand-constructs `MetaRpc`) mid-edit + * by a concurrent Plan 4 task and could not touch its construction site, + * so the fix stays entirely inside the repository it already owns. + * Returns whether the write happened; `false` means a NEW key was blocked + * by the cap (the caller maps that to `BadParams`, same message as before). + */ + async upsertValueCapped(pluginId: string, entityType: string, entityId: number, key: string, value: string, maxKeys: number): Promise { + return this.getEntityManager().transactional(async () => { + const exists = (await this.findValue(pluginId, entityType, entityId, key)) !== null; + if (!exists) { + const n = await this.countForEntity(pluginId, entityType, entityId); + if (n >= maxKeys) return false; + } + await this.upsertValue(pluginId, entityType, entityId, key, value); + return true; + }); + } } diff --git a/server/src/db/repositories/PluginScheduledTasks.repository.ts b/server/src/db/repositories/PluginScheduledTasks.repository.ts index bf2801bfe..ea3312573 100644 --- a/server/src/db/repositories/PluginScheduledTasks.repository.ts +++ b/server/src/db/repositories/PluginScheduledTasks.repository.ts @@ -113,4 +113,29 @@ export class PluginScheduledTasksRepository extends TrekRepository 0; } + + /** + * Plan 3j Task 7 fix wave, must-land 3 (task-7-review.md): HR5 (the + * pre-write existence check), HR6 (the `SCHED_MAX` count gate) and HR7 + * (the upsert) made atomic — the SAME shape and SAME reason as + * `PluginEntityMetadataRepository.upsertValueCapped` (must-land 2): two + * concurrent `scheduler.set` calls for two different new task names on the + * same plugin could both read `count < max` before either's write landed + * (live: 130 rows on a 100 cap). `this.getEntityManager().transactional` — + * no new constructor dependency on `HostSurfaceRpc`, for the same + * `tests/helpers/plugin-host.ts` (dirty, mid-edit by a concurrent Plan 4 + * task) reason `upsertValueCapped`'s docstring gives. Returns whether the + * write happened; `false` means a NEW task name was blocked by the cap. + */ + async upsertTaskCapped(input: PluginScheduledTaskUpsertInput, maxTasks: number): Promise { + return this.getEntityManager().transactional(async () => { + const exists = await this.existsForPluginAndName(input.plugin_id, input.name); + if (!exists) { + const n = await this.countForPlugin(input.plugin_id); + if (n >= maxTasks) return false; + } + await this.upsertTask(input); + return true; + }); + } } diff --git a/server/src/nest/plugins/host/plugin-audit.ts b/server/src/nest/plugins/host/plugin-audit.ts index 6b46c8fc2..8bf83a2f2 100644 --- a/server/src/nest/plugins/host/plugin-audit.ts +++ b/server/src/nest/plugins/host/plugin-audit.ts @@ -82,6 +82,57 @@ const MAX_AUDIT_ROWS = envInt('TREK_PLUGIN_AUDIT_MAX_ROWS', 20_000); const PRUNE_EVERY = 500; // amortise the COUNT/DELETE over this many appends per plugin const appendsSincePrune = new Map(); +/** + * Plan 3j Task 7 fix wave, must-land 1 (task-7-review.md R-hash-chain): a + * per-plugin promise tail serializing `appendAudit`'s read-previous-hash + + * insert pair. Before this, two RPCs from the SAME plugin racing through + * `appendAudit` concurrently could both read the same `lastHash` before + * either's INSERT committed, forking the chain — a live 40-way burst left + * 108 of 120 rows unlinked. Queuing per plugin id (the SAME module-mutable- + * state shape `appendsSincePrune` above already uses for the amortised + * prune counter) fixes it: a plugin's Nth append always reads the (N-1)th's + * committed hash, while a DIFFERENT plugin's chain never waits on this one's. + * + * `uow.transactional` (wrapping `lastHash` + `insertRow` in a real SQLite + * transaction) was the fix brief's other option, but `isAuditable` covers + * EVERY core-data/broadcast method, so it runs on nearly every granted RPC + * call — wrapping each one in a transaction would serialize the app's single + * connection mutex (`unit-of-work.ts`'s own docstring: "an open transaction + * holds it for every other request") behind the audit write on every + * authorized plugin call, a far larger blast radius than queuing one + * plugin's own chain in JS. No DB transaction opens here, so rule 24 ("a + * transactional body awaits DB work only") is not engaged either way. The + * IPC answer-then-throw ruling (`rpc-host.ts#dispatch` computes the RPC + * answer FIRST, then `try { await this.deps.audit(...) } catch { /* + * auditing must never break a call * / }` afterward) is preserved unchanged: + * a queued append's rejection surfaces only to the dispatch() call that + * queued it, through that same swallow — never to an unrelated later call. + */ +const auditAppendTails = new Map>(); + +/** + * Runs `run` after every earlier queued append for `pluginId` has settled, + * whether that one resolved or rejected — `.then(run, run)`, never a bare + * `.then(run)`, which would leave every later append for this plugin + * permanently queued behind one rejected promise. The stored tail is a + * swallowed copy (`.then(() => undefined, () => undefined)`) so a later + * call's own `auditAppendTails.get(pluginId)` read never itself rejects; + * the returned promise is the UNswallowed one, so the caller that queued + * THIS run still sees its own failure. + */ +function serializePerPlugin(pluginId: string, run: () => Promise): Promise { + const prevTail = auditAppendTails.get(pluginId) ?? Promise.resolve(); + const result = prevTail.then(run, run); + auditAppendTails.set( + pluginId, + result.then( + () => undefined, + () => undefined, + ), + ); + return result; +} + function envInt(name: string, def: number): number { const raw = process.env[name]; if (raw === undefined || raw.trim() === '') return def; @@ -110,19 +161,23 @@ export async function pruneAudit(audit: PluginCapabilityAuditRepository, pluginI * handle. */ export async function appendAudit(audit: PluginCapabilityAuditRepository, e: AuditEntry): Promise { - const prev = (await audit.lastHash(e.pluginId)) ?? ''; - const ts = new Date().toISOString(); - const row = JSON.stringify([e.pluginId, e.actingUserId ?? null, e.method, e.resource ?? null, e.code, ts]); - const hash = crypto.createHash('sha256').update(prev + row).digest('hex'); - await audit.insertRow({ - plugin_id: e.pluginId, - acting_user_id: e.actingUserId ?? null, - method: e.method, - resource: e.resource ?? null, - code: e.code, - ts, - prev_hash: prev || null, - hash, + // The read-previous-hash + insert pair, serialized per plugin — see + // `serializePerPlugin`'s docstring above (must-land 1). + await serializePerPlugin(e.pluginId, async () => { + const prev = (await audit.lastHash(e.pluginId)) ?? ''; + const ts = new Date().toISOString(); + const row = JSON.stringify([e.pluginId, e.actingUserId ?? null, e.method, e.resource ?? null, e.code, ts]); + const hash = crypto.createHash('sha256').update(prev + row).digest('hex'); + await audit.insertRow({ + plugin_id: e.pluginId, + acting_user_id: e.actingUserId ?? null, + method: e.method, + resource: e.resource ?? null, + code: e.code, + ts, + prev_hash: prev || null, + hash, + }); }); // Amortised retention: prune roughly every PRUNE_EVERY appends per plugin. const n = (appendsSincePrune.get(e.pluginId) ?? 0) + 1; diff --git a/server/src/nest/plugins/host/plugin-host-state.ts b/server/src/nest/plugins/host/plugin-host-state.ts index b4f9261fe..7c2bbb90c 100644 --- a/server/src/nest/plugins/host/plugin-host-state.ts +++ b/server/src/nest/plugins/host/plugin-host-state.ts @@ -37,6 +37,7 @@ export function closePluginDataDb(id: string): void { dataDbs.get(id)?.close(); dataDbs.delete(id); budgets.delete(id); + seedingBudgets.delete(id); } // Per-plugin daily broker budgets (ai/notify). Lazily created + seeded from the @@ -45,23 +46,45 @@ export function closePluginDataDb(id: string): void { // nothing persisted or phoned home. const budgets = new Map(); -export async function budgetFor(id: string, audit: PluginCapabilityAuditRepository): Promise { - let b = budgets.get(id); - if (!b) { - const now = Date.now(); - const since = new Date(now).toISOString().slice(0, 10) + 'T00:00:00'; - const rows = await audit.budgetSeed(id, since); - let ai = 0, notify = 0; - for (const r of rows) { - if (r.method === 'notify.send') notify += r.n; - else ai += r.n; // ai.complete + ai.extract - } - b = new DailyBudget(DEFAULT_DAILY_BUDGET, now, { ai, notify }); - budgets.set(id, b); +// Plan 3j Task 7 fix wave, should-land 7 (task-7-review.md "budget seed race"): +// `budgetFor` used to check `budgets.get(id)` and, on a miss, `await +// audit.budgetSeed(...)` before setting the map — an `await` between the read +// and the write. Two concurrent first calls for the same plugin (e.g. a burst +// of `notify.send`/`ai.complete` RPCs right after activation) could both miss, +// both seed, and each end up with its OWN `DailyBudget` instance — one +// overwriting the other in `budgets`, so usage tracked against the discarded +// instance is invisible to the surviving one and the daily cap under-counts. +// Not reproduced live (task-7-review.md: a 20-way burst with a cap of 5 came +// back 5/5 on both sides) but real given the `await` gap, so fixed defensively: +// concurrent first callers now share the SAME in-flight seeding promise (and +// therefore the SAME `DailyBudget` instance) instead of racing to seed twice. +const seedingBudgets = new Map>(); + +async function seedBudget(id: string, audit: PluginCapabilityAuditRepository): Promise { + const now = Date.now(); + const since = new Date(now).toISOString().slice(0, 10) + 'T00:00:00'; + const rows = await audit.budgetSeed(id, since); + let ai = 0, notify = 0; + for (const r of rows) { + if (r.method === 'notify.send') notify += r.n; + else ai += r.n; // ai.complete + ai.extract } + const b = new DailyBudget(DEFAULT_DAILY_BUDGET, now, { ai, notify }); + budgets.set(id, b); return b; } +export async function budgetFor(id: string, audit: PluginCapabilityAuditRepository): Promise { + const existing = budgets.get(id); + if (existing) return existing; + let pending = seedingBudgets.get(id); + if (pending === undefined) { + pending = seedBudget(id, audit).finally(() => seedingBudgets.delete(id)); + seedingBudgets.set(id, pending); + } + return pending; +} + /** Today's broker usage for one plugin (admin view). Seeds the counter if unseen. */ export async function pluginBudgetUsage(id: string, audit: PluginCapabilityAuditRepository): Promise> { return (await budgetFor(id, audit)).used(Date.now()); diff --git a/server/src/nest/plugins/host/rpc/host-surface.rpc.ts b/server/src/nest/plugins/host/rpc/host-surface.rpc.ts index b0c52f809..1eaf98ee7 100644 --- a/server/src/nest/plugins/host/rpc/host-surface.rpc.ts +++ b/server/src/nest/plugins/host/rpc/host-surface.rpc.ts @@ -204,20 +204,14 @@ export class HostSurfaceRpc { } const json = JSON.stringify(params.payload ?? null); if (json.length > SCHED_PAYLOAD_MAX) throw new BadParams(`scheduler payload too large (max ${SCHED_PAYLOAD_MAX} bytes)`); - const existing = await this.scheduledTasks.existsForPluginAndName(ctx.pluginId, name); // HR5 — Plan 3j - if (!existing) { - const n = await this.scheduledTasks.countForPlugin(ctx.pluginId); // HR6 — Plan 3j - if (n >= SCHED_MAX) throw new BadParams(`too many scheduled tasks (max ${SCHED_MAX})`); - } + // HR5/HR6/HR7 — Plan 3j Task 7 fix (must-land 3): one atomic call, not + // three separately-awaited ones (task-7-review.md's concurrent-cap-bypass). // Upsert by (plugin, name): re-scheduling the same name replaces it. - await this.scheduledTasks.upsertTask({ - // HR7 — Plan 3j - plugin_id: ctx.pluginId, - name, - due_at: Math.max(dueAt, Date.now()), - payload: json, - every_ms: everyMs ?? null, - }); + const written = await this.scheduledTasks.upsertTaskCapped( + { plugin_id: ctx.pluginId, name, due_at: Math.max(dueAt, Date.now()), payload: json, every_ms: everyMs ?? null }, + SCHED_MAX, + ); + if (!written) throw new BadParams(`too many scheduled tasks (max ${SCHED_MAX})`); return { scheduled: true }; } diff --git a/server/src/nest/plugins/host/rpc/meta.rpc.ts b/server/src/nest/plugins/host/rpc/meta.rpc.ts index 777c92e50..7d8bb3cf7 100644 --- a/server/src/nest/plugins/host/rpc/meta.rpc.ts +++ b/server/src/nest/plugins/host/rpc/meta.rpc.ts @@ -78,12 +78,10 @@ export class MetaRpc { if (key.length > META_KEY_MAX) throw new BadParams(`metadata key too long (>${META_KEY_MAX} chars)`); const json = JSON.stringify(params.value ?? null); if (json.length > META_VALUE_MAX) throw new BadParams(`metadata value too large (>${META_VALUE_MAX} bytes)`); - const exists = (await this.meta.findValue(ctx.pluginId, entityType, entityId, key)) !== null; // MR2 — Plan 3j - if (!exists) { - const n = await this.meta.countForEntity(ctx.pluginId, entityType, entityId); // MR3 — Plan 3j - if (n >= META_KEYS_MAX) throw new BadParams(`too many metadata keys on this ${entityType} (max ${META_KEYS_MAX})`); - } - await this.meta.upsertValue(ctx.pluginId, entityType, entityId, key, json); // MR4 — Plan 3j + // MR2/MR3/MR4 — Plan 3j Task 7 fix (must-land 2): one atomic call, not + // three separately-awaited ones (task-7-review.md's concurrent-cap-bypass). + const written = await this.meta.upsertValueCapped(ctx.pluginId, entityType, entityId, key, json, META_KEYS_MAX); + if (!written) throw new BadParams(`too many metadata keys on this ${entityType} (max ${META_KEYS_MAX})`); return { key, value: params.value ?? null }; } diff --git a/server/src/nest/plugins/plugin-runtime.service.ts b/server/src/nest/plugins/plugin-runtime.service.ts index aae5e5125..fa360f3ac 100644 --- a/server/src/nest/plugins/plugin-runtime.service.ts +++ b/server/src/nest/plugins/plugin-runtime.service.ts @@ -814,10 +814,13 @@ export class PluginRuntimeService implements OnApplicationBootstrap, OnModuleDes if (!clean.includes(host)) clean.push(host); } if (!this.uow) throw new Error('UnitOfWork not provided — tests that set egress hosts must pass one'); - // PR22/PR23, R-uninstall's ONE named transaction: the delete-then-insert-loop is now - // atomic (`PluginEgressHostsRepository.replaceAllForPlugin`) — a crash mid-write - // leaves the OLD host set intact instead of a plugin with zero egress hosts, a - // deliberate behavior change ("For the user" in the plan document), not silent parity. + // PR22/PR23: the delete-then-insert-loop is atomic (`PluginEgressHostsRepository + // .replaceAllForPlugin`) — a crash mid-write leaves the OLD host set intact instead + // of a plugin with zero egress hosts. Plan 3j Task 7 fix (should-land 6, + // task-7-review.md): this transaction is PARITY, not a new behaviour — the legacy + // code already wrapped this delete+insert in a transaction (base: line 679). It was + // wrongly reported as a deliberate deviation; corrected here, and the egress "For + // the user" item is dropped (no decision needed — nothing changed). await this.uow.transactional(async () => { await this.pluginEgressHosts.replaceAllForPlugin(id, clean); }); diff --git a/server/tests/integration/plugins/plugin-runtime.task2-conversion.test.ts b/server/tests/integration/plugins/plugin-runtime.task2-conversion.test.ts index c188985e4..6125fbac6 100644 --- a/server/tests/integration/plugins/plugin-runtime.task2-conversion.test.ts +++ b/server/tests/integration/plugins/plugin-runtime.task2-conversion.test.ts @@ -301,21 +301,28 @@ describe('Plan 3j Task 2 — R-uninstall\'s ONE named transaction: PR22/PR23 egr await rt.setOperatorEgressHosts(id, ['old-a.example.com', 'old-b.example.com']); expect((await rt.operatorEgressHosts(id)).sort()).toEqual(['old-a.example.com', 'old-b.example.com']); - // Force the INSERT half of the DELETE+loop-INSERT to throw on the second host, - // inside the SAME uow.transactional call the runtime already wraps this in. + // Plan 3j Task 7 fix (must-land 4a, task-7-review.md): mocking the WHOLE + // `replaceAllForPlugin` call (the old shape of this test) throws before the real + // DELETE ever runs, so it proves nothing about the transaction — M13 (deleting the + // `uow.transactional` wrapper entirely) still passed against it. Spy on the per-host + // `upsert` call INSIDE `replaceAllForPlugin` instead, and only throw on the SECOND + // host: the real DELETE and the real first-host INSERT both already happened by the + // time this throws, so the assertion below can only pass if the transaction actually + // rolled both of them back, not just skipped a write that was never attempted. // eslint-disable-next-line @typescript-eslint/no-explicit-any const egressRepo = (rt as any).pluginEgressHosts; - const originalReplace = egressRepo.replaceAllForPlugin.bind(egressRepo); - const spy = vi.spyOn(egressRepo, 'replaceAllForPlugin').mockImplementation(async (pluginId: string, hosts: string[]) => { - if (hosts.includes('boom')) throw new Error('simulated mid-write failure'); - return originalReplace(pluginId, hosts); + const originalUpsert = egressRepo.upsert.bind(egressRepo); + const spy = vi.spyOn(egressRepo, 'upsert').mockImplementation(async (data: { plugin_id: string; host: string }, ...rest: unknown[]) => { + if (data.host === 'boom') throw new Error('simulated mid-write failure'); + return originalUpsert(data, ...rest); }); await expect(rt.setOperatorEgressHosts(id, ['new-a.example.com', 'boom'])).rejects.toThrow(/simulated mid-write failure/); spy.mockRestore(); - // The OLD set survives untouched — the transaction rolled the DELETE back too, - // not a partially-cleared allow-list. + // The OLD set survives untouched — the transaction rolled the DELETE AND the + // already-succeeded 'new-a.example.com' insert back too, not a partially-cleared or + // partially-replaced allow-list. expect((await rt.operatorEgressHosts(id)).sort()).toEqual(['old-a.example.com', 'old-b.example.com']); }); }); @@ -384,6 +391,45 @@ describe('Plan 3j Task 2 — two racing ticks (scheduler claim/re-arm + erasure } }); + it('RACE-SCHED-002: two concurrent fireDueScheduled passes deliver a due RECURRING task exactly once (the re-arm claim guard, unproven by RACE-SCHED-001)', async () => { + // Plan 3j Task 7 fix (must-land 4b, task-7-review.md): RACE-SCHED-001 above only + // exercises the ONE-SHOT branch (`deleteById`'s claim guard) — the recurring branch + // (`rearm`'s claim guard) went untested, and both mutations dropping IT survived + // (M6/M6b). Same shape as RACE-SCHED-001, on a task with `every_ms` set instead. + const pluginId = 'race-sched-recur'; + seedPlugin(pluginId, { enabled: 1 }); + const everyMs = 60_000; + testDb.prepare('INSERT INTO plugin_scheduled_tasks (plugin_id, name, due_at, every_ms) VALUES (?, ?, ?, ?)').run(pluginId, 'recurring', Date.now() - 1000, everyMs); + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const supervisor = (rt as any).supervisor; + supervisor.running.set(pluginId, { id: pluginId, status: 'active' }); + let delivered = 0; + const invokeSpy = vi.spyOn(supervisor, 'invoke').mockImplementation(async () => { + delivered += 1; + // Yield, so the second concurrent pass's own claim-read genuinely overlaps + // this one's window rather than running strictly after it. + await new Promise((r) => setTimeout(r, 5)); + return {}; + }); + const before = Date.now(); + + try { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + await Promise.all([(rt as any).fireDueScheduled(), (rt as any).fireDueScheduled()]); + // Without the `due_at <= now` guard on `rearm`, BOTH concurrent passes would read + // `claimed = true` and both invoke — this is the assertion RACE-SCHED-001's own + // shape proves for the one-shot branch, now proven for the recurring one. + expect(delivered).toBe(1); + const row = testDb.prepare('SELECT due_at FROM plugin_scheduled_tasks WHERE plugin_id=? AND name=?').get(pluginId, 'recurring') as { due_at: number } | undefined; + expect(row).toBeDefined(); // a recurring task is re-armed, never deleted + expect(row?.due_at).toBeGreaterThanOrEqual(before + everyMs); // moved forward, not left due in the past + } finally { + invokeSpy.mockRestore(); + supervisor.running.delete(pluginId); + } + }); + it('RACE-ERASURE-001: two concurrent drains coalesce onto the SAME in-flight pass (drainUserErasures\' own promise-sharing) — one ACK, one drop', async () => { const pluginId = 'race-erasure'; seedPlugin(pluginId, { enabled: 1 }); diff --git a/server/tests/unit/plugins/host-rpc.test.ts b/server/tests/unit/plugins/host-rpc.test.ts index 218f6adab..38c9aef0c 100644 --- a/server/tests/unit/plugins/host-rpc.test.ts +++ b/server/tests/unit/plugins/host-rpc.test.ts @@ -63,6 +63,7 @@ import { MetaRpc } from '../../../src/nest/plugins/host/rpc/meta.rpc'; import { HostSurfaceRpc } from '../../../src/nest/plugins/host/rpc/host-surface.rpc'; import { UnreadableLlmResponse } from '../../../src/nest/llm-parse/clients/openai-compatible.client'; import { getPluginDataDb, closePluginDataDb } from '../../../src/nest/plugins/host/plugin-host-state'; +import { verifyChain } from '../../../src/nest/plugins/host/plugin-audit'; import { db as mockDb } from '../../../src/db/database'; import { DatabaseService } from '../../../src/nest/database/database.service'; import { RealtimeService } from '../../../src/nest/realtime/realtime.service'; @@ -226,6 +227,24 @@ const metaRepo = { async deleteValue(pluginId: string, entityType: string, entityId: number, key: string) { return raw.prepare('DELETE FROM plugin_entity_metadata WHERE plugin_id=? AND entity_type=? AND entity_id=? AND key=?').run(pluginId, entityType, entityId, key).changes > 0; }, + // Plan 3j Task 7 fix (must-land 2) — `meta.rpc.ts#set` now calls this ONE method + // instead of `findValue`+`countForEntity`+`upsertValue` separately. One synchronous + // better-sqlite3 statement (no `await` gap between the cap check and the write), + // the same atomicity property the real `PluginEntityMetadataRepository + // .upsertValueCapped` gets from `getEntityManager().transactional`. + async upsertValueCapped(pluginId: string, entityType: string, entityId: number, key: string, value: string, maxKeys: number) { + const result = raw + .prepare( + `INSERT INTO plugin_entity_metadata (plugin_id, entity_type, entity_id, key, value, updated_at) + SELECT ?, ?, ?, ?, ?, datetime('now') + WHERE EXISTS (SELECT 1 FROM plugin_entity_metadata WHERE plugin_id=? AND entity_type=? AND entity_id=? AND key=?) + OR (SELECT COUNT(*) FROM plugin_entity_metadata WHERE plugin_id=? AND entity_type=? AND entity_id=?) < ? + ON CONFLICT(plugin_id, entity_type, entity_id, key) + DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`, + ) + .run(pluginId, entityType, entityId, key, value, pluginId, entityType, entityId, key, pluginId, entityType, entityId, maxKeys); + return result.changes > 0; + }, } as unknown as PluginEntityMetadataRepository; const scheduledTasksRepo = { @@ -246,6 +265,22 @@ const scheduledTasksRepo = { async deleteByPluginAndName(pluginId: string, name: string) { return raw.prepare('DELETE FROM plugin_scheduled_tasks WHERE plugin_id = ? AND name = ?').run(pluginId, name).changes > 0; }, + // Plan 3j Task 7 fix (must-land 3) — `host-surface.rpc.ts#schedulerSet` now calls + // this ONE method instead of `existsForPluginAndName`+`countForPlugin`+`upsertTask` + // separately. Same one-statement atomicity shape as `metaRepo.upsertValueCapped` + // above, mirroring the real `PluginScheduledTasksRepository.upsertTaskCapped`. + async upsertTaskCapped(input: { plugin_id: string; name: string; due_at: number; payload: string; every_ms: number | null }, maxTasks: number) { + const result = raw + .prepare( + `INSERT INTO plugin_scheduled_tasks (plugin_id, name, due_at, payload, every_ms) + SELECT ?, ?, ?, ?, ? + WHERE EXISTS (SELECT 1 FROM plugin_scheduled_tasks WHERE plugin_id=? AND name=?) + OR (SELECT COUNT(*) FROM plugin_scheduled_tasks WHERE plugin_id=?) < ? + ON CONFLICT (plugin_id, name) DO UPDATE SET due_at = excluded.due_at, payload = excluded.payload, every_ms = excluded.every_ms`, + ) + .run(input.plugin_id, input.name, input.due_at, input.payload, input.every_ms, input.plugin_id, input.name, input.plugin_id, maxTasks); + return result.changes > 0; + }, } as unknown as PluginScheduledTasksRepository; const dbs = new DatabaseService(mockDb); @@ -562,6 +597,30 @@ describe('HostSurfaceRpc — users, broadcasts, notify, ai, oauth, scheduler', ( expect((await call(host, 'users.getById', { id: 6 }, undefined)).error?.code).toBe('RESOURCE_FORBIDDEN'); }); + it('HOSTRPC-033 a burst of concurrent auditable RPC calls for the SAME plugin never forks the capability-audit hash chain (Plan 3j Task 7 fix, must-land 1)', async () => { + // The REAL production path, not a direct `appendAudit` unit call: `dispatch()` + // audits every `isAuditable` method AFTER computing its answer (`users.getById` + // here, permission `db:read:users` != `db:own`), through the SAME per-plugin + // promise-tail `appendAudit` now serializes on — task-7-review.md's SDK default of + // 16 in-flight RPCs is what made this reachable live. + const host = makeHost('auditburst', 'db:read:users'); + const N = 20; + const results = await Promise.all(Array.from({ length: N }, () => call(host, 'users.getById', { id: 5 }, 5))); + expect(results.every((r) => r.ok)).toBe(true); + const rows = ( + mockDb as unknown as { + prepare(s: string): { all(...a: unknown[]): Array<{ plugin_id: string; acting_user_id: number | null; method: string; resource: string | null; code: string; ts: string; prev_hash: string | null; hash: string }> }; + } + ) + .prepare("SELECT plugin_id, acting_user_id, method, resource, code, ts, prev_hash, hash FROM plugin_capability_audit WHERE plugin_id = 'auditburst' ORDER BY id ASC") + .all(); + expect(rows).toHaveLength(N); + expect(verifyChain(rows)).toBe(true); + const prevHashes = rows.map((r) => r.prev_hash ?? ''); + expect(new Set(prevHashes).size).toBe(prevHashes.length); // no two rows read the same "previous" tip + closePluginDataDb('auditburst'); + }); + it('HOSTRPC-020 trip broadcasts are force-namespaced and membership-gated', async () => { const host = makeHost('surface', 'ws:broadcast:trip'); expect((await call(host, 'ws.broadcastToTrip', { tripId: 1, event: 'ping', data: { a: 1 } }, 5)).ok).toBe(true); diff --git a/server/tests/unit/plugins/plugin-audit.test.ts b/server/tests/unit/plugins/plugin-audit.test.ts index 552b872eb..dd1686452 100644 --- a/server/tests/unit/plugins/plugin-audit.test.ts +++ b/server/tests/unit/plugins/plugin-audit.test.ts @@ -142,6 +142,29 @@ describe('appendAudit hash chain (converted, through PluginCapabilityAuditReposi }); }); +describe('R-hash-chain: concurrency (Plan 3j Task 7 fix wave, must-land 1)', () => { + it('R3J-CHAIN-002: 10 concurrent appendAudit calls for the SAME plugin produce a linear, gap-free chain (red without the per-plugin serialization fix)', async () => { + // The exact real code path production uses: `appendAudit`, racing N callers, not a + // sequential loop — task-7-review.md's own live 40-way burst found 108 of 120 rows + // unlinked (108 duplicate prev_hash values) before this fix; 9 of 10 broken in an + // earlier, in-process 10-way version of this same test. + await Promise.all( + Array.from({ length: 10 }, (_, i) => + appendAudit(audit, { pluginId: 'burst', actingUserId: 1, method: 'trips.getById', resource: `trip:${i}`, code: 'ok' }), + ), + ); + const rows = chainRows('burst'); + expect(rows).toHaveLength(10); + expect(verifyChain(rows)).toBe(true); // every row's prev_hash === the previous row's hash + // No two rows share a prev_hash — the exact shape a forked chain (two racers reading + // the SAME "previous" tip) would produce. + const prevHashes = rows.map((r) => r.prev_hash ?? ''); + expect(new Set(prevHashes).size).toBe(prevHashes.length); + // Exactly one row has no predecessor (the genesis of this plugin's chain). + expect(prevHashes.filter((h) => h === '').length).toBe(1); + }); +}); + describe('R-hash-chain: replay, extension, mutation', () => { it('AUDITCHAIN-001 REPLAY+EXTENSION: verifies a legacy-produced chain, then extends it through the converted appendAudit and the WHOLE chain stays consistent', async () => { // 1. Write three rows the way the PRE-CONVERSION code did: raw SQL, no repository. diff --git a/server/tests/unit/plugins/plugin-repository-sql-parity.test.ts b/server/tests/unit/plugins/plugin-repository-sql-parity.test.ts new file mode 100644 index 000000000..ee0df2040 --- /dev/null +++ b/server/tests/unit/plugins/plugin-repository-sql-parity.test.ts @@ -0,0 +1,212 @@ +/** + * Real-SQL parity tests for the plugin-adjacent repository methods Plan 3j Task 5 + * converted but never covered against raw SQL (task-7-review.md, must-land 4c): the + * "Reviewer's full-seed parity" fixture (`r3j-parity.test.ts`, run in-process under the + * review's own isolated `r3j-mut` export, 5/5 green there) — ported here per the fix + * brief ("donate it, never import from the scratchpad"), not merely referenced. + * + * Each assertion below compares the converted repository method's result against the + * SAME statement run directly on the raw better-sqlite3 handle — the shape that catches + * M14 (`findPublicIdentity`'s miss returning `null` instead of `undefined`), M15 (its + * key order), M16 (`sharesTripWith` always `true`, a security gate) and M17 + * (`existsById` ignored) from the review's mutation table, none of which any other + * suite killed. + */ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { createSnapshotTestDb } from '../../helpers/db-mock'; +import { resetTestDb } from '../../helpers/test-db'; +import { createTestOrm, type TestOrm } from '../../helpers/test-orm'; +import { Trips } from '../../../src/db/entities/Trips.entity'; +import { Users } from '../../../src/db/entities/Users.entity'; +import { Days } from '../../../src/db/entities/Days.entity'; +import { PluginEntityMetadata } from '../../../src/db/entities/PluginEntityMetadata.entity'; +import { PluginScheduledTasks } from '../../../src/db/entities/PluginScheduledTasks.entity'; + +const testDb = createSnapshotTestDb(); +let t: TestOrm; +beforeAll(async () => { + t = await createTestOrm(testDb); +}); +beforeEach(() => { + resetTestDb(testDb); + t.clear(); +}); +afterAll(async () => { + await t.close(); + testDb.close(); +}); + +function seed() { + for (let i = 1; i <= 6; i++) { + testDb + .prepare("INSERT INTO users (id, username, email, password_hash, display_name, avatar) VALUES (?, ?, ?, 'x', ?, ?)") + .run(i, 'u' + i, `u${i}@x`, i % 2 ? 'D' + i : null, i === 3 ? 'a.png' : null); + } + const trips: Array<[number, number]> = [ + [1, 1], + [2, 2], + [3, 1], + [4, 5], + ]; + for (const [id, owner] of trips) testDb.prepare('INSERT INTO trips (id, user_id, title) VALUES (?, ?, ?)').run(id, owner, 't'); + for (const [trip, user] of [ + [1, 2], + [1, 3], + [2, 3], + [3, 4], + [4, 1], + ]) { + testDb.prepare('INSERT INTO trip_members (trip_id, user_id) VALUES (?, ?)').run(trip, user); + } + for (let d = 1; d <= 5; d++) testDb.prepare('INSERT INTO days (id, trip_id, day_number) VALUES (?, ?, ?)').run(10 + d, d % 2 ? 1 : 2, d); +} + +describe('R3J-PARITY (Plan 3j Task 7 fix wave, must-land 4c)', () => { + it('HR9 Trips.sharesTripWith == legacy bilateral SQL for every ordered user pair (incl. missing users)', async () => { + seed(); + const repo = t.repo(Trips); + const legacy = testDb.prepare(`SELECT 1 FROM trips t + LEFT JOIN trip_members m1 ON m1.trip_id = t.id AND m1.user_id = ? + LEFT JOIN trip_members m2 ON m2.trip_id = t.id AND m2.user_id = ? + WHERE (t.user_id = ? OR m1.user_id IS NOT NULL) + AND (t.user_id = ? OR m2.user_id IS NOT NULL) + LIMIT 1`); + const diffs: string[] = []; + for (let a = 1; a <= 7; a++) { + for (let b = 1; b <= 7; b++) { + const legacyResult = !!legacy.get(a, b, a, b); + const converted = await repo.sharesTripWith(a, b); + if (legacyResult !== converted) diffs.push(`${a},${b}: legacy=${legacyResult} new=${converted}`); + } + } + expect(diffs).toEqual([]); + }); + + it('Trips.existsById == legacy row-exists SQL, including a miss', async () => { + seed(); + const repo = t.repo(Trips); + for (const id of [1, 2, 3, 4, 99]) { + const legacy = !!testDb.prepare('SELECT 1 FROM trips WHERE id = ?').get(id); + expect(await repo.existsById(id)).toBe(legacy); + } + }); + + it('HR1 Users.findPublicIdentity == legacy row (full key, key order) and miss', async () => { + seed(); + const repo = t.repo(Users); + for (let i = 1; i <= 7; i++) { + const legacy = testDb.prepare('SELECT id, username, display_name, avatar FROM users WHERE id = ?').get(i); + const converted = (await repo.findPublicIdentity(i)) ?? undefined; + // JSON.stringify: same keys, same order, same values — catches M15 (key order) + // as well as a value drift a plain `toEqual` on two possibly-undefined objects + // would not distinguish from M14 (miss shape) below. + expect(JSON.stringify(converted)).toBe(JSON.stringify(legacy)); + } + // M14: a miss returns `null` (this repository's own convention, see + // `host-surface.rpc.ts#getUser`'s `?? undefined` docstring) — the CALLER maps + // that to `undefined` to match the legacy `better-sqlite3` `.get()` shape, which + // the `JSON.stringify` comparison above already exercises for id 7 (a miss). + expect(await repo.findPublicIdentity(999)).toBeNull(); + }); + + it('CT1/CT2 Days.listIdsByTrip + MR9 Days.findTripId == legacy', async () => { + seed(); + const repo = t.repo(Days); + for (const trip of [1, 2, 3, 99]) { + const legacy = (testDb.prepare('SELECT id FROM days WHERE trip_id = ?').all(trip) as Array<{ id: number }>).map((r) => r.id); + expect(await repo.listIdsByTrip(trip)).toEqual(legacy); + } + for (const id of [11, 12, 99]) { + const legacy = (testDb.prepare('SELECT trip_id FROM days WHERE id = ?').get(id) as { trip_id: number } | undefined)?.trip_id; + expect(await repo.findTripId(id)).toEqual(legacy); + } + }); + + it('MR1-MR6 PluginEntityMetadata repository == legacy statements', async () => { + seed(); + const repo = t.repo(PluginEntityMetadata); + await repo.upsertValue('p', 'trip', 1, 'b', '"1"'); + await repo.upsertValue('p', 'trip', 1, 'a', '"2"'); + await repo.upsertValue('q', 'trip', 1, 'a', '"x"'); + const before = testDb.prepare("SELECT updated_at FROM plugin_entity_metadata WHERE plugin_id='p' AND key='b'").get() as { updated_at: string }; + testDb.prepare("UPDATE plugin_entity_metadata SET updated_at='2000-01-01 00:00:00' WHERE plugin_id='p' AND key='b'").run(); + await repo.upsertValue('p', 'trip', 1, 'b', '"3"'); + const after = testDb.prepare("SELECT id, value, updated_at FROM plugin_entity_metadata WHERE plugin_id='p' AND key='b'").get() as { + id: number; + value: string; + updated_at: string; + }; + expect(after.value).toBe('"3"'); // MR4: upsert replaces the value + expect(after.updated_at).not.toBe('2000-01-01 00:00:00'); // MR4: updated_at refreshed on conflict + expect(after.updated_at).toMatch(/^\d{4}-\d\d-\d\d \d\d:\d\d:\d\d$/); + expect(before.updated_at).toMatch(/^\d{4}-\d\d-\d\d \d\d:\d\d:\d\d$/); + expect(await repo.findValue('p', 'trip', 1, 'b')).toBe('"3"'); // MR1 + expect(await repo.findValue('p', 'trip', 1, 'zz')).toBeNull(); // MR1 miss + expect(await repo.countForEntity('p', 'trip', 1)).toBe(2); // MR3 + expect(await repo.listForEntity('p', 'trip', 1)).toEqual( + testDb.prepare("SELECT key, value FROM plugin_entity_metadata WHERE plugin_id=? AND entity_type=? AND entity_id=? ORDER BY key").all('p', 'trip', 1), + ); // MR5, ORDER BY key + expect(await repo.deleteValue('p', 'trip', 1, 'a')).toBe(true); // MR6 + expect(await repo.deleteValue('p', 'trip', 1, 'a')).toBe(false); // MR6, already gone + expect(testDb.prepare('SELECT COUNT(*) c FROM plugin_entity_metadata').get()).toEqual({ c: 2 }); // 'p'/'b' + 'q'/'a' survive + }); + + it('HR5-HR8 PluginScheduledTasks repository == legacy statements (upsert keeps id, replaces fields)', async () => { + const repo = t.repo(PluginScheduledTasks); + await repo.upsertTask({ plugin_id: 'p', name: 'n', due_at: 5, payload: '1', every_ms: null }); + const id1 = (testDb.prepare("SELECT id FROM plugin_scheduled_tasks WHERE name='n'").get() as { id: number }).id; + await repo.upsertTask({ plugin_id: 'p', name: 'n', due_at: 9, payload: '2', every_ms: 60000 }); + expect(testDb.prepare('SELECT id, plugin_id, name, due_at, payload, every_ms FROM plugin_scheduled_tasks').all()).toEqual([ + { id: id1, plugin_id: 'p', name: 'n', due_at: 9, payload: '2', every_ms: 60000 }, + ]); // HR7: same id, replaced fields — an upsert, not a delete+insert + expect(await repo.existsForPluginAndName('p', 'n')).toBeTruthy(); // HR5 + expect(await repo.existsForPluginAndName('p', 'x')).toBeFalsy(); // HR5 miss + expect(await repo.countForPlugin('p')).toBe(1); // HR6 + expect(await repo.deleteByPluginAndName('p', 'n')).toBe(true); // HR8 + expect(await repo.deleteByPluginAndName('p', 'n')).toBe(false); // HR8, already gone + }); + + it("HR5-HR7 re-arm's claim guard: rearm only claims a row whose due_at is still <= the claimed-from time", async () => { + const repo = t.repo(PluginScheduledTasks); + const now = Date.now(); + await repo.upsertTask({ plugin_id: 'p', name: 'claim', due_at: now - 1000, payload: '1', every_ms: 60000 }); + const id = (testDb.prepare("SELECT id FROM plugin_scheduled_tasks WHERE name='claim'").get() as { id: number }).id; + // The winning claim: due_at <= claimedFromDueAt (the row's own current due_at). + expect(await repo.rearm(id, now + 60000, now - 1000)).toBe(true); + // A second, losing claim against the SAME stale claimedFromDueAt now fails — the + // row's due_at already moved forward past it (RACE-SCHED-001/002's guard). + expect(await repo.rearm(id, now + 120000, now - 1000)).toBe(false); + const row = testDb.prepare('SELECT due_at FROM plugin_scheduled_tasks WHERE id = ?').get(id) as { due_at: number }; + expect(row.due_at).toBe(now + 60000); // the losing rearm never wrote + // deleteById carries the identical guard shape for the one-shot branch. + expect(await repo.deleteById(id, now - 1000)).toBe(false); // stale claim, no-op + expect(await repo.deleteById(id, now + 60000)).toBe(true); // current due_at, claims + deletes + }); + + it('must-land 2: 130 concurrent PluginEntityMetadata.upsertValueCapped calls for the SAME (plugin, entity), 130 DIFFERENT new keys, never exceed the 100 cap', async () => { + // The REAL repository method, raced through Promise.all — not a sequential loop — + // so the exists+count+write really overlaps (task-7-review.md: live, 130 rows landed + // on a 100 cap before this fix). Red without `upsertValueCapped`'s + // `getEntityManager().transactional` wrap: the old three-separately-awaited-calls + // shape let every racer's stale `count < 100` read pass before any of their writes + // landed. + const repo = t.repo(PluginEntityMetadata); + const results = await Promise.all( + Array.from({ length: 130 }, (_, i) => repo.upsertValueCapped('capbomb', 'trip', 1, `k${i}`, String(i), 100)), + ); + expect(results.filter(Boolean).length).toBe(100); // exactly 100 writes accepted + expect(await repo.countForEntity('capbomb', 'trip', 1)).toBe(100); // never more than 100 rows + expect(testDb.prepare("SELECT COUNT(*) c FROM plugin_entity_metadata WHERE plugin_id='capbomb'").get()).toEqual({ c: 100 }); + }); + + it('must-land 3: 130 concurrent PluginScheduledTasks.upsertTaskCapped calls for the SAME plugin, 130 DIFFERENT new names, never exceed the 100 cap', async () => { + const repo = t.repo(PluginScheduledTasks); + const due = Date.now() + 60_000; + const results = await Promise.all( + Array.from({ length: 130 }, (_, i) => repo.upsertTaskCapped({ plugin_id: 'schedbomb', name: `t${i}`, due_at: due, payload: '1', every_ms: null }, 100)), + ); + expect(results.filter(Boolean).length).toBe(100); // exactly 100 writes accepted + expect(await repo.countForPlugin('schedbomb')).toBe(100); // never more than 100 rows + expect(testDb.prepare("SELECT COUNT(*) c FROM plugin_scheduled_tasks WHERE plugin_id='schedbomb'").get()).toEqual({ c: 100 }); + }); +});