fix(api): make daily submission quota race-free

README.md

@@ -304,7 +304,8 @@ is published; do not rely on the dev fallback locally if you plan to promote tha
 
 Master key rotation does not retire a source's Signing Identity. Signing Identity rotation retires
 the old public identity, creates a new encrypted one, and requires a new NEAR binding without
-modifying historical events. See [`docs/activity-protocol.md`](docs/activity-protocol.md#source-credentials-and-signing)
+modifying historical events. After rotation, ingestion returns `403` until the new identity is
+re-bound on-chain. See [`docs/activity-protocol.md`](docs/activity-protocol.md#source-credentials-and-signing)
 for the per-identity signing history.
 
 #### NEAR binding variables

api/src/services/activity-ingestion.ts

@@ -1,6 +1,6 @@
 import { createHash } from "node:crypto";
 import { ORPCError } from "@orpc/server";
-import { and, count, eq, gte, isNull } from "drizzle-orm";
+import { and, count, eq, gte, isNull, sql } from "drizzle-orm";
 import { type Event, validateEvent, verifyEvent } from "nostr-tools/pure";
 import { ACTIVITY_EVENT_KIND, type ActivityRelay } from "../activity/activity-relay";
 import { ACTIVITY_TEST_EVENT_TYPE } from "../contract";
@@ -113,38 +113,6 @@ export class ActivityIngestionService {
     };
   }
 
-  async #enforceDailySubmissionLimit(
-    sourceRecordId: string,
-    idempotencyKey: string,
-    limit: number,
-    message: string,
-  ) {
-    const [existing] = await this.#db
-      .select({ id: submissionsTable.id })
-      .from(submissionsTable)
-      .where(
-        and(
-          eq(submissionsTable.sourceRecordId, sourceRecordId),
-          eq(submissionsTable.idempotencyKey, idempotencyKey),
-        ),
-      )
-      .limit(1);
-    if (existing) return;
-    const since = new Date(this.#now().getTime() - 24 * 60 * 60 * 1_000);
-    const [recent] = await this.#db
-      .select({ total: count() })
-      .from(submissionsTable)
-      .where(
-        and(
-          eq(submissionsTable.sourceRecordId, sourceRecordId),
-          gte(submissionsTable.createdAt, since),
-        ),
-      );
-    if ((recent?.total ?? 0) >= limit) {
-      throw new ORPCError("TOO_MANY_REQUESTS", { message });
-    }
-  }
-
   async #submitWithCredential(
     credential: ActivityEventWriteCredential,
     input: ActivityEventSubmission,
@@ -182,38 +150,59 @@ export class ActivityIngestionService {
       throw new ORPCError("NOT_FOUND", { message: "Activity Source not found" });
     }
     const quota = this.#resolveDailySubmissionQuota(source);
-    if (quota) {
-      await this.#enforceDailySubmissionLimit(
-        sourceRow.id,
-        input.idempotencyKey,
-        quota.limit,
-        quota.message,
-      );
-    }
     const requestHash = hashSubmission(input);
-    const [created] = await this.#db
-      .insert(submissionsTable)
-      .values({
-        sourceRecordId: sourceRow.id,
-        idempotencyKey: input.idempotencyKey,
-        requestHash,
-      })
-      .onConflictDoNothing()
-      .returning();
-    const submission =
-      created ??
-      (
-        await this.#db
-          .select()
+    const submission = await this.#db.transaction(async (tx) => {
+      if (quota) {
+        await tx.execute(sql`select pg_advisory_xact_lock(hashtext(${sourceRow.id}))`);
+      }
+      const [existing] = await tx
+        .select()
+        .from(submissionsTable)
+        .where(
+          and(
+            eq(submissionsTable.sourceRecordId, sourceRow.id),
+            eq(submissionsTable.idempotencyKey, input.idempotencyKey),
+          ),
+        )
+        .limit(1);
+      if (existing) return existing;
+      if (quota) {
+        const since = new Date(this.#now().getTime() - 24 * 60 * 60 * 1_000);
+        const [recent] = await tx
+          .select({ total: count() })
           .from(submissionsTable)
           .where(
             and(
               eq(submissionsTable.sourceRecordId, sourceRow.id),
-              eq(submissionsTable.idempotencyKey, input.idempotencyKey),
+              gte(submissionsTable.createdAt, since),
             ),
-          )
-          .limit(1)
-      )[0];
+          );
+        if ((recent?.total ?? 0) >= quota.limit) {
+          throw new ORPCError("TOO_MANY_REQUESTS", { message: quota.message });
+        }
+      }
+      const [created] = await tx
+        .insert(submissionsTable)
+        .values({
+          sourceRecordId: sourceRow.id,
+          idempotencyKey: input.idempotencyKey,
+          requestHash,
+        })
+        .onConflictDoNothing()
+        .returning();
+      if (created) return created;
+      const [reserved] = await tx
+        .select()
+        .from(submissionsTable)
+        .where(
+          and(
+            eq(submissionsTable.sourceRecordId, sourceRow.id),
+            eq(submissionsTable.idempotencyKey, input.idempotencyKey),
+          ),
+        )
+        .limit(1);
+      return reserved;
+    });
     if (!submission) {
       throw new ORPCError("INTERNAL_SERVER_ERROR", {
         message: "Activity event reservation failed",

api/tests/integration/activity-ingestion.test.ts

@@ -406,6 +406,44 @@ describe("Activity event ingestion", () => {
     );
   });
 
+  it("enforces the pending daily limit for concurrent distinct submissions", async () => {
+    const { secret } = await provisionIngestionSource({
+      sourceId: "concurrent-quota-source",
+      ownerId: "concurrent-quota-owner",
+      organizationId: "org-concurrent-quota",
+      eventType: "quota.recorded",
+      approve: false,
+    });
+    const gateway = await getPluginClient(undefined, {
+      authorization: `Bearer ${secret}`,
+    });
+    const submit = (sequence: number) =>
+      gateway.submitActivityEvent({
+        eventType: "quota.recorded",
+        actor: "quota-actor.near",
+        idempotencyKey: `quota:${sequence}`,
+        payload: { sequence },
+      });
+    resetTestRelayEvents();
+
+    await submit(1);
+    await submit(2);
+    const results = await Promise.allSettled([submit(3), submit(4), submit(5)]);
+
+    expect(results.filter(({ status }) => status === "fulfilled")).toHaveLength(1);
+    const rejected = results.filter(
+      (result): result is PromiseRejectedResult => result.status === "rejected",
+    );
+    expect(rejected).toHaveLength(2);
+    expect(rejected.every(({ reason }) => reason?.code === "TOO_MANY_REQUESTS")).toBe(true);
+    expect(
+      rejected.every(({ reason }) =>
+        String(reason).includes("Activity Sources under review may submit up to 3 events per day"),
+      ),
+    ).toBe(true);
+    expect(getTestRelayEvents()).toHaveLength(3);
+  });
+
   it("enforces a daily limit for approved standard sources", async () => {
     const { secret } = await provisionIngestionSource({
       sourceId: "approved-ingestion-source",