diff --git a/packages/control-plane/src/github/api.ts b/packages/control-plane/src/github/api.ts index 30e1e9432..9468222e3 100644 --- a/packages/control-plane/src/github/api.ts +++ b/packages/control-plane/src/github/api.ts @@ -62,8 +62,32 @@ export interface GithubRequestOpts { bigIdsAsStrings?: boolean } +/** One page of a paginated GitHub list: the body plus the `rel="next"` cursor. */ +export interface GithubPage { + data: T + /** Next page as a path under the SAME base — a `next` pointing anywhere else + * is dropped rather than followed. */ + nextPath?: string +} + +/** `rel="next"` from a Link header, relative to `baseUrl`. */ +function nextPathFrom(link: string | null, baseUrl: string): string | undefined { + if (!link) return undefined + for (const part of link.split(',')) { + const url = /<([^>]+)>\s*;\s*rel="next"/.exec(part.trim())?.[1] + if (url?.startsWith(baseUrl)) return url.slice(baseUrl.length) + } + return undefined +} + /** One GitHub REST call → parsed JSON. Throws `GithubApiError` on any non-2xx. */ export async function githubRequest(path: string, opts: GithubRequestOpts): Promise { + return (await githubRequestPage(path, opts)).data +} + +/** {@link githubRequest} keeping the pagination cursor — for the list endpoints + * whose first page is not the whole answer. */ +export async function githubRequestPage(path: string, opts: GithubRequestOpts): Promise> { const fetchImpl = opts.fetchImpl ?? (fetch as FetchLike) let res: Response try { @@ -83,11 +107,12 @@ export async function githubRequest(path: string, opts: GithubRequestOpts): P } if (res.ok) { const text = await res.text() - if (!text) return undefined as T // 202-style empty success (e.g. redelivery accepted) + const nextPath = nextPathFrom(res.headers.get('link'), opts.baseUrl ?? API_BASE) + if (!text) return { data: undefined as T } // 202-style empty success (e.g. redelivery accepted) // Only `id` keys with ≥15 digits are re-quoted — small numeric ids // (repositories, installations) stay numbers for existing callers. const safe = opts.bigIdsAsStrings ? text.replace(/"id"\s*:\s*(\d{15,})/g, '"id":"$1"') : text - return JSON.parse(safe) as T + return { data: JSON.parse(safe) as T, ...(nextPath ? { nextPath } : {}) } } // Read the message for diagnostics; GitHub error bodies carry no secrets. diff --git a/packages/control-plane/src/github/github.test.ts b/packages/control-plane/src/github/github.test.ts index 5c41c11f3..af3e48ea7 100644 --- a/packages/control-plane/src/github/github.test.ts +++ b/packages/control-plane/src/github/github.test.ts @@ -10,7 +10,7 @@ import type { GitCredCapability } from '@agentconnect.md/protocol' import { FakeClock } from '../../test/fakes/fake-clock.js' import type { AgentRepoAuthorizationRecord, GithubInstallationRecord } from '../persistence/ports.js' import { githubAppBotIdentity, resolveGithubAppConfig, type GithubAppConfig } from './config.js' -import { GithubApiError, githubRequest, mintAppJwt } from './api.js' +import { GithubApiError, githubRequest, mintAppJwt, type FetchLike } from './api.js' import { InstallationTokenInvalidatedError, InstallationTokenService } from './installation-token.service.js' import { deriveInstallStateKey, mintInstallState, verifyInstallState, INSTALL_STATE_TTL_MS } from './install-state.js' import { GithubService } from './service.js' @@ -152,6 +152,90 @@ describe('githubRequest', () => { }) }) +describe('GithubService.listHookDeliveries', () => { + const deliveryPage = (guids: string[], at: string): string => + JSON.stringify( + guids.map((guid) => ({ + id: '1234567890123456789', + guid, + delivered_at: at, + event: 'pull_request', + action: 'opened', + repository_id: 42, + installation_id: 7 + })) + ) + const svc = (fetchImpl: FetchLike) => + new GithubService({ + cfg: cfg(), + clock: new FakeClock(1_700_000_000_000), + installations: {} as never, + installState: { put: async () => {}, consume: async () => true }, + pepper: 'p'.repeat(32), + fetchImpl + }) + + it('walks the cursor until a page reaches past the floor', async () => { + const paths: string[] = [] + const fetchImpl = vi.fn(async (url: string) => { + paths.push(url) + const second = url.includes('cursor=next') + return new Response( + second ? deliveryPage(['old'], '2023-11-14T21:40:00.000Z') : deliveryPage(['new'], '2023-11-14T22:10:00.000Z'), + { + status: 200, + headers: { + 'content-type': 'application/json', + link: '; rel="next"' + } + } + ) + }) + + const page = await svc(fetchImpl).listHookDeliveries({ deliveredSince: new Date('2023-11-14T21:50:00.000Z') }) + expect(page.deliveries.map((d) => d.guid)).toEqual(['new', 'old']) + expect(page.truncated).toBe(false) // the second page reaches past the floor + expect(paths[1]).toContain('cursor=next') + }) + + it('reports truncation when the page budget runs out before the floor', async () => { + const fetchImpl = vi.fn( + async () => + new Response(deliveryPage(['g'], '2023-11-14T22:10:00.000Z'), { + status: 200, + headers: { + 'content-type': 'application/json', + link: '; rel="next"' + } + }) + ) + + const page = await svc(fetchImpl).listHookDeliveries({ + maxPages: 3, + deliveredSince: new Date('2023-11-14T20:00:00.000Z') + }) + expect(fetchImpl).toHaveBeenCalledTimes(3) + expect(page.truncated).toBe(true) + }) + + it('never follows a next cursor that points off the API base', async () => { + const fetchImpl = vi.fn( + async () => + new Response(deliveryPage(['g'], '2023-11-14T22:10:00.000Z'), { + status: 200, + headers: { + 'content-type': 'application/json', + link: '; rel="next"' + } + }) + ) + + const page = await svc(fetchImpl).listHookDeliveries({ deliveredSince: new Date('2023-11-14T20:00:00.000Z') }) + expect(fetchImpl).toHaveBeenCalledTimes(1) + expect(page.truncated).toBe(false) + }) +}) + describe('GithubService comment authorization lookups', () => { const installation: GithubInstallationRecord = { id: 'installation-row', diff --git a/packages/control-plane/src/github/service.ts b/packages/control-plane/src/github/service.ts index c2278d33d..730ab5be4 100644 --- a/packages/control-plane/src/github/service.ts +++ b/packages/control-plane/src/github/service.ts @@ -23,7 +23,7 @@ import type { AgentRepoAuthorizationRepo, RepoAccess } from '../persistence/ports.js' -import { githubRequest, mintAppJwt, GithubApiError, type FetchLike } from './api.js' +import { githubRequest, githubRequestPage, mintAppJwt, GithubApiError, type FetchLike, type GithubPage } from './api.js' import { githubAppBotIdentity, type GithubAppConfig } from './config.js' import { InstallationTokenService, type CapabilityLevels, type MintedGitCred } from './installation-token.service.js' import { deriveInstallStateKey, mintInstallState, verifyInstallState } from './install-state.js' @@ -76,6 +76,15 @@ export interface GhHookDelivery { installation_id: number | null } +/** A delivery listing plus how honest it is about its own reach. */ +export interface GhHookDeliveryPage { + /** Newest first, across every page walked. */ + deliveries: GhHookDelivery[] + /** The walk stopped on its page budget with a cursor left — the caller has + * NOT seen everything back to `deliveredSince`, and must not claim it did. */ + truncated: boolean +} + export interface GithubServiceDeps { cfg: GithubAppConfig clock: Clock @@ -100,6 +109,8 @@ export interface GithubServiceDeps { const OUTDATED_INSTALLATIONS_CACHE_MS = 30_000 const REPO_PAGE_CACHE_MS = 5 * 60_000 const MAX_REPO_PAGE_CACHE_ENTRIES = 1_000 +/** Delivery-list pages one sweep may walk — the firehose bound, not the window. */ +const MAX_DELIVERY_PAGES = 10 type RepoPageLookup = { ins: GithubInstallationRecord; page: number; perPage: number } @@ -469,13 +480,29 @@ export class GithubService { /** * The App's recent webhook deliveries, newest first (P2.5 redelivery - * reconciliation). ONE page of `perPage` — the deliveries cursor rides a Link - * header `githubRequest` doesn't surface; the reconciler logs when the page - * is full so an under-covered window is visible, never silent. + * reconciliation). Follows the Link cursor until `deliveredSince` is reached + * — one page is a few minutes of traffic on a busy App, far less than the + * reconciler's look-back window — and stops at `maxPages` so a firehose + * bounds the sweep instead of the sweep bounding itself. The caller sees how + * far back the result actually reaches and reports the shortfall. */ - async listHookDeliveries(perPage = 100): Promise { + async listHookDeliveries( + opts: { perPage?: number; maxPages?: number; deliveredSince?: Date } = {} + ): Promise { + const floorMs = opts.deliveredSince?.getTime() + const deliveries: GhHookDelivery[] = [] // bigIdsAsStrings: delivery ids overflow Number.MAX_SAFE_INTEGER — see GhHookDelivery.id. - return this.appRequest(`/app/hook/deliveries?per_page=${perPage}`, 'GET', true) + let path = `/app/hook/deliveries?per_page=${opts.perPage ?? 100}` + for (let page = 0; page < (opts.maxPages ?? MAX_DELIVERY_PAGES); page++) { + const rep: GithubPage = await this.appRequestPage(path, true) + deliveries.push(...rep.data) + const oldest = rep.data[rep.data.length - 1] + // No cursor left, or this page already reaches past the floor: complete. + if (!rep.nextPath || !oldest) return { deliveries, truncated: false } + if (floorMs !== undefined && Date.parse(oldest.delivered_at) <= floorMs) return { deliveries, truncated: false } + path = rep.nextPath + } + return { deliveries, truncated: true } } /** Ask GitHub to redeliver one delivery (202; lands on the relay pool again — @@ -1063,8 +1090,17 @@ export class GithubService { method: 'GET' | 'POST' | 'DELETE' = 'GET', bigIdsAsStrings = false ): Promise { + return (await this.appRequestPage(path, bigIdsAsStrings, method)).data + } + + /** {@link appRequest} keeping the `rel="next"` cursor (paginated list reads). */ + private async appRequestPage( + path: string, + bigIdsAsStrings = false, + method: 'GET' | 'POST' | 'DELETE' = 'GET' + ): Promise> { const jwt = await mintAppJwt(this.deps.cfg) - return githubRequest(path, { + return githubRequestPage(path, { method, auth: jwt, fetchImpl: this.deps.fetchImpl, diff --git a/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.test.ts b/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.test.ts index 77dc819f5..21aee2f99 100644 --- a/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.test.ts +++ b/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.test.ts @@ -73,6 +73,8 @@ function delivery(over: Partial = {}): GhHookDelivery { function make(opts: { hooks?: HookRecord[] deliveries?: GhHookDelivery[] | (() => GhHookDelivery[]) + /** The listing walked its page budget without reaching `deliveredSince`. */ + truncated?: boolean landed?: string[] relaysAlive?: boolean | (() => boolean) redeliverError?: boolean @@ -99,11 +101,13 @@ function make(opts: { ) const settleMock = vi.fn(async () => 0) const reviewFanoutClaimMock = vi.fn(async () => opts.reviewFanoutClaim ?? false) + const listMock = vi.fn(async (_opts?: { deliveredSince?: Date }) => ({ + deliveries: typeof opts.deliveries === 'function' ? opts.deliveries() : (opts.deliveries ?? [delivery()]), + truncated: opts.truncated ?? false + })) const reconciler = new HookRedeliveryReconciler( { - listHookDeliveries: vi.fn(async () => - typeof opts.deliveries === 'function' ? opts.deliveries() : (opts.deliveries ?? [delivery()]) - ), + listHookDeliveries: listMock, redeliverHookDelivery: redeliverMock }, { @@ -120,9 +124,12 @@ function make(opts: { // Ticks stay MANUAL: tick()'s finally re-arms a timer, and clock.advance() // would fire those into async sweeps racing the test's own tick() calls. reconciler.stop() - return { reconciler, redelivered, redeliverMock, reviewFanoutClaimMock, claimMock, settleMock, clock } + return { reconciler, redelivered, redeliverMock, reviewFanoutClaimMock, claimMock, settleMock, listMock, clock } } +/** Let a clock-fired sweep run to completion. */ +const flush = (): Promise => new Promise((r) => setTimeout(r, 0)) + describe('HookRedeliveryReconciler', () => { it('redelivers a matching, unlanded GUID', async () => { const h = make({}) @@ -337,4 +344,65 @@ describe('HookRedeliveryReconciler', () => { expect(h.redelivered).toEqual([]) // never succeeded… expect(h.redeliverMock).toHaveBeenCalledTimes(3) // …retried next ticks, then the cap ended the calls }) + + it('asks the delivery listing to reach the whole look-back window', async () => { + const h = make({}) + await h.reconciler.tick() + expect(h.listMock.mock.calls[0]?.[0]?.deliveredSince).toEqual(new Date(NOW - CFG.windowMs)) + }) + + it('a truncated listing covers only what it listed, and the next sweep resumes there', async () => { + // The busy-App case: the page budget runs out long before the window does. + // Anything older than the oldest listed delivery was never looked at. + const listedFloor = NOW - 4 * 60 * 1000 + const h = make({ + truncated: true, + deliveries: () => [delivery({ delivered_at: new Date(listedFloor).toISOString() })] + }) + await h.reconciler.tick() + h.clock.advance(CFG.intervalMs) + await h.reconciler.tick() + + // Not `now − windowMs` (which would slide past the unlisted slice) and not + // `newest` — the second sweep starts exactly where the first one stopped. + expect(h.listMock.mock.calls[1]?.[0]?.deliveredSince).toEqual(new Date(listedFloor)) + }) + + it('a listing truncated entirely inside the grace window covers no more than the ceiling it evaluated', async () => { + // Every listed delivery is younger than `graceMs`, so none of them were + // evaluated. Coverage must stop at that ceiling — carrying it up to the + // oldest listed delivery would skip everything in between once those + // deliveries age into eligibility. + const h = make({ + truncated: true, + deliveries: () => [delivery({ delivered_at: new Date(NOW - 30_000).toISOString() })] + }) + await h.reconciler.tick() + h.clock.advance(CFG.intervalMs) + await h.reconciler.tick() + expect(h.listMock.mock.calls[1]?.[0]?.deliveredSince).toEqual(new Date(NOW - CFG.graceMs)) + }) + + it('a complete listing that is simply short still advances coverage', async () => { + const h = make({ deliveries: () => [delivery()] }) // truncated: false — quiet App + await h.reconciler.tick() + h.clock.advance(CFG.intervalMs) + await h.reconciler.tick() + // The first sweep covered its whole window, so the second resumes at that + // sweep's ceiling instead of re-listing an interval it already saw. + expect(h.listMock.mock.calls[1]?.[0]?.deliveredSince).toEqual(new Date(NOW - CFG.graceMs)) + }) + + it('runs its first sweep early — a CP that restarts every few minutes still sweeps', async () => { + const h = make({}) + h.reconciler.start() + + h.clock.advance(30_000) + await flush() + expect(h.listMock).not.toHaveBeenCalled() + h.clock.advance(30_000) // 60s after boot, not a full interval + await flush() + expect(h.listMock).toHaveBeenCalledOnce() + h.reconciler.stop() + }) }) diff --git a/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.ts b/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.ts index eab6422a0..319bb3f6e 100644 --- a/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.ts +++ b/packages/control-plane/src/orchestrator/hookRedeliveryReconciler.ts @@ -16,20 +16,25 @@ * waste — the next tick retries once a relay is back); * - a per-GUID attempt cap breaks the no-HookRun re-list loop. Retryable * delivery-stage HookRuns instead carry their bounded due schedule in - * Postgres, so a CP restart cannot reset their budget. + * Postgres, so a CP restart cannot reset their budget; + * - coverage is only ever claimed for what was actually listed, and the first + * sweep of a process runs early — the outage worth recovering is usually the + * one that restarted this process. * * Same Clock-driven self-rescheduling shape as {@link CronRunReaper}; armed by * `startBackground()` only when the GitHub App is configured. */ import type { Clock, TimerHandle } from '../domain/clock.js' import type { HookId } from '../domain/ids.js' -import type { GhHookDelivery } from '../github/service.js' +import type { GhHookDeliveryPage } from '../github/service.js' import type { HookRecord, RelayRecord } from '../persistence/ports.js' /** The families the relay matches (everything else never produces a run). */ const SUBSCRIPTION_EVENTS = new Set(['issues', 'pull_request', 'issue_comment', 'pull_request_review_comment', 'push']) /** Redeliveries requested per GUID before giving up (loop breaker). */ const MAX_ATTEMPTS = 3 +/** Delay before the first sweep of a process — see {@link HookRedeliveryReconciler.start}. */ +const FIRST_SWEEP_DELAY_MS = 60_000 /** Durable due gates for HookRuns that landed as a definite pre-dispatch * failure. Ambiguous dispatch timeouts are deliberately terminal until an * end-to-end admission fence or cross-daemon idempotency exists. The 10-minute @@ -42,12 +47,13 @@ export const FAILED_DELIVERY_BACKOFF_MS = [30_000] as const /** Attempt-map bound (flush-at-cap, the daemon dedup-map precedent). */ const MAX_TRACKED = 5_000 /** Hard ceiling on how far back a post-outage catch-up may reach. GitHub's own - * redelivery window is 3 days; one 100-item page bounds the practical reach - * anyway (a full page is warned about, never silent). */ + * redelivery window is 3 days; the delivery listing walks its cursor to the + * window floor and, when a firehose exhausts its page budget first, coverage + * stops where the listing stopped and the next sweep resumes from there. */ const MAX_LOOKBACK_MS = 24 * 60 * 60 * 1000 export interface HookRedeliveryGithub { - listHookDeliveries(perPage?: number): Promise + listHookDeliveries(opts?: { perPage?: number; maxPages?: number; deliveredSince?: Date }): Promise redeliverHookDelivery(deliveryId: string): Promise } @@ -137,10 +143,13 @@ export class HookRedeliveryReconciler { private readonly log?: ReconcilerLog ) {} - /** Arm the periodic sweep. Idempotent — a second call re-arms from now. */ + /** Arm the periodic sweep. Idempotent — a second call re-arms from now. The + * FIRST sweep comes early: the window a fresh process most needs to recover + * is the one its own restart interrupted, and a deployment that restarts the + * CP more often than `intervalMs` would otherwise never sweep at all. */ start(): void { this.stopped = false - this.arm() + this.arm(Math.min(FIRST_SWEEP_DELAY_MS, this.cfg.intervalMs)) } stop(): void { @@ -151,10 +160,10 @@ export class HookRedeliveryReconciler { } } - private arm(): void { + private arm(delayMs = this.cfg.intervalMs): void { if (this.stopped) return if (this.timer !== undefined) this.clock.clearTimeout(this.timer) - this.timer = this.clock.setTimeout(() => void this.tick(), this.cfg.intervalMs) + this.timer = this.clock.setTimeout(() => void this.tick(), delayMs) } /** One sweep. Errors are logged and swallowed — a GitHub/DB blip must never @@ -210,13 +219,22 @@ export class HookRedeliveryReconciler { return } - const deliveries = await this.github.listHookDeliveries() - if (deliveries.length >= 100) { - // One page only (the cursor rides a Link header we don't surface) — say - // so instead of silently under-covering the window. + const { deliveries, truncated } = await this.github.listHookDeliveries({ deliveredSince: new Date(oldest) }) + // A truncated walk saw everything down to its oldest entry and nothing + // below it. Coverage stops THERE — advancing the cursor to `newest` would + // declare a slice swept that was never listed, which is exactly how a lost + // delivery stays lost. + const oldestListedMs = deliveries.length > 0 ? Date.parse(deliveries[deliveries.length - 1]!.delivered_at) : NaN + const truncatedFloor = + truncated && !Number.isNaN(oldestListedMs) && oldestListedMs > oldest ? oldestListedMs : undefined + // …and never PAST `newest`: everything above that ceiling is inside the + // grace window and was skipped below, so a budget exhausted entirely inside + // the grace period must not carry coverage over the deliveries it withheld. + const sweptUntil = truncatedFloor === undefined ? newest : Math.min(newest, truncatedFloor) + if (sweptUntil !== newest) { this.log?.warn( - { count: deliveries.length }, - 'hook-redelivery: delivery page full — the look-back window may be under-covered' + { count: deliveries.length, reachedBack: new Date(sweptUntil).toISOString() }, + 'hook-redelivery: delivery list ran out before the look-back window — the rest retries next sweep' ) } @@ -235,7 +253,7 @@ export class HookRedeliveryReconciler { return true }) if (matching.length === 0) { - this.coveredUntilMs = newest + this.coveredUntilMs = sweptUntil return } @@ -280,6 +298,6 @@ export class HookRedeliveryReconciler { if (redelivered > 0) { this.log?.info({ redelivered, scanned: deliveries.length }, 'hook-redelivery: re-posted lost deliveries') } - this.coveredUntilMs = oldestFailedAt !== undefined ? Math.min(newest, oldestFailedAt - 1) : newest + this.coveredUntilMs = oldestFailedAt !== undefined ? Math.min(sweptUntil, oldestFailedAt - 1) : sweptUntil } }