fix(connections): snapshot relay triggers and lease abandoned handoffs

Delivery retries reloaded the connection's currently-enabled triggers, so a
trigger added or re-enabled between attempts could receive an envelope for an
event that predated it. Capture the enabled trigger set atomically with the
delivery claim (written in the INSERT branch, never rebuilt on reclaim) and
route every attempt to that stored snapshot, so recovery can never reconstruct
it from current configuration.

Dispatch also moved a delivery to `forwarded` before invoking any destination,
but the reclaim path only matched `failed` rows. A crash while `forwarded` left
the row stuck and every later connector attempt returned `duplicate`, silently
dropping the remaining trigger work. Add a lease (`lease_expires_at`) set on
claim and renewed on forward; a later attempt may reclaim an abandoned
in-progress delivery only once its lease has expired, so it never races a live
worker. Per-trigger idempotency keys remain the correctness backstop.

Covered with embedded-postgres tests: retries honor the original snapshot and
skip newly enabled triggers; an abandoned forwarded delivery is dropped as a
duplicate while its lease is live, reclaimed once it expires, and a trigger
enabled after the crash never receives the recovered envelope.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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,

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,

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<string, unknown>;
+};
+
 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<Record<string, unknown>>().notNull(),
+    triggerSnapshot: jsonb("trigger_snapshot").$type<ConnectionTriggerSnapshot[] | null>(),
     completedTriggerIds: jsonb("completed_trigger_ids").$type<string[]>().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(),

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<RelayEnvelope> = {}) {
   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<string>; statuses: string[]; completedTriggerIds: Set<string> } {
   const deliveries = new Set<string>();
+  const snapshots = new Map<string, RelayTrigger[]>();
   const statuses: string[] = [];
   const completedTriggerIds = new Set<string>();
   return {
@@ -32,10 +35,12 @@ function store(): ConnectionRelayStore & { deliveries: Set<string>; 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<string>();
     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]);
+  });
 });

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<string, unknown>;
 };
 
+// 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<RelayTrigger[]>;
   updateDelivery?(input: { connectionId: string; deliveryId: string; status: "forwarded" | "delivered" | "failed" | "dead_letter"; error?: string | null; now?: Date }): Promise<void>;
   markTriggerCompleted?(input: { connectionId: string; deliveryId: string; triggerId: string; now?: Date }): Promise<void>;
@@ -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<typeof processConnectionRelay>[1]) {