Bound event verification before outbound requests

Reserve expiring subscription admission leases before remote verification and retain failed attempts for shared rate limits. Avoid holding database transactions during network calls and reject cancelled, expired, or revoked verification before activation. Cover quota rejection without callbacks, concurrent replica admission, free database locks, cancellation, expiry, and revocation with real PostgreSQL tests.

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

doc/public-mcp.md

@@ -198,6 +198,14 @@ Rescan the plugin's MCP server to discover the event catalog. Installing or
 connecting alone does not start monitoring. Claude and other clients without
 Events support can continue retrieving results through the read tools.
 
+Callback and hosted-authority verification reserve durable capacity before any
+remote request. Across replicas, at most 2 verifications per grant, 8 per company
+and 32 per instance can run concurrently. Attempts are bounded to 30 per grant,
+200 per company and 1,000 per instance per minute, including failed verification.
+New monitors reserve subscription quota before verification; leases expire after
+one minute. Network waits hold no database transaction, and expired, cancelled
+or revoked requests cannot finalize a subscription.
+
 Delivery uses a verified HTTPS callback, Standard Webhooks HMAC signatures,
 public-address DNS pinning on every connection, and no redirects. Callback URLs,
 current/previous signing keys and hosted authorization proofs are encrypted with

packages/db/src/migrations/0296_slippery_dracula.sql

@@ -0,0 +1,24 @@
+CREATE TABLE IF NOT EXISTS "mcp_event_admissions" (
+	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+	"subscription_id" text NOT NULL,
+	"company_id" uuid NOT NULL,
+	"grant_id" uuid NOT NULL,
+	"reserves_subscription" boolean NOT NULL,
+	"expires_at" timestamp with time zone NOT NULL,
+	"finished_at" timestamp with time zone,
+	"created_at" timestamp with time zone DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+DO $$ BEGIN
+ALTER TABLE "mcp_event_admissions" ADD CONSTRAINT "mcp_event_admissions_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;
+EXCEPTION WHEN duplicate_object THEN NULL;
+END $$;
+--> statement-breakpoint
+DO $$ BEGIN
+ALTER TABLE "mcp_event_admissions" ADD CONSTRAINT "mcp_event_admissions_grant_id_mcp_oauth_grants_id_fk" FOREIGN KEY ("grant_id") REFERENCES "public"."mcp_oauth_grants"("id") ON DELETE cascade ON UPDATE no action;
+EXCEPTION WHEN duplicate_object THEN NULL;
+END $$;
+--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "mcp_event_admissions_expiry_idx" ON "mcp_event_admissions" USING btree ("expires_at");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "mcp_event_admissions_grant_idx" ON "mcp_event_admissions" USING btree ("grant_id");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "mcp_event_admissions_company_idx" ON "mcp_event_admissions" USING btree ("company_id");

packages/db/src/migrations/meta/_journal.json

@@ -2059,6 +2059,13 @@
       "when": 1790951906069,
       "tag": "0295_numerous_slapstick",
       "breakpoints": true
+    },
+    {
+      "idx": 296,
+      "version": "7",
+      "when": 1791055277082,
+      "tag": "0296_slippery_dracula",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/public_mcp.ts

@@ -1,4 +1,4 @@
-import { index, integer, jsonb, pgTable, text, timestamp, uniqueIndex, uuid } from "drizzle-orm/pg-core";
+import { boolean, index, integer, jsonb, pgTable, text, timestamp, uniqueIndex, uuid } from "drizzle-orm/pg-core";
 import { authUsers } from "./auth.js";
 import { companies } from "./companies.js";
 
@@ -91,6 +91,23 @@ export const mcpEventSubscriptions = pgTable("mcp_event_subscriptions", {
   scannedAt: timestamp("scanned_at", { withTimezone: true }).notNull().defaultNow(),
 }, (t) => [index("mcp_event_subscriptions_expiry_idx").on(t.expiresAt), index("mcp_event_subscriptions_company_idx").on(t.companyId)]);
 
+// Short-lived admission leases bound remote verification across replicas. Finished
+// attempts remain until expiry so failed callbacks cannot bypass rate limits.
+export const mcpEventAdmissions = pgTable("mcp_event_admissions", {
+  id: uuid("id").primaryKey().defaultRandom(),
+  subscriptionId: text("subscription_id").notNull(),
+  companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }),
+  grantId: uuid("grant_id").notNull().references(() => mcpOauthGrants.id, { onDelete: "cascade" }),
+  reservesSubscription: boolean("reserves_subscription").notNull(),
+  expiresAt: timestamp("expires_at", { withTimezone: true }).notNull(),
+  finishedAt: timestamp("finished_at", { withTimezone: true }),
+  createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
+}, (t) => [
+  index("mcp_event_admissions_expiry_idx").on(t.expiresAt),
+  index("mcp_event_admissions_grant_idx").on(t.grantId),
+  index("mcp_event_admissions_company_idx").on(t.companyId),
+]);
+
 export const mcpEventDeliveries = pgTable("mcp_event_deliveries", {
   id: uuid("id").primaryKey().defaultRandom(),
   subscriptionId: text("subscription_id").notNull().references(() => mcpEventSubscriptions.id, { onDelete: "cascade" }),

server/src/__tests__/public-mcp.test.ts

@@ -1,9 +1,9 @@
 import { createHmac, createHash, randomBytes, randomUUID } from "node:crypto";
 import express, { type Request } from "express";
 import request from "supertest";
-import { eq } from "drizzle-orm";
+import { eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
-import { activityLog, mcpEventSubscriptions, createDb, authUsers, companies, companyMemberships, mcpOauthTokens, mcpOauthRequests, mcpOauthGrants, mcpOauthClients, mcpMutationReceipts, agents, issues, issueComments, instanceUserRoles } from "@paperclipai/db";
+import { activityLog, mcpEventAdmissions, mcpEventSubscriptions, createDb, authUsers, companies, companyMemberships, mcpOauthTokens, mcpOauthRequests, mcpOauthGrants, mcpOauthClients, mcpMutationReceipts, agents, issues, issueComments, instanceUserRoles } from "@paperclipai/db";
 import { createPublicMcpOAuth, publicMcpConfig, hashMcpSecret } from "../services/public-mcp/oauth.js";
 import { McpApiError, createMcpApiDispatch, createPublicMcpExecutor, publicMcpCapabilities } from "../services/public-mcp/capabilities.js";
 import { publicMcpIngressRoutes, publicMcpManagementRoutes } from "../routes/public-mcp.js";
@@ -460,7 +460,9 @@ describe.skipIf(!support.supported)("public MCP OAuth and tool boundary", () =>
     const f = await eventFixture();
     const makeInput = (index: number) => ({ ...f.input, delivery: { ...f.input.delivery, url: f.input.delivery.url + "/" + index } });
     for (let i = 0; i < 20; i++) await f.service.subscribe(f.principal, makeInput(i));
+    const verified = f.received.length;
     await expect(f.service.subscribe(f.principal, makeInput(20))).rejects.toMatchObject({ code: -32602 });
+    expect(f.received).toHaveLength(verified);
     await db.update(mcpEventSubscriptions).set({ stoppedAt: new Date(f.now()) }).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id));
     await expect(f.service.subscribe(f.principal, makeInput(20))).resolves.toHaveProperty("id");
     expect(await db.select().from(mcpEventSubscriptions).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id))).toHaveLength(1);
@@ -470,6 +472,57 @@ describe.skipIf(!support.supported)("public MCP OAuth and tool boundary", () =>
     await f.service.unsubscribe(f.principal, makeInput(21));
   });
 
+  it("bounds failed callback attempts across service replicas before doing network work", async () => {
+    const f = await eventFixture(); f.badChallenge();
+    const replica = createPublicMcpEvents(db, oauth, f.dispatch, f.options);
+    const input = (index: number) => ({ ...f.input, delivery: { ...f.input.delivery, url: f.input.delivery.url + "/" + index } });
+    for (let i = 0; i < 30; i++) await expect((i % 2 ? replica : f.service).subscribe(f.principal, input(i))).rejects.toMatchObject({ reason: "challenge_failed" });
+    expect(f.received).toHaveLength(30);
+    await expect(replica.subscribe(f.principal, input(30))).rejects.toThrow("capacity reached");
+    expect(f.received).toHaveLength(30);
+    expect(await db.select().from(mcpEventAdmissions).where(eq(mcpEventAdmissions.grantId, f.principal.grant.id))).toHaveLength(30);
+    f.advance(60_001);
+    await expect(replica.subscribe(f.principal, input(31))).rejects.toMatchObject({ reason: "challenge_failed" });
+    expect(f.received).toHaveLength(31);
+    expect(await db.select().from(mcpEventAdmissions).where(eq(mcpEventAdmissions.grantId, f.principal.grant.id))).toHaveLength(1);
+  });
+
+  it("reserves bounded verification capacity without holding transactions and respects unsubscribe", async () => {
+    const f = await eventFixture();
+    const replica = createPublicMcpEvents(db, oauth, f.dispatch, f.options);
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    f.duringVerification(() => gate);
+    const input = (index: number) => ({ ...f.input, delivery: { ...f.input.delivery, url: f.input.delivery.url + "/" + index } });
+    const pending = Promise.allSettled([f.service.subscribe(f.principal, input(0)), replica.subscribe(f.principal, input(1))]);
+    try {
+      await vi.waitFor(() => expect(f.received).toHaveLength(2));
+      const held = await db.select().from(mcpEventAdmissions).where(eq(mcpEventAdmissions.grantId, f.principal.grant.id));
+      expect(held).toHaveLength(2);
+      await db.transaction(async tx => {
+        const [locks] = await tx.execute(sql`select pg_try_advisory_xact_lock(736721043) as global_free, pg_try_advisory_xact_lock(hashtextextended(${held[0]!.subscriptionId}, 0)) as subscription_free`);
+        expect(locks).toMatchObject({ global_free: true, subscription_free: true });
+      });
+      await expect(replica.subscribe(f.principal, input(2))).rejects.toThrow("capacity reached");
+      expect(f.received).toHaveLength(2);
+      await f.service.unsubscribe(f.principal, input(0));
+    } finally { release(); }
+    const results = await pending;
+    expect(results.map(result => result.status)).toEqual(["rejected", "fulfilled"]);
+    expect(await db.select().from(mcpEventSubscriptions).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id))).toHaveLength(1);
+    await replica.unsubscribe(f.principal, input(1));
+  });
+
+  it.each(["lease", "revocation"])("rejects stale %s authority after callback verification", async reason => {
+    const f = await eventFixture();
+    f.duringVerification(async () => {
+      if (reason === "lease") f.advance(60_001);
+      else await oauth.revokeConnection(f.principal.grant.id, f.principal.grant.userId);
+    });
+    await expect(f.service.subscribe(f.principal, f.input)).rejects.toThrow();
+    expect(await db.select().from(mcpEventSubscriptions).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id))).toHaveLength(0);
+  });
+
   it("filters unchanged statuses and includes native checkout and release transitions", async () => {
     const f = await eventFixture();
     const input = { ...f.input, arguments: { companyId: f.company.id, taskId: f.task.id } };

server/src/services/public-mcp/events.ts

@@ -1,7 +1,7 @@
 import { createHash } from "node:crypto";
 import { and, asc, count, eq, gt, gte, isNotNull, isNull, lt, lte, or, sql } from "drizzle-orm";
 import { z } from "zod";
-import { companies, activityLog, mcpEventDeliveries as deliveries, mcpEventSubscriptions as subscriptions, type Db } from "@paperclipai/db";
+import { companies, activityLog, mcpEventAdmissions as admissions, mcpEventDeliveries as deliveries, mcpEventSubscriptions as subscriptions, type Db } from "@paperclipai/db";
 import { ISSUE_STATUSES } from "@paperclipai/shared";
 import { localEncryptedProvider } from "../../secrets/local-encrypted-provider.js";
 import { logActivity } from "../activity-log.js";
@@ -81,51 +81,89 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis
     if (!input.delivery.secret) throw new McpEventError(-32602, "A webhook signing secret is required.");
     signingKey(input.delivery.secret);
     await authorize(principal, input.arguments);
-    await authorizeCloud(principal, cloud);
     const id = identity(principal, input);
-    // Serialize admission, refresh and unsubscribe; the bounded verification has
-    // no application data. This also prevents duplicate challenges on concurrent retries.
-    return db.transaction(async tx => {
+    const url = callbackUrl(input.delivery.url);
+    const secret = input.delivery.secret;
+    // Reserve capacity in a short transaction before either remote authority or
+    // callback verification. Leases and attempt counts are shared across replicas.
+    const reservation = await db.transaction(async tx => {
+      await tx.execute(sql`select pg_advisory_xact_lock(736721043)`);
       await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${id}, 0))`);
+      await tx.delete(admissions).where(lte(admissions.expiresAt, new Date(now())));
+      await tx.delete(subscriptions).where(or(lte(subscriptions.expiresAt, new Date(now())), isNotNull(subscriptions.stoppedAt)));
       const [existing] = await tx.select().from(subscriptions).where(eq(subscriptions.id, id));
       const previous = existing ? await decrypt(existing) : null;
-      const url = callbackUrl(input.delivery.url);
-      const secret = input.delivery.secret!;
-      const verify = !existing || existing.stoppedAt || existing.expiresAt.getTime() <= now() || existing.verifiedAt.getTime() + rotationMs <= now() || previous?.secret !== secret;
-      if (verify) await verifyCallback(fetcher, id, url, secret, now());
-      // Network verification holds only this subscription's lock. The global
-      // quota lock covers short admission writes, never a remote callback.
-      await tx.execute(sql`select pg_advisory_xact_lock(736721043)`);
-      const active = existing && !existing.stoppedAt && existing.expiresAt.getTime() > now();
-      // No replay is promised for expired/stopped subscriptions. Reclaim these
-      // rows (and their receipts) before admission so historical monitors cannot
-      // exhaust active capacity or grow storage without bound.
-      await tx.delete(subscriptions).where(or(lte(subscriptions.expiresAt, new Date(now())), isNotNull(subscriptions.stoppedAt)));
-      if (!active) {
+      const [pending] = await tx.select().from(admissions).where(and(eq(admissions.subscriptionId, id), isNull(admissions.finishedAt)));
+      if (pending) throw new McpEventError(-32602, "Subscription verification is in progress. Retry shortly.");
+      const verify = !existing || existing.verifiedAt.getTime() + rotationMs <= now() || previous?.secret !== secret;
+      if (!existing) {
         const [total] = await tx.select({ n: count() }).from(subscriptions);
         const [company] = await tx.select({ n: count() }).from(subscriptions).where(eq(subscriptions.companyId, principal.grant.companyId));
         const [grant] = await tx.select({ n: count() }).from(subscriptions).where(eq(subscriptions.grantId, principal.grant.id));
-        if (total!.n >= 1000 || company!.n >= 100 || grant!.n >= 20) throw new McpEventError(-32602, "Subscription limit reached. Stop an existing monitor first.");
+        const reserved = await tx.select().from(admissions).where(and(isNull(admissions.finishedAt), eq(admissions.reservesSubscription, true)));
+        if (total!.n + reserved.length >= 1000 || company!.n + reserved.filter(r => r.companyId === principal.grant.companyId).length >= 100 || grant!.n + reserved.filter(r => r.grantId === principal.grant.id).length >= 20) {
+          throw new McpEventError(-32602, "Subscription limit reached. Stop an existing monitor first.");
+        }
       }
-      const expiresAt = new Date(Math.min(now() + Math.min(Math.max(input.ttlMs ?? lifetime, 30_000), lifetime), cloudOrigin ? Math.min(now() + rotationMs, cloud!.expiresAt) : Infinity));
-      const destination: Destination = { url, secret, ...(cloudOrigin ? { cloud } : {}),
-        ...(previous && previous.secret !== secret ? { previousSecret: previous.secret, previousUntil: now() + rotationMs }
-          : previous?.previousUntil && previous.previousUntil > now() ? { previousSecret: previous.previousSecret, previousUntil: previous.previousUntil } : {}) };
-      if (existing && !active) await tx.delete(deliveries).where(eq(deliveries.subscriptionId, id));
-      const value = { companyId: principal.grant.companyId, grantId: principal.grant.id, name: input.name, taskId: input.arguments.taskId,
-        arguments: input.arguments, deliveryMaterial: await encrypt(destination), expiresAt, stoppedAt: null,
-        verifiedAt: verify ? new Date(now()) : existing!.verifiedAt, startsAt: active ? existing.startsAt : requestedAt, scannedAt: new Date(now()) };
-      await tx.insert(subscriptions).values({ id, ...value }).onConflictDoUpdate({ target: subscriptions.id, set: value });
-      await logActivity(tx as unknown as Db, { companyId: principal.grant.companyId, actorType: "user", actorId: principal.grant.userId,
-        action: "mcp.event_subscribed", entityType: "mcp_subscription", entityId: id, details: { name: input.name, taskId: input.arguments.taskId, expiresAt: expiresAt.toISOString() } });
-      return { id, refreshBefore: expiresAt.toISOString(), cursor: null, truncated: false };
+      if (!verify && !cloudOrigin) return { existing, previous, verify, lease: null };
+      const attempts = await tx.select().from(admissions);
+      const companyAttempts = attempts.filter(r => r.companyId === principal.grant.companyId);
+      const grantAttempts = companyAttempts.filter(r => r.grantId === principal.grant.id);
+      const pendingCount = (rows: typeof attempts) => rows.filter(r => !r.finishedAt).length;
+      if (attempts.length >= 1000 || companyAttempts.length >= 200 || grantAttempts.length >= 30
+        || pendingCount(attempts) >= 32 || pendingCount(companyAttempts) >= 8 || pendingCount(grantAttempts) >= 2) {
+        throw new McpEventError(-32602, "Subscription verification capacity reached. Retry later.");
+      }
+      const [lease] = await tx.insert(admissions).values({ subscriptionId: id, companyId: principal.grant.companyId, grantId: principal.grant.id,
+        reservesSubscription: !existing, createdAt: new Date(now()), expiresAt: new Date(now() + 60_000) }).returning();
+      return { existing, previous, verify, lease: lease! };
     });
+    const { existing, previous, verify, lease } = reservation;
+    try {
+      await authorizeCloud(principal, cloud);
+      if (verify) await verifyCallback(fetcher, id, url, secret, now());
+      // Remote waits must not retain database connections or stale authority.
+      const current = await oauth.authorizeGrant(principal.grant.id);
+      await authorize(current, input.arguments);
+      const material = !verify && !cloudOrigin ? existing!.deliveryMaterial : await encrypt({ url, secret, ...(cloudOrigin ? { cloud } : {}),
+        ...(previous && previous.secret !== secret ? { previousSecret: previous.secret, previousUntil: now() + rotationMs }
+          : previous?.previousUntil && previous.previousUntil > now() ? { previousSecret: previous.previousSecret, previousUntil: previous.previousUntil } : {}) });
+      return await db.transaction(async tx => {
+        await tx.execute(sql`select pg_advisory_xact_lock(736721043)`);
+        await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${id}, 0))`);
+        if (lease) {
+          const [held] = await tx.update(admissions).set({ finishedAt: new Date(now()) }).where(and(eq(admissions.id, lease.id), isNull(admissions.finishedAt), gt(admissions.expiresAt, new Date(now())))).returning();
+          if (!held) throw new McpEventError(-32602, "Subscription verification expired or was stopped. Reconnect the monitor.");
+          if (existing) {
+            const [active] = await tx.select().from(subscriptions).where(and(eq(subscriptions.id, id), isNull(subscriptions.stoppedAt), gt(subscriptions.expiresAt, new Date(now()))));
+            if (!active) throw new McpEventError(-32602, "The monitor expired or was stopped. Subscribe again.");
+          }
+        } else {
+          // A cached refresh cannot recreate a subscription removed while it was awaiting authority.
+          const [held] = await tx.select().from(subscriptions).where(eq(subscriptions.id, id));
+          const [pending] = await tx.select().from(admissions).where(and(eq(admissions.subscriptionId, id), isNull(admissions.finishedAt), gt(admissions.expiresAt, new Date(now()))));
+          if (!held || held.stoppedAt || held.expiresAt.getTime() <= now() || pending || canonical(held.deliveryMaterial) !== canonical(existing!.deliveryMaterial)) throw new McpEventError(-32602, "The monitor changed or was stopped. Retry the subscription.");
+        }
+        if (cloudOrigin && cloud!.expiresAt <= now()) throw new McpEventError(-32602, "Refresh the hosted connection before subscribing.");
+        const expiresAt = new Date(Math.min(now() + Math.min(Math.max(input.ttlMs ?? lifetime, 30_000), lifetime), cloudOrigin ? Math.min(now() + rotationMs, cloud!.expiresAt) : Infinity));
+        const value = { companyId: principal.grant.companyId, grantId: principal.grant.id, name: input.name, taskId: input.arguments.taskId,
+          arguments: input.arguments, deliveryMaterial: material, expiresAt, stoppedAt: null,
+          verifiedAt: verify ? new Date(now()) : existing!.verifiedAt, startsAt: existing?.startsAt ?? requestedAt, scannedAt: new Date(now()) };
+        await tx.insert(subscriptions).values({ id, ...value }).onConflictDoUpdate({ target: subscriptions.id, set: value });
+        await logActivity(tx as unknown as Db, { companyId: principal.grant.companyId, actorType: "user", actorId: principal.grant.userId,
+          action: "mcp.event_subscribed", entityType: "mcp_subscription", entityId: id, details: { name: input.name, taskId: input.arguments.taskId, expiresAt: expiresAt.toISOString() } });
+        return { id, refreshBefore: expiresAt.toISOString(), cursor: null, truncated: false };
+      });
+    } finally {
+      if (lease) await db.update(admissions).set({ finishedAt: new Date(now()) }).where(and(eq(admissions.id, lease.id), isNull(admissions.finishedAt)));
+    }
   }
   async function unsubscribe(principal: McpPrincipal, raw: unknown) {
     const input = unsubscribeSchema.parse(raw); validate(input);
     const id = identity(principal, input);
     await db.transaction(async tx => {
       await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${id}, 0))`);
+      await tx.update(admissions).set({ finishedAt: new Date(now()) }).where(and(eq(admissions.subscriptionId, id), isNull(admissions.finishedAt)));
       await tx.delete(subscriptions).where(and(eq(subscriptions.id, id), eq(subscriptions.grantId, principal.grant.id)));
       await logActivity(tx as unknown as Db, { companyId: principal.grant.companyId, actorType: "user", actorId: principal.grant.userId,
         action: "mcp.event_unsubscribed", entityType: "mcp_subscription", entityId: id, details: { name: input.name, taskId: input.arguments.taskId } });
@@ -207,6 +245,7 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis
   }
   let running: Promise<void> | null = null;
   const tick = () => running ?? (running = (async () => {
+    await db.delete(admissions).where(lte(admissions.expiresAt, new Date(now())));
     await db.delete(subscriptions).where(lt(subscriptions.expiresAt, new Date(now() - 7 * 24 * hour)));
     const active = await db.select().from(subscriptions).where(and(isNull(subscriptions.stoppedAt), gt(subscriptions.expiresAt, new Date(now())))).orderBy(asc(subscriptions.scannedAt)).limit(20);
     for (const s of active) await enqueue(s);