From 7edb163a80d1aca60e6c780c9472befa523fcd5a Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:49:17 -0500 Subject: [PATCH] fix(server): cancel merge observation HTTP requests on shutdown Bound GitHub object and repository probes, keep readiness independent of external reads, and page pending cards through the owning parser. Co-Authored-By: Paperclip --- ...026-10-30-merged-pr-confirmations-sleep.md | 34 ++++++- .../external-objects-service.test.ts | 9 +- .../merged-pr-confirmation-work.test.ts | 88 +++++++++++++++++++ .../server-startup-feedback-export.test.ts | 17 +++- server/src/index.ts | 4 +- server/src/services/external-objects.ts | 1 + .../github-external-object-provider.ts | 18 ++-- .../src/services/github-pull-request-merge.ts | 17 ++-- .../src/services/issue-thread-interactions.ts | 38 +++++--- .../services/merged-pr-confirmation-work.ts | 14 ++- 10 files changed, 200 insertions(+), 40 deletions(-) diff --git a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md index fb8b091f841..7b8699456fb 100644 --- a/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md +++ b/doc/plans/2026-10-30-merged-pr-confirmations-sleep.md @@ -13,10 +13,12 @@ it receives no GitHub payloads. No database migration is required. ## Durable truth and producers Pending `request_confirmation` interaction rows remain the durable work source. -The candidate query excludes terminal and hidden issues, linked tool actions, -and text without merge intent or a PR reference. The existing fail-closed parser -then excludes additional actions, governed actions, and document/attachment targets. -No accepted or cancelled history keeps a deadline alive. +The candidate query pages pending cards in bounded groups of 100 and excludes +terminal and hidden issues and linked tool actions. The existing fail-closed +parser is the sole text eligibility source: it requires merge intent and a PR +reference and excludes additional actions, governed actions, and document/attachment targets. +No accepted or cancelled history keeps a deadline alive. A full page without any +merge candidates still advances its keyset cursor; later candidates are not lost. Ordinary creation registers the topic before inserting a confirmation in its transaction. Existing confirmation-resolution notifications invalidate its issue @@ -53,6 +55,14 @@ This is candidate-backed external observation. It does not eliminate all polling register a Cloud wake alarm, or prove that the host/database suspends. Existing sleep-safety blockers remain in force. +Startup registers this observation without waiting for live GitHub requests. +The internal resolver invocation accepts an optional host-owned abort signal; +that signal is not part of external-object records or plugin/RPC payloads. +The GitHub owner combines cancellation with a ten-second request budget and +passes it to the actual object fetch and repository-visibility probes, including +response body reads. Shutdown aborts the request and waits for its cancellation; +no post-abort acceptance is committed. A timeout leaves the durable card pending. + ## Acceptance evidence Focused tests use real PostgreSQL, the actual service producers, transaction @@ -68,6 +78,22 @@ provider failure across drain; standby/resume; restart; tenant-scoped and duplic merge hints; latest/superseded cards during provider observation; hidden/terminal issues; URL targets; no-continuation policy; and shutdown during an in-flight read. Existing startup and normalized GitHub event tests exercise the changed consumers. +Controlled local HTTP tests hold headers, partial response bodies, 404 probes, +and public-visibility probes. They observe server-side response close after +shutdown within one second, rather than only racing a promise. A shortened +request budget times out a real request; the saved card then accepts after +restart when that same HTTP fixture serves a merged response. Existing GitHub +provider and external-object service tests preserve their authorization behavior. + +## Verification follow-up + +The original head passed full-workspace typecheck and build. A later required +Linux CI shard failed the unchanged managed-skill maintenance CLI fixture in +`managed-skill-content.test.ts`: it returned `unavailable_or_corrupt_package` +where `local_storage` was expected. Both the isolated case and the complete +unchanged 17-test file pass against fresh local PostgreSQL. The Linux cause is +not established; this is retained as an unresolved CI failure, not called a flake. +The current-head required CI result remains the completion gate. ## Existing limitations outside this migration diff --git a/server/src/__tests__/external-objects-service.test.ts b/server/src/__tests__/external-objects-service.test.ts index e9022599779..af3b4667deb 100644 --- a/server/src/__tests__/external-objects-service.test.ts +++ b/server/src/__tests__/external-objects-service.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; @@ -26,7 +27,8 @@ import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -709,8 +711,8 @@ describeEmbeddedPostgres("externalObjectService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-"); + db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 }); }, 20_000); afterEach(async () => { @@ -725,6 +727,7 @@ describeEmbeddedPostgres("externalObjectService", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/merged-pr-confirmation-work.test.ts b/server/src/__tests__/merged-pr-confirmation-work.test.ts index 51195e6157b..95565fd96e7 100644 --- a/server/src/__tests__/merged-pr-confirmation-work.test.ts +++ b/server/src/__tests__/merged-pr-confirmation-work.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; +import { createPullRequestMergeStateResolver } from "../services/github-pull-request-merge.js"; import { randomUUID } from "node:crypto"; import { setTimeout as delay } from "node:timers/promises"; import { eq, sql } from "drizzle-orm"; @@ -262,6 +264,92 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); }); + it.each(["headers", "body", "not-found probe", "public visibility probe"] as const)("cancels an actual unresponsive HTTP %s read promptly at stop", async phase => { + vi.useRealTimers(); + const card = await create(); + let observed!: () => void, closed!: () => void; + const requested = new Promise(resolve => { observed = resolve; }); + const disconnected = new Promise(resolve => { closed = resolve; }); + const signals: AbortSignal[] = []; + const server = createServer((request, response) => { + if (phase === "not-found probe" && request.url?.includes("/pulls/")) { response.writeHead(404).end(); return; } + response.on("close", closed); + if (phase === "body") { response.writeHead(200, { "content-type": "application/json" }); response.write('{"merged":'); } + observed(); + }); + await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Expected test HTTP address"); + const resolveState = createPullRequestMergeStateResolver(db, { + requestTimeoutMs: 10_000, + credentialProvider: () => phase === "public visibility probe" ? { token: "fixture", source: "server_env", secretName: null } : null, + fetch: (url, init) => { signals.push(init!.signal!); return fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init); }, + }); + const real = issueThreadInteractionService(db, { resolvePullRequestState: resolveState, wakeup }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const startupAt = performance.now(); + const worker = registerMergeConfirmationWork(coordinator, real, () => true); + expect(performance.now() - startupAt).toBeLessThan(100); + await Promise.race([requested, delay(2000).then(() => { throw new Error("HTTP observation did not start"); })]); + const stoppingAt = performance.now(); + await coordinator.stop(); coordinator = undefined; + expect(performance.now() - stoppingAt).toBeLessThan(1000); + await worker.ready; + await Promise.race([disconnected, delay(1000).then(() => { throw new Error("HTTP request remained open after stop"); })]); + expect(signals.at(-1)?.aborted).toBe(true); + expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled(); expect(errors).toEqual([]); + } finally { + await coordinator?.stop(); coordinator = undefined; + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); + } + }); + + it("times out a real unresponsive HTTP read and recovers its durable candidate on restart", async () => { + vi.useRealTimers(); const card = await create(); + let serveMerge = false, closed!: () => void; + const disconnected = new Promise(resolve => { closed = resolve; }); + const server = createServer((_request, response) => { + if (serveMerge) { response.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify({ merged: true, state: "closed" })); return; } + response.on("close", closed); + }); + await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected test HTTP address"); + const real = issueThreadInteractionService(db, { resolvePullRequestState: createPullRequestMergeStateResolver(db, { + requestTimeoutMs: 100, tokenProvider: () => null, + fetch: (url, init) => fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init), + }), wakeup }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const startedAt = performance.now(); + await registerMergeConfirmationWork(coordinator, real, () => true).ready; + expect(performance.now() - startedAt).toBeLessThan(2000); + await Promise.race([disconnected, delay(1000).then(() => { throw new Error("Timed out HTTP request remained open"); })]); + expect(errors).toHaveLength(1); expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled(); + expect(coordinator.nextWakeAt()).not.toBeNull(); + await coordinator.stop(); coordinator = undefined; serveMerge = true; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerMergeConfirmationWork(coordinator, real, () => true).ready; + expect((await stored(card.id)).status).toBe("accepted"); expect(wakeup).toHaveBeenCalledTimes(1); + } finally { + await coordinator?.stop(); coordinator = undefined; + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); + } + }); + + it("pages beyond a full pending-card page without duplicating the merge parser in SQL", async () => { + const card = await create(); + const id = BigInt(`0x${card.id.replaceAll("-", "")}`); + await db.insert(issueThreadInteractions).values(Array.from({ length: 100 }, (_, index) => { + const hex = (id - BigInt(index + 1)).toString(16).padStart(32, "0"); + return { id: `${hex.slice(0,8)}-${hex.slice(8,12)}-${hex.slice(12,16)}-${hex.slice(16,20)}-${hex.slice(20)}`, + companyId, issueId, kind: "request_confirmation", title: "Approve the plan?", payload: { version: 1 as const, prompt: "Approve the plan?" } }; + })); + await start(); expect(provider).not.toHaveBeenCalled(); + await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet(); + expect(provider).toHaveBeenCalledTimes(1); expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + }); + it("limits normalized merge hints to the owning company and matching PR and tolerates duplicates", async () => { provider.mockResolvedValue("open"); const own = await create(); const foreignCompany = randomUUID(), foreignIssue = randomUUID(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 9fc7234b2e6..4e7e3562156 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -180,6 +180,7 @@ const { const heartbeatServiceFactoryMock = vi.fn(() => heartbeatServiceMock); const issueThreadInteractionServiceMock = { sweepSupersededPendingRequestConfirmations: vi.fn(async () => ({ expired: 0 })), + mergedPullRequestConfirmationPage: vi.fn(async () => ({ rows: [], complete: true })), mergedPullRequestConfirmationCandidates: vi.fn(async () => []), sweepMergedPullRequestConfirmations: vi.fn(async () => ({ checked: 0, @@ -1335,7 +1336,7 @@ describe("startServer feedback export wiring", () => { expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); - expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationCandidates).toHaveBeenCalledTimes(1); + expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalledTimes(1); expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).not.toHaveBeenCalled(); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.mergedPullRequestConfirmations)).toBe(true); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); @@ -1424,6 +1425,20 @@ describe("startServer feedback export wiring", () => { } }); + it("becomes ready without waiting for the merge-confirmation observation", async () => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); + let release!: () => void; + const blocked = new Promise<{ rows: never[]; complete: boolean }>(resolve => { release = () => resolve({ rows: [], complete: true }); }); + issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage.mockReturnValueOnce(blocked); + const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(() => 1 as unknown as ReturnType); + let started = false; + const starting = startServer().then(server => { started = true; return server; }); + try { + await vi.waitFor(() => expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalled()); + await settleStartupHousekeeping(); expect(started).toBe(true); expect(getStartupRecoveryState().phase).toBe("ready"); + } finally { release(); await starting; timer.mockRestore(); } + }); + it.each(["retained cleanup", "workspace finalization", "environment settlement", "native run projection", "native result commit", "native recovery notices", "external conversation settlement"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); let releaseCleanup!: () => void; diff --git a/server/src/index.ts b/server/src/index.ts index c1804998ef5..86074131dc3 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1601,8 +1601,8 @@ async function startServerWithDatabaseTeardown( : config.workspaceCompactionDelayHours * 60 * 60 * 1000, terminalWorkspaceInspectionBackoffMs: TERMINAL_WORKSPACE_RETRY_MS, }); - await registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations, - () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); const gitRestorePointWorker = registerGitRestorePointWork(deliveryWork, gitRestorePointRetentionService(db as any, { storage: storageService, retentionDays: config.gitRestorePointRetentionDays, diff --git a/server/src/services/external-objects.ts b/server/src/services/external-objects.ts index 56a8fb91834..306f043d753 100644 --- a/server/src/services/external-objects.ts +++ b/server/src/services/external-objects.ts @@ -84,6 +84,7 @@ export interface ExternalObjectResolver { resolve(input: { companyId: string; object: ExternalObjectRecord; + signal?: AbortSignal; }): Promise; } diff --git a/server/src/services/github-external-object-provider.ts b/server/src/services/github-external-object-provider.ts index 6dedac97efb..83b47354035 100644 --- a/server/src/services/github-external-object-provider.ts +++ b/server/src/services/github-external-object-provider.ts @@ -15,6 +15,7 @@ type FetchLike = (url: string, init?: RequestInit) => Promise; export interface GitHubExternalObjectProviderOptions { fetch?: FetchLike; + requestTimeoutMs?: number; tokenProvider?: (companyId: string) => Promise | string | null; /** * Resolve the credential and its source. Prefer this over `tokenProvider` when the @@ -243,12 +244,12 @@ type RepoProbeOutcome = { async function probeRepoVisibility( identity: GitHubObjectIdentity, - opts: { fetch: FetchLike; headers?: Record | undefined }, + opts: { fetch: FetchLike; headers?: Record | undefined; signal?: AbortSignal }, ): Promise { try { const probe = await opts.fetch( `${gitHubApiBase(identity.host)}/repos/${encodeURIComponent(identity.owner)}/${encodeURIComponent(identity.repo)}`, - { headers: opts.headers }, + { headers: opts.headers, signal: opts.signal }, ); if (probe.ok) { const body = await safeJson(probe); @@ -427,12 +428,13 @@ export function createGitHubExternalObjectProvider( companyId: string, identity: GitHubObjectIdentity, headers: Record, + signal: AbortSignal, ): Promise { const credentialFingerprint = createHash("sha256").update(headers.authorization ?? "anonymous").digest("hex"); const key = `${companyId}\u0000${credentialFingerprint}\u0000${identity.owner.toLowerCase()}\u0000${identity.repo.toLowerCase()}`; const cached = probeCache.get(key); if (cached && cached.expiresAt > Date.now()) return cached.outcome; - const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (outcome.visibility === "visible" || outcome.visibility === "hidden") { probeCache.set(key, { outcome, expiresAt: Date.now() + GITHUB_OBJECT_TTL_SECONDS * 1000 }); } @@ -464,7 +466,9 @@ export function createGitHubExternalObjectProvider( return { providerKey: "github", objectType, - async resolve({ companyId, object }) { + async resolve({ companyId, object, signal: ownerSignal }) { + const deadline = AbortSignal.timeout(opts.requestTimeoutMs ?? 10_000); + const signal = ownerSignal ? AbortSignal.any([ownerSignal, deadline]) : deadline; const identity = parseGitHubObject(object); if (!identity || identity.objectType !== objectType) { return { @@ -509,7 +513,7 @@ export function createGitHubExternalObjectProvider( // "public" verdict must not authorize a later read of its private contents. // The cache is still used below to classify a 404, where a stale verdict can // only make the object look deleted — it can never expose private data. - const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers }); + const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal }); if (probe.visibility === "hidden") { return { ok: false, @@ -554,7 +558,7 @@ export function createGitHubExternalObjectProvider( let response: Response; try { - response = await fetchImpl(url, { headers }); + response = await fetchImpl(url, { headers, signal }); } catch { return { ok: false, @@ -573,7 +577,7 @@ export function createGitHubExternalObjectProvider( // Probe the repository metadata to classify: if the repo itself is visible // the object truly is gone; if it is not, surface credentials-needed // instead of permanently freezing a reference as "Not found". - const probe = await cachedProbeRepoVisibility(companyId, identity, headers); + const probe = await cachedProbeRepoVisibility(companyId, identity, headers, signal); if (probe.visibility === "visible") { return { ok: true, snapshot: notFoundSnapshot(identity, etag) }; } diff --git a/server/src/services/github-pull-request-merge.ts b/server/src/services/github-pull-request-merge.ts index bb92d37d7c2..1cdff01dc1f 100644 --- a/server/src/services/github-pull-request-merge.ts +++ b/server/src/services/github-pull-request-merge.ts @@ -1,5 +1,5 @@ import type { Db } from "@paperclipai/db"; -import { createGitHubExternalObjectProvider } from "./github-external-object-provider.js"; +import { createGitHubExternalObjectProvider, type GitHubExternalObjectProviderOptions } from "./github-external-object-provider.js"; export type GitHubPullRequestReference = { host: "github.com"; @@ -25,11 +25,13 @@ export type PullRequestMergeDetails = { export type PullRequestMergeStateResolver = ( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ) => Promise; export type PullRequestMergeDetailsResolver = ( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ) => Promise; export const PULL_REQUEST_CACHE_MAX_ENTRIES = 1_000; @@ -86,14 +88,15 @@ function readRecord(value: unknown): Record | null { : null; } -export function createPullRequestMergeDetailsResolver(db: Db): PullRequestMergeDetailsResolver { - const resolver = createGitHubExternalObjectProvider(db).resolvers +export function createPullRequestMergeDetailsResolver(db: Db, options?: GitHubExternalObjectProviderOptions): PullRequestMergeDetailsResolver { + const resolver = createGitHubExternalObjectProvider(db, options).resolvers .find((candidate) => candidate.objectType === "pull_request") ?? null; - return async (companyId, reference) => { + return async (companyId, reference, signal) => { if (!resolver) return { state: "unknown", headRef: null, headSha: null }; const result = await resolver.resolve({ companyId, + signal, object: { externalId: `${reference.owner}/${reference.repo}#pull/${reference.number}`, sanitizedCanonicalUrl: `https://github.com/${reference.owner}/${reference.repo}/pull/${reference.number}`, @@ -123,7 +126,7 @@ export function createPullRequestMergeDetailsResolver(db: Db): PullRequestMergeD }; } -export function createPullRequestMergeStateResolver(db: Db): PullRequestMergeStateResolver { - const resolveDetails = createPullRequestMergeDetailsResolver(db); - return async (companyId, reference) => (await resolveDetails(companyId, reference)).state; +export function createPullRequestMergeStateResolver(db: Db, options?: GitHubExternalObjectProviderOptions): PullRequestMergeStateResolver { + const resolveDetails = createPullRequestMergeDetailsResolver(db, options); + return async (companyId, reference, signal) => (await resolveDetails(companyId, reference, signal)).state; } diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index b306f637e27..b49ca48d5c1 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -204,6 +204,7 @@ export type IssueThreadInteractionServiceOptions = { resolvePullRequestState?: ( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ) => Promise; wakeup?: InteractionWakeup; pullRequestCacheTtlMs?: number; @@ -2045,10 +2046,11 @@ export function issueThreadInteractionService( async function resolvePullRequestState( companyId: string, reference: GitHubPullRequestReference, + signal?: AbortSignal, ): Promise { if (opts.resolvePullRequestState) - return opts.resolvePullRequestState(companyId, reference); - return defaultPullRequestStateResolver?.(companyId, reference) ?? "unknown"; + return opts.resolvePullRequestState(companyId, reference, signal); + return defaultPullRequestStateResolver?.(companyId, reference, signal) ?? "unknown"; } async function resolvePullRequestStates( @@ -2057,6 +2059,7 @@ export function issueThreadInteractionService( companyId: string; reference: GitHubPullRequestReference; }>, + signal?: AbortSignal, ) { const states = new Map(); const pending = entries.slice(); @@ -2064,11 +2067,13 @@ export function issueThreadInteractionService( await Promise.all( Array.from({ length: workerCount }, async () => { while (pending.length > 0) { + signal?.throwIfAborted(); const entry = pending.shift(); if (!entry) return; const state = await resolvePullRequestState( entry.companyId, entry.reference, + signal, ); states.set(entry.key, state); } @@ -2076,11 +2081,7 @@ export function issueThreadInteractionService( ); return states; } - async function mergedPullRequestConfirmationCandidates(scope?: MergeConfirmationScope) { - const text = sql`concat_ws(' ', ${issueThreadInteractions.title}, ${issueThreadInteractions.summary}, - ${issueThreadInteractions.payload}->>'prompt', ${issueThreadInteractions.payload}->>'acceptLabel', - ${issueThreadInteractions.payload}->>'detailsMarkdown', ${issueThreadInteractions.payload}->'target'->>'key', - ${issueThreadInteractions.payload}->'target'->>'label', ${issueThreadInteractions.payload}->'target'->>'href')`; + async function mergedPullRequestConfirmationPage(scope?: MergeConfirmationScope, after?: string) { const rows = await db.select({ interaction: issueThreadInteractions, issue: { id: issues.id, companyId: issues.companyId, projectId: issues.projectId, goalId: issues.goalId, status: issues.status, assigneeAgentId: issues.assigneeAgentId, @@ -2092,14 +2093,24 @@ export function issueThreadInteractionService( scope ? eq(issueThreadInteractions.companyId, scope.companyId) : undefined, scope?.issueId ? eq(issueThreadInteractions.issueId, scope.issueId) : undefined, scope?.interactionId ? eq(issueThreadInteractions.id, scope.interactionId) : undefined, - sql`${text} ~* '(merge|merging)'`, - sql`${text} ~* '(github\\.com/[^ /]+/[^ /]+/pull/[1-9][0-9]*|[a-z0-9_.-]+/[a-z0-9_.-]+#[1-9][0-9]*)'`, + after ? sql`${issueThreadInteractions.id} > ${after}::uuid` : undefined, sql`NOT EXISTS (SELECT 1 FROM ${toolActionRequests} WHERE ${toolActionRequests.interactionId} = ${issueThreadInteractions.id})`, - )); - return rows.flatMap(row => { + )).orderBy(asc(issueThreadInteractions.id)).limit(100); + return { rows: rows.flatMap(row => { const references = getMergeConfirmationPullRequestReferences(row.interaction); return references.length ? [{ ...row, references }] : []; - }); + }), after: rows.at(-1)?.interaction.id, complete: rows.length < 100 }; + } + + async function mergedPullRequestConfirmationCandidates(scope?: MergeConfirmationScope) { + const candidates: Awaited>["rows"] = []; + let after: string | undefined; + while (true) { + const page = await mergedPullRequestConfirmationPage(scope, after); + candidates.push(...page.rows); + if (page.complete) return candidates; + after = page.after; + } } async function getIdempotentInteraction(args: { @@ -3012,6 +3023,7 @@ export function issueThreadInteractionService( await emitInteractionResolvedTelemetry(db, interaction); return interaction; }, + mergedPullRequestConfirmationPage, mergedPullRequestConfirmationCandidates, sweepMergedPullRequestConfirmations: async ( mergedHints: Array<{ @@ -3064,7 +3076,7 @@ export function issueThreadInteractionService( const refreshedStates = await resolvePullRequestStates([ ...uniqueReferences.values(), - ]); + ], signal); if (scope && [...refreshedStates.values()].some(state => state === "unknown")) { throw new Error("Merge confirmation provider state is unavailable"); } diff --git a/server/src/services/merged-pr-confirmation-work.ts b/server/src/services/merged-pr-confirmation-work.ts index bfccaeea79f..6a754d5a88a 100644 --- a/server/src/services/merged-pr-confirmation-work.ts +++ b/server/src/services/merged-pr-confirmation-work.ts @@ -9,16 +9,24 @@ export const MERGE_CONFIRMATION_RECONCILE_MS = 5 * 60_000; export function registerMergeConfirmationWork( coordinator: ReturnType, service: Pick, - "mergedPullRequestConfirmationCandidates" | "sweepMergedPullRequestConfirmations">, + "mergedPullRequestConfirmationCandidates" | "mergedPullRequestConfirmationPage" | "sweepMergedPullRequestConfirmations">, canRun: () => boolean, ) { + const cursors = new Map(); return registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.mergedPullRequestConfirmations, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as MergeConfirmationScope, matches: (entry, scope?: MergeConfirmationScope) => !scope || (entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.interactionId || entry.id === scope.interactionId)), - load: async (scope?: MergeConfirmationScope) => (await service.mergedPullRequestConfirmationCandidates(scope)) - .map(({ interaction }) => ({ id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: Date.now() })), + load: async (scope?: MergeConfirmationScope) => { + const key = scope ? JSON.stringify(scope) : "*"; + const page = await service.mergedPullRequestConfirmationPage(scope, cursors.get(key)); + if (page.complete) cursors.delete(key); + else cursors.set(key, page.after!); + return { complete: page.complete, rows: page.rows.map(({ interaction }) => ({ + id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: Date.now(), + })) }; + }, check: async (entry, signal) => { const scope = { companyId: entry.companyId, issueId: entry.issueId, interactionId: entry.id }; await service.sweepMergedPullRequestConfirmations([], scope, signal);