From c003d6f0daa920a6741521a2c24f12ebef3678d5 Mon Sep 17 00:00:00 2001 From: Postil Maintainer Date: Sun, 18 Oct 2026 15:06:19 +0000 Subject: [PATCH] fix: reconcile feedback beyond 20 historical review threads --- src/lib/github/publication-threads.ts | 1 - tests/review-feedback-queue.test.ts | 166 ++++++++++++++++++++++- tests/review-feedback.test.ts | 186 +++++++++++++++++++++++++- 3 files changed, 350 insertions(+), 3 deletions(-) diff --git a/src/lib/github/publication-threads.ts b/src/lib/github/publication-threads.ts index 67c5ed4b..6417d87d 100644 --- a/src/lib/github/publication-threads.ts +++ b/src/lib/github/publication-threads.ts @@ -201,7 +201,6 @@ export async function readGitHubReviewFeedback( thread.comments?.pageInfo?.hasNextPage !== false || !Array.isArray(comments)) { throw new Error("review feedback published thread is incomplete"); } - if (threads.length >= 20) throw new Error("review feedback exceeds 20 published threads"); const replies: ObservedFeedbackThread["comments"] = []; for (const comment of comments.slice(1)) { const author = human(comment.author); diff --git a/tests/review-feedback-queue.test.ts b/tests/review-feedback-queue.test.ts index 5de8ed9e..72ece91a 100644 --- a/tests/review-feedback-queue.test.ts +++ b/tests/review-feedback-queue.test.ts @@ -1,5 +1,9 @@ import { afterAll, beforeAll, describe, expect, mock, test } from "bun:test"; import { Pool } from "pg"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { ReviewJobPayload } from "@/lib/queue"; import { createEphemeralDatabase, type EphemeralDatabase } from "./ephemeral-database"; const describeDb = process.env.POSTIL_TEST_DATABASE_URL ? describe : describe.skip; @@ -34,7 +38,7 @@ mock.module("@/lib/private-repository-entitlement", () => ({ ...realEntitlement, canProcessRepositoryInference: async () => ({ allowed: true }), })); -const { admitReviewFeedbackEvent, reconcileReviewFeedback, reviewFeedbackDigest, reviewFeedbackEnabled, scheduleReviewFeedbackReconciliationJobs } = await import("@/lib/review-feedback"); +const { admitReviewFeedbackEvent, reconcileReviewFeedback, reviewFeedbackDigest, reviewFeedbackEnabled, scheduleReviewFeedbackReconciliationJobs, serializeReviewFeedback, writeReviewFeedbackFile } = await import("@/lib/review-feedback"); const { claimJob, completeJob, enqueueJob, enqueueReviewJobOnce, failJob, requeueJobsOwnedBy } = await import("@/lib/queue"); const { deferHostedReviewForRelease, activateHostedInferenceRelease } = await import("@/lib/release-job-rollout"); const { closeDb } = await import("@/lib/db"); @@ -845,4 +849,164 @@ describeDb("durable review feedback admission", () => { await limitedPool.end(); } }, 15_000); + + test("complete historical observations preserve exact admission, failure atomicity and actor limits", async () => { + const pool = database.pool; + const prNumber = 301; + const payload = { githubRepoId: 72, installationId: 82, prNumber }; + const historyOrg = (await pool.query(`INSERT INTO organizations (slug, name, github_org_id) + VALUES ('history-fixture', 'History fixture', 92) RETURNING id`)).rows[0].id; + const historyInstallation = (await pool.query(`INSERT INTO installations + (github_installation_id, org_id, account_login, account_type) + VALUES (82, $1, 'octo', 'Organization') RETURNING id`, [historyOrg])).rows[0].id; + const historyRepository = (await pool.query(`INSERT INTO repositories + (installation_id, github_repo_id, full_name, enabled, private) + VALUES ($1, 72, 'octo/history', true, false) RETURNING id`, [historyInstallation])).rows[0].id; + const review = await pool.query(`INSERT INTO reviews + (repository_id, source_org_id, source_installation_id, source_github_installation_id, source_github_repo_id, + source_repo_full_name, pr_number, head_sha, base_sha, status, author_github_id, author_login, finished_at) + VALUES ($1, $2, $3, 82, 72, 'octo/history', $4, $5, $6, 'completed', 51, 'maintainer', now()) RETURNING id`, + [historyRepository, historyOrg, historyInstallation, prNumber, headSha, baseSha]); + await pool.query(`INSERT INTO finding_publications + (review_id, finding_id, stable_identity, initial_state, current_state, github_comment_id) + SELECT $1, 'history-' || i, true, 'inline', 'resolved', ($2::bigint + i * 100)::text + FROM generate_series(0, 99) i`, [review.rows[0].id, rootId]); + await pool.query("UPDATE review_feedback_control SET mode = 'enabled' WHERE id = 1"); + const actor = { __typename: "User", databaseId: 51, login: "maintainer" }; + const root = (i: number) => ({ isResolved: false, resolvedBy: null as typeof actor | null, + comments: { nodes: [{ databaseId: String(rootId + i * 100), author: { __typename: "Bot", login: "postil-dev[bot]" } }] as Array>, + pageInfo: { hasNextPage: false } } }); + const comment = (i: number, id = 51, body = "The recovery procedure covers this path.") => ({ + databaseId: String(rootId + i * 100 + 1), author: { ...actor, databaseId: id, login: id === 51 ? "maintainer" : `visitor-${id}` }, body, updatedAt, + }); + let nodes = Array.from({ length: 100 }, (_, i) => root(i)); + let failPage = false; + const savedFetch = globalThis.fetch; + const counts = async () => ({ + receipts: (await pool.query("SELECT count(*)::int AS n FROM review_feedback_requests WHERE pr_number = $1", [prNumber])).rows[0].n, + jobs: (await pool.query("SELECT count(*)::int AS n FROM jobs WHERE kind = 'review-feedback' AND payload->>'prNumber' = $1", [String(prNumber)])).rows[0].n, + }); + const poll = async () => (await pool.query("SELECT last_error, next_poll_at FROM review_feedback_polls WHERE github_repo_id = 72 AND pr_number = $1", [prNumber])).rows[0]; + const noAdmission = async (message: string) => { + const before = await counts(); + await expect(reconcileReviewFeedback(payload, pool)).rejects.toThrow(message); + expect(await counts()).toEqual(before); + expect((await poll()).last_error).toContain(message); + }; + try { + globalThis.fetch = (async (_url, options) => { + const cursor = JSON.parse(String(options!.body)).variables.cursor; + const offset = cursor === null ? 0 : Number(cursor); + if (failPage && offset > 0) return new Response(null, { status: 503 }); + return Response.json({ data: { repository: { databaseId: 72, nameWithOwner: "octo/history", pullRequest: { + headRefOid: headSha, state: "OPEN", isDraft: false, reviewThreads: { nodes: nodes.slice(offset, offset + 50), + pageInfo: { hasNextPage: offset + 50 < nodes.length, endCursor: String(offset + 50) } }, + } } } }); + }) as typeof fetch; + await pool.query(`INSERT INTO review_feedback_polls (github_repo_id, pr_number, next_poll_at, last_error) + VALUES (72, $1, now(), 'review feedback exceeds 20 published threads')`, [prNumber]); + await reconcileReviewFeedback(payload, pool); + expect(await counts()).toEqual({ receipts: 0, jobs: 0 }); + expect((await poll()).last_error).toBeNull(); + expect((await poll()).next_poll_at.getTime()).toBeGreaterThan(Date.now()); + + nodes[99]!.comments.nodes.push(comment(99), { ...comment(99), databaseId: String(rootId + 9902) }); + nodes[98]!.isResolved = true; + nodes[98]!.resolvedBy = actor; + for (const [index, body] of ["Thanks!", "@postil review", "@postil approve history-2", "@postil dismiss history-3"].entries()) { + nodes[index]!.comments.nodes.push(comment(index, 100 + index, body)); + } + nodes[4]!.comments.nodes.push(comment(4, 104)); + const expected = { version: 1 as const, repository: "octo/history", prNumber, headSha, threads: [ + { findingId: "history-98", rootCommentId: rootId + 9800, resolved: true, comments: [] }, + { findingId: "history-99", rootCommentId: rootId + 9900, resolved: false, comments: [ + { commentId: rootId + 9901, author: { id: 51, login: "maintainer" }, body: comment(99).body, updatedAt }, + { commentId: rootId + 9902, author: { id: 51, login: "maintainer" }, body: comment(99).body, updatedAt }, + ] }, + ] }; + let seed = 71; + for (let trial = 0; trial < 8; trial++) { + for (let i = nodes.length - 1; i > 0; i--) { + seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0; + const j = seed % (i + 1); + [nodes[i], nodes[j]] = [nodes[j]!, nodes[i]!]; + } + for (const node of nodes) node.comments.nodes.splice(1, node.comments.nodes.length - 1, ...node.comments.nodes.slice(1).reverse()); + await Promise.all(Array.from({ length: 3 }, () => reconcileReviewFeedback(payload, pool))); + expect(await counts()).toEqual({ receipts: 1, jobs: 1 }); + const job = (await pool.query("SELECT payload FROM jobs WHERE kind = 'review-feedback' AND payload->>'prNumber' = $1", [String(prNumber)])).rows[0].payload; + expect(serializeReviewFeedback(job.reviewFeedback)).toBe(serializeReviewFeedback(expected)); + expect(job.trigger.feedbackDigest).toBe(reviewFeedbackDigest(expected)); + } + await pool.query("UPDATE jobs SET status = 'done' WHERE status = 'queued' AND payload->>'prNumber' <> $1", [String(prNumber)]); + const claimed = await claimJob(pool, "history-consumer", ["review-feedback"]); + expect(claimed?.payload.prNumber).toBe(prNumber); + const directory = await mkdtemp(join(tmpdir(), "postil-feedback-")); + try { + const path = join(directory, "feedback.json"); + expect(await writeReviewFeedbackFile(path, claimed!.payload as ReviewJobPayload)).toBe(true); + expect(await readFile(path, "utf8")).toBe(serializeReviewFeedback(expected)); + } finally { await rm(directory, { recursive: true }); } + expect(await completeJob(pool, { id: claimed!.id, lockedBy: "history-consumer" })).toBe("done"); + + nodes = Array.from({ length: 100 }, (_, i) => root(i)); + nodes[0]!.comments.nodes.push(comment(0)); + failPage = true; + await noAdmission("HTTP 503"); + failPage = false; + nodes[99]!.comments.pageInfo.hasNextPage = true; + await noAdmission("published thread is incomplete"); + nodes[99]!.comments.pageInfo.hasNextPage = false; + nodes[99] = nodes[0]!; + await noAdmission("thread identity is invalid"); + nodes = Array.from({ length: 100 }, (_, i) => root(i)); + for (let i = 0; i < 21; i++) nodes[i]!.comments.nodes.push(comment(i, 100 + i)); + nodes[99]!.comments.nodes.push(comment(99)); + await noAdmission("exceeds 20 human actors"); + nodes = Array.from({ length: 21 }, (_, i) => ({ ...root(i), isResolved: true, resolvedBy: actor })); + await noAdmission("context identity or bounds"); + for (const [threads, replies, body, error] of [ + [1, 1, "λ".repeat(2049), "context identity or bounds"], + [8, 17, "Evidence.", "exceeds 128 comments"], + [8, 1, "x".repeat(4096), "exceeds 32 KiB"], + ] as const) { + nodes = Array.from({ length: threads }, (_, i) => root(i)); + for (const [i, node] of nodes.entries()) for (let j = 0; j < replies; j++) { + node.comments.nodes.push({ ...comment(i, 51, body), databaseId: String(rootId + i * 100 + j + 1) }); + } + await noAdmission(error); + } + nodes = Array.from({ length: 100 }, (_, i) => root(i)); + nodes[99]!.comments.nodes.push(comment(99)); + authority = "unavailable"; + await noAdmission("actor authority is unavailable"); + authority = "authorized"; + + // Interrupt each admission stage and observe real PostgreSQL rollback before retrying. + const before = await counts(); + for (const point of ["INSERT INTO review_feedback_requests", "FROM jobs", "INSERT INTO jobs"]) { + const failingPool = new Proxy(pool, { get(target, property) { + if (property !== "connect") { + const value = Reflect.get(target, property); + return typeof value === "function" ? value.bind(target) : value; + } + return async () => { + const client = await target.connect(); + return new Proxy(client, { get(connection, key) { + const value = Reflect.get(connection, key); + if (key === "query") return (sql: string, ...args: unknown[]) => { + if (sql.includes(point)) throw new Error("fixture transaction failure"); + return Reflect.apply(connection.query, connection, [sql, ...args]); + }; + return typeof value === "function" ? value.bind(connection) : value; + } }); + }; + } }); + await expect(reconcileReviewFeedback(payload, failingPool)).rejects.toThrow("fixture transaction failure"); + expect(await counts()).toEqual(before); + } + await reconcileReviewFeedback(payload, pool); + expect(await counts()).toEqual({ receipts: before.receipts + 1, jobs: before.jobs + 1 }); + } finally { authority = "authorized"; globalThis.fetch = savedFetch; } + }, 30_000); }); diff --git a/tests/review-feedback.test.ts b/tests/review-feedback.test.ts index cfbd3919..35e5606d 100644 --- a/tests/review-feedback.test.ts +++ b/tests/review-feedback.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, spyOn, test } from "bun:test"; import type { Pool } from "pg"; import { readGitHubReviewFeedback } from "@/lib/github/publication-threads"; @@ -29,9 +29,193 @@ function response(overrides: Record = {}): Response { ], pageInfo: { hasNextPage: false } }, ...overrides }], pageInfo: { hasNextPage: false } } } } } }); } +function historyRoot(index: number) { + return { isResolved: false, resolvedBy: null, comments: { nodes: [ + { databaseId: String(rootCommentId + index * 100), author: { __typename: "Bot", login: "postil-dev[bot]" } }, + ] as Array>, pageInfo: { hasNextPage: false } } }; +} + +function historyPage(nodes: ReturnType[], hasNextPage = false, endCursor: string | null = null) { + return { data: { repository: { databaseId: 71, nameWithOwner: "octo/repository", pullRequest: { + headRefOid: headSha, state: "OPEN", isDraft: false, reviewThreads: { nodes, pageInfo: { hasNextPage, endCursor } }, + } } } }; +} + afterEach(() => { globalThis.fetch = originalFetch; }); describe("bounded review feedback context", () => { + test("retains actionable thread, reply and exact UTF-8 context boundaries", () => { + const many = (threads: number, replies: number) => ({ ...context(), threads: Array.from({ length: threads }, (_, i) => ({ + ...context().threads[0]!, rootCommentId: rootCommentId + i * 100, + comments: Array.from({ length: replies }, (_, j) => ({ ...context().threads[0]!.comments[0]!, commentId: rootCommentId + i * 100 + j + 1, body: "" })), + })) }); + expect(() => serializeReviewFeedback(many(20, 0))).not.toThrow(); + expect(() => serializeReviewFeedback(many(21, 0))).toThrow("bounds"); + expect(() => serializeReviewFeedback(many(1, 20))).not.toThrow(); + expect(() => serializeReviewFeedback(many(1, 21))).toThrow("bounds"); + expect(() => serializeReviewFeedback(many(8, 16))).not.toThrow(); + const total = many(8, 16); + total.threads[0]!.comments.push({ ...total.threads[0]!.comments[0]!, commentId: rootCommentId + 90 }); + expect(() => serializeReviewFeedback(total)).toThrow("128 comments"); + const body = context(); + body.threads[0]!.comments[0]!.body = "λ".repeat(2048); + expect(() => serializeReviewFeedback(body)).not.toThrow(); + body.threads[0]!.comments[0]!.body += "x"; + expect(() => serializeReviewFeedback(body)).toThrow("bounds"); + const aggregate = many(1, 16); + let remaining = 32 * 1024 - Buffer.byteLength(serializeReviewFeedback(aggregate)); + for (const comment of aggregate.threads[0]!.comments) { + const bytes = Math.min(4095, remaining); + comment.body = "x".repeat(bytes); + remaining -= bytes; + } + expect(remaining).toBe(0); + expect(Buffer.byteLength(serializeReviewFeedback(aggregate))).toBe(32 * 1024); + aggregate.threads[0]!.comments[0]!.body += "x"; + expect(() => serializeReviewFeedback(aggregate)).toThrow("32 KiB"); + const duplicate = many(2, 1); + duplicate.threads[1]!.comments[0]!.commentId = duplicate.threads[0]!.comments[0]!.commentId; + expect(() => serializeReviewFeedback(duplicate)).toThrow("reply identity"); + }); + + test("completes generated bounded histories across page partitions and root permutations", async () => { + let seed = 71; + const random = () => (seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0); + for (const count of [20, 21, 51, 100, 200]) { + const roots = Array.from({ length: count }, (_, index) => historyRoot(index)); + const known = new Set(roots.slice(0, 100).map((root) => Number(root.comments.nodes[0]!.databaseId))); + roots[20 % count]!.comments.nodes.push({ databaseId: rootCommentId + 2001, author: human, + body: "The recovery procedure covers this failure.", updatedAt: "2026-09-01T12:00:00Z" }); + for (let trial = 0; trial < 8; trial++) { + const shuffled = [...roots]; + for (let i = shuffled.length - 1; i > 0; i--) { + const j = random() % (i + 1); + [shuffled[i], shuffled[j]] = [shuffled[j]!, shuffled[i]!]; + } + const pages: typeof roots[] = []; + while (shuffled.length) { + const remainingPages = 4 - pages.length; + const minimum = Math.max(1, shuffled.length - (remainingPages - 1) * 50); + pages.push(shuffled.splice(0, minimum + random() % (Math.min(50, shuffled.length) - minimum + 1))); + } + let calls = 0; + let inFlight = 0; + const signals: AbortSignal[] = []; + globalThis.fetch = (async (_url: RequestInfo | URL, options?: RequestInit) => { + expect(++inFlight).toBe(1); + expect(JSON.parse(String(options!.body)).variables.cursor).toBe(calls === 0 ? null : `page-${calls}`); + signals.push(options!.signal!); + const page = calls++; + await Promise.resolve(); + inFlight--; + return Response.json(historyPage(pages[page]!, calls < pages.length, `page-${calls}`)); + }) as unknown as typeof fetch; + const result = await readGitHubReviewFeedback(crypto.randomUUID(), { id: 71, fullName: "octo/repository" }, 17, known); + expect(result.threads.map((thread) => thread.rootCommentId).sort()).toEqual([...known].sort()); + expect(result.threads.flatMap((thread) => thread.comments).map((comment) => comment.body)) + .toEqual(["The recovery procedure covers this failure."]); + expect(calls).toBe(pages.length); + expect(signals.every((signal) => signal === signals[0])).toBe(true); + } + } + }); + + test("rejects later-page failures without returning an actionable prefix or fetching page five", async () => { + for (const failure of ["http", "graphql", "head", "cursor", "comments", "nodes", "page-limit"]) { + let calls = 0; + const first = historyRoot(0); + first.comments.nodes.push({ databaseId: rootCommentId + 1, author: human, body: "Actionable evidence.", updatedAt: "2026-09-01T12:00:00Z" }); + globalThis.fetch = (async () => { + const page = calls++; + expect(calls).toBeLessThanOrEqual(4); + const value = historyPage([first], true, `page-${calls}`); + if (page > 0) { + if (failure === "http") return new Response(null, { status: 503 }); + if (failure === "graphql") return Response.json({ ...value, errors: [{ message: "fixture failure" }] }); + if (failure === "head") value.data.repository.pullRequest.headRefOid = "b".repeat(40); + if (failure === "cursor") value.data.repository.pullRequest.reviewThreads.pageInfo.endCursor = null; + if (failure === "comments") first.comments.pageInfo.hasNextPage = true; + if (failure === "nodes") return Response.json({ data: { repository: { ...value.data.repository, + pullRequest: { ...value.data.repository.pullRequest, reviewThreads: { pageInfo: { hasNextPage: false } } } } } }); + } + return Response.json(value); + }) as unknown as typeof fetch; + await expect(readGitHubReviewFeedback(crypto.randomUUID(), { id: 71, fullName: "octo/repository" }, 17, new Set([rootCommentId]))) + .rejects.toThrow(); + expect(calls).toBe(failure === "page-limit" ? 4 : 2); + } + }); + + test("retains response identity, shape and reply validation failures", async () => { + for (const kind of ["repository", "name", "head", "graphql", "json", "body", "pagination", "author", "reply"]) { + const node = historyRoot(0); + const value = historyPage([node]); + if (kind === "repository") value.data.repository.databaseId++; + if (kind === "name") value.data.repository.nameWithOwner = "octo/other"; + if (kind === "head") value.data.repository.pullRequest.headRefOid = "invalid"; + if (kind === "author") node.comments.nodes[0]!.author = { __typename: "Bot", login: "another-app" }; + if (kind === "reply") node.comments.nodes.push({ databaseId: "invalid", author: human, body: "Evidence.", updatedAt: "invalid" }); + globalThis.fetch = (async () => { + if (kind === "graphql") return Response.json({ ...value, errors: [{ message: "fixture error" }] }); + if (kind === "json") return new Response("{"); + if (kind === "body") return new Response(null); + if (kind === "pagination") return Response.json({ data: { repository: { ...value.data.repository, + pullRequest: { ...value.data.repository.pullRequest, reviewThreads: {} } } } }); + return Response.json(value); + }) as unknown as typeof fetch; + await expect(readGitHubReviewFeedback(crypto.randomUUID(), { id: 71, fullName: "octo/repository" }, 17, new Set([rootCommentId]))) + .rejects.toThrow(); + } + }); + + test("accepts exactly 512 KiB and rejects the next streamed byte with cancellation", async () => { + for (const extra of [0, 1]) { + const json = JSON.stringify(historyPage([historyRoot(0)])); + const bytes = new TextEncoder().encode(json + " ".repeat(512 * 1024 - Buffer.byteLength(json) + extra)); + let cancelled = false; + const body = new ReadableStream({ start(controller) { + controller.enqueue(bytes.subarray(0, 512 * 1024)); + controller.enqueue(bytes.subarray(512 * 1024)); + if (!extra) controller.close(); + }, cancel() { cancelled = true; } }); + globalThis.fetch = (async () => new Response(body)) as unknown as typeof fetch; + const result = readGitHubReviewFeedback(crypto.randomUUID(), { id: 71, fullName: "octo/repository" }, 17, new Set([rootCommentId])); + if (extra) await expect(result).rejects.toThrow("byte bound"); + else expect((await result).threads).toHaveLength(1); + expect(cancelled).toBe(Boolean(extra)); + expect(body.locked).toBe(false); + } + }); + + test("retains one reader deadline and caller cancellation across later fetch and body reads", async () => { + for (const source of ["caller", "deadline"]) for (const phase of ["fetch", "body"]) { + const caller = new AbortController(); + const deadline = new AbortController(); + const timeout = spyOn(AbortSignal, "timeout").mockReturnValue(deadline.signal); + let calls = 0; + try { + globalThis.fetch = (async (_url: RequestInfo | URL, options?: RequestInit) => { + if (++calls === 1) return Response.json(historyPage([historyRoot(0)], true, "next")); + const signal = options!.signal!; + const abort = () => (source === "caller" ? caller : deadline).abort(new DOMException("fixture aborted", "AbortError")); + if (phase === "fetch") return new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + abort(); + }); + return new Response(new ReadableStream({ start(controller) { + signal.addEventListener("abort", () => controller.error(signal.reason), { once: true }); + abort(); + } })); + }) as unknown as typeof fetch; + await expect(readGitHubReviewFeedback(crypto.randomUUID(), { id: 71, fullName: "octo/repository" }, 17, + new Set([rootCommentId]), caller.signal)).rejects.toThrow("fixture aborted"); + expect(timeout).toHaveBeenCalledTimes(1); + expect(timeout).toHaveBeenCalledWith(30_000); + expect(calls).toBe(2); + } finally { timeout.mockRestore(); } + } + }); + test("rejects malformed webhook actor logins before database or authority lookup", async () => { const enabled = process.env.POSTIL_REVIEW_FEEDBACK_ENABLED; process.env.POSTIL_REVIEW_FEEDBACK_ENABLED = "1";