diff --git a/packages/db/src/migrations/0184_connection_trigger_relay.sql b/packages/db/src/migrations/0184_connection_trigger_relay.sql index ebfc35cfe05..9af406ec380 100644 --- a/packages/db/src/migrations/0184_connection_trigger_relay.sql +++ b/packages/db/src/migrations/0184_connection_trigger_relay.sql @@ -20,10 +20,12 @@ CREATE TABLE "connection_trigger_deliveries" ( "status" text DEFAULT 'received' NOT NULL, "attempt" integer DEFAULT 1 NOT NULL, "envelope" jsonb NOT NULL, + "trigger_snapshot" jsonb, "completed_trigger_ids" jsonb DEFAULT '[]'::jsonb NOT NULL, "received_at" timestamp with time zone NOT NULL, "forwarded_at" timestamp with time zone, "delivered_at" timestamp with time zone, + "lease_expires_at" timestamp with time zone, "last_error" text, "created_at" timestamp with time zone DEFAULT now() NOT NULL, "updated_at" timestamp with time zone DEFAULT now() NOT NULL, diff --git a/packages/db/src/schema/index.ts b/packages/db/src/schema/index.ts index c7eba8c495d..5faeb0ea620 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -133,6 +133,7 @@ export { toolGatewayRateLimitCounters, toolAccessAuditEvents, } from "./tool_access.js"; +export type { ConnectionTriggerSnapshot } from "./tool_access.js"; export { companySkills, companySkillVersions, diff --git a/packages/db/src/schema/tool_access.ts b/packages/db/src/schema/tool_access.ts index 23d75e5b76c..e68cbf7d293 100644 --- a/packages/db/src/schema/tool_access.ts +++ b/packages/db/src/schema/tool_access.ts @@ -169,6 +169,14 @@ export const connectionTriggers = pgTable( ], ); +export type ConnectionTriggerSnapshot = { + id: string; + companyId: string; + destinationType: "routine" | "issue_wake" | "plugin_worker"; + destinationId: string; + config?: Record; +}; + export const connectionTriggerDeliveries = pgTable( "connection_trigger_deliveries", { @@ -180,10 +188,12 @@ export const connectionTriggerDeliveries = pgTable( status: text("status").$type<"received" | "forwarded" | "delivered" | "failed" | "dead_letter">().notNull().default("received"), attempt: integer("attempt").notNull().default(1), envelope: jsonb("envelope").$type>().notNull(), + triggerSnapshot: jsonb("trigger_snapshot").$type(), completedTriggerIds: jsonb("completed_trigger_ids").$type().notNull().default([]), receivedAt: timestamp("received_at", { withTimezone: true }).notNull(), forwardedAt: timestamp("forwarded_at", { withTimezone: true }), deliveredAt: timestamp("delivered_at", { withTimezone: true }), + leaseExpiresAt: timestamp("lease_expires_at", { withTimezone: true }), lastError: text("last_error"), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), diff --git a/server/src/services/connection-relay.test.ts b/server/src/services/connection-relay.test.ts index 2e95efa5766..fbcfbdd3789 100644 --- a/server/src/services/connection-relay.test.ts +++ b/server/src/services/connection-relay.test.ts @@ -1,11 +1,12 @@ import { randomBytes } from "node:crypto"; +import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { createRelaySignature, type RelayEnvelope } from "@paperclip/connect-protocol"; import { companies, connectionTriggerDeliveries, connectionTriggers, createDb, toolApplications, toolConnections } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; -import { connectionRelayStore, pollRelayChannel, processAndDispatchConnectionRelay, processConnectionRelay, type ConnectionRelayStore } from "./connection-relay.js"; +import { connectionRelayStore, pollRelayChannel, processAndDispatchConnectionRelay, processConnectionRelay, RELAY_DELIVERY_LEASE_MS, type ConnectionRelayStore, type RelayTrigger } from "./connection-relay.js"; -function fixture() { +function fixture(overrides: Partial = {}) { const envelope: RelayEnvelope = { v: 1, deliveryId: "dl_01K0EXAMPLE", @@ -15,6 +16,7 @@ function fixture() { attempt: 1, provider: { headers: { "x-vercel-signature": "provider-signature" }, bodyB64: "eyJvayI6dHJ1ZX0=" }, verification: { profile: "vercel@1", result: "verified", keyId: "primary" }, + ...overrides, }; const rawBody = Buffer.from(JSON.stringify(envelope)); const relaySecret = randomBytes(32); @@ -23,6 +25,7 @@ function fixture() { function store(): ConnectionRelayStore & { deliveries: Set; statuses: string[]; completedTriggerIds: Set } { const deliveries = new Set(); + const snapshots = new Map(); const statuses: string[] = []; const completedTriggerIds = new Set(); return { @@ -32,10 +35,12 @@ function store(): ConnectionRelayStore & { deliveries: Set; statuses: st async findConnectionByPublicRef(publicRef) { return publicRef === "cn_01K0EXAMPLE" ? { id: "connection-1", companyId: "company-1", enabled: true } : null; }, - async claimDelivery({ envelope }) { - if (deliveries.has(envelope.deliveryId)) return { claimed: false, completedTriggerIds: [] }; + async claimDelivery({ envelope, triggerSnapshot }) { + // Mirror the store: the snapshot is captured once, on the first claim, and reused thereafter. + if (deliveries.has(envelope.deliveryId)) return { claimed: false, completedTriggerIds: [], triggerSnapshot: snapshots.get(envelope.deliveryId) ?? null }; deliveries.add(envelope.deliveryId); - return { claimed: true, completedTriggerIds: [...completedTriggerIds] }; + snapshots.set(envelope.deliveryId, triggerSnapshot); + return { claimed: true, completedTriggerIds: [...completedTriggerIds], triggerSnapshot }; }, async listEnabledTriggers() { return [{ id: "trigger-1", companyId: "company-1", destinationType: "routine", destinationId: "routine-1" }]; @@ -131,16 +136,18 @@ describe("processConnectionRelay", () => { const { rawBody: firstBody, relaySecret } = fixture(); let attempt = 0; let status = "received"; + let snapshot: RelayTrigger[] | null = null; const completed = new Set(); const retryStore: ConnectionRelayStore = { async findConnectionByPublicRef() { return { id: "connection-1", companyId: "company-1", enabled: true }; }, - async claimDelivery({ envelope }) { + async claimDelivery({ envelope, triggerSnapshot }) { if (attempt === 0 || (status === "failed" && envelope.attempt > attempt)) { + if (snapshot === null) snapshot = triggerSnapshot; attempt = envelope.attempt; status = "received"; - return { claimed: true, completedTriggerIds: [...completed] }; + return { claimed: true, completedTriggerIds: [...completed], triggerSnapshot: snapshot }; } - return { claimed: false, completedTriggerIds: [] }; + return { claimed: false, completedTriggerIds: [], triggerSnapshot: snapshot }; }, async listEnabledTriggers() { return [ @@ -264,4 +271,122 @@ describeEmbeddedPostgres("connection relay persistence", () => { expect(rows).toHaveLength(1); expect(rows[0]).toMatchObject({ deliveryId: "dl_01K0EXAMPLE", status: "delivered", connectionId: connection.id }); }); + + it("routes retries to the trigger set captured on the first attempt, not newly enabled triggers", async () => { + const company = await db.insert(companies).values({ name: "Relay Snapshot", issuePrefix: "RLS" }).returning().then((rows) => rows[0]!); + const application = await db.insert(toolApplications).values({ companyId: company.id, name: "Vercel", type: "mcp_http" }).returning().then((rows) => rows[0]!); + const connection = await db.insert(toolConnections).values({ + companyId: company.id, + applicationId: application.id, + name: "Vercel Relay", + uid: "vercel-relay-snapshot", + transport: "rest_api", + enabled: true, + config: { relay: { publicRef: "cn_01K0SNAPSHOT" } }, + }).returning().then((rows) => rows[0]!); + const originalDestination = "11111111-1111-4111-8111-111111111111"; + const lateDestination = "22222222-2222-4222-8222-222222222222"; + await db.insert(connectionTriggers).values({ companyId: company.id, connectionId: connection.id, destinationType: "routine", destinationId: originalDestination }); + + const { rawBody: firstBody, relaySecret } = fixture({ connectionPublicRef: "cn_01K0SNAPSHOT", deliveryId: "dl_01K0SNAPSHOT" }); + const now = new Date("2026-07-21T18:04:05.000Z"); + const timestamp = String(Math.floor(now.getTime() / 1000)); + const store = connectionRelayStore(db); + const fired: string[] = []; + const dispatcher = { + routine: async (trigger: { destinationId: string }) => { + fired.push(trigger.destinationId); + // Fail the very first dispatch so the delivery is retried; succeed on the retry. + if (trigger.destinationId === originalDestination && fired.filter((id) => id === originalDestination).length === 1) throw new Error("temporary failure"); + }, + issue_wake: async () => {}, + plugin_worker: async () => {}, + }; + + const first = await processAndDispatchConnectionRelay(store, dispatcher, { + rawBody: firstBody, + signature: createRelaySignature({ body: firstBody, relaySecret, timestamp }), + timestamp, + relaySecret, + now, + }); + expect(first.status).toBe("failed"); + + // Operator enables a brand-new trigger after the envelope was already captured. + await db.insert(connectionTriggers).values({ companyId: company.id, connectionId: connection.id, destinationType: "routine", destinationId: lateDestination }); + + const secondEnvelope = { ...JSON.parse(firstBody.toString("utf8")), attempt: 2 }; + const secondBody = Buffer.from(JSON.stringify(secondEnvelope)); + const second = await processAndDispatchConnectionRelay(store, dispatcher, { + rawBody: secondBody, + signature: createRelaySignature({ body: secondBody, relaySecret, timestamp }), + timestamp, + relaySecret, + now, + }); + + expect(second.status).toBe("delivered"); + // The late trigger never receives the pre-existing envelope; only the snapshotted destination fires. + expect(fired).toEqual([originalDestination, originalDestination]); + const rows = await db.select().from(connectionTriggerDeliveries).where(eq(connectionTriggerDeliveries.connectionId, connection.id)); + expect(rows).toHaveLength(1); + expect((rows[0]!.triggerSnapshot ?? []).map((trigger) => trigger.destinationId)).toEqual([originalDestination]); + }); + + it("reclaims an abandoned forwarded delivery only after its lease expires", async () => { + const company = await db.insert(companies).values({ name: "Relay Lease", issuePrefix: "RLL" }).returning().then((rows) => rows[0]!); + const application = await db.insert(toolApplications).values({ companyId: company.id, name: "Vercel", type: "mcp_http" }).returning().then((rows) => rows[0]!); + const connection = await db.insert(toolConnections).values({ + companyId: company.id, + applicationId: application.id, + name: "Vercel Relay", + uid: "vercel-relay-lease", + transport: "rest_api", + enabled: true, + config: { relay: { publicRef: "cn_01K0LEASE" } }, + }).returning().then((rows) => rows[0]!); + const destination = "33333333-3333-4333-8333-333333333333"; + await db.insert(connectionTriggers).values({ companyId: company.id, connectionId: connection.id, destinationType: "routine", destinationId: destination }); + + const { rawBody, relaySecret } = fixture({ connectionPublicRef: "cn_01K0LEASE", deliveryId: "dl_01K0LEASE" }); + const envelope = JSON.parse(rawBody.toString("utf8")) as RelayEnvelope; + const store = connectionRelayStore(db); + + // A worker claims the delivery and marks it `forwarded`, then crashes before dispatching. + const claimedAt = new Date("2026-07-21T18:04:05.000Z"); + const capturedTriggers = await store.listEnabledTriggers(connection.id); + await store.claimDelivery({ companyId: company.id, connectionId: connection.id, envelope, triggerSnapshot: capturedTriggers, now: claimedAt }); + await store.updateDelivery!({ connectionId: connection.id, deliveryId: envelope.deliveryId, status: "forwarded", now: claimedAt }); + + const dispatchAt = async (now: Date) => { + const timestamp = String(Math.floor(now.getTime() / 1000)); + const fired: string[] = []; + const result = await processAndDispatchConnectionRelay(store, { + routine: async (trigger) => { fired.push(trigger.destinationId); }, + issue_wake: async () => {}, + plugin_worker: async () => {}, + }, { rawBody, signature: createRelaySignature({ body: rawBody, relaySecret, timestamp }), timestamp, relaySecret, now }); + return { result, fired }; + }; + + // While the lease is still live the row must not be reclaimed — an active worker could still + // be running, so a competing attempt is dropped as a duplicate and nothing is dispatched. + const duringLease = await dispatchAt(new Date(claimedAt.getTime() + RELAY_DELIVERY_LEASE_MS / 2)); + expect(duringLease.result.status).toBe("duplicate"); + expect(duringLease.fired).toEqual([]); + + // A trigger enabled after the crash must not receive this in-flight envelope: recovery reuses + // the snapshot captured with the claim, never the connection's current trigger configuration. + await db.insert(connectionTriggers).values({ companyId: company.id, connectionId: connection.id, destinationType: "routine", destinationId: "44444444-4444-4444-8444-444444444444" }); + + // Once the lease has expired the abandoned delivery is safely reclaimed and the work runs. + const afterLease = await dispatchAt(new Date(claimedAt.getTime() + RELAY_DELIVERY_LEASE_MS + 60_000)); + expect(afterLease.result.status).toBe("delivered"); + expect(afterLease.fired).toEqual([destination]); + + const rows = await db.select().from(connectionTriggerDeliveries).where(eq(connectionTriggerDeliveries.connectionId, connection.id)); + expect(rows).toHaveLength(1); + expect(rows[0]!.status).toBe("delivered"); + expect((rows[0]!.triggerSnapshot ?? []).map((trigger) => trigger.destinationId)).toEqual([destination]); + }); }); diff --git a/server/src/services/connection-relay.ts b/server/src/services/connection-relay.ts index 7d2ad1fa8d5..0fe817ee8f6 100644 --- a/server/src/services/connection-relay.ts +++ b/server/src/services/connection-relay.ts @@ -1,4 +1,4 @@ -import { and, eq, sql } from "drizzle-orm"; +import { and, eq, inArray, isNotNull, lt, lte, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { connectionTriggerDeliveries, connectionTriggers, issues, pluginCompanySettings, plugins, routines as routinesTable, toolConnections } from "@paperclipai/db"; import { @@ -20,9 +20,15 @@ export type RelayTrigger = { config?: Record; }; +// How long a claimed delivery may stay in an in-progress state (`received`/`forwarded`) +// before a later attempt is allowed to reclaim it. This must comfortably exceed the time +// a healthy worker needs to dispatch every trigger, so an active worker is never raced; +// per-trigger idempotency keys are the correctness backstop if a slow worker overruns it. +export const RELAY_DELIVERY_LEASE_MS = 5 * 60 * 1000; + export type ConnectionRelayStore = { findConnectionByPublicRef(publicRef: string): Promise<{ id: string; companyId: string; enabled: boolean } | null>; - claimDelivery(input: { companyId: string; connectionId: string; envelope: RelayEnvelope }): Promise<{ claimed: boolean; completedTriggerIds: string[] }>; + claimDelivery(input: { companyId: string; connectionId: string; envelope: RelayEnvelope; triggerSnapshot: RelayTrigger[]; now?: Date }): Promise<{ claimed: boolean; completedTriggerIds: string[]; triggerSnapshot: RelayTrigger[] | null }>; listEnabledTriggers(connectionId: string): Promise; updateDelivery?(input: { connectionId: string; deliveryId: string; status: "forwarded" | "delivered" | "failed" | "dead_letter"; error?: string | null; now?: Date }): Promise; markTriggerCompleted?(input: { connectionId: string; deliveryId: string; triggerId: string; now?: Date }): Promise; @@ -36,19 +42,45 @@ export function connectionRelayStore(db: Db): ConnectionRelayStore { const rows = await db.select({ id: toolConnections.id, companyId: toolConnections.companyId, enabled: toolConnections.enabled, config: toolConnections.config }).from(toolConnections); return rows.find((row) => (row.config as RelayConnectionConfig).relay?.publicRef === publicRef) ?? null; }, - async claimDelivery({ companyId, connectionId, envelope }) { - const rows = await db.insert(connectionTriggerDeliveries).values({ companyId, connectionId, deliveryId: envelope.deliveryId, providerSlug: envelope.providerSlug, attempt: envelope.attempt, envelope, receivedAt: new Date(envelope.receivedAt) }).onConflictDoUpdate({ + async claimDelivery({ companyId, connectionId, envelope, triggerSnapshot, now = new Date() }) { + const leaseExpiresAt = new Date(now.getTime() + RELAY_DELIVERY_LEASE_MS); + // The trigger snapshot is written in the INSERT branch only, so it is persisted atomically + // with the row's creation — there is no window where a delivery exists without its snapshot. + // The reclaim `set` deliberately omits `triggerSnapshot`, so retries and crash recovery reuse + // the set captured at first claim and never rebuild it from current configuration. + const rows = await db.insert(connectionTriggerDeliveries).values({ companyId, connectionId, deliveryId: envelope.deliveryId, providerSlug: envelope.providerSlug, attempt: envelope.attempt, envelope, triggerSnapshot, receivedAt: new Date(envelope.receivedAt), leaseExpiresAt }).onConflictDoUpdate({ target: [connectionTriggerDeliveries.connectionId, connectionTriggerDeliveries.deliveryId], - set: { attempt: envelope.attempt, envelope, status: "received", lastError: null, updatedAt: new Date() }, - setWhere: and(eq(connectionTriggerDeliveries.status, "failed"), sql`${connectionTriggerDeliveries.attempt} < ${envelope.attempt}`), - }).returning({ completedTriggerIds: connectionTriggerDeliveries.completedTriggerIds }); - return { claimed: rows.length === 1, completedTriggerIds: rows[0]?.completedTriggerIds ?? [] }; + set: { attempt: envelope.attempt, envelope, status: "received", lastError: null, leaseExpiresAt, updatedAt: now }, + // Reclaim either an explicitly failed delivery on a strictly newer attempt, or an + // in-progress (`received`/`forwarded`) delivery whose lease has expired — the latter + // is how an abandoned handoff from a crashed worker is safely recovered without + // racing a worker that still holds a live lease. + setWhere: or( + and(eq(connectionTriggerDeliveries.status, "failed"), lt(connectionTriggerDeliveries.attempt, envelope.attempt)), + and( + inArray(connectionTriggerDeliveries.status, ["received", "forwarded"]), + isNotNull(connectionTriggerDeliveries.leaseExpiresAt), + lt(connectionTriggerDeliveries.leaseExpiresAt, now), + lte(connectionTriggerDeliveries.attempt, envelope.attempt), + ), + ), + }).returning({ completedTriggerIds: connectionTriggerDeliveries.completedTriggerIds, triggerSnapshot: connectionTriggerDeliveries.triggerSnapshot }); + return { claimed: rows.length === 1, completedTriggerIds: rows[0]?.completedTriggerIds ?? [], triggerSnapshot: rows[0]?.triggerSnapshot ?? null }; }, async listEnabledTriggers(connectionId) { return db.select({ id: connectionTriggers.id, companyId: connectionTriggers.companyId, destinationType: connectionTriggers.destinationType, destinationId: connectionTriggers.destinationId, config: connectionTriggers.config }).from(connectionTriggers).where(and(eq(connectionTriggers.connectionId, connectionId), eq(connectionTriggers.enabled, true))); }, async updateDelivery({ connectionId, deliveryId, status, error, now = new Date() }) { - await db.update(connectionTriggerDeliveries).set({ status, forwardedAt: status === "forwarded" ? now : undefined, deliveredAt: status === "delivered" ? now : undefined, lastError: error ?? null, updatedAt: now }).where(and(eq(connectionTriggerDeliveries.connectionId, connectionId), eq(connectionTriggerDeliveries.deliveryId, deliveryId))); + await db.update(connectionTriggerDeliveries).set({ + status, + forwardedAt: status === "forwarded" ? now : undefined, + deliveredAt: status === "delivered" ? now : undefined, + // Renew the lease while the handoff is actively in flight; clear it on terminal + // states so a delivered/dead-letter row can never be mistaken for an abandoned one. + leaseExpiresAt: status === "forwarded" ? new Date(now.getTime() + RELAY_DELIVERY_LEASE_MS) : status === "delivered" || status === "dead_letter" ? null : undefined, + lastError: error ?? null, + updatedAt: now, + }).where(and(eq(connectionTriggerDeliveries.connectionId, connectionId), eq(connectionTriggerDeliveries.deliveryId, deliveryId))); }, async markTriggerCompleted({ connectionId, deliveryId, triggerId, now = new Date() }) { await db.update(connectionTriggerDeliveries).set({ @@ -172,10 +204,15 @@ export async function processConnectionRelay( throw new Error("Relay connection not found"); } - const claim = await store.claimDelivery({ companyId: connection.companyId, connectionId: connection.id, envelope }); + // Resolve the currently-enabled triggers up front so they can be captured atomically with the + // claim. The claim only persists this set on the very first insert; every later attempt reuses + // the stored snapshot. This guarantees a trigger added or re-enabled after the first claim never + // receives an envelope for an event that predates it — even across a crash-and-recover cycle. + const currentTriggers = await store.listEnabledTriggers(connection.id); + const claim = await store.claimDelivery({ companyId: connection.companyId, connectionId: connection.id, envelope, triggerSnapshot: currentTriggers, now: input.now }); if (!claim.claimed) return { status: "duplicate" as const, envelope, connection, triggers: [] as RelayTrigger[], completedTriggerIds: [] as string[] }; - return { status: "accepted" as const, envelope, connection, triggers: await store.listEnabledTriggers(connection.id), completedTriggerIds: claim.completedTriggerIds }; + return { status: "accepted" as const, envelope, connection, triggers: claim.triggerSnapshot ?? currentTriggers, completedTriggerIds: claim.completedTriggerIds }; } export async function processAndDispatchConnectionRelay(store: ConnectionRelayStore, dispatcher: ConnectionRelayDispatcher, input: Parameters[1]) {