fix(selfhost): bound notification queue coalescing (#3218)

* fix(selfhost): bound notification queue coalescing

* test(selfhost): fix a notify-evaluate coalesce-key assertion stale from the digest switch

This PR changed jobCoalesceKey's notify-evaluate case to a fixed-length
digest of the sorted dedup-key set, but one assertion in the general
"coalesces by stable id" suite still expected the old raw-key format,
failing CI. Matches the shape assertion already used by the dedicated
digest test added elsewhere in this PR.

* fix(selfhost): sort notify-evaluate events before chunking, not after

jobCoalesceKey already hashes each chunk's OWN sorted dedup-key set,
so a chunk's coalesce key was order-independent WITHIN that chunk --
but chunk MEMBERSHIP was still built from notificationEvents' raw
arrival order, so a redelivery whose events resolved in a different
order could split across a different 100-event boundary and never
coalesce with the earlier attempt. Sorting by dedupKey before
chunking makes membership a pure function of the detected event set,
restoring the "same full set in any order" coalescing guarantee
across chunk boundaries too. Added a regression test that reproduces
the split with 205 reordered watchers and fails without the fix.

Author: JSONbored

src/queue/processors.ts

@@ -480,6 +480,7 @@ const PR_PUBLIC_SURFACE_ACTIONS = new Set([
 ]);
 const PR_GATE_CLOSED_ACTIONS = new Set(["closed"]);
 const ISSUE_PLAN_COOLDOWN_MS = 10 * 60 * 1000;
+const NOTIFY_EVALUATE_EVENTS_PER_JOB = 100;
 
 type RequiredStatusContextsLookup = { requiredContexts: Set<string> | null; resolved: boolean };
 
@@ -511,6 +512,14 @@ function liveFactTokenPart(token: string | undefined): string {
   return `token:${token.length}:${(hash >>> 0).toString(16).padStart(8, "0")}`;
 }
 
+function chunkNotificationEvents(events: DetectedNotificationEvent[]): DetectedNotificationEvent[][] {
+  const chunks: DetectedNotificationEvent[][] = [];
+  for (let start = 0; start < events.length; start += NOTIFY_EVALUATE_EVENTS_PER_JOB) {
+    chunks.push(events.slice(start, start + NOTIFY_EVALUATE_EVENTS_PER_JOB));
+  }
+  return chunks;
+}
+
 function githubAdmissionKeyForToken(
   env: Env,
   installationId: number | null | undefined,
@@ -5602,15 +5611,20 @@ async function processGitHubWebhook(
         },
       });
     }
-    // Batched (#selfhost-maintenance-self-pin): every event this ONE webhook delivery detected rides in a
-    // single notify-evaluate job instead of one job per event -- the audit trail above still records each
-    // event individually, so nothing about observability changes, only how many maintenance-lane rows a
-    // multi-watcher issue (or a review event landing alongside issue-watch matches) creates.
-    if (notificationEvents.length > 0) {
+    // Batched, but bounded (#selfhost-maintenance-self-pin): a popular issue can have thousands of watchers,
+    // so keep queue payloads comfortably below backend message limits while still avoiding one row per event.
+    // Sorted by dedupKey BEFORE chunking (#3218 review): jobCoalesceKey hashes each chunk's OWN sorted dedup-key
+    // set, so it's already order-independent WITHIN a chunk -- but chunk MEMBERSHIP itself was still built from
+    // notificationEvents' arrival order, so a redelivery whose events resolved in a different order could split
+    // across a different 100-event boundary and never coalesce with the earlier attempt. Sorting first makes
+    // chunk membership a pure function of the detected event SET, not its arrival order, restoring the "same
+    // full set in any order" coalescing guarantee across chunk boundaries too.
+    const notificationEventsForChunking = [...notificationEvents].sort((a, b) => a.dedupKey.localeCompare(b.dedupKey));
+    for (const events of chunkNotificationEvents(notificationEventsForChunking)) {
       await env.JOBS.send({
         type: "notify-evaluate",
         requestedBy: "webhook",
-        events: notificationEvents,
+        events,
       });
     }
 

src/selfhost/queue-common.ts

@@ -996,15 +996,15 @@ export function jobCoalesceKey(payload: string): string | null {
       }
       case "notify-evaluate": {
         // A batched job carries every event from one webhook delivery (#selfhost-maintenance-self-pin) --
-        // coalescing keys off the FULL sorted set of dedup keys, so a redelivery of the identical batch still
-        // coalesces (same events -> same key) while any batch with even one different event gets its own key.
+        // coalescing keys off a digest of the FULL sorted set of dedup keys, so a redelivery of the identical
+        // batch still coalesces without placing an attacker-sized concatenation into the indexed job_key column.
         // If ANY event is missing its dedup key (a malformed payload), the whole batch is left uncoalesced
         // (null) rather than keying off a partial set that could collide with an unrelated batch and silently
         // drop the malformed event's work -- same rule as the other event-id-keyed types above.
         if (!Array.isArray(message.events) || message.events.length === 0) return null;
         const dedupKeys = message.events.map((event) => normalizedId(event?.dedupKey));
         if (dedupKeys.some((dedupKey) => dedupKey === null)) return null;
-        return keyOf(type, [...(dedupKeys as string[])].sort().join(","));
+        return keyOf(type, stableStringDigest([...(dedupKeys as string[])].sort()));
       }
       case "submit-draft": {
         const draftId = normalizedId(message.draftId);
@@ -1118,6 +1118,10 @@ function normalizedPathScope(value: unknown): string | null {
   return `sha256:${createHash("sha256").update(JSON.stringify(paths)).digest("hex")}`;
 }
 
+function stableStringDigest(values: string[]): string {
+  return `sha256:${createHash("sha256").update(JSON.stringify(values)).digest("hex")}`;
+}
+
 function boolFlag(value: unknown): string {
   return value === true ? "1" : "0";
 }

test/unit/queue.test.ts

@@ -8,6 +8,7 @@ import * as repositoriesModule from "../../src/db/repositories";
 import * as repositorySettingsModule from "../../src/settings/repository-settings";
 import * as sentryModule from "../../src/selfhost/sentry";
 import { renderMetrics, resetMetrics } from "../../src/selfhost/metrics";
+import { jobCoalesceKey } from "../../src/selfhost/queue-common";
 import {
   listCollisionEdges,
   createAgentRun,
@@ -16946,8 +16947,10 @@ describe("queue processors", () => {
     const enqueued: Array<{ type: string; events?: Array<{ eventType: string; recipientLogin: string; pullNumber: number }> }> = [];
     const env = createTestEnv({ JOBS: { async send(message: { type: string }) { enqueued.push(message); } } as unknown as Queue });
     vi.stubGlobal("fetch", async () => new Response("not found", { status: 404 })); // no .gittensory.yml → empty manifest
-    await upsertIssueWatchSubscription(env, { login: "watcher-one", repoFullName: "JSONbored/gittensory" });
-    await upsertIssueWatchSubscription(env, { login: "watcher-two", repoFullName: "JSONbored/gittensory" });
+    const watcherLogins = Array.from({ length: 205 }, (_, index) => `watcher-${String(index + 1).padStart(3, "0")}`);
+    for (const login of watcherLogins) {
+      await upsertIssueWatchSubscription(env, { login, repoFullName: "JSONbored/gittensory" });
+    }
     await upsertIssueWatchSubscription(env, { login: "maintainer", repoFullName: "JSONbored/gittensory" }); // the author — should be skipped
 
     await processJob(env, {
@@ -16962,17 +16965,58 @@ describe("queue processors", () => {
       },
     });
 
-    // Batched (#selfhost-maintenance-self-pin): both watcher matches from this ONE webhook delivery ride in a
-    // SINGLE notify-evaluate job, not one job per watcher -- that fan-out was flooding the self-host maintenance
-    // lane with a job per watcher on a popular issue.
+    // Batched but bounded (#selfhost-maintenance-self-pin): watcher matches from this ONE webhook delivery ride in
+    // chunked notify-evaluate jobs, not one job per watcher and not one unbounded queue payload.
     const evaluateJobs = enqueued.filter((m): m is { type: "notify-evaluate"; events: Array<{ eventType: string; recipientLogin: string; pullNumber: number }> } => m.type === "notify-evaluate");
-    expect(evaluateJobs).toHaveLength(1);
-    const watchEvents = evaluateJobs[0]!.events.filter((event) => event.eventType === "issue_watch_match");
-    expect(watchEvents.map((event) => event.recipientLogin).sort()).toEqual(["watcher-one", "watcher-two"]); // maintainer (author) skipped
+    expect(evaluateJobs.map((job) => job.events).map((events) => events.length)).toEqual([100, 100, 5]);
+    const watchEvents = evaluateJobs.flatMap((job) => job.events).filter((event) => event.eventType === "issue_watch_match");
+    expect(watchEvents.map((event) => event.recipientLogin).sort()).toEqual(watcherLogins); // maintainer (author) skipped
     expect(watchEvents.every((event) => event.pullNumber === 91)).toBe(true);
 
-    const detected = await env.DB.prepare("select metadata_json from audit_events where event_type = 'notification.event_detected' and target_key = ?").bind("watcher-one").first<{ metadata_json: string }>();
-    expect(JSON.parse(detected!.metadata_json)).toMatchObject({ eventType: "issue_watch_match", recipientLogin: "watcher-one", repoFullName: "JSONbored/gittensory" });
+    const detected = await env.DB.prepare("select metadata_json from audit_events where event_type = 'notification.event_detected' and target_key = ?").bind("watcher-001").first<{ metadata_json: string }>();
+    expect(JSON.parse(detected!.metadata_json)).toMatchObject({ eventType: "issue_watch_match", recipientLogin: "watcher-001", repoFullName: "JSONbored/gittensory" });
+  });
+
+  it("REGRESSION (#3218 review): chunk membership across a >100-watcher batch is order-independent -- the SAME watcher set in a different arrival order still produces the SAME set of chunk coalesce keys", async () => {
+    const watcherLogins = Array.from({ length: 205 }, (_, index) => `watcher-${String(index + 1).padStart(3, "0")}`);
+
+    const enqueueNotifyEvaluateJobs = async (loginOrder: string[]): Promise<Array<{ type: string; events: Array<{ dedupKey: string }> }>> => {
+      const enqueued: Array<{ type: string; events?: Array<{ dedupKey: string }> }> = [];
+      const env = createTestEnv({ JOBS: { async send(message: { type: string }) { enqueued.push(message); } } as unknown as Queue });
+      vi.stubGlobal("fetch", async () => new Response("not found", { status: 404 })); // no .gittensory.yml → empty manifest
+      // listIssueWatchersForRepo has no ORDER BY -- insertion order IS read-back order, so inserting in a
+      // different order here genuinely reproduces two logically-identical detection passes disagreeing on
+      // notificationEvents' arrival order, exactly the redelivery scenario the review is concerned about.
+      for (const login of loginOrder) {
+        await upsertIssueWatchSubscription(env, { login, repoFullName: "JSONbored/gittensory" });
+      }
+      await processJob(env, {
+        type: "github-webhook",
+        deliveryId: `issue-watch-open-${loginOrder[0]}`,
+        eventName: "issues",
+        payload: {
+          action: "opened",
+          installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" } },
+          repository: { name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } },
+          issue: { number: 91, title: "Add caching to the registry sync", state: "open", user: { login: "maintainer" }, author_association: "OWNER", body: "We should cache the registry fetch." },
+        },
+      });
+      vi.unstubAllGlobals();
+      return enqueued.filter((m): m is { type: "notify-evaluate"; events: Array<{ dedupKey: string }> } => m.type === "notify-evaluate");
+    };
+
+    const coalesceKeysFor = (jobs: Array<{ type: string; events: Array<{ dedupKey: string }> }>): Array<string | null> =>
+      jobs.map((job) => jobCoalesceKey(JSON.stringify(job))).sort();
+
+    const forwardJobs = await enqueueNotifyEvaluateJobs(watcherLogins);
+    const reversedJobs = await enqueueNotifyEvaluateJobs([...watcherLogins].reverse());
+
+    // Same chunk SIZES either way (chunking itself is unaffected -- only membership was the risk).
+    expect(forwardJobs.map((job) => job.events.length)).toEqual([100, 100, 5]);
+    expect(reversedJobs.map((job) => job.events.length)).toEqual([100, 100, 5]);
+    // The set of chunk-level coalesce keys must match -- proving a redelivery whose events resolve in a
+    // different order still coalesces with the original batch instead of silently re-running as "new" work.
+    expect(coalesceKeysFor(reversedJobs)).toEqual(coalesceKeysFor(forwardJobs));
   });
 
   it("appends issue-side slop findings to the issue advisory only when slop is opted in (#533)", async () => {

test/unit/selfhost-queue-common.test.ts

@@ -1052,11 +1052,14 @@ describe("self-host queue common helpers", () => {
     expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "github_comment", runId: "run-abc123" }))).toBe("run-agent:run-abc123");
     expect(jobCoalesceKey(payload({ type: "notify-deliver", requestedBy: "notify-evaluate", deliveryId: "del-77" }))).toBe("notify-deliver:del-77");
     expect(jobCoalesceKey(payload({ type: "submit-draft", requestedBy: "api", draftId: "draft-9" }))).toBe("submit-draft:draft-9");
+    // notify-evaluate keys off a fixed-length digest of the batch's dedup keys, not the raw keys themselves
+    // (#selfhost-maintenance-self-pin, tested for shape/order-independence/collision-avoidance below) --
+    // still a stable per-invocation key, so it belongs in this "coalesces by stable id" suite too.
     expect(
       jobCoalesceKey(
         payload({ type: "notify-evaluate", requestedBy: "webhook", events: [{ dedupKey: "review_requested:o/r#3:bob" }] }),
       ),
-    ).toBe("notify-evaluate:review_requested:o/r#3:bob");
+    ).toMatch(/^notify-evaluate:sha256:[a-f0-9]{64}$/);
     // Two DISTINCT invocations have distinct ids → distinct keys, so they never merge.
     expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "github_comment", runId: "run-xyz789" }))).toBe("run-agent:run-xyz789");
     // A payload missing its id → null (uncoalesced), never a shared key that could drop a distinct job.
@@ -1068,7 +1071,7 @@ describe("self-host queue common helpers", () => {
     expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "test", events: [{}] }))).toBeNull();
   });
 
-  it("batches a notify-evaluate job's coalesce key off the FULL sorted set of dedup keys (#selfhost-maintenance-self-pin)", () => {
+  it("batches a notify-evaluate job's coalesce key off a fixed-length digest of the FULL sorted set of dedup keys (#selfhost-maintenance-self-pin)", () => {
     // Order-independent: the same two events in either order produce the same key, so a redelivery with the
     // events reordered still coalesces.
     const forward = jobCoalesceKey(
@@ -1085,7 +1088,7 @@ describe("self-host queue common helpers", () => {
         events: [{ dedupKey: "issue_watch_match:o/r#9:alice" }, { dedupKey: "review_requested:o/r#3:bob" }],
       }),
     );
-    expect(forward).toBe("notify-evaluate:issue_watch_match:o/r#9:alice,review_requested:o/r#3:bob");
+    expect(forward).toMatch(/^notify-evaluate:sha256:[a-f0-9]{64}$/);
     expect(reversed).toBe(forward);
     // A batch with even one different event gets a DIFFERENT key -- never silently merges with an unrelated batch.
     const differentBatch = jobCoalesceKey(
@@ -1096,6 +1099,18 @@ describe("self-host queue common helpers", () => {
       }),
     );
     expect(differentBatch).not.toBe(forward);
+    expect(differentBatch).toMatch(/^notify-evaluate:sha256:[a-f0-9]{64}$/);
+    const many = jobCoalesceKey(
+      payload({
+        type: "notify-evaluate",
+        requestedBy: "webhook",
+        events: Array.from({ length: 5_000 }, (_, index) => ({
+          dedupKey: `issue_watch_match:owner/repo#9:watcher-${String(index).padStart(4, "0")}`,
+        })),
+      }),
+    );
+    expect(many).toMatch(/^notify-evaluate:sha256:[a-f0-9]{64}$/);
+    expect(many!.length).toBe("notify-evaluate:sha256:".length + 64);
     // If ANY event in the batch is missing its dedup key, the whole batch is left uncoalesced (null) rather than
     // key off a partial set that could collide with -- and silently drop the malformed event from -- an
     // unrelated batch.