diff --git a/apps/docs/docs/guides/runners.md b/apps/docs/docs/guides/runners.md index b89c25c5..069d7bd2 100644 --- a/apps/docs/docs/guides/runners.md +++ b/apps/docs/docs/guides/runners.md @@ -44,7 +44,7 @@ A runner operation that succeeded means the VM ran and was cleaned up. It does n - **App:** open a connection in **Repositories** to see its recent jobs. - **CLI:** `vectis job list --json` returns the latest 100 jobs with their GitHub status and conclusion. -Job states arrive through GitHub webhooks. If one seems missing or stale: +Job states arrive through GitHub webhooks. Vectis checks for failed deliveries every two minutes and requests one automatic redelivery per webhook, usually recovering missed updates within a few minutes without a connected Mac. If one still seems missing or stale: ```sh vectis job scan --wait # find queued and running jobs Vectis missed diff --git a/convex/_generated/api.d.ts b/convex/_generated/api.d.ts index 6486965c..516bda35 100644 --- a/convex/_generated/api.d.ts +++ b/convex/_generated/api.d.ts @@ -28,6 +28,8 @@ import type * as githubIdentity from "../githubIdentity.js"; import type * as githubJobs from "../githubJobs.js"; import type * as githubMigrations from "../githubMigrations.js"; import type * as githubOAuth from "../githubOAuth.js"; +import type * as githubRedelivery from "../githubRedelivery.js"; +import type * as githubRedeliveryState from "../githubRedeliveryState.js"; import type * as githubRepositories from "../githubRepositories.js"; import type * as githubRunnerProbe from "../githubRunnerProbe.js"; import type * as githubRunners from "../githubRunners.js"; @@ -78,6 +80,8 @@ declare const fullApi: ApiFromModules<{ githubJobs: typeof githubJobs; githubMigrations: typeof githubMigrations; githubOAuth: typeof githubOAuth; + githubRedelivery: typeof githubRedelivery; + githubRedeliveryState: typeof githubRedeliveryState; githubRepositories: typeof githubRepositories; githubRunnerProbe: typeof githubRunnerProbe; githubRunners: typeof githubRunners; diff --git a/convex/crons.ts b/convex/crons.ts index 4dfe5baf..48cc0251 100644 --- a/convex/crons.ts +++ b/convex/crons.ts @@ -2,5 +2,6 @@ import { cronJobs } from "convex/server"; import { internal } from "./_generated/api.js"; const crons = cronJobs(); +crons.interval("GitHub webhook redelivery", { minutes: 2 }, internal.githubRedelivery.scan, {}); crons.daily("retention sweep", { hourUTC: 3, minuteUTC: 0 }, internal.retention.sweep, {}); export default crons; diff --git a/convex/githubRedelivery.ts b/convex/githubRedelivery.ts new file mode 100644 index 00000000..8b4ba49d --- /dev/null +++ b/convex/githubRedelivery.ts @@ -0,0 +1,144 @@ +"use node"; +import { Schema } from "effect"; +import { createGitHubAppJwt, GitHubApiError } from "../packages/github/src/app-client.js"; +import { internal } from "./_generated/api.js"; +import { internalAction } from "./_generated/server.js"; + +const minute = 60_000; +const maxAge = 3 * 24 * 60 * minute; +const events = new Set(["workflow_job", "installation", "installation_repositories", "repository"]); +const deliveries = Schema.Array( + Schema.Struct({ + id: Schema.String.check(Schema.isPattern(/^\d{1,30}$/)), + guid: Schema.NonEmptyString, + delivered_at: Schema.String, + status_code: Schema.NullOr(Schema.Int), + event: Schema.String, + }), +); + +function nextCursor(link: string | null) { + if (!link) return null; + for (const part of link.split(",")) { + const match = part.match(/<([^>]+)>;\s*rel="next"/); + const url = match?.[1]; + if (url) { + const cursor = new URL(url).searchParams.get("cursor"); + if (!cursor) throw new Error("GitHub delivery pagination is missing its cursor."); + return cursor; + } + } + return null; +} + +export const scan = internalAction({ + args: {}, + handler: async (ctx): Promise => { + const state = await ctx.runQuery(internal.githubRedeliveryState.load, {}); + if (!state) return; + const now = Date.now(); + const since = Math.max(state.checkpoint?.deliveredAt ?? now - 60 * minute, now - maxAge); + const through = now - minute; + const deadline = now + 90_000; + const signal = AbortSignal.timeout(90_000); + const headers = { + Authorization: `Bearer ${createGitHubAppJwt(state.app)}`, + Accept: "application/vnd.github+json", + "X-GitHub-Api-Version": "2026-03-10", + }; + const request = async (path: string, method: "GET" | "POST") => { + const response = await fetch(`https://api.github.com${path}`, { + method, + headers, + redirect: "error", + signal: AbortSignal.any([signal, AbortSignal.timeout(10_000)]), + }); + if (!response.ok) { + await response.body?.cancel(); + throw new GitHubApiError(response.status); + } + return response; + }; + + const attempts = new Map(); + let complete = false; + let cursor: string | null = null; + try { + for (let page = 0; page < 10; page++) { + const params = new URLSearchParams({ per_page: "100" }); + if (cursor) params.set("cursor", cursor); + const response = await request(`/app/hook/deliveries?${params}`, "GET"); + // Delivery IDs exceed Number.MAX_SAFE_INTEGER, so they must stay strings. + const text = (await response.text()).replace(/"id":\s*(\d+)/g, '"id":"$1"'); + const rows = Schema.decodeUnknownSync(deliveries)(JSON.parse(text)); + if (rows.length > 100) throw new Error("GitHub exceeded the delivery page limit."); + for (const row of rows) { + const deliveredAt = Date.parse(row.delivered_at); + if (!Number.isFinite(deliveredAt)) throw new Error("Invalid GitHub delivery timestamp."); + if (deliveredAt < since) { + complete = true; + break; + } + if (!events.has(row.event)) continue; + const successful = + row.status_code !== null && row.status_code >= 200 && row.status_code < 300; + const existing = attempts.get(row.guid); + if (!existing || deliveredAt > existing.deliveredAt) + attempts.set(row.guid, { + id: row.id, + deliveredAt, + successful: successful || existing?.successful === true, + }); + else existing.successful ||= successful; + } + cursor = nextCursor(response.headers.get("link")); + if (complete || !cursor || rows.length === 0) { + complete = true; + break; + } + } + } catch (error) { + console.warn("GitHub webhook delivery listing failed; checkpoint unchanged.", String(error)); + return; + } + if (!complete) { + console.warn("GitHub webhook delivery page limit reached; checkpoint unchanged."); + return; + } + + const candidates = [...attempts.entries()] + .filter(([, attempt]) => !attempt.successful && attempt.deliveredAt <= through) + .sort((a, b) => a[1].deliveredAt - b[1].deliveredAt); + let requested = 0; + for (const [guid, attempt] of candidates) { + if (requested >= 20 || Date.now() >= deadline || signal.aborted) { + complete = false; + break; + } + if (!(await ctx.runMutation(internal.githubRedeliveryState.claim, { guid }))) continue; + requested++; + try { + const response = await request(`/app/hook/deliveries/${attempt.id}/attempts`, "POST"); + await response.body?.cancel(); + } catch (error) { + console.warn("GitHub webhook redelivery request failed; it will not be repeated.", { + guid, + deliveryId: attempt.id, + error, + }); + complete = false; + break; + } + } + if (!complete) { + console.warn("GitHub webhook redelivery scan incomplete; checkpoint unchanged."); + return; + } + // Fresh attempts remain in the next scan, even when this run saw newer successful deliveries. + await ctx.runMutation(internal.githubRedeliveryState.advance, { + id: state.checkpoint?._id ?? null, + deliveredAt: through, + scannedAt: now, + }); + }, +}); diff --git a/convex/githubRedeliveryState.ts b/convex/githubRedeliveryState.ts new file mode 100644 index 00000000..9b49d15a --- /dev/null +++ b/convex/githubRedeliveryState.ts @@ -0,0 +1,47 @@ +import { v } from "convex/values"; +import { internalMutation, internalQuery } from "./_generated/server.js"; + +export const load = internalQuery({ + args: {}, + handler: async (ctx) => { + const app = await ctx.db.query("githubApps").first(); + if (!app) return null; + return { + app: { clientId: app.clientId, privateKey: app.privateKey }, + checkpoint: await ctx.db.query("githubRedeliveryCheckpoint").first(), + }; + }, +}); + +export const advance = internalMutation({ + args: { + id: v.union(v.id("githubRedeliveryCheckpoint"), v.null()), + deliveredAt: v.number(), + scannedAt: v.number(), + }, + handler: async (ctx, { id, ...checkpoint }) => { + if (id) { + await ctx.db.patch("githubRedeliveryCheckpoint", id, checkpoint); + return; + } + const existing = await ctx.db.query("githubRedeliveryCheckpoint").first(); + if (existing) { + if (existing.deliveredAt < checkpoint.deliveredAt) + await ctx.db.patch("githubRedeliveryCheckpoint", existing._id, checkpoint); + } else await ctx.db.insert("githubRedeliveryCheckpoint", checkpoint); + }, +}); + +export const claim = internalMutation({ + args: { guid: v.string() }, + handler: async (ctx, { guid }) => { + const existing = await ctx.db + .query("githubRedeliveryRequests") + .withIndex("by_guid", (q) => q.eq("guid", guid)) + .unique(); + if (existing) return false; + // Reserve before POST: a timeout may still mean GitHub accepted the request. + await ctx.db.insert("githubRedeliveryRequests", { guid, requestedAt: Date.now() }); + return true; + }, +}); diff --git a/convex/retention.ts b/convex/retention.ts index 62af13e1..38ecca56 100644 --- a/convex/retention.ts +++ b/convex/retention.ts @@ -6,6 +6,7 @@ import { internalMutation, type MutationCtx } from "./_generated/server.js"; const day = 24 * 60 * 60 * 1000; export const windows = { deliveries: 30 * day, + redeliveryRequests: 3 * day, jobs: 90 * day, operations: 90 * day, leases: 30 * day, @@ -32,6 +33,14 @@ async function sweepTable( export async function sweepOnce(ctx: MutationCtx, now: number) { let deleted = 0; + deleted += await sweepTable( + await ctx.db + .query("githubRedeliveryRequests") + .withIndex("by_requested", (q) => q.lt("requestedAt", now - windows.redeliveryRequests)) + .take(limit), + () => false, + async (row) => ctx.db.delete("githubRedeliveryRequests", row._id), + ); deleted += await sweepTable( await ctx.db .query("githubDeliveries") diff --git a/convex/schema.ts b/convex/schema.ts index c997db5a..2ba60702 100644 --- a/convex/schema.ts +++ b/convex/schema.ts @@ -229,6 +229,16 @@ export default defineSchema({ }) .index("by_delivery", ["deliveryId"]) .index("by_received", ["receivedAt"]), + githubRedeliveryCheckpoint: defineTable({ + deliveredAt: v.number(), + scannedAt: v.number(), + }), + githubRedeliveryRequests: defineTable({ + guid: v.string(), + requestedAt: v.number(), + }) + .index("by_guid", ["guid"]) + .index("by_requested", ["requestedAt"]), githubAppSetups: defineTable({ stateDigest: v.string(), ownerId: v.number(), diff --git a/packages/github/src/app-client.ts b/packages/github/src/app-client.ts index afd4bd03..d0e09fde 100644 --- a/packages/github/src/app-client.ts +++ b/packages/github/src/app-client.ts @@ -92,13 +92,16 @@ export class GitHubPermissionError extends GitHubApiError { this.message = "The Vectis GitHub App needs updated Actions write permission."; } } +export function createGitHubAppJwt(app: { clientId: string; privateKey: string }) { + const now = Math.floor(Date.now() / 1000); + const encode = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url"); + const data = `${encode({ alg: "RS256", typ: "JWT" })}.${encode({ iss: app.clientId, iat: now - 60, exp: now + 300 })}`; + return `${data}.${sign("RSA-SHA256", Buffer.from(data), createPrivateKey(app.privateKey)).toString("base64url")}`; +} export class GitHubAppClient { constructor(private readonly app: { appId: number; clientId: string; privateKey: string }) {} private jwt() { - const now = Math.floor(Date.now() / 1000); - const encode = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url"); - const data = `${encode({ alg: "RS256", typ: "JWT" })}.${encode({ iss: this.app.clientId, iat: now - 60, exp: now + 300 })}`; - return `${data}.${sign("RSA-SHA256", Buffer.from(data), createPrivateKey(this.app.privateKey)).toString("base64url")}`; + return createGitHubAppJwt(this.app); } async request( token: string, diff --git a/tests/convex/githubRedelivery.test.ts b/tests/convex/githubRedelivery.test.ts new file mode 100644 index 00000000..b777e3d4 --- /dev/null +++ b/tests/convex/githubRedelivery.test.ts @@ -0,0 +1,316 @@ +import { generateKeyPairSync } from "node:crypto"; +import { convexTest } from "convex-test"; +import { afterEach, beforeEach, expect, test, vi } from "vitest"; +import schema from "../../convex/schema.js"; +import { internal } from "../../convex/_generated/api.js"; + +const modules = { + "../../convex/githubRedelivery.ts": () => import("../../convex/githubRedelivery.js"), + "../../convex/githubRedeliveryState.ts": () => import("../../convex/githubRedeliveryState.js"), + "../../convex/retention.ts": () => import("../../convex/retention.js"), + "../../convex/_generated/server.js": () => import("../../convex/_generated/server.js"), +}; +const minute = 60_000; +const day = 24 * 60 * minute; +const now = Date.parse("2026-10-16T12:00:00Z"); +const privateKey = generateKeyPairSync("rsa", { modulusLength: 2048 }) + .privateKey.export({ format: "pem", type: "pkcs8" }) + .toString(); +const delivery = ( + id: number, + guid = `guid-${id}`, + age = 2 * minute, + status = 500, + event = "workflow_job", +) => ({ + id, + guid, + delivered_at: new Date(Date.now() - age).toISOString(), + status_code: status, + status: status === 200 ? "OK" : "Internal Server Error", + event, + action: "queued", + redelivery: false, +}); + +beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(now); + vi.spyOn(console, "warn").mockImplementation(() => {}); + vi.stubGlobal( + "fetch", + vi.fn(() => { + throw new Error("Unexpected unstubbed GitHub request."); + }), + ); +}); +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); +}); + +async function fixture(rows: ReturnType[] = []) { + const t = convexTest(schema, modules); + await t.run(async (ctx) => { + await ctx.db.insert("githubApps", { + appId: 42, + slug: "test", + ownerId: 7, + clientId: "test-client", + privateKey, + clientSecret: "test", + webhookSecret: "test", + }); + }); + const fetch = vi.fn(async (_url: string, init?: RequestInit) => + init?.method === "POST" ? new Response(null, { status: 202 }) : Response.json(rows), + ); + vi.stubGlobal("fetch", fetch); + return { + t, + fetch, + scan: () => t.action(internal.githubRedelivery.scan, {}), + checkpoint: () => t.run((ctx) => ctx.db.query("githubRedeliveryCheckpoint").first()), + requests: () => t.run((ctx) => ctx.db.query("githubRedeliveryRequests").collect()), + posts: () => fetch.mock.calls.filter(([, init]) => init?.method === "POST"), + }; +} + +test("redelivers a failed guid once using the App JWT and advances the checkpoint", async () => { + const f = await fixture([delivery(2, "failed"), delivery(1, "failed", 3 * minute)]); + await f.scan(); + expect(f.posts()).toHaveLength(1); + expect(f.posts()[0]?.[0]).toBe("https://api.github.com/app/hook/deliveries/2/attempts"); + const headers = new Headers(f.fetch.mock.calls[0]?.[1]?.headers); + const token = headers.get("Authorization")?.split(" ")[1]; + const payload = token?.split(".")[1]; + expect(payload && JSON.parse(Buffer.from(payload, "base64url").toString())).toMatchObject({ + iss: "test-client", + exp: now / 1000 + 300, + }); + expect(await f.requests()).toMatchObject([{ guid: "failed", requestedAt: now }]); + expect(await f.checkpoint()).toMatchObject({ deliveredAt: now - minute, scannedAt: now }); +}); + +test("a later successful attempt suppresses redelivery across cursor pages", async () => { + const f = await fixture(); + f.fetch.mockImplementation(async (url) => { + const cursor = new URL(url).searchParams.get("cursor"); + if (cursor === null) + return Response.json([delivery(2, "recovered", 2 * minute, 200)], { + headers: { + link: '; rel="next"', + }, + }); + expect(cursor).toBe("opaque+next"); + return Response.json([delivery(1, "recovered", 3 * minute)]); + }); + await f.scan(); + expect(f.fetch).toHaveBeenCalledTimes(2); + expect(f.posts()).toHaveLength(0); + expect(await f.requests()).toHaveLength(0); +}); + +test("any 2xx attempt suppresses redelivery, even when the newest attempt failed", async () => { + const f = await fixture([delivery(2, "recovered"), delivery(1, "recovered", 3 * minute, 204)]); + await f.scan(); + expect(f.posts()).toHaveLength(0); +}); + +test("fresh failures stay eligible for the next scan", async () => { + const row = delivery(1, "fresh", 30_000); + const f = await fixture([row]); + await f.scan(); + expect(f.posts()).toHaveLength(0); + expect(await f.checkpoint()).toMatchObject({ deliveredAt: now - minute }); + vi.setSystemTime(now + 2 * minute); + await f.scan(); + expect(f.posts()).toHaveLength(1); +}); + +test.each([403, 429, 500, 503])("listing failure %s preserves the checkpoint", async (status) => { + const f = await fixture(); + await f.scan(); + const checkpoint = await f.checkpoint(); + vi.setSystemTime(now + 2 * minute); + f.fetch.mockResolvedValue(Response.json({ message: "Unavailable" }, { status })); + await f.scan(); + expect(await f.checkpoint()).toEqual(checkpoint); + expect(f.posts()).toHaveLength(0); + expect(console.warn).toHaveBeenCalled(); +}); + +test("a later page failure makes no requests and creates no checkpoint", async () => { + const f = await fixture(); + f.fetch + .mockResolvedValueOnce( + Response.json([delivery(1)], { + headers: { link: '; rel="next"' }, + }), + ) + .mockRejectedValueOnce(new Error("Network unavailable")); + await f.scan(); + expect(await f.checkpoint()).toBeNull(); + expect(await f.requests()).toHaveLength(0); +}); + +test("guid dedupe survives repeated scans and new failed delivery IDs", async () => { + const f = await fixture([delivery(1, "failed")]); + await f.scan(); + vi.setSystemTime(now + 2 * minute); + f.fetch.mockImplementation(async (_url, init) => + init?.method === "POST" + ? new Response(null, { status: 202 }) + : Response.json([delivery(2, "failed")]), + ); + await f.scan(); + expect(f.posts()).toHaveLength(1); + expect(await f.requests()).toHaveLength(1); +}); + +test("concurrent claims cannot both request redelivery for one guid", async () => { + const f = await fixture(); + const claims = await Promise.all([ + f.t.mutation(internal.githubRedeliveryState.claim, { guid: "same" }), + f.t.mutation(internal.githubRedeliveryState.claim, { guid: "same" }), + ]); + expect(claims.sort()).toEqual([false, true]); + expect(await f.requests()).toHaveLength(1); +}); + +test("no configured App means no fetch and no checkpoint", async () => { + const t = convexTest(schema, modules); + await t.action(internal.githubRedelivery.scan, {}); + expect(fetch).not.toHaveBeenCalled(); + expect(await t.run((ctx) => ctx.db.query("githubRedeliveryCheckpoint").first())).toBeNull(); + expect(console.warn).not.toHaveBeenCalled(); +}); + +test("only Vectis events within the initial one-hour window are redelivered", async () => { + const f = await fixture([ + delivery(6, "unsupported", minute, 500, "push"), + ...["workflow_job", "installation", "installation_repositories", "repository"].map( + (event, index) => delivery(5 - index, event, 2 * minute, 500, event), + ), + delivery(1, "too-old", 61 * minute), + ]); + await f.scan(); + expect((await f.requests()).map((row) => row.guid).sort()).toEqual([ + "installation", + "installation_repositories", + "repository", + "workflow_job", + ]); +}); + +test("stale checkpoints never extend eligibility beyond three days", async () => { + const f = await fixture([delivery(2, "eligible", 2 * day), delivery(1, "expired", 4 * day)]); + await f.t.run((ctx) => + ctx.db.insert("githubRedeliveryCheckpoint", { + deliveredAt: now - 5 * day, + scannedAt: now - 5 * day, + }), + ); + await f.scan(); + expect(await f.requests()).toMatchObject([{ guid: "eligible" }]); +}); + +test("scanning stops at the checkpoint without fetching the next page", async () => { + const f = await fixture(); + await f.t.run((ctx) => + ctx.db.insert("githubRedeliveryCheckpoint", { + deliveredAt: now - 3 * minute, + scannedAt: now - 2 * minute, + }), + ); + f.fetch.mockResolvedValue( + Response.json([delivery(1, "old", 4 * minute)], { + headers: { link: '; rel="next"' }, + }), + ); + await f.scan(); + expect(f.fetch).toHaveBeenCalledTimes(1); + expect(f.posts()).toHaveLength(0); +}); + +test("the redelivery limit drains remaining guids on a later run without skipping them", async () => { + const f = await fixture(Array.from({ length: 25 }, (_, index) => delivery(index + 1))); + await f.scan(); + expect(f.posts()).toHaveLength(20); + expect(await f.checkpoint()).toBeNull(); + await f.scan(); + expect(f.posts()).toHaveLength(25); + expect(await f.requests()).toHaveLength(25); + expect(await f.checkpoint()).not.toBeNull(); +}); + +test("the page limit preserves the checkpoint rather than skipping unseen attempts", async () => { + const f = await fixture(); + f.fetch.mockImplementation(async () => + Response.json([delivery(1)], { + headers: { link: '; rel="next"' }, + }), + ); + await f.scan(); + expect(f.fetch).toHaveBeenCalledTimes(10); + expect(await f.checkpoint()).toBeNull(); + expect(f.posts()).toHaveLength(0); +}); + +test("a POST error is logged and never repeated, including after an uncertain timeout", async () => { + const f = await fixture([delivery(1)]); + f.fetch.mockImplementation(async (_url, init) => { + if (init?.method === "POST") throw new Error("Timed out after sending"); + return Response.json([delivery(1)]); + }); + await f.scan(); + expect(await f.checkpoint()).toBeNull(); + await f.scan(); + expect(f.posts()).toHaveLength(1); + expect(await f.checkpoint()).not.toBeNull(); + expect(console.warn).toHaveBeenCalled(); +}); + +test("the action time budget leaves unrequested guids available for the next tick", async () => { + const f = await fixture([delivery(1)]); + f.fetch.mockImplementation(async () => { + vi.setSystemTime(now + 90_000); + return Response.json([delivery(1, "guid-1", 3 * minute)]); + }); + await f.scan(); + expect(await f.requests()).toHaveLength(0); + expect(await f.checkpoint()).toBeNull(); +}); + +test("retention expires redelivery receipts after three days in bounded batches", async () => { + const f = await fixture(); + await f.t.run(async (ctx) => { + for (let index = 0; index < 501; index++) + await ctx.db.insert("githubRedeliveryRequests", { + guid: `old-${index}`, + requestedAt: now - 3 * day - 1, + }); + await ctx.db.insert("githubRedeliveryRequests", { + guid: "boundary", + requestedAt: now - 3 * day, + }); + await ctx.db.insert("githubRedeliveryRequests", { guid: "recent", requestedAt: now }); + }); + await f.t.mutation(internal.retention.sweep, {}); + expect(await f.requests()).toHaveLength(3); + await f.t.finishAllScheduledFunctions(vi.runAllTimers); + expect((await f.requests()).map((row) => row.guid).sort()).toEqual(["boundary", "recent"]); +}); + +test("delivery IDs beyond the safe integer range are redelivered exactly", async () => { + const f = await fixture(); + const body = JSON.stringify([delivery(1)]).replace('"id":1', '"id":98765432109876543210'); + f.fetch.mockImplementation(async (_url: string, init?: RequestInit) => + init?.method === "POST" ? new Response(null, { status: 202 }) : new Response(body), + ); + await f.scan(); + expect(f.posts()[0]?.[0]).toBe( + "https://api.github.com/app/hook/deliveries/98765432109876543210/attempts", + ); +});