Skip to content

Commit a39c458

Browse files
JSONboredJSONbored
committed
fix(queue): wake missed siblings on out-of-order contributor-cap delivery (#2270)
Webhook delivery order is not guaranteed to match PR creation order, so a sibling PR's webhook can process before a later-created PR exists in the DB and wrongly conclude the author is within the open-PR cap. Nothing else would ever re-evaluate that sibling, permanently bypassing the cap. Now that a later delivery has the complete picture, wake any other still-open over-cap sibling so its own next pass self-corrects.
1 parent a32a483 commit a39c458

2 files changed

Lines changed: 259 additions & 1 deletion

File tree

‎src/queue/processors.ts‎

Lines changed: 57 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1738,7 +1738,7 @@ async function runAgentMaintenancePlanAndExecute(
17381738
liveFacts: LiveGithubFacts;
17391739
},
17401740
): Promise<void> {
1741-
const { installationId, repoFullName, pr, settings, otherOpenPullRequests, gate } = args;
1741+
const { installationId, repoFullName, pr, settings, otherOpenPullRequests, deliveryId, gate } = args;
17421742

17431743
// Convergence safety: feed the planner the PR's changed paths + the repo's hard-guardrail globs so guarded
17441744
// paths force manual review, and flag owner-authored PRs so they are never auto-closed (standing rule).
@@ -1882,6 +1882,17 @@ async function runAgentMaintenancePlanAndExecute(
18821882
if (overCapNumbers.has(pr.number)) {
18831883
contributorCapMatch = { matched: true, authorLogin: pr.authorLogin, openCount: authorOpenPrNumbers.length, cap: contributorOpenPrCap };
18841884
}
1885+
// Webhook-delivery-order guard (#2479 gate finding): delivery order is not guaranteed to match PR creation
1886+
// order, so a sibling PR's own webhook can process before THIS PR exists in the DB and wrongly conclude the
1887+
// author is within the cap — nothing else would ever re-evaluate it, permanently bypassing the cap for that
1888+
// sibling. Now that THIS delivery has the complete picture, wake any OTHER still-open sibling that's also
1889+
// in the over-cap set so its own next pass re-evaluates against the complete set and self-corrects.
1890+
const otherOverCapSiblingNumbers = otherOpenPullRequests
1891+
.filter((other) => (other.authorLogin ?? "").toLowerCase() === authorLoginLower && overCapNumbers.has(other.number))
1892+
.map((other) => other.number);
1893+
if (otherOverCapSiblingNumbers.length > 0) {
1894+
await wakeOverCapSiblingPullRequests(env, deliveryId, installationId, repoFullName, otherOverCapSiblingNumbers);
1895+
}
18851896
}
18861897

18871898
const planned = planAgentMaintenanceActions({
@@ -2525,6 +2536,51 @@ async function scheduleTrailingIssueLinkedReReview(
25252536
await putTransientKey(env, key, "1", CI_COALESCE_WINDOW_SECONDS);
25262537
}
25272538

2539+
/** Best-effort wake for sibling PRs discovered to be over the per-contributor cap by a LATER delivery (#2270,
2540+
* #2479 gate finding): webhook delivery order isn't guaranteed to match PR creation order, so a sibling's own
2541+
* webhook can fire before this one exists in the DB and wrongly conclude the author is within the cap — with
2542+
* nothing else to ever re-evaluate it, that verdict would otherwise stand forever. Reuses the existing
2543+
* agent-regate-pr sweep-unit job (already rate-limit-aware and retried) — the SAME "wake and fully
2544+
* re-evaluate" entry point the linked-issue-wake feature (#2259) uses for an identical class of problem, so
2545+
* the sibling gets its own live-head/CI-freshness re-check before anything acts on it, not a shortcut based
2546+
* on this delivery's now-possibly-stale snapshot. Coalesced per sibling PR (mirrors
2547+
* scheduleTrailingIssueLinkedReReview's check-then-claim-after-success shape) so a burst of N over-cap
2548+
* siblings each discovering the same others doesn't fan out into an O(N^2) job storm. */
2549+
async function wakeOverCapSiblingPullRequests(
2550+
env: Env,
2551+
deliveryId: string,
2552+
installationId: number,
2553+
repoFullName: string,
2554+
siblingPrNumbers: number[],
2555+
): Promise<void> {
2556+
await Promise.all(
2557+
siblingPrNumbers.map(async (prNumber) => {
2558+
const key = `contributor-cap-wake:${repoFullName.toLowerCase()}#${prNumber}`;
2559+
if (await getTransientKey(env, key)) return;
2560+
try {
2561+
await env.JOBS.send({
2562+
type: "agent-regate-pr",
2563+
deliveryId,
2564+
repoFullName,
2565+
prNumber,
2566+
installationId,
2567+
});
2568+
} catch (error) {
2569+
console.log(
2570+
JSON.stringify({
2571+
ev: "contributor_cap_wake_enqueue_failed",
2572+
repoFullName,
2573+
pull: prNumber,
2574+
message: errorMessage(error).slice(0, 120),
2575+
}),
2576+
);
2577+
return; // do NOT claim — a later discovery should retry the enqueue
2578+
}
2579+
await putTransientKey(env, key, "1", CI_COALESCE_WINDOW_SECONDS);
2580+
}),
2581+
);
2582+
}
2583+
25282584
async function ciHeadShaResolutionCoalesced(
25292585
env: Env,
25302586
repoFullName: string,

‎test/unit/queue.test.ts‎

Lines changed: 202 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5905,6 +5905,208 @@ describe("queue processors", () => {
59055905
expect(seen.closed).toBe(false);
59065906
});
59075907

5908+
it("contributor open-PR cap (#2270): an author-less (ghost) open PR among the repo's others is excluded from the count and the sibling-wake scan, not crashed on", async () => {
5909+
const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() });
5910+
await upsertInstallation(env, {
5911+
installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" }, target_type: "User", repository_selection: "all", permissions: { metadata: "read", pull_requests: "write", issues: "write" }, events: ["pull_request"] },
5912+
repositories: [{ name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } }],
5913+
});
5914+
// A ghost PR with no `user` at all (authorLogin ends up null) — must not match farmer99's count, and must
5915+
// not crash the sibling-wake scan, which runs the identical (authorLogin ?? "") fallback.
5916+
await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 50, title: "Ghost PR", state: "open", head: { sha: "ghost50" }, labels: [], body: "z" });
5917+
await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 53, title: "Farmer PR one", state: "open", user: { login: "farmer99" }, head: { sha: "f53" }, labels: [], body: "x" });
5918+
await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 54, title: "Farmer PR two", state: "open", user: { login: "farmer99" }, head: { sha: "f54" }, labels: [], body: "y" });
5919+
await upsertRepositorySettings(env, {
5920+
repoFullName: "JSONbored/gittensory",
5921+
commentMode: "all_prs",
5922+
publicSurface: "comment_only",
5923+
checkRunMode: "off",
5924+
gateCheckMode: "enabled",
5925+
aiReviewMode: "advisory",
5926+
autonomy: { close: "auto", label: "auto" },
5927+
contributorOpenPrCap: 2,
5928+
});
5929+
const seen = { closed: false };
5930+
vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => {
5931+
const url = input.toString();
5932+
const method = init?.method ?? "GET";
5933+
if (url === "https://api.gittensor.io/miners") return Response.json([]);
5934+
if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" });
5935+
if (url.includes("/pulls/55/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+const ok = true;" }]);
5936+
if (url.includes("/pulls/55/reviews")) return Response.json([]);
5937+
if (url.includes("/pulls/55/commits")) return Response.json([]);
5938+
if (url.endsWith("/pulls/55") && method === "PATCH") { seen.closed = JSON.parse(String(init?.body ?? "{}")).state === "closed"; return Response.json({ number: 55, state: "closed" }); }
5939+
if (url.endsWith("/pulls/55")) return Response.json({ number: 55, state: "open", user: { login: "farmer99" }, head: { sha: "f55" }, mergeable_state: "clean" });
5940+
if (url.includes("/commits/f55/check-runs")) return Response.json({ total_count: 0, check_runs: [] });
5941+
if (url.includes("/commits/f55/status")) return Response.json({ state: "success", statuses: [] });
5942+
if (url.includes("/issues/55/labels") && method === "GET") return Response.json([]);
5943+
if (url.includes("/issues/55/labels") && method === "POST") return Response.json([]);
5944+
if (url.includes("/issues/55/comments") && method === "POST") return Response.json({ id: 1 }, { status: 201 });
5945+
if (url.includes("/issues/55/comments")) return Response.json([]);
5946+
return Response.json({});
5947+
});
5948+
5949+
await processJob(env, {
5950+
type: "github-webhook",
5951+
deliveryId: "contributor-cap-ghost-author",
5952+
eventName: "pull_request",
5953+
payload: {
5954+
action: "opened",
5955+
installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" } },
5956+
repository: { name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } },
5957+
pull_request: { number: 55, title: "Farmer's 3rd PR", state: "open", user: { login: "farmer99" }, head: { sha: "f55" }, labels: [], body: "x", mergeable_state: "clean", reviewDecision: "APPROVED" },
5958+
},
5959+
});
5960+
5961+
// Ghost PR's null authorLogin never matches "farmer99" — the count is still exactly 3 (farmer99's own).
5962+
expect(seen.closed).toBe(true);
5963+
});
5964+
5965+
it("contributor open-PR cap (#2270): out-of-order webhook delivery wakes and self-corrects the missed sibling (regression, gate finding on #2479)", async () => {
5966+
// PR56 (the NEWER PR) is delivered BEFORE PR55 exists in the DB — a real possibility under concurrent/
5967+
// retried webhook delivery. At that moment PR56 only sees {54, 56} (2 total, AT the cap of 2, not over),
5968+
// so it correctly stays open — but a naive "only ever check myself" implementation would leave it open
5969+
// FOREVER, since nothing else ever re-evaluates PR56 again. This pins the fix: once PR55's delivery later
5970+
// sees the COMPLETE set {54, 55, 56}, it must wake PR56 (not just decide for itself) so PR56 gets a fresh,
5971+
// fully-gated re-evaluation and self-corrects.
5972+
const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() });
5973+
await upsertInstallation(env, {
5974+
installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" }, target_type: "User", repository_selection: "all", permissions: { metadata: "read", pull_requests: "write", issues: "write" }, events: ["pull_request"] },
5975+
repositories: [{ name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } }],
5976+
});
5977+
await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 54, title: "Farmer PR zero", state: "open", user: { login: "farmer99" }, head: { sha: "f54" }, labels: [], body: "w" });
5978+
await upsertRepositorySettings(env, {
5979+
repoFullName: "JSONbored/gittensory",
5980+
commentMode: "all_prs",
5981+
publicSurface: "comment_only",
5982+
checkRunMode: "off",
5983+
gateCheckMode: "enabled",
5984+
aiReviewMode: "advisory",
5985+
autonomy: { close: "auto", label: "auto" },
5986+
contributorOpenPrCap: 2,
5987+
});
5988+
const closedNumbers = new Set<number>();
5989+
const fanned: import("../../src/types").JobMessage[] = [];
5990+
const realSend = env.JOBS.send.bind(env.JOBS);
5991+
env.JOBS.send = (async (message: import("../../src/types").JobMessage, options?: QueueSendOptions) => {
5992+
if (message.type === "agent-regate-pr") fanned.push(message);
5993+
return realSend(message, options);
5994+
}) as typeof env.JOBS.send;
5995+
vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => {
5996+
const url = input.toString();
5997+
const method = init?.method ?? "GET";
5998+
if (url === "https://api.gittensor.io/miners") return Response.json([]);
5999+
if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" });
6000+
for (const [n, sha] of [[55, "f55"], [56, "f56"]] as const) {
6001+
if (url.includes(`/pulls/${n}/files`)) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+const ok = true;" }]);
6002+
if (url.includes(`/pulls/${n}/reviews`)) return Response.json([]);
6003+
if (url.includes(`/pulls/${n}/commits`)) return Response.json([]);
6004+
if (url.endsWith(`/pulls/${n}`) && method === "PATCH") { closedNumbers.add(n); return Response.json({ number: n, state: "closed" }); }
6005+
if (url.endsWith(`/pulls/${n}`)) return Response.json({ number: n, state: closedNumbers.has(n) ? "closed" : "open", user: { login: "farmer99" }, head: { sha }, mergeable_state: "clean" });
6006+
if (url.includes(`/commits/${sha}/check-runs`)) return Response.json({ total_count: 0, check_runs: [] });
6007+
if (url.includes(`/commits/${sha}/status`)) return Response.json({ state: "success", statuses: [] });
6008+
if (url.includes(`/issues/${n}/labels`) && method === "GET") return Response.json([]);
6009+
if (url.includes(`/issues/${n}/labels`) && method === "POST") return Response.json([]);
6010+
if (url.includes(`/issues/${n}/comments`) && method === "POST") return Response.json({ id: 1 }, { status: 201 });
6011+
if (url.includes(`/issues/${n}/comments`)) return Response.json([]);
6012+
}
6013+
return Response.json({});
6014+
});
6015+
6016+
// PR56 arrives FIRST — PR55 does not exist yet, so PR56 sees only {54, 56}: at the cap, not over.
6017+
await processJob(env, {
6018+
type: "github-webhook",
6019+
deliveryId: "burst-pr56-first",
6020+
eventName: "pull_request",
6021+
payload: {
6022+
action: "opened",
6023+
installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" } },
6024+
repository: { name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } },
6025+
pull_request: { number: 56, title: "Farmer PR two (out of order)", state: "open", user: { login: "farmer99" }, head: { sha: "f56" }, labels: [], body: "y", mergeable_state: "clean", reviewDecision: "APPROVED" },
6026+
},
6027+
});
6028+
expect(closedNumbers.has(56)).toBe(false); // correctly not closed YET — the set looked complete at the time
6029+
6030+
// PR55 arrives SECOND — now the complete set {54, 55, 56} is visible. PR55 itself ranks within the cap
6031+
// (oldest 2 of 3), so it stays open — but PR56 is now discoverably over-cap and must be woken.
6032+
await processJob(env, {
6033+
type: "github-webhook",
6034+
deliveryId: "burst-pr55-second",
6035+
eventName: "pull_request",
6036+
payload: {
6037+
action: "opened",
6038+
installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" } },
6039+
repository: { name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } },
6040+
pull_request: { number: 55, title: "Farmer PR one", state: "open", user: { login: "farmer99" }, head: { sha: "f55" }, labels: [], body: "x", mergeable_state: "clean", reviewDecision: "APPROVED" },
6041+
},
6042+
});
6043+
expect(closedNumbers.has(55)).toBe(false); // PR55 itself is within the cap
6044+
expect(fanned.some((job) => job.type === "agent-regate-pr" && job.prNumber === 56)).toBe(true); // sibling woken
6045+
6046+
// Drain the woken job — PR56's OWN fresh re-evaluation now sees the complete set and self-corrects.
6047+
env.JOBS.send = realSend;
6048+
for (const job of fanned) await processJob(env, job);
6049+
expect(closedNumbers.has(56)).toBe(true);
6050+
});
6051+
6052+
it("contributor open-PR cap (#2270): a re-delivered sibling-wake is coalesced — the second discovery does not re-enqueue", async () => {
6053+
const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() });
6054+
await upsertInstallation(env, {
6055+
installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" }, target_type: "User", repository_selection: "all", permissions: { metadata: "read", pull_requests: "write", issues: "write" }, events: ["pull_request"] },
6056+
repositories: [{ name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } }],
6057+
});
6058+
// Pre-seed the coalescing key for PR56 exactly as wakeOverCapSiblingPullRequests itself would after a
6059+
// first, already-successful enqueue — proving the SECOND discovery within the window skips re-enqueueing.
6060+
await env.SELFHOST_TRANSIENT_CACHE?.set("contributor-cap-wake:jsonbored/gittensory#56", "1", 60);
6061+
await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 54, title: "Farmer PR zero", state: "open", user: { login: "farmer99" }, head: { sha: "f54" }, labels: [], body: "w" });
6062+
await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 56, title: "Farmer PR two (already over cap)", state: "open", user: { login: "farmer99" }, head: { sha: "f56" }, labels: [], body: "y" });
6063+
await upsertRepositorySettings(env, {
6064+
repoFullName: "JSONbored/gittensory",
6065+
commentMode: "all_prs",
6066+
publicSurface: "comment_only",
6067+
checkRunMode: "off",
6068+
gateCheckMode: "enabled",
6069+
aiReviewMode: "advisory",
6070+
autonomy: { close: "auto", label: "auto" },
6071+
contributorOpenPrCap: 2,
6072+
});
6073+
const fanned: import("../../src/types").JobMessage[] = [];
6074+
const realSend = env.JOBS.send.bind(env.JOBS);
6075+
env.JOBS.send = (async (message: import("../../src/types").JobMessage, options?: QueueSendOptions) => {
6076+
if (message.type === "agent-regate-pr") fanned.push(message);
6077+
return realSend(message, options);
6078+
}) as typeof env.JOBS.send;
6079+
vi.stubGlobal("fetch", async (input: RequestInfo | URL) => {
6080+
const url = input.toString();
6081+
if (url === "https://api.gittensor.io/miners") return Response.json([]);
6082+
if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" });
6083+
if (url.includes("/pulls/55/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+const ok = true;" }]);
6084+
if (url.includes("/pulls/55/reviews")) return Response.json([]);
6085+
if (url.includes("/pulls/55/commits")) return Response.json([]);
6086+
if (url.endsWith("/pulls/55")) return Response.json({ number: 55, state: "open", user: { login: "farmer99" }, head: { sha: "f55" }, mergeable_state: "clean" });
6087+
if (url.includes("/commits/f55/check-runs")) return Response.json({ total_count: 0, check_runs: [] });
6088+
if (url.includes("/commits/f55/status")) return Response.json({ state: "success", statuses: [] });
6089+
if (url.includes("/issues/55/labels")) return Response.json([]);
6090+
if (url.includes("/issues/55/comments")) return Response.json([]);
6091+
return Response.json({});
6092+
});
6093+
6094+
// PR55 arrives and independently discovers PR56 is over cap — but the wake was already claimed.
6095+
await processJob(env, {
6096+
type: "github-webhook",
6097+
deliveryId: "wake-coalesce-second",
6098+
eventName: "pull_request",
6099+
payload: {
6100+
action: "opened",
6101+
installation: { id: 123, account: { login: "JSONbored", id: 1, type: "User" } },
6102+
repository: { name: "gittensory", full_name: "JSONbored/gittensory", private: false, owner: { login: "JSONbored" } },
6103+
pull_request: { number: 55, title: "Farmer PR one", state: "open", user: { login: "farmer99" }, head: { sha: "f55" }, labels: [], body: "x", mergeable_state: "clean", reviewDecision: "APPROVED" },
6104+
},
6105+
});
6106+
6107+
expect(fanned).toEqual([]); // coalesced — no duplicate wake enqueued
6108+
});
6109+
59086110
// #1092: prReadyForReview rebases a BEHIND-base PR through the agent executor (gated by update_branch autonomy
59096111
// + pull_requests:write) before reviewing, then defers — the synchronize on the new head re-runs review.
59106112
async function seedBehindRepo(env: Env, over: { autonomy?: Record<string, string>; agentPaused?: boolean; perms?: Record<string, string>; noInstall?: boolean } = {}) {

0 commit comments

Comments
 (0)