From c0f4654bb9672cbdd555f8ce0e2fe596152b4e46 Mon Sep 17 00:00:00 2001 From: Hardik Bhatia Date: Sat, 26 Sep 2026 21:31:40 +0530 Subject: [PATCH] fix: keep recovered decisions and outbox events consistent --- apps/control-plane/src/app.ts | 31 +++--------- apps/control-plane/test/reviews.test.ts | 14 +++++- apps/gateway/src/app.ts | 61 +++++++---------------- packages/storage/src/event-replay.test.ts | 17 +++++++ packages/storage/src/index.ts | 7 +-- 5 files changed, 60 insertions(+), 70 deletions(-) create mode 100644 packages/storage/src/event-replay.test.ts diff --git a/apps/control-plane/src/app.ts b/apps/control-plane/src/app.ts index 89d3f27..a7061d3 100644 --- a/apps/control-plane/src/app.ts +++ b/apps/control-plane/src/app.ts @@ -75,32 +75,15 @@ export async function buildControlPlane(config: ControlPlaneConfig): Promise { - const parsed = AppSchema.safeParse(record); - return parsed.success ? [parsed.data] : []; + await appsStore.update((stored) => { + const valid = stored.flatMap((record) => { const parsed = AppSchema.safeParse(record); return parsed.success ? [parsed.data] : []; }); + return valid.length ? valid : [createDefaultApp()]; }); - const validApps = normalizedApps.length ? normalizedApps : [createDefaultApp()]; - if (JSON.stringify(validApps) !== JSON.stringify(storedApps)) await appsStore.write(validApps); - const storedKeys = await keysStore.read(); - if (storedKeys.some((key) => !key.appId)) { - await keysStore.write(storedKeys.map((key) => ({ ...key, appId: key.appId ?? "default" }))); - } - const storedProfiles = await profilesStore.read(); - const normalizedProfiles = storedProfiles.flatMap((profile) => { - const parsed = ProfileSchema.safeParse(profile); - return parsed.success ? [parsed.data] : []; + await keysStore.update((stored) => stored.map((key) => ({ ...key, appId: key.appId ?? "default" }))); + await settingsStore.update((stored) => { + const parsed = ProviderSettingsSchema.safeParse(stored); + return parsed.success ? parsed.data : { ...createDefaultProviderSettings(), endpoint: config.typesafeEndpoint, model: config.typesafeModel }; }); - if (normalizedProfiles.length !== storedProfiles.length || JSON.stringify(normalizedProfiles) !== JSON.stringify(storedProfiles)) { - await profilesStore.write(normalizedProfiles.length ? normalizedProfiles : [createDefaultProfile()]); - } - const storedSettings = await settingsStore.read(); - const normalizedSettings = ProviderSettingsSchema.safeParse(storedSettings); - if (normalizedSettings.success && JSON.stringify(normalizedSettings.data) !== JSON.stringify(storedSettings)) { - await settingsStore.write(normalizedSettings.data); - } else if (!normalizedSettings.success) { - await settingsStore.write({ ...createDefaultProviderSettings(), endpoint: config.typesafeEndpoint, model: config.typesafeModel }); - } const requireSession = accessGuard(database); diff --git a/apps/control-plane/test/reviews.test.ts b/apps/control-plane/test/reviews.test.ts index a0f4288..a1f47aa 100644 --- a/apps/control-plane/test/reviews.test.ts +++ b/apps/control-plane/test/reviews.test.ts @@ -1,7 +1,8 @@ import assert from "node:assert/strict"; import test from "node:test"; import { randomUUID } from "node:crypto"; -import { openDatabase } from "@pyro/storage"; +import { encryptText, openDatabase } from "@pyro/storage"; +import { DeliveryWorker, signWebhook } from "@pyro/integrations"; import { buildControlPlane } from "../src/app.js"; test("review resolution is attributed, immutable, concurrency-safe and does not change a decision", async (t) => { const config = { host: "127.0.0.1", port: 0, databaseUrl: `memory://reviews-${randomUUID()}`, adminPassword: "correct-horse-battery-staple", controlPlaneSecret: "control-plane-test-secret", gatewayInternalUrl: "http://127.0.0.1:1", gatewayApiKey: "test-key", typesafeEndpoint: "https://api.typesafe.ai/v1/systemone", typesafeModel: "jev-latest" }; @@ -9,11 +10,22 @@ test("review resolution is attributed, immutable, concurrency-safe and does not const login = await app.inject({ method: "POST", url: "/api/auth/login", payload: { password: config.adminPassword } }); const cookie = login.headers["set-cookie"]!.split(";")[0]!; const event = { id: "review-one", appId: "default", createdAt: new Date().toISOString(), profileId: "default", action: "review" as const, verdict: "suspicious" as const, risk: .7, confidence: .7, reason: "review", detectors: [], model: "local", provider: "local-rules", latencyMs: 0, queueMs: 0, inputHash: "hash" }; await db.events.append(event); + const integrationId = randomUUID(), signingSecret = "qa-webhook-signing-secret"; + await db.document("integrations", () => []).write([{ id: integrationId, name: "Review callback", type: "webhook", enabled: true, reviewResolutions: true, actions: ["review"], minimumRisk: 0, profileIds: [], appIds: ["default"], allowPrivateNetwork: true, destination: encryptText("http://127.0.0.1/events", config.controlPlaneSecret), signingSecret: encryptText(signingSecret, config.controlPlaneSecret), createdAt: event.createdAt, updatedAt: event.createdAt }] as never[]); const edit = (disposition: string) => app.inject({ method: "PUT", url: "/api/reviews/review-one", headers: { cookie }, payload: { expectedRevision: 0, disposition, comment: "Redacted feedback" } }); const responses = await Promise.all([edit("true_positive"), edit("false_positive")]); assert.deepEqual(responses.map((r) => r.statusCode).sort(), [200, 409]); assert.deepEqual(await db.events.findById(event.id), event); const review = (await app.inject({ method: "GET", url: "/api/reviews/review-one", headers: { cookie } })).json().review; + for (let i = 0; i < 40 && !(await db.deliveries.list(integrationId)).length; i++) await new Promise((r) => setTimeout(r, 50)); + const callbacks = await db.deliveries.list(integrationId); + assert.equal(callbacks.length, 1); assert.equal(callbacks[0]!.payload.type, "review.resolved"); + assert.equal(callbacks[0]!.payload.data.action, "review"); + const transport = new DeliveryWorker(db, config.controlPlaneSecret, async (_url, body, headers) => { + assert.equal(headers["x-pyro-signature"], signWebhook(signingSecret, headers["x-pyro-timestamp"]!, body)); + assert.equal(JSON.parse(body).data.review.revision, 1); return { status: 204 }; + }); + await transport.tick(); assert.equal((await db.deliveries.list(integrationId))[0]!.status, "delivered"); assert.equal(review.revision, 1); assert.equal(review.status, "resolved"); assert.equal(review.resolvedBy, login.json().user.id); assert.equal(review.comments.length, 1); const viewer = (await app.inject({ method: "POST", url: "/api/team", headers: { cookie }, payload: { username: "viewer", role: "viewer", appIds: ["default"] } })).json(); const session = await app.inject({ method: "POST", url: "/api/auth/login", payload: { username: "viewer", password: viewer.password } }); diff --git a/apps/gateway/src/app.ts b/apps/gateway/src/app.ts index 2de1d8e..46da275 100644 --- a/apps/gateway/src/app.ts +++ b/apps/gateway/src/app.ts @@ -105,45 +105,16 @@ export async function buildGateway(config: GatewayConfig): Promise { - const parsed = ProfileSchema.safeParse(profile); - return parsed.success ? [{ ...profile, ...parsed.data }] : []; + await appsStore.update((stored) => { + const valid = stored.flatMap((record) => { const parsed = AppSchema.safeParse(record); return parsed.success ? [parsed.data] : []; }); + return valid.length ? valid : [createDefaultApp()]; }); - if (normalizedProfiles.length !== storedProfiles.length || JSON.stringify(normalizedProfiles) !== JSON.stringify(storedProfiles)) { - await profilesStore.write(normalizedProfiles.length ? normalizedProfiles : [createDefaultProfile()]); - } - const storedApps = await appsStore.read(); - const normalizedApps = storedApps.flatMap((record) => { - const parsed = AppSchema.safeParse(record); - return parsed.success ? [parsed.data] : []; + await settingsStore.update((stored) => { const parsed = ProviderSettingsSchema.safeParse(stored); return parsed.success ? parsed.data : createDefaultProviderSettings(); }); + await keysStore.update((stored) => { + const normalized = stored.map((key) => ({ ...key, appId: key.appId ?? "default" })); + if (normalized.length || !config.bootstrapApiKey) return normalized; + return [{ id: randomUUID(), name: "Local development", prefix: config.bootstrapApiKey.slice(0, 8), hash: hash(config.bootstrapApiKey), appId: "default", createdAt: new Date().toISOString() }]; }); - const validApps = normalizedApps.length ? normalizedApps : [createDefaultApp()]; - if (JSON.stringify(validApps) !== JSON.stringify(storedApps)) await appsStore.write(validApps); - const storedSettings = await settingsStore.read(); - const normalizedSettings = ProviderSettingsSchema.safeParse(storedSettings); - if (normalizedSettings.success && JSON.stringify(normalizedSettings.data) !== JSON.stringify(storedSettings)) { - await settingsStore.write(normalizedSettings.data); - } else if (!normalizedSettings.success) { - await settingsStore.write(createDefaultProviderSettings()); - } - const storedKeys = await keysStore.read(); - if (storedKeys.some((key) => !key.appId)) { - await keysStore.write(storedKeys.map((key) => ({ ...key, appId: key.appId ?? "default" }))); - } - if ((await keysStore.read()).length === 0 && config.bootstrapApiKey) { - await keysStore.write([ - { - id: randomUUID(), - name: "Local development", - prefix: config.bootstrapApiKey.slice(0, 8), - hash: hash(config.bootstrapApiKey), - appId: "default", - createdAt: new Date().toISOString(), - }, - ]); - keys.invalidate(); - } const queue = new ConcurrentQueue(config.queueConcurrency, config.queueMaxDepth); const metrics = new GatewayMetrics(); @@ -228,13 +199,15 @@ export async function buildGateway(config: GatewayConfig): Promise profileId !== profile.id) - .map((profileId) => allProfiles.find((candidate) => candidate.id === profileId)) + .map((profileId) => { const record = allProfiles.find((candidate) => candidate.id === profileId); return record ? revisionOf(record, firewallApp.profileRevisions?.[profileId]) : undefined; }) .filter((candidate): candidate is Profile => Boolean(candidate)) .slice(0, 3); const shadows = await Promise.all(shadowProfiles.map(async (shadowProfile) => { const result = await evaluate(`${id}:shadow:${shadowProfile.id}`, envelope, shadowProfile, 0, traceId, firewallApp); return { profileId: shadowProfile.id, + policyRevision: shadowProfile.revision, + policyHash: shadowProfile.contentHash ?? policyHash(shadowProfile), verdict: result.decision.verdict, action: result.decision.action, risk: result.decision.risk, @@ -263,8 +236,12 @@ export async function buildGateway(config: GatewayConfig): Promise ({ name: "Pyro gateway", - version: "0.2.0", + version: "0.3.0-beta.1", endpoints: ["POST /v1/classify", "POST /v1/jobs", "GET /v1/jobs/:id", "WS /v1/events"], })); @@ -312,7 +289,7 @@ export async function buildGateway(config: GatewayConfig): Promise !profile.archived) - .filter((profile) => firewallApp.allowedProfileIds.length === 0 || firewallApp.allowedProfileIds.includes(profile.id)) + .filter((profile) => profileAllowed(profile.id, request.apiKey!, firewallApp)) .map(({ id, name, description }) => ({ id, name, description })); }); diff --git a/packages/storage/src/event-replay.test.ts b/packages/storage/src/event-replay.test.ts new file mode 100644 index 0000000..dd1f54f --- /dev/null +++ b/packages/storage/src/event-replay.test.ts @@ -0,0 +1,17 @@ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import test from "node:test"; +import type { ClassificationEvent, Delivery } from "@pyro/contracts"; +import { openDatabase } from "./index.js"; +for (const url of [`memory://replay-${randomUUID()}`, ...(process.env.TEST_DATABASE_URL ? [process.env.TEST_DATABASE_URL] : [])]) { + test(`first event and its webhook outbox are authoritative: ${url.startsWith("memory") ? "memory" : "PostgreSQL"}`, async (t) => { + const db = await openDatabase(url); t.after(() => db.close()); + const id = randomUUID(), integrationId = randomUUID(), at = new Date().toISOString(); + const event: ClassificationEvent = { id, appId: id, profileId: "default", createdAt: at, action: "allow", verdict: "safe", risk: 0, confidence: 1, reason: "first", detectors: [], model: "local", provider: "local-rules", latencyMs: 0, queueMs: 0, inputHash: "hash" }; + const delivery: Delivery = { id: randomUUID(), integrationId, eventId: id, createdAt: at, status: "pending", attempts: 0, nextAttemptAt: at, payload: { id, type: "decision.created", createdAt: at, data: { id, action: "allow" } } }; + await db.events.append(event, [delivery]); + await db.events.append({ ...event, action: "block", reason: "late worker" }, [{ ...delivery, id: randomUUID() }]); + assert.deepEqual(await db.events.findById(id), event); + assert.equal((await db.deliveries.list(integrationId)).length, 1); + }); +} diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index 97c682d..7d49d8c 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -311,7 +311,7 @@ class PostgresEvents implements EventStore { const client = await this.pool.connect(); try { await client.query("BEGIN"); - await client.query( + const inserted = await client.query( `INSERT INTO pyro_events (id, created_at, app_id, profile_id, verdict, action, risk, provider, api_key_id, labels, payload) VALUES ($1, $2::timestamptz, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11::jsonb) @@ -319,7 +319,7 @@ class PostgresEvents implements EventStore { [event.id, event.createdAt, event.appId ?? null, event.profileId, event.verdict, event.action, event.risk, event.provider, event.apiKeyId ?? null, JSON.stringify(event.labels ?? {}), JSON.stringify(event)], ); - for (const delivery of deliveries) await insertDelivery(client, delivery); + if (inserted.rowCount) for (const delivery of deliveries) await insertDelivery(client, delivery); await client.query("COMMIT"); } catch (error) { await client.query("ROLLBACK"); throw error; } finally { client.release(); } @@ -689,7 +689,8 @@ class MemoryDatabase implements Database { }; readonly events: EventStore = { append: async (event, deliveries = []) => { - if (!this.eventRows.some((item) => item.id === event.id)) this.eventRows.push(structuredClone(event)); + if (this.eventRows.some((item) => item.id === event.id)) return; + this.eventRows.push(structuredClone(event)); for (const delivery of deliveries) await this.deliveries.enqueue(delivery); }, readRecent: async (limit = 100) => this.eventRows.slice(-Math.max(1, limit)).reverse().map((event) => structuredClone(event)),