From a06190eafe4ae1ef7c649d1a39d483aa975417ca Mon Sep 17 00:00:00 2001 From: Omar Ramadan Date: Wed, 12 Aug 2026 01:25:45 +0000 Subject: [PATCH] fix(outbox): preserve local FIFO event order Co-Authored-By: Paperclip --- .../src/__tests__/plugin-event-outbox.test.ts | 61 ++++++++++++++++++- server/src/services/activity-log.ts | 42 +++++++++---- 2 files changed, 88 insertions(+), 15 deletions(-) diff --git a/server/src/__tests__/plugin-event-outbox.test.ts b/server/src/__tests__/plugin-event-outbox.test.ts index 1aaff0b5615b..154343d0e8df 100644 --- a/server/src/__tests__/plugin-event-outbox.test.ts +++ b/server/src/__tests__/plugin-event-outbox.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; -import { companies, createDb, pluginEventOutbox } from "@paperclipai/db"; -import { eq } from "drizzle-orm"; +import { activityLog, companies, createDb, pluginEventOutbox, type Db } from "@paperclipai/db"; +import { asc, eq } from "drizzle-orm"; import type { PluginEvent } from "@paperclipai/plugin-sdk"; import { getEmbeddedPostgresTestSupport, @@ -9,6 +9,7 @@ import { } from "./helpers/embedded-postgres.js"; import { createPluginEventBus } from "../services/plugin-event-bus.js"; import { pollOnce, resetStaleProcessing } from "../services/plugin-event-outbox.js"; +import { logActivity, setPluginEventOutboxDb } from "../services/activity-log.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -40,6 +41,8 @@ describeEmbeddedPostgres("plugin event outbox", () => { }); beforeEach(async () => { + setPluginEventOutboxDb(db); + await db.delete(activityLog); await db.delete(pluginEventOutbox); }); @@ -103,6 +106,60 @@ describeEmbeddedPostgres("plugin event outbox", () => { expect(order).toEqual(["approval.created", "approval.decided"]); }); + it("preserves sequential logActivity order when an outbox insert is delayed", async () => { + const pendingInserts: Promise[] = []; + let resolveSecondInsertScheduled!: () => void; + const secondInsertScheduled = new Promise((resolve) => { + resolveSecondInsertScheduled = resolve; + }); + let insertCount = 0; + let delayFirstInsert = true; + const delayedOutboxDb = { + insert: (table: typeof pluginEventOutbox) => { + const builder = db.insert(table); + return { + values: (values: typeof pluginEventOutbox.$inferInsert) => { + const query = builder.values(values); + const completion = delayFirstInsert + ? new Promise((resolve) => setImmediate(resolve)).then(() => query) + : Promise.resolve(query); + insertCount += 1; + if (insertCount === 2) resolveSecondInsertScheduled(); + delayFirstInsert = false; + pendingInserts.push(completion); + return completion; + }, + }; + }, + } as unknown as Db; + setPluginEventOutboxDb(delayedOutboxDb); + + await logActivity(db, { + companyId, + actorType: "system", + actorId: "test", + action: "issue.created", + entityType: "issue", + entityId: randomUUID(), + }); + await logActivity(db, { + companyId, + actorType: "system", + actorId: "test", + action: "issue.updated", + entityType: "issue", + entityId: randomUUID(), + }); + await secondInsertScheduled; + await Promise.all(pendingInserts); + + const rows = await db + .select({ eventType: pluginEventOutbox.eventType }) + .from(pluginEventOutbox) + .orderBy(asc(pluginEventOutbox.seq)); + expect(rows.map((row) => row.eventType)).toEqual(["issue.created", "issue.updated"]); + }); + it("marks processed even when a handler throws (no poison loop)", async () => { const bus = createPluginEventBus(); bus.forPlugin("test").subscribe("approval.created", async () => { diff --git a/server/src/services/activity-log.ts b/server/src/services/activity-log.ts index 49833061a157..c461a24f35f3 100644 --- a/server/src/services/activity-log.ts +++ b/server/src/services/activity-log.ts @@ -36,6 +36,7 @@ const ACTIVITY_ACTION_TO_PLUGIN_EVENT: Readonly> let _pluginEventBus: PluginEventBus | null = null; let _outboxDb: Db | null = null; +let _outboxEnqueueTail: Promise | null = null; /** Wire the plugin event bus so domain events are forwarded to plugins. */ export function setPluginEventBus(bus: PluginEventBus): void { @@ -68,27 +69,42 @@ function eventTypeForActivityAction(action: string): PluginEventType | null { * in-process: the worker-tier poller (plugin-event-outbox.ts) is the sole * emitter, so events raised on any tier (notably the API tier, where plugins * are not loaded) reliably reach subscribed plugins. One writer + one emitter - * ⇒ no double-delivery. Fire-and-forget to keep the signature synchronous. + * ⇒ no double-delivery. Writes are serialized within this process so the + * database `seq` reflects call order even though this remains fire-and-forget. + * Each insert is still best-effort: a failure is logged and does not block the + * next event or surface to the activity caller. */ export function publishPluginDomainEvent(event: PluginEvent): void { - if (!_outboxDb) { + const outboxDb = _outboxDb; + if (!outboxDb) { logger.warn( { eventType: event.eventType, eventId: event.eventId }, "plugin event outbox db not set; dropping event", ); return; } - void _outboxDb - .insert(pluginEventOutbox) - .values({ - eventId: event.eventId, - companyId: event.companyId, - eventType: event.eventType, - payload: event as unknown as Record, - }) - .catch((err) => - logger.warn({ err, eventType: event.eventType }, "failed to enqueue plugin event to outbox"), - ); + const values = { + eventId: event.eventId, + companyId: event.companyId, + eventType: event.eventType, + payload: event as unknown as Record, + }; + + const enqueue = async () => { + try { + await outboxDb.insert(pluginEventOutbox).values(values); + } catch (err) { + logger.warn({ err, eventType: event.eventType }, "failed to enqueue plugin event to outbox"); + } + }; + // Start the first writer immediately, preserving the original detached timing. + // Later calls chain behind it, so they cannot claim an earlier `seq`. + const pending = _outboxEnqueueTail + ? _outboxEnqueueTail.then(enqueue, enqueue) + : enqueue(); + // Keep an unexpected failure from poisoning the chain or becoming unhandled; + // normal insert failures are already logged and swallowed above. + _outboxEnqueueTail = pending.catch(() => {}); } /**