From ddfb4c30ace168701de69f2513565358c01aabd3 Mon Sep 17 00:00:00 2001 From: Ogden-2688 Date: Wed, 21 Oct 2026 11:57:00 -0400 Subject: [PATCH 1/2] Polling fallback: read each list by updated_at and keep a cursor per list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The three `…Since` polls could lose rows silently, three ways, each shown by a probe on main 6b5f8e2 (the real adapter against the test file's fake GitHub server): - A page shorter than 100 ended the read even when GitHub's `Link` header said more followed (149 rows, page 1 of 99 with rel="next": 99 read). - The 3-page cap ended a pass, the cursor moved to the pass's start, and the rows past the cap were never asked for again, though the log said the next tick would pick them up (350 rows: 300 delivered, 50 never). - A row updated while the list was being paged moved to the end, so an unread row slid onto a page already read and was skipped (150 rows, #5 updated between pages 1 and 2: #101 never served). The adapter now reads a list by `updated_at` (pollSince, replacing pollPages): each request after the first asks `since` = the last row's `updated_at`, page 1, and drops the rows of that second it reads again by id; a page wholly in one second goes on to that second's next page. Only `Link: rel="next"` says more follow. After POLL_MAX_PAGES requests it stops, going on only to finish a second it is part way through: the stop second's rows are left out and returned as `resumeAt`, which the next read asks from. GitHub's `since` includes its own second (checked with `gh api` on cli/cli's comments: since = a row's own stamp returns it, one second later does not). POLL_MAX_ENTRIES goes, since the request bound now bounds the rows. The three list methods return `PollRead` ({ rows, resumeAt? }). The poller keeps a cursor per repository and list in the state (`pollCursors`, additive to version 1) rather than one shared cursor: a list that finished never re-reads after another's stop, so the 2,000-key ring need not cover other lists' re-reads. A list with no cursor of its own starts from `lastPollAt`, which a state file from before this carries, and so does a repository newly in GITHUB_REPOS. Cursors are written only when the whole pass delivered, as before, and a repository no longer polled drops out of them. Cursors are whole seconds. GitHub's `since` part way through a second leaves out every row stamped in that second (probed on cli/cli: a comment's own second returned it; that second plus 1, 100 or 999 ms left out all six tried), so the pass-start cursor, with milliseconds, missed rows changed later in the second a pass started in. A list read to the end now starts next at its pass's start second, which is read again. lastPollAt is the earliest of the pass start and every stop, so every list has delivered what changed before it: a release from before this change, which reads only lastPollAt, reads on from the earliest unread point. A first pass stores the window it reads as lastPollAt before it reads, so a pass that is refused or throws is retried from the same start. Stored positions are read as whole seconds, so main's millisecond lastPollAt starts at its second, and the loader drops a lastPollAt, cursor since or resumeAt that isn't an instant. A stopped issues read keeps its window's start for amendment 2a's "opened" rule, so an issue opened in the window but changed past the stop still arrives as opened. A row read again inside its window (after a stop, a refused delivery, or in the second a pass started in) must not be opened twice, so the state remembers each row announced as new (`pollNew`, id -> created_at) when its event is accepted, beside the head marker and the dedup key, until its created_at falls before its list's window; normalizePolledIssue takes `announced`. Head markers could not do this: a stopped read is about 300 issue rows, past POLL_HEAD_MEMORY's 200. getCheckRuns reads on past a short page while total_count says more runs follow, instead of leaving such a head undecided. Without a total_count, which main read one page of, it reads on past a full page, and a read the page bound stops is undecided. A lastPollAt that isn't an instant loads as none, with a warning. Tests: probes A, B and C became tests (adapter level, and B through a real Poller), with tests for the keyset requests, a second wider than the request bound, a page with no stamp, a Link header without next, the stop's resumeAt for all three lists, opened rows across a stop, a refused pass's announced rows and a refused opening, whole-second cursors and the second a pass starts in, stored positions with milliseconds, the earliest stop as lastPollAt, a first pass refused or thrown, short check-run pages and a missing total_count, the migration from lastPollAt, a failed delivery, and the state file's cursors and remembered rows. The old 300-row bound test goes. Two existing tests that rewind lastPollAt to re-read now clear the per-list cursors too. --- changelog.d/fix-poll-paging.fixed.md | 20 ++ docs/contract-notes.md | 20 +- src/github-adapter.ts | 156 +++++++---- src/normalize.ts | 16 +- src/poller.ts | 149 +++++++++-- src/state.ts | 60 ++++- src/types.ts | 70 ++++- test/githubAdapter.test.ts | 383 +++++++++++++++++++++++++-- test/harness.ts | 13 +- test/integration.test.ts | 2 + test/poller.test.ts | 186 ++++++++++++- test/smoke.test.ts | 12 +- test/state.test.ts | 57 ++++ test/webhook.test.ts | 12 +- 14 files changed, 1030 insertions(+), 126 deletions(-) create mode 100644 changelog.d/fix-poll-paging.fixed.md diff --git a/changelog.d/fix-poll-paging.fixed.md b/changelog.d/fix-poll-paging.fixed.md new file mode 100644 index 0000000..e951e04 --- /dev/null +++ b/changelog.d/fix-poll-paging.fixed.md @@ -0,0 +1,20 @@ +- The polling fallback no longer loses rows in three ways it did. A page + shorter than 100 no longer ends a list while `Link: rel="next"` says more + follow. A read stopped at the three-request bound keeps that list's cursor + where it stopped, and a later pass reads on from there, instead of moving + past the rows it never read. And each list is read by `updated_at` rather + than by page offset, so a row updated mid-read can't push an unread one onto + a page already read, except among rows that share one `updated_at` second, + which are still paged by offset. +- Cursors are whole seconds. GitHub's `since` part way through a second leaves + out every row stamped in that second, so a cursor at the pass's start made + the next pass miss rows changed later in the second the pass started in. +- Each repository's three lists keep a cursor of their own in the state file + (`pollCursors`), starting from `lastPollAt` until they have one. + `lastPollAt` never passes where a list stopped. +- A row announced as opened isn't announced as opened again when it's read + again inside its window, after a stopped read or a refused delivery + (`pollNew` in the state file). A first pass that a delivery refused keeps + the window it read. +- `getCheckRuns` reads on past a short page while `total_count` says more + runs follow, so such a head is no longer left undecided. diff --git a/docs/contract-notes.md b/docs/contract-notes.md index 5ddd932..1f9414e 100644 --- a/docs/contract-notes.md +++ b/docs/contract-notes.md @@ -33,7 +33,11 @@ camel case, so a reviewer can map a contract bullet to a type by sight. `issues?state=all&sort=updated&since=`), `listIssueCommentsSince` (`issues/comments?sort=updated&since=`) and `listReviewCommentsSince` (`pulls/comments?sort=updated&since=`), returning `PolledIssue` / - `PolledComment` rows. The poller (.3) keeps the cursor `lastPollAt`, dedups + `PolledComment` rows, each list read by `updated_at` rather than by page + offset, three requests a pass and more only to finish a second. The poller + (.3) keeps a cursor for each repository's list in whole seconds + (`pollCursors`, starting from `lastPollAt` until it has its own), which a + read stopped at that bound leaves where it stopped, dedups with the key `::` through `StateStore.seenPollKey` (ring of `POLL_DEDUP_RING` = 2000 keys, .7), names poll events `poll:::`, and asks `getCheckRuns` for the heads of @@ -45,11 +49,17 @@ camel case, so a reviewer can map a contract bullet to a type by sight. and a recipe keyed on `github:issue=opened` could never fire on it (found by the WP10 end-to-end run). A `since` query still carries no action, but it does carry `created_at`, and that is enough: `PolledIssue` gains `createdAt` - and `normalizePolledIssue(repo, row, receivedAt, since, previousHeadSha?)` + and `normalizePolledIssue(repo, row, receivedAt, since, previousHeadSha?, announced?)` derives the kind from it. - - An **issue** row created at or after `since` is `issue.opened`; otherwise - `issue.edited`, as before. Closed rows stay `issue.closed`. - - A **pull-request** row created at or after `since` is `pull.opened`. An + - An **issue** row created at or after `since`, and not announced as new + already (`announced`), is `issue.opened`; otherwise + `issue.edited`, as before. Closed rows stay `issue.closed`. `since` is + the start of the list's window, which stays put while a stopped read is + caught up. The state also keeps the rows announced as new while their + `created_at` is inside the window (`pollNew`), so a row read again there + is not opened twice. + - A **pull-request** row created at or after `since`, and not announced + as new already, is `pull.opened`. An open row seen again is `pull.synchronize` when its head sha moved away from the one the last pass recorded, and is **dropped** (null) when it did not: every comment on a pull request bumps its row, and the comment and diff --git a/src/github-adapter.ts b/src/github-adapter.ts index 1c4d991..fddfc3b 100644 --- a/src/github-adapter.ts +++ b/src/github-adapter.ts @@ -55,6 +55,7 @@ import { type Logger, type PolledComment, type PolledIssue, + type PollRead, type PullRequest, type PullRequestFile, type RepoRef, @@ -111,13 +112,13 @@ const CHECK_RUN_PAGE = 100; */ const CHECK_RUN_MAX_PAGES = 5; /** - * Documented exception to the 50-entry cap: the three `…Since` polls (contract - * amendment 1) feed the event pipeline, not a rendered list, so dropping rows - * drops events. They page 100 at a time up to this many rows per call. + * Requests one `…Since` poll makes before it stops, past which it goes on + * only to finish a second it is part way through, and leaves the rest for a + * later pass, which reads on from where it stopped. A documented exception to + * the 50-entry cap: these lists feed the event pipeline, not a rendered list, + * so they read 100 rows a request. */ -export const POLL_MAX_ENTRIES = 300; -/** Pages a `…Since` poll will ask for before it stops, whatever is left. */ -const POLL_MAX_PAGES = 3; +export const POLL_MAX_PAGES = 3; /** * Amendment 1: "at most 20 PRs per poll". `listIssuesSince` fills `headSha` * for at most this many open pull requests; the rest come back without one. @@ -412,6 +413,15 @@ function numberFromUrl(url: string | null | undefined): number { return m ? Number(m[1]) : 0; } +/** + * Whether a response's `Link` header names a next page, which is GitHub's + * signal that more follow. A page's length is not: a page shorter than + * `per_page` is no proof of the end. + */ +function hasNextPage(link: string | undefined): boolean { + return /(?:^|,)\s*<[^>]*>\s*;\s*rel="next"/.test(link ?? ''); +} + /** * Roll a sha's check runs up to one conclusion. A run that has not completed * makes the whole sha undecided (`null`); otherwise the worst conclusion wins, @@ -1041,7 +1051,8 @@ export class GitHubAdapter implements GitHubApi { * * The endpoint is paginated, and a busy commit can carry hundreds of runs: * reading one page and calling it `success` is how a failure on page two - * becomes a green light. So pages are followed to {@link CHECK_RUN_MAX_PAGES}, + * becomes a green light. So pages are followed until `total_count` runs are + * read, up to {@link CHECK_RUN_MAX_PAGES} pages, whatever a page's length; * `total` is GitHub's own `total_count` rather than what we managed to read, * and a sha whose runs did not all fit is never reported as a definite * success — its conclusion is `null`, the same "not decided" the contract @@ -1050,6 +1061,8 @@ export class GitHubAdapter implements GitHubApi { async getCheckRuns(repo: RepoRef, sha: string): Promise { const runs: CheckRunRow[] = []; let totalCount = 0; + // Whether runs were left unread when the loop ended: then the sha is undecided. + let more = false; for (let page = 1; page <= CHECK_RUN_MAX_PAGES; page++) { const batch = await this.call('getCheckRuns', async () => { @@ -1065,10 +1078,13 @@ export class GitHubAdapter implements GitHubApi { }); if (page === 1) totalCount = batch.total; runs.push(...batch.rows); - if (batch.rows.length < CHECK_RUN_PAGE || runs.length >= totalCount) break; + // `total_count` says how many there are, so a short page is no end; + // a response without one leaves a full page as the sign of more. + more = totalCount > 0 ? runs.length < totalCount : batch.rows.length === CHECK_RUN_PAGE; + if (!more) break; } - const truncated = runs.length < totalCount; + const truncated = more; if (truncated) { this.log('warn', 'check runs did not fit in the page bound; the sha is reported undecided', { sha, @@ -1350,24 +1366,26 @@ export class GitHubAdapter implements GitHubApi { // ─────────────────────────────────────────────────────────────────────────── /** - * Issues and pull requests touched at or after `since`, oldest first so the - * poller's cursor only ever moves forward. Bounded by - * {@link POLL_MAX_ENTRIES} rather than by the 50-entry list cap: these rows - * become events, and a dropped row is a dropped event. + * Issues and pull requests touched at or after `since`, in the order read: + * oldest change first, except that a row changed while the list is read + * keeps the place it was first read at (the poller sorts what it + * delivers). Bounded by {@link POLL_MAX_PAGES} requests rather than by the + * 50-entry list cap: these rows become events, so a read the bound stops + * says where it stopped (`resumeAt`) and the next one reads on from there. * * `headSha` is filled for at most {@link POLL_PR_HEAD_LIMIT} open pull * requests (amendment 1's "at most 20 PRs per poll"); the issues endpoint * does not carry it, so each one costs a read. */ - async listIssuesSince(repo: RepoRef, since: string): Promise { - const rows = await this.pollPages('listIssuesSince', (page) => + async listIssuesSince(repo: RepoRef, since: string): Promise> { + const { rows, resumeAt } = await this.pollSince('listIssuesSince', since, (from, page) => this.octokit.request('GET /repos/{owner}/{repo}/issues', { owner: repo.owner, repo: repo.name, state: 'all', sort: 'updated', direction: 'asc', - since, + since: from, per_page: 100, page, }), @@ -1399,39 +1417,41 @@ export class GitHubAdapter implements GitHubApi { }); if (head) pull.headSha = head; } - return polled; + return resumeAt === undefined ? { rows: polled } : { rows: polled, resumeAt }; } /** Issue comments updated at or after `since`. Bounded like the issues poll. */ - async listIssueCommentsSince(repo: RepoRef, since: string): Promise { - const rows = await this.pollPages('listIssueCommentsSince', (page) => + async listIssueCommentsSince(repo: RepoRef, since: string): Promise> { + const { rows, resumeAt } = await this.pollSince('listIssueCommentsSince', since, (from, page) => this.octokit.request('GET /repos/{owner}/{repo}/issues/comments', { owner: repo.owner, repo: repo.name, sort: 'updated', direction: 'asc', - since, + since: from, per_page: 100, page, }), ); - return rows.map((row) => mapPolledComment(row, numberFromUrl(row.issue_url))); + const polled = rows.map((row) => mapPolledComment(row, numberFromUrl(row.issue_url))); + return resumeAt === undefined ? { rows: polled } : { rows: polled, resumeAt }; } /** Review comments updated at or after `since`. Bounded like the issues poll. */ - async listReviewCommentsSince(repo: RepoRef, since: string): Promise { - const rows = await this.pollPages('listReviewCommentsSince', (page) => + async listReviewCommentsSince(repo: RepoRef, since: string): Promise> { + const { rows, resumeAt } = await this.pollSince('listReviewCommentsSince', since, (from, page) => this.octokit.request('GET /repos/{owner}/{repo}/pulls/comments', { owner: repo.owner, repo: repo.name, sort: 'updated', direction: 'asc', - since, + since: from, per_page: 100, page, }), ); - return rows.map((row) => mapPolledComment(row, numberFromUrl(row.pull_request_url))); + const polled = rows.map((row) => mapPolledComment(row, numberFromUrl(row.pull_request_url))); + return resumeAt === undefined ? { rows: polled } : { rows: polled, resumeAt }; } /** @@ -1469,41 +1489,67 @@ export class GitHubAdapter implements GitHubApi { } /** - * Page a `…Since` poll: pages of 100 until a short page, {@link POLL_MAX_PAGES} - * pages, or {@link POLL_MAX_ENTRIES} rows, whichever comes first. + * Read a `…Since` list, which GitHub sorts by `updated_at`, oldest first. + * + * Each request after the first asks from the last row's `updated_at` + * rather than for the next page, so a row updated while the list is being + * read, which moves to the end, can only move into what is still to be + * read: an offset would slide an unread row onto a page already read. The + * rows of that second come back again and are dropped by id, keeping the + * later copy. When a whole page shares one second, the next request asks + * that second's next page; that paging is by offset, so it holds while + * GitHub keeps one second's rows in the same order from page to page, + * which hasn't been checked, and a row of that second changed meanwhile can + * move another past a page. Only {@link hasNextPage} says whether more are + * waiting. + * + * After {@link POLL_MAX_PAGES} requests the read stops, but never inside a + * second: the rows of the second it stopped in are left out, and + * `resumeAt` names that second for the next read to ask from. GitHub's + * `since` includes its own second, so nothing between is skipped, and no + * row this read returns comes back unchanged in the next. */ - private async pollPages( + private async pollSince( operation: string, - fetchPage: (page: number) => Promise<{ data: unknown }>, - ): Promise { - const rows: T[] = []; - let truncated = false; - for (let page = 1; ; page++) { - const batch = await this.call(operation, async () => { - const res = await fetchPage(page); - return (Array.isArray(res.data) ? res.data : []) as T[]; - }); - rows.push(...batch); - // A short page is the end of the list; a full one means more is waiting. - if (batch.length < 100) break; - if (page >= POLL_MAX_PAGES || rows.length >= POLL_MAX_ENTRIES) { - truncated = true; - break; + since: string, + fetchPage: (from: string, page: number) => Promise<{ data: unknown; headers: { link?: string } }>, + ): Promise<{ rows: T[]; resumeAt?: string }> { + const rows = new Map(); + let from = since; + let page = 1; + // True while the requests page through one `since`, which a stop must not cut. + let byPage = false; + for (let requests = 0; ; requests++) { + if (requests >= POLL_MAX_PAGES && !byPage) { + const stop = Date.parse(from); + for (const [id, row] of rows) { + if (Date.parse(row.updated_at ?? '') >= stop) rows.delete(id); + } + this.log('warn', 'poll page cap reached; a later pass reads the rest from resumeAt', { + operation, + kept: rows.size, + resumeAt: from, + }); + return { rows: [...rows.values()], resumeAt: from }; } - } - if (rows.length > POLL_MAX_ENTRIES) { - rows.length = POLL_MAX_ENTRIES; - truncated = true; - } - if (truncated) { - // The poller's cursor only advances over rows it saw, so the rest come - // back next tick rather than being lost. - this.log('warn', 'poll page cap reached; the cursor will pick the rest up next tick', { - operation, - kept: rows.length, + const { batch, more } = await this.call(operation, async () => { + const res = await fetchPage(from, page); + return { batch: (Array.isArray(res.data) ? res.data : []) as T[], more: hasNextPage(res.headers.link) }; }); + // A row read again replaces its earlier copy. + for (const row of batch) rows.set(row.id ?? 0, row); + if (!more) return { rows: [...rows.values()] }; + const lastAt = batch.at(-1)?.updated_at ?? ''; + const last = Date.parse(lastAt); + // One second's rows fill the page, or there is no stamp to read on from: by page. + byPage = !Number.isFinite(last) || last === Date.parse(from); + if (byPage) { + page += 1; + } else { + from = lastAt; + page = 1; + } } - return rows; } /** A pull request payload plus the check runs on its head sha. */ diff --git a/src/normalize.ts b/src/normalize.ts index 718cd47..eca7592 100644 --- a/src/normalize.ts +++ b/src/normalize.ts @@ -457,7 +457,7 @@ function normalizeCheckRun(base: EventBase, payload: unknown, actor: EventActor * later one. An unparseable stamp answers false, which is the fail-safe * direction: the row keeps the kind it had before amendment 2a. */ -function atOrAfter(at: string, since: string): boolean { +export function atOrAfter(at: string, since: string): boolean { const a = Date.parse(at); const b = Date.parse(since); return Number.isFinite(a) && Number.isFinite(b) && a >= b; @@ -468,10 +468,15 @@ function atOrAfter(at: string, since: string): boolean { * * A `since` query reports *that* a row changed, never *how* — there is no * action — so amendment 2a derives the one from the row's own `created_at` - * against the cursor the pass queried with: + * against `since`, the start of the window the poller has not finished + * reading (`PollCursor.since` in types.ts): * - * - **opened** when the row was created at or after `since`: this pass is - * the first that could have seen it, so `issue.opened` / `pull.opened`. + * - **opened** when the row was created at or after `since` and the poller + * hasn't announced it as new already (`announced`): this pass is the + * first that could have seen it, so `issue.opened` / `pull.opened`. A + * row is read again inside its window after a stopped read, a refused + * delivery, or in the second a pass starts in, and `announced` keeps it + * from being opened twice. * - otherwise an **issue** row is `issue.edited` (or `issue.closed`, below) * when its state, title or body differs from what the last pass recorded * for it ({@link polledMarker}), and is **dropped** when they do not. A @@ -517,9 +522,10 @@ export function normalizePolledIssue( receivedAt: string, since: string, previousMarker?: string, + announced = false, ): NormalizedEvent | null { const key = pollKey('issue', row.id, row.updatedAt); - const isNew = atOrAfter(row.createdAt, since); + const isNew = !announced && atOrAfter(row.createdAt, since); const closed = row.state === 'closed'; const changed = previousMarker === undefined || polledMarker(row) !== previousMarker; let kind: NormalizedEventKind; diff --git a/src/poller.ts b/src/poller.ts index f2d91b4..3b2ce11 100644 --- a/src/poller.ts +++ b/src/poller.ts @@ -17,11 +17,16 @@ * - **Record only after delivery.** The dedup ring is read once per pass * and a key is written back only once its event has been handed to the * listener without throwing, so a failed delivery is re-attempted on the - * next pass instead of being silently consumed. The cursor advances for + * next pass instead of being silently consumed. The cursors advance for * the same reason: only when the whole pass delivered. + * + * Each repository's three lists keep a cursor of their own + * ({@link PollCursor}), so a list whose read the page bound stopped reads on + * from where it stopped, while the rest move to the start of the pass. */ import { + atOrAfter, CheckGate, checkMarkOf, checksPollKey, @@ -44,6 +49,8 @@ import { type NormalizedEvent, type PolledComment, type PolledIssue, + type PollCursor, + type PollRead, type RepoRef, type ThreadRef, } from './types.js'; @@ -91,6 +98,13 @@ export type PollerApi = Pick< export interface IngressState { lastPollAt(): string | null; setLastPollAt(when: string): void; + /** Each list's own cursor, keyed by {@link pollCursorKey}. */ + pollCursors(): Readonly>; + /** Replace them all: a list left out starts from `lastPollAt` next time. */ + setPollCursors(cursors: Record): void; + /** Per list, the rows announced as new in its window, id → created_at. */ + polledNew(): Readonly>>; + setPolledNew(rows: Record>): void; /** Records `key` in the ring and reports whether it was already there. */ seenPollKey(key: string): boolean; /** The head sha the last pass recorded for this pull request, or null. */ @@ -122,9 +136,21 @@ export interface PollerOptions { checkGate?: CheckGate; } +/** The three lists the poller reads from each repository. */ +export type PollList = 'issues' | 'issue_comments' | 'review_comments'; + +/** `/:`, the key of a list's cursor in the state. */ +export function pollCursorKey(repo: RepoRef, list: PollList): string { + return `${repoSlug(repo)}:${list}`; +} + /** What one pass did, for the caller's diagnostics and for the tests. */ export interface PollResult { - /** The `since` the pass queried with. */ + /** + * The start of a list with no cursor of its own: the stored `lastPollAt` + * as kept, or one interval back in whole seconds. The list asks GitHub from + * its whole second. + */ since: string; /** Events handed to the listener, in order. */ delivered: NormalizedEvent[]; @@ -132,7 +158,10 @@ export interface PollResult { duplicates: number; /** Events whose listener threw; their keys were not recorded. */ failed: number; - /** Whether the cursor was advanced to the start of this pass. */ + /** + * Whether the cursors moved: each list's to the start of this pass, or, for + * a read the page bound stopped, to where it stopped. + */ cursorAdvanced: boolean; /** `getCheckRuns` calls this pass made; never above the cap. */ checkLookups: number; @@ -188,6 +217,26 @@ interface Candidate { * at collection time loses nothing. */ head?: { channel: string; sha: string }; + /** + * An issue row announced as new, remembered in `pollNew` on the same terms + * as {@link Candidate.head}, so a refused pass's accepted openings stay + * remembered when the retry reads them again. + */ + fresh?: { list: string; id: string; createdAt: string }; +} + +/** + * An instant as a whole second, in the cursor's form. GitHub stamps rows to + * the second, and a `since` part way through a second leaves out the rows + * stamped in it. + */ +function wholeSecond(ms: number): string { + return new Date(Math.floor(ms / 1000) * 1000).toISOString(); +} + +/** A stored position as a whole second, whatever spelling or precision it was kept in. */ +function atSecond(iso: string): string { + return wholeSecond(Date.parse(iso)); } export class Poller { @@ -303,26 +352,37 @@ export class Poller { * One pass: query every allow-listed repository since the cursor, drain as * much of the check backlog as the per-pass cap allows, drop what the dedup * ring already holds, deliver the rest oldest-first, record each key after - * its delivery, and advance the cursor if no delivery failed. + * its delivery, and advance the cursors if no delivery failed. * * A failed *lookup* does not hold the cursor back: the head stays in the * backlog instead, which is the thing that gets retried. A failed * *delivery* does hold it back, so its row is queried again. * * For that retry to mean anything, nothing a refused event would be - * recognised by on the next pass may be written before it is accepted. Two - * things are therefore committed inside the delivery loop and nowhere else: - * the dedup key, and the pull-request head a `pull.synchronize` announces - * ({@link Candidate.head}). + * recognised by on the next pass may be written before it is accepted. + * Three things are therefore committed inside the delivery loop and nowhere + * else: the dedup key, the pull-request head a `pull.synchronize` announces + * ({@link Candidate.head}), and a row announced as new ({@link Candidate.fresh}). */ async pollOnce(): Promise { const startedAt = this.now(); - const since = this.state.lastPollAt() ?? new Date(startedAt.getTime() - this.effectiveIntervalMs).toISOString(); + const stored = this.state.lastPollAt(); + const since = stored ?? wholeSecond(startedAt.getTime() - this.effectiveIntervalMs); + // A first pass keeps the window it reads before it reads, so however the + // pass ends, refused or thrown, the next asks from the same start rather + // than one the clock has moved on. + if (stored === null) this.state.setLastPollAt(since); const receivedAt = startedAt.toISOString(); + // Where a list read to the end starts next: this pass's start, so a row + // changed after its read is asked for, in whole seconds, so its second is too. + const startSecond = wholeSecond(startedAt.getTime()); const candidates: Candidate[] = []; + // Where each list reads from next, written only if the pass delivers. A + // repository no longer in GITHUB_REPOS drops out of it. + const cursors: Record = {}; for (const repo of this.repos) { - candidates.push(...(await this.collectRows(repo, since, receivedAt))); + candidates.push(...(await this.collectRows(repo, since, startSecond, receivedAt, cursors))); } // One drain for every repository together, so the cap is a cap on the // pass and not on each repository in turn. @@ -355,6 +415,12 @@ export class Poller { // event, or the ring says an earlier pass already got it through. const commitHead = (): void => { if (candidate.head) this.state.setPolledHead(candidate.head.channel, candidate.head.sha); + if (candidate.fresh) { + const { list, id, createdAt } = candidate.fresh; + const all = { ...this.state.polledNew() }; + all[list] = { ...all[list], [id]: createdAt }; + this.state.setPolledNew(all); + } }; if (alreadySeen.has(candidate.key) || deliveredThisPass.has(candidate.key)) { @@ -402,7 +468,20 @@ export class Poller { result.checksBacklog = this.backlog.size; if (result.failed === 0) { - this.state.setLastPollAt(receivedAt); + this.state.setPollCursors(cursors); + // Every list has delivered what changed before this pass's start, or + // before where a stopped read stopped, whichever is earlier. + const stops = Object.values(cursors).flatMap((cursor) => cursor.resumeAt ?? []); + this.state.setLastPollAt([startSecond, ...stops].reduce((a, b) => (Date.parse(b) < Date.parse(a) ? b : a))); + // A row stays remembered as announced new while it is created inside its + // list's window, the only rows the window's start would call new again. + const remembered: Record> = {}; + for (const [list, rows] of Object.entries(this.state.polledNew())) { + const window = cursors[list]?.since; + const kept = Object.entries(rows).filter(([, createdAt]) => window !== undefined && atOrAfter(createdAt, window)); + if (kept.length > 0) remembered[list] = Object.fromEntries(kept); + } + this.state.setPolledNew(remembered); result.cursorAdvanced = true; } else { this.logger.warn('poll cursor held back: a delivery failed in this pass', { @@ -425,20 +504,45 @@ export class Poller { /** * The issue, comment and review-comment rows one repository changed since - * the cursor. Every open pull-request head it saw is *enqueued* for a check - * lookup rather than looked up here, so the cap and the retries belong to - * the backlog and not to one repository's turn. + * each list's cursor. Every open pull-request head it saw is *enqueued* for + * a check lookup rather than looked up here, so the cap and the retries + * belong to the backlog and not to one repository's turn. + * + * Each list's next cursor goes into `next`: the start of this pass when + * the read reached the end of the list, and otherwise where it stopped, + * with the window's start kept, since the rows past the stop are still + * to be read. */ - private async collectRows(repo: RepoRef, since: string, receivedAt: string): Promise { + private async collectRows( + repo: RepoRef, + fallback: string, + startSecond: string, + receivedAt: string, + next: Record, + ): Promise { + const cursors = this.state.pollCursors(); + const read = async (list: PollList, ask: (from: string) => Promise>) => { + const key = pollCursorKey(repo, list); + // Stored positions are read as whole seconds, so one kept with + // milliseconds, as by a release before this, starts at its second. + const since = atSecond(cursors[key]?.since ?? fallback); + const stop = cursors[key]?.resumeAt; + const { rows, resumeAt } = await ask(stop === undefined ? since : atSecond(stop)); + next[key] = resumeAt === undefined ? { since: startSecond } : { since, resumeAt }; + return { rows, since, key }; + }; const out: Candidate[] = []; - const issues = await this.api.listIssuesSince(repo, since); - for (const row of issues) { + const issues = await read('issues', (from) => this.api.listIssuesSince(repo, from)); + const announced = this.state.polledNew()[issues.key] ?? {}; + for (const row of issues.rows) { // What the last pass recorded for this thread (amendment 2a for a pull // request's head, a state-and-text fingerprint for an issue) turns a // real change into an event and drops the row a comment only bumped. const channel = channelId({ repo, number: row.number, kind: row.isPullRequest ? 'pull' : 'issue' }); - const event = normalizePolledIssue(repo, row, receivedAt, since, this.state.polledHead(channel) ?? undefined); + const seen = announced[String(row.id)] !== undefined; + const event = normalizePolledIssue(repo, row, receivedAt, issues.since, this.state.polledHead(channel) ?? undefined, seen); + const fresh = !seen && atOrAfter(row.createdAt, issues.since); const marker = polledMarker(row); const openHead = row.isPullRequest && row.state === 'open' && row.headSha ? row.headSha : undefined; if (event) { @@ -452,6 +556,7 @@ export class Poller { event, at: row.updatedAt, ...(marker === undefined ? {} : { head: { channel, sha: marker } }), + ...(fresh ? { fresh: { list: issues.key, id: String(row.id), createdAt: row.createdAt } } : {}), }); } else if (marker !== undefined) { // No event: this row was dropped *because* the marker is unchanged, so @@ -464,10 +569,10 @@ export class Poller { if (openHead !== undefined) this.enqueueHead(repo, row); } - out.push(...this.commentCandidates(repo, await this.api.listIssueCommentsSince(repo, since), 'issue_comment', receivedAt)); - out.push( - ...this.commentCandidates(repo, await this.api.listReviewCommentsSince(repo, since), 'review_comment', receivedAt), - ); + const comments = await read('issue_comments', (from) => this.api.listIssueCommentsSince(repo, from)); + out.push(...this.commentCandidates(repo, comments.rows, 'issue_comment', receivedAt)); + const reviewComments = await read('review_comments', (from) => this.api.listReviewCommentsSince(repo, from)); + out.push(...this.commentCandidates(repo, reviewComments.rows, 'review_comment', receivedAt)); return out; } diff --git a/src/state.ts b/src/state.ts index 09aab53..1833a5b 100644 --- a/src/state.ts +++ b/src/state.ts @@ -14,8 +14,11 @@ * - `rollback(checkpointId)` computes what has to be undone (delete the * created, restore the edited) and returns null for an unknown checkpoint * so the caller can answer -32005 (§8.7); - * - the polling cursor `lastPollAt` and the dedup ring of the last - * {@link POLL_DEDUP_RING} poll keys (contract amendment 1), and the last + * - the polling cursors, one per repository and list in `pollCursors` with + * `lastPollAt` where a list with none of its own starts, the rows + * announced as new while inside their list's window (`pollNew`), the + * dedup ring of the last {@link POLL_DEDUP_RING} poll keys (contract + * amendment 1), and the last * polled head sha of each open pull request, bounded by * {@link POLL_HEAD_MEMORY} (amendment 2a). * @@ -42,6 +45,7 @@ import { type CreatedComment, type EditedComment, type Logger, + type PollCursor, type RollbackPlan, type StateSnapshot, type StateStore, @@ -137,6 +141,8 @@ export function emptySnapshot(): GithubStateSnapshot { edited: {}, deleted: {}, lastPollAt: null, + pollCursors: {}, + pollNew: {}, pollKeys: [], pollHeads: {}, }; @@ -195,6 +201,10 @@ export class GithubState implements GithubStateStore { logger.warn(`state: ${file} is not a version ${STATE_VERSION} snapshot; starting empty`); return GithubState.empty(file, logger); } + const kept = (parsed as { lastPollAt?: unknown }).lastPollAt; + if (kept !== undefined && kept !== null && snap.lastPollAt === null) { + logger.warn(`state: ${file} has a lastPollAt that isn't a time (${JSON.stringify(kept)}); polling starts one interval back`); + } return new GithubState(file, snap, logger); } @@ -313,6 +323,22 @@ export class GithubState implements GithubStateStore { this.snap.lastPollAt = when; } + pollCursors(): Readonly> { + return this.snap.pollCursors; + } + + setPollCursors(cursors: Record): void { + this.snap.pollCursors = cursors; + } + + polledNew(): Readonly>> { + return this.snap.pollNew; + } + + setPolledNew(rows: Record>): void { + this.snap.pollNew = rows; + } + seenPollKey(key: string): boolean { if (this.snap.pollKeys.includes(key)) return true; this.snap.pollKeys.push(key); @@ -396,7 +422,16 @@ function normalize(parsed: unknown): GithubStateSnapshot | null { snap.created = buckets(raw.created); snap.edited = buckets(raw.edited); snap.deleted = buckets(raw.deleted); - snap.lastPollAt = typeof raw.lastPollAt === 'string' ? raw.lastPollAt : null; + // Only an instant: it is the start of every list with no cursor of its own. + snap.lastPollAt = + typeof raw.lastPollAt === 'string' && Number.isFinite(Date.parse(raw.lastPollAt)) ? raw.lastPollAt : null; + // Additive like `pollHeads`: a file written before them starts every list + // from `lastPollAt`, with no row remembered as announced new. + snap.pollCursors = cursors(raw.pollCursors); + snap.pollNew = {}; + if (raw.pollNew && typeof raw.pollNew === 'object' && !Array.isArray(raw.pollNew)) { + for (const [list, rows] of Object.entries(raw.pollNew)) snap.pollNew[list] = strings(rows); + } snap.pollKeys = Array.isArray(raw.pollKeys) ? raw.pollKeys.filter((k): k is string => typeof k === 'string') : []; // Additive since amendment 2a, so a state file written before it is still a // valid version-1 snapshot: an absent map just means no head is known yet. @@ -404,6 +439,25 @@ function normalize(parsed: unknown): GithubStateSnapshot | null { return snap; } +/** + * Poll cursors from untrusted JSON. An entry whose `since` isn't an instant is + * dropped, and its list starts from `lastPollAt`, which is never later than + * where any list stopped; a `resumeAt` that isn't one is left out, so the + * list reads from its `since`. + */ +function cursors(raw: unknown): Record { + if (!raw || typeof raw !== 'object' || Array.isArray(raw)) return {}; + const instant = (value: unknown): value is string => typeof value === 'string' && Number.isFinite(Date.parse(value)); + const out: Record = {}; + for (const [key, value] of Object.entries(raw as Record)) { + if (!value || typeof value !== 'object') continue; + const { since, resumeAt } = value as { since?: unknown; resumeAt?: unknown }; + if (!instant(since)) continue; + out[key] = instant(resumeAt) ? { since, resumeAt } : { since }; + } + return out; +} + /** A `Record` from untrusted JSON; non-string values dropped. */ function strings(raw: unknown): Record { if (!raw || typeof raw !== 'object' || Array.isArray(raw)) return {}; diff --git a/src/types.ts b/src/types.ts index 1a2167a..cfd314f 100644 --- a/src/types.ts +++ b/src/types.ts @@ -353,8 +353,9 @@ export interface PolledIssue { author: EventActor; /** * When the row was opened (amendment 2a). A `since` query says only *that* a - * row changed; comparing this against the cursor the pass queried with is - * what tells an opened issue from an edited one. + * row changed; comparing this against the start of its list's window + * (`PollCursor.since`), for a row not yet announced as new, is what tells + * an opened issue from an edited one. */ createdAt: string; updatedAt: string; @@ -382,6 +383,36 @@ export interface PolledComment { inReplyTo?: number; } +/** One read of a poll list: its rows, and where a read the page bound stopped. */ +export interface PollRead { + rows: T[]; + /** + * Set when the read stopped before the end of the list: the `updated_at` + * second it stopped in. The rows of that second and later were left for + * the next read, which asks from here. + */ + resumeAt?: string; +} + +/** + * Where one poll list (a repository's issues, its issue comments or its + * review comments) reads from, keyed `/:`. + */ +export interface PollCursor { + /** + * The start of the window this list has not finished: the start of the + * last pass that read it to the end with every delivery accepted, in whole + * seconds. A row created at or after it is new (amendment 2a), unless + * `pollNew` says it was announced as new already. + */ + since: string; + /** + * Set while a read the page bound stopped is being caught up: the second + * it stopped in, which the next read asks from instead of `since`. + */ + resumeAt?: string; +} + export interface BotIdentity { login: string; id: number; @@ -425,11 +456,11 @@ export interface GitHubApi { // ── Polling fallback (amendment 1) and identity ── /** `GET /repos/{o}/{r}/issues?state=all&sort=updated&since=`: issues and PRs. */ - listIssuesSince(repo: RepoRef, since: string): Promise; + listIssuesSince(repo: RepoRef, since: string): Promise>; /** `GET /repos/{o}/{r}/issues/comments?sort=updated&since=`. */ - listIssueCommentsSince(repo: RepoRef, since: string): Promise; + listIssueCommentsSince(repo: RepoRef, since: string): Promise>; /** `GET /repos/{o}/{r}/pulls/comments?sort=updated&since=`. */ - listReviewCommentsSince(repo: RepoRef, since: string): Promise; + listReviewCommentsSince(repo: RepoRef, since: string): Promise>; getAuthenticatedBot(): Promise; } @@ -840,7 +871,29 @@ export interface StateSnapshot { current: string | null; created: Record; edited: Record; + /** + * Every list of the repositories polled has delivered what changed before + * this: the start of the last pass whose every delivery was accepted, or + * where a read it stopped stopped, if earlier; before any pass has + * delivered, the window the first pass read. A list with no cursor of its + * own in `pollCursors` starts here, such as one of a repository newly in + * GITHUB_REPOS, or every list of a state file written before `pollCursors` + * existed. Read as a whole second. + */ lastPollAt: string | null; + /** + * Each poll list's own cursor, for the repositories the last pass read. + * **Additive**: a state file without it is still a valid version-1 snapshot. + */ + pollCursors: Record; + /** + * Per list, the rows announced as new (id → created_at), so a row read + * again inside its list's window, after a stopped read, a refused delivery + * or in the second a pass started in, is not announced as new twice. + * Written when the row's event is accepted, and pruned to the rows created + * inside the window when a pass delivers. **Additive**, like `pollCursors`. + */ + pollNew: Record>; /** Ring of the last {@link POLL_DEDUP_RING} `::` keys (amendment 1). */ pollKeys: string[]; /** @@ -885,6 +938,13 @@ export interface StateStore { checkpointStatus(checkpointId: string): CheckpointStatus; lastPollAt(): string | null; setLastPollAt(when: string): void; + /** Each poll list's own cursor, keyed `/:`. */ + pollCursors(): Readonly>; + /** Replace them all: a list left out has no cursor of its own. */ + setPollCursors(cursors: Record): void; + /** Per list, the rows announced as new in its window (`StateSnapshot.pollNew`). */ + polledNew(): Readonly>>; + setPolledNew(rows: Record>): void; /** True when the key was already seen; otherwise records it in the ring and returns false. */ seenPollKey(key: string): boolean; /** diff --git a/test/githubAdapter.test.ts b/test/githubAdapter.test.ts index 5e73202..fd0222a 100644 --- a/test/githubAdapter.test.ts +++ b/test/githubAdapter.test.ts @@ -24,15 +24,19 @@ import { after, test } from 'node:test'; import { GitHubAdapter, GitHubApiError, - POLL_MAX_ENTRIES, + POLL_MAX_PAGES, POLL_PR_HEAD_LIMIT, REDACTED, redactSecrets, summarizeChecks, } from '../src/github-adapter.js'; +import { Poller } from '../src/poller.js'; +import { GithubState } from '../src/state.js'; import { GITHUB_API_METHODS, readConfig, + silentLogger, + type NormalizedEvent, type Config, type Logger, type RepoRef, @@ -998,7 +1002,7 @@ test('listIssuesSince asks the documented query and fills the head sha of open P ], }); - const rows = await r.adapter.listIssuesSince(REPO, since); + const { rows } = await r.adapter.listIssuesSince(REPO, since); // The list carries closed_by; a closed row is attributed to its closer. assert.deepEqual(rows[2].closedBy, { login: 'rayforce-clone[bot]', isBot: true }); assert.equal(rows[0].closedBy, undefined); @@ -1020,22 +1024,327 @@ test('listIssuesSince asks the documented query and fills the head sha of open P assert.equal(r.fake.for('GET /repos/anima-research/rayforce/pulls/12').length, 1); }); -test('the issues poll pages up to its documented bound and no further', async () => { +/** An issue row for a paging test: its number is its id, created before any cursor below. */ +function pagedIssue(id: number, updatedAt: string) { + return { ...ISSUE_PAYLOAD, id, number: id, created_at: '2026-10-01T00:00:00Z', updated_at: updatedAt }; +} + +/** `2026-10-10T01:00:00Z` plus `s` seconds, as GitHub stamps a row. */ +function second(s: number): string { + return new Date(Date.parse('2026-10-10T01:00:00Z') + s * 1000).toISOString().replace('.000Z', 'Z'); +} + +/** + * Serve `rows` as GitHub serves a `…Since` list: the rows updated at or after + * `since` (its own second included), oldest change first and then by id, 100 + * to a page by offset, with `Link: rel="next"` while more follow. `rows` is + * read at each request, so a test can update a row between two of them. + */ +function serveSince(r: Rig, route: string, rows: { id: number; updated_at: string }[]): void { + r.fake.on(route, (req) => { + const since = Date.parse(String(req.query.since)); + const page = Number(req.query.page ?? '1'); + const match = rows + .filter((row) => Date.parse(row.updated_at) >= since) + .sort((a, b) => Date.parse(a.updated_at) - Date.parse(b.updated_at) || a.id - b.id); + const more = page * 100 < match.length; + return { + body: match.slice((page - 1) * 100, page * 100), + ...(more ? { headers: { link: `<${r.fake.baseUrl}${route.split(' ')[1]}?page=${page + 1}>; rel="next"` } } : {}), + }; + }); +} + +const ISSUES = 'GET /repos/anima-research/rayforce/issues'; + +test('a poll follows rel="next" past a short page', async () => { + // GitHub answered 99 rows and said more follow: the 99 are not the end. + const r = await rig(); + r.fake.on(ISSUES, (req) => + req.query.page === '1' && req.query.since === '2026-10-10T00:00:00Z' + ? { + body: Array.from({ length: 99 }, (_, i) => pagedIssue(1 + i, second(i))), + headers: { link: `<${r.fake.baseUrl}/x?page=2>; rel="next"` }, + } + : { body: Array.from({ length: 50 }, (_, i) => pagedIssue(100 + i, second(99 + i))) }, + ); + + const { rows, resumeAt } = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + assert.equal(rows.length, 149); + assert.equal(resumeAt, undefined); + assert.equal(r.fake.for(ISSUES).length, 2); +}); + +test('each request after the first asks from the last row it read, and keeps one copy of each row', async () => { + const r = await rig(); + const all = Array.from({ length: 250 }, (_, i) => pagedIssue(1 + i, second(i))); + serveSince(r, ISSUES, all); + + const { rows, resumeAt } = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + assert.deepEqual( + r.fake.for(ISSUES).map((c) => [c.query.since, c.query.page]), + [ + ['2026-10-10T00:00:00Z', '1'], + [second(99), '1'], + [second(198), '1'], + ], + ); + // The last row of each page comes back first on the next one, and is kept once. + assert.deepEqual( + rows.map((row) => row.id), + all.map((row) => row.id), + ); + assert.equal(resumeAt, undefined); +}); + +test('a row updated between two requests cannot push an unread row onto a page already read', async () => { const r = await rig(); - const page = (n: number) => - Array.from({ length: 100 }, (_, i) => ({ ...ISSUE_PAYLOAD, id: n * 1000 + i, number: n * 1000 + i })); - r.fake.on('GET /repos/anima-research/rayforce/issues', (req) => ({ body: page(Number(req.query.page)) })); + const all = Array.from({ length: 150 }, (_, i) => pagedIssue(1 + i, second(i))); + r.fake.on(ISSUES, (req, nth) => { + // Someone comments on #5 once page 1 has been read: it moves to the end. + if (nth === 1) all[4] = { ...all[4], updated_at: second(500) }; + const since = Date.parse(String(req.query.since)); + const page = Number(req.query.page ?? '1'); + const match = all + .filter((row) => Date.parse(row.updated_at) >= since) + .sort((a, b) => Date.parse(a.updated_at) - Date.parse(b.updated_at)); + const more = page * 100 < match.length; + return { + body: match.slice((page - 1) * 100, page * 100), + ...(more ? { headers: { link: `<${r.fake.baseUrl}/x?page=${page + 1}>; rel="next"` } } : {}), + }; + }); - const rows = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); - assert.equal(rows.length, POLL_MAX_ENTRIES); + const { rows } = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + const ids = new Set(rows.map((row) => row.id)); assert.deepEqual( - r.fake.for('GET /repos/anima-research/rayforce/issues').map((c) => c.query.page), - ['1', '2', '3'], + all.map((row) => row.id).filter((id) => !ids.has(id)), + [], + 'every row is read, #101 included, which page 2 by offset would have skipped', + ); + assert.equal(rows.find((row) => row.id === 5)?.updatedAt, second(500), 'the later copy of #5 is the one kept'); +}); + +test('a read never stops inside a second, even past its request bound', async () => { + // 350 rows share one second: the read pages through it, then stops nowhere inside it. + const r = await rig(); + const all = [ + ...Array.from({ length: 350 }, (_, i) => pagedIssue(1 + i, second(1))), + ...Array.from({ length: 5 }, (_, i) => pagedIssue(400 + i, second(2 + i))), + ]; + serveSince(r, ISSUES, all); + + const { rows, resumeAt } = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + assert.equal(rows.length, 355); + assert.equal(resumeAt, undefined); + assert.deepEqual( + r.fake.for(ISSUES).map((c) => [c.query.since, c.query.page]), + [ + ['2026-10-10T00:00:00Z', '1'], + [second(1), '1'], + [second(1), '2'], + [second(1), '3'], + [second(1), '4'], + ], + ); +}); + +test('a Link header without rel="next" ends the read', async () => { + // GitHub's last page names only prev and first. + const r = await rig(); + r.fake.on(ISSUES, (_req, nth) => + nth === 0 + ? { + body: Array.from({ length: 100 }, (_, i) => pagedIssue(1 + i, second(i))), + headers: { link: `<${r.fake.baseUrl}/x?page=1>; rel="prev", <${r.fake.baseUrl}/x?page=1>; rel="first"` }, + } + : { body: [] }, + ); + + const { rows } = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + assert.equal(rows.length, 100); + assert.equal(r.fake.for(ISSUES).length, 1); +}); + +test('both comment lists stop at the request bound and say where', async () => { + const r = await rig(); + const comment = (id: number) => ({ + ...COMMENT_PAYLOAD, + id, + updated_at: second(id - 1), + issue_url: 'https://api.github.com/repos/anima-research/rayforce/issues/7', + pull_request_url: 'https://api.github.com/repos/anima-research/rayforce/pulls/12', + }); + const all = Array.from({ length: 350 }, (_, i) => comment(1 + i)); + serveSince(r, 'GET /repos/anima-research/rayforce/issues/comments', all); + serveSince(r, 'GET /repos/anima-research/rayforce/pulls/comments', all); + + for (const read of [ + await r.adapter.listIssueCommentsSince(REPO, '2026-10-10T00:00:00Z'), + await r.adapter.listReviewCommentsSince(REPO, '2026-10-10T00:00:00Z'), + ]) { + assert.equal(read.resumeAt, second(297)); + assert.equal(read.rows.length, 297); + } +}); + +test('a page with no stamp to read on from is followed by the next page of the same since', async () => { + const r = await rig(); + r.fake.on(ISSUES, (req) => + req.query.page === '1' + ? { + body: Array.from({ length: 100 }, (_, i) => ({ ...pagedIssue(1 + i, ''), updated_at: null })), + headers: { link: `<${r.fake.baseUrl}/x?page=2>; rel="next"` }, + } + : { body: Array.from({ length: 3 }, (_, i) => pagedIssue(200 + i, second(i))) }, + ); + + const { rows } = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + assert.equal(rows.length, 103); + assert.deepEqual( + r.fake.for(ISSUES).map((c) => [c.query.since, c.query.page]), + [ + ['2026-10-10T00:00:00Z', '1'], + ['2026-10-10T00:00:00Z', '2'], + ], + ); +}); + +test('a read the request bound stops leaves the second it stopped in for the next read', async () => { + const r = await rig(); + const all = Array.from({ length: 350 }, (_, i) => pagedIssue(1 + i, second(i))); + serveSince(r, ISSUES, all); + + const first = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + assert.equal(r.fake.for(ISSUES).length, POLL_MAX_PAGES); + // The third request ended at row 298's second, so that row waits for the next read. + assert.equal(first.resumeAt, second(297)); + assert.deepEqual( + first.rows.map((row) => row.id), + all.slice(0, 297).map((row) => row.id), ); assert.ok( - r.logs.some((l) => l.startsWith('warn poll page cap reached')), - 'a capped poll says so rather than dropping rows silently', + r.logs.some((l) => l.startsWith('warn poll page cap reached; a later pass reads the rest from resumeAt')), + 'a stopped read says so', + ); + + const second2 = await r.adapter.listIssuesSince(REPO, first.resumeAt as string); + assert.deepEqual( + [...first.rows, ...second2.rows].map((row) => row.id), + all.map((row) => row.id), + 'the two reads together return every row once', ); + assert.equal(second2.resumeAt, undefined); +}); + +/** A poller over a rig's adapter, its pass start set by `at`. */ +function pollerOver(r: Rig, state: GithubState, at: { now: string }, delivered: NormalizedEvent[]): Poller { + serveSince(r, 'GET /repos/anima-research/rayforce/issues/comments', []); + serveSince(r, 'GET /repos/anima-research/rayforce/pulls/comments', []); + return new Poller({ + api: r.adapter, + repos: [REPO], + state, + intervalMs: 60_000, + logger: silentLogger(), + now: () => new Date(at.now), + onEvent: async (event) => { + delivered.push(event); + }, + }); +} + +test('a pass the request bound stops is read on from where it stopped, and every row arrives once', async () => { + const r = await rig(); + const all = Array.from({ length: 350 }, (_, i) => pagedIssue(1 + i, second(i))); + serveSince(r, ISSUES, all); + const state = GithubState.empty(null, silentLogger()); + state.setLastPollAt('2026-10-10T00:00:00.000Z'); + const at = { now: '2026-10-10T02:00:00.000Z' }; + const delivered: NormalizedEvent[] = []; + const poller = pollerOver(r, state, at, delivered); + + const first = await poller.pollOnce(); + assert.equal(first.delivered.length, 297); + assert.deepEqual(state.pollCursors()['anima-research/rayforce:issues'], { + since: '2026-10-10T00:00:00.000Z', + resumeAt: second(297), + }); + assert.deepEqual(state.pollCursors()['anima-research/rayforce:issue_comments'], { since: at.now }); + assert.equal(state.lastPollAt(), second(297), 'every list has delivered what changed before the stop'); + + at.now = '2026-10-10T02:01:00.000Z'; + const second2 = await poller.pollOnce(); + assert.equal( + Date.parse(String(r.fake.for(ISSUES).at(-1)?.query.since)), + Date.parse(second(297)), + 'the next pass asks from where the last stopped', + ); + assert.equal(second2.delivered.length, 53); + assert.deepEqual( + delivered.map((e) => e.thread?.number), + all.map((row) => row.id), + 'every row once, in order', + ); + assert.deepEqual(state.pollCursors()['anima-research/rayforce:issues'], { since: at.now }); + assert.equal(state.lastPollAt(), at.now); +}); + +test('a row changed later in the second a pass started in arrives next pass, and one opened then is opened once', async () => { + const r = await rig(); + const all = [{ ...pagedIssue(1, '2026-10-10T02:00:00Z'), created_at: '2026-10-10T02:00:00Z' }]; + serveSince(r, ISSUES, all); + const state = GithubState.empty(null, silentLogger()); + state.setLastPollAt('2026-10-10T01:59:00.000Z'); + const at = { now: '2026-10-10T02:00:00.700Z' }; + const delivered: NormalizedEvent[] = []; + const poller = pollerOver(r, state, at, delivered); + + await poller.pollOnce(); + assert.deepEqual(delivered.map((e) => [e.thread?.number, e.kind]), [[1, 'issue.opened']]); + assert.deepEqual(state.pollCursors()['anima-research/rayforce:issues'], { since: '2026-10-10T02:00:00.000Z' }); + assert.deepEqual(state.polledNew()['anima-research/rayforce:issues'], { '1': '2026-10-10T02:00:00Z' }, 'created inside the next window, so kept'); + + // Still inside that second, after the read: #2 changes. Then #1 is edited. + all.push(pagedIssue(2, '2026-10-10T02:00:00Z')); + all[0] = { ...all[0], body: 'edited', updated_at: '2026-10-10T02:00:30Z' }; + at.now = '2026-10-10T02:01:00.000Z'; + await poller.pollOnce(); + assert.equal(r.fake.for(ISSUES).at(-1)?.query.since, '2026-10-10T02:00:00.000Z'); + assert.deepEqual( + delivered.slice(1).map((e) => [e.thread?.number, e.kind]), + [ + [2, 'issue.edited'], + [1, 'issue.edited'], + ], + ); +}); + +test('across a stopped read, a row opened in the window is opened once', async () => { + const r = await rig(); + const all: ReturnType[] = Array.from({ length: 350 }, (_, i) => pagedIssue(1 + i, second(i))); + // #10 is opened in the window and read before the stop; #320 is opened in + // the window but changed again past the stop, so only the next pass reads it. + all[9] = { ...all[9], created_at: second(0) }; + all[319] = { ...all[319], created_at: second(5) }; + serveSince(r, ISSUES, all); + const state = GithubState.empty(null, silentLogger()); + state.setLastPollAt('2026-10-10T00:00:00.000Z'); + const at = { now: '2026-10-10T02:00:00.000Z' }; + const delivered: NormalizedEvent[] = []; + const poller = pollerOver(r, state, at, delivered); + + await poller.pollOnce(); + assert.equal(delivered.find((e) => e.thread?.number === 10)?.kind, 'issue.opened'); + assert.deepEqual(state.polledNew()['anima-research/rayforce:issues'], { '10': second(0) }); + + // #10 is edited before the next pass reads on. + all[9] = { ...all[9], body: 'edited', updated_at: second(400) }; + at.now = '2026-10-10T02:01:00.000Z'; + const next = await poller.pollOnce(); + const kinds = Object.fromEntries(next.delivered.map((e) => [e.thread?.number, e.kind])); + assert.equal(kinds[320], 'issue.opened', 'opened in the window, though read after the stop'); + assert.equal(kinds[10], 'issue.edited', 'already announced as opened, so not again'); }); test('a poll never reads more than the amendment’s 20 PR heads', async () => { @@ -1055,7 +1364,7 @@ test('a poll never reads more than the amendment’s 20 PR heads', async () => { }); } - const rows = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + const { rows } = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); assert.equal(rows.filter((row) => row.headSha !== undefined).length, POLL_PR_HEAD_LIMIT); }); @@ -1077,14 +1386,14 @@ test('the comment polls carry the issue or pull number the comment hangs on', as ], }); - const issueComments = await r.adapter.listIssueCommentsSince(REPO, since); + const { rows: issueComments } = await r.adapter.listIssueCommentsSince(REPO, since); assert.equal(issueComments.length, 1); assert.equal(issueComments[0].number, 7); assert.equal(issueComments[0].path, undefined); assert.equal(r.fake.for('GET /repos/anima-research/rayforce/issues/comments')[0].query.sort, 'updated'); assert.equal(r.fake.for('GET /repos/anima-research/rayforce/issues/comments')[0].query.since, since); - const reviewComments = await r.adapter.listReviewCommentsSince(REPO, since); + const { rows: reviewComments } = await r.adapter.listReviewCommentsSince(REPO, since); assert.equal(reviewComments[0].number, 12); assert.equal(reviewComments[0].path, 'src/a.ts'); assert.equal(reviewComments[0].inReplyTo, 555); @@ -1282,6 +1591,50 @@ test('a single complete page still stops at one request', async () => { assert.equal(r.fake.for(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`).length, 1); }); +test('getCheckRuns reads on past a short page while total_count says more runs follow', async () => { + const r = await rig(); + r.fake.on(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`, (req) => ({ + body: { + total_count: 100, + check_runs: + Number(req.query.page) === 1 + ? successRuns(0, 99) + : [{ name: 'late-failure', status: 'completed', conclusion: 'failure', html_url: 'https://ci/late' }], + }, + })); + + const summary = await r.adapter.getCheckRuns(REPO, HEAD_SHA); + assert.equal(summary.conclusion, 'failure', 'a page of 99 was not the last'); + assert.equal(r.fake.for(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`).length, 2); +}); + +test('getCheckRuns without a total_count reads on past a full page', async () => { + const r = await rig(); + r.fake.on(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`, (req) => ({ + body: { + check_runs: + Number(req.query.page) === 1 + ? successRuns(0, 100) + : [{ name: 'late-failure', status: 'completed', conclusion: 'failure', html_url: 'https://ci/late' }], + }, + })); + + const summary = await r.adapter.getCheckRuns(REPO, HEAD_SHA); + assert.equal(summary.conclusion, 'failure'); + assert.equal(r.fake.for(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`).length, 2); +}); + +test('getCheckRuns without a total_count is undecided when the page bound stops it', async () => { + const r = await rig(); + r.fake.on(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`, (req) => ({ + body: { check_runs: successRuns((Number(req.query.page) - 1) * 100, 100) }, + })); + + const summary = await r.adapter.getCheckRuns(REPO, HEAD_SHA); + assert.equal(summary.conclusion, null, 'what was not read cannot be called green'); + assert.equal(r.fake.for(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`).length, 5); +}); + test('summarizeChecks refuses to call a truncated read success', () => { const complete = summarizeChecks('s', successRuns(0, 3)); assert.equal(complete.conclusion, 'success'); diff --git a/test/harness.ts b/test/harness.ts index 146194f..e2afa12 100644 --- a/test/harness.ts +++ b/test/harness.ts @@ -41,6 +41,7 @@ import { type IssueSearchHit, type PolledComment, type PolledIssue, + type PollRead, type PullRequest, type PullRequestFile, type RepoRef, @@ -555,7 +556,7 @@ export class FakeGitHubApi implements GitHubApi { return threadKey.startsWith(`${repoSlug(repo)}#`); } - async listIssuesSince(repo: RepoRef, since: string): Promise { + async listIssuesSince(repo: RepoRef, since: string): Promise> { this.record('listIssuesSince', repo, since); const rows: PolledIssue[] = []; for (const [k, i] of this.issues) { @@ -568,10 +569,10 @@ export class FakeGitHubApi implements GitHubApi { if (!this.inRepo(repo, k) || p.updatedAt <= since) continue; rows.push({ id: p.number, number: p.number, isPullRequest: true, title: p.title, body: p.body, state: p.state, author: p.author, createdAt: p.createdAt, updatedAt: p.updatedAt, html_url: p.html_url, headSha: p.head.sha }); } - return rows.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)); + return { rows: rows.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)) }; } - async listIssueCommentsSince(repo: RepoRef, since: string): Promise { + async listIssueCommentsSince(repo: RepoRef, since: string): Promise> { this.record('listIssueCommentsSince', repo, since); const rows: PolledComment[] = []; for (const [k, list] of this.comments) { @@ -581,10 +582,10 @@ export class FakeGitHubApi implements GitHubApi { if (c.updatedAt > since) rows.push({ id: c.id, number, author: c.author, body: c.body, updatedAt: c.updatedAt, html_url: c.html_url }); } } - return rows.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)); + return { rows: rows.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)) }; } - async listReviewCommentsSince(repo: RepoRef, since: string): Promise { + async listReviewCommentsSince(repo: RepoRef, since: string): Promise> { this.record('listReviewCommentsSince', repo, since); const rows: PolledComment[] = []; for (const [k, threads] of this.reviewThreads) { @@ -599,7 +600,7 @@ export class FakeGitHubApi implements GitHubApi { } } } - return rows.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)); + return { rows: rows.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)) }; } async getAuthenticatedBot(): Promise { diff --git a/test/integration.test.ts b/test/integration.test.ts index 7d0990b..0e342a9 100644 --- a/test/integration.test.ts +++ b/test/integration.test.ts @@ -536,7 +536,9 @@ test('amendment 2a: a polled pull-request row with an unmoved head never reaches // `receivedAt`, which is the real clock here, and these rows are stamped in // the fixture's past. The pass under test is "the second time this poller // sees this pull request", and that is what the head record carries. + // Each list's own cursor is cleared, so every list starts from it. state.setLastPollAt('2026-10-13T13:00:00.000Z'); + state.setPollCursors({}); await poller.pollOnce(); await until(() => pushes(h).length === afterFirst + 1, 'the comment, and nothing else'); assert.equal((pushes(h)[afterFirst].origin as { kind: string }).kind, 'comment.created'); diff --git a/test/poller.test.ts b/test/poller.test.ts index 73fd8d5..64fa3d3 100644 --- a/test/poller.test.ts +++ b/test/poller.test.ts @@ -27,6 +27,7 @@ import { POLL_DEDUP_RING, type ChecksSummary, type NormalizedEvent, + type PollCursor, type RepoRef, type StateStore, } from '../src/types.js'; @@ -77,6 +78,26 @@ class MemoryIngressState implements IngressState { snapshot(): { pollKeys: string[] } { return { pollKeys: [...this.keys] }; } + + cursors: Record = {}; + + pollCursors(): Readonly> { + return this.cursors; + } + + setPollCursors(cursors: Record): void { + this.cursors = { ...cursors }; + } + + fresh: Record> = {}; + + polledNew(): Readonly>> { + return this.fresh; + } + + setPolledNew(rows: Record>): void { + this.fresh = rows; + } } interface Rig { @@ -600,6 +621,7 @@ test('a row already in the gate is not delivered twice', async () => { // marker as an eviction would: now the gate, and not the cursor or the // marker, is the only thing that can stop the duplicate. r.state.setLastPollAt('2026-10-13T00:00:00.000Z'); + r.state.setPollCursors({}); r.state.heads.clear(); const second = await r.poller.pollOnce(); assert.deepEqual(second.delivered, []); @@ -773,11 +795,169 @@ test('IngressState is a structural subset of the frozen StateStore', () => { // And a runtime one, over the member names the poller actually calls. const state = new MemoryIngressState(); - for (const member of ['lastPollAt', 'setLastPollAt', 'seenPollKey', 'snapshot'] as const) { + for (const member of ['lastPollAt', 'setLastPollAt', 'pollCursors', 'setPollCursors', 'polledNew', 'setPolledNew', 'seenPollKey', 'snapshot'] as const) { assert.equal(typeof state[member], 'function', member); } }); +test('each list starts from lastPollAt until it has a cursor of its own, then keeps it', async () => { + // A state file from before per-list cursors carries `lastPollAt` alone, and + // one cursor for a repository no longer polled. + const r = rig({ cursor: '2026-10-13T00:00:00.000Z', repos: [TEST_REPO, OTHER_REPO] }); + r.state.setPollCursors({ 'gone/repo:issues': { since: '2026-10-01T00:00:00.000Z' } }); + + await r.poller.pollOnce(); + const asked = () => r.api.calls.filter((c) => String(c.method).endsWith('Since')).map((c) => c.args[1]); + assert.deepEqual(asked(), Array(6).fill('2026-10-13T00:00:00.000Z')); + const start = '2026-10-14T00:00:00.000Z'; + assert.deepEqual(r.state.pollCursors(), { + 'anima-research/rayforce:issues': { since: start }, + 'anima-research/rayforce:issue_comments': { since: start }, + 'anima-research/rayforce:review_comments': { since: start }, + 'other-org/other-repo:issues': { since: start }, + 'other-org/other-repo:issue_comments': { since: start }, + 'other-org/other-repo:review_comments': { since: start }, + }); + + // A list catching up reads from where it stopped; the others from their own. + r.state.setPollCursors({ + ...r.state.pollCursors(), + 'anima-research/rayforce:issues': { since: '2026-10-13T00:00:00.000Z', resumeAt: '2026-10-13T06:00:00Z' }, + }); + r.api.calls.length = 0; + await r.poller.pollOnce(); + assert.deepEqual(asked(), ['2026-10-13T06:00:00.000Z', start, start, start, start, start]); +}); + +test('a failed delivery leaves every list’s cursor where it was', async () => { + const api = new FakeGitHubApi(); + api.seedIssue({ updatedAt: '2026-10-13T12:00:00.000Z' }); + const r = rig({ + api, + onEvent: () => { + throw new Error('host refused the event'); + }, + }); + const before = { 'anima-research/rayforce:issues': { since: '2026-10-13T00:00:00.000Z', resumeAt: '2026-10-13T06:00:00Z' } }; + r.state.setPollCursors(before); + + const result = await r.poller.pollOnce(); + assert.equal(result.failed, 1); + assert.deepEqual(r.state.pollCursors(), before); +}); + +test('a refused pass keeps the rows it announced as new, so a retry does not open them again', async () => { + const api = new FakeGitHubApi(); + const issue = api.seedIssue({ number: 41, createdAt: '2026-10-13T12:00:00.000Z', updatedAt: '2026-10-13T12:00:00.000Z' }); + const pull = api.seedPullRequest({ number: 58, createdAt: '2026-10-13T12:00:00.000Z', updatedAt: '2026-10-13T12:00:00.000Z' }); + api.seedComment({ repo: TEST_REPO, number: 41, kind: 'issue' }, { id: 7001, updatedAt: '2026-10-13T12:00:01.000Z' }); + let refusals = 0; + const r = rig({ + api, + onEvent: (event) => { + if (event.kind === 'comment.created' && refusals++ === 0) throw new Error('host refused the event'); + }, + }); + + const first = await r.poller.pollOnce(); + assert.deepEqual(first.delivered.map((e) => e.kind).sort(), ['issue.opened', 'pull.opened']); + assert.equal(first.cursorAdvanced, false); + assert.deepEqual(r.state.fresh, { + 'anima-research/rayforce:issues': { '41': '2026-10-13T12:00:00.000Z', '58': '2026-10-13T12:00:00.000Z' }, + }); + + // Both change before the retry: the issue is edited, the pull request pushed. + issue.body = 'edited'; + issue.updatedAt = '2026-10-13T13:00:00.000Z'; + pull.head = { ...pull.head, sha: 'f'.repeat(40) }; + pull.updatedAt = '2026-10-13T13:00:00.000Z'; + const second = await r.poller.pollOnce(); + assert.deepEqual(second.delivered.map((e) => e.kind).sort(), ['comment.created', 'issue.edited', 'pull.synchronize']); + // The pass delivered, and its window starts after both were created. + assert.deepEqual(r.state.fresh, {}); +}); + +test('a refused opening leaves nothing remembered, so its retry announces it as opened', async () => { + const api = new FakeGitHubApi(); + api.seedIssue({ number: 41, createdAt: '2026-10-13T12:00:00.000Z', updatedAt: '2026-10-13T12:00:00.000Z' }); + let refusals = 0; + const r = rig({ + api, + onEvent: (event) => { + if (event.kind === 'issue.opened' && refusals++ === 0) throw new Error('host refused the event'); + }, + }); + + await r.poller.pollOnce(); + assert.deepEqual(r.state.fresh, {}); + const second = await r.poller.pollOnce(); + assert.deepEqual(second.delivered.map((e) => e.kind), ['issue.opened']); +}); + +test('lastPollAt is the earliest place a list stopped, and never past the pass start', async () => { + const api = new FakeGitHubApi(); + const r = rig({ api }); + api.listIssuesSince = async () => ({ rows: [], resumeAt: '2026-10-13T08:00:00Z' }); + api.listIssueCommentsSince = async () => ({ rows: [], resumeAt: '2026-10-13T07:00:00Z' }); + await r.poller.pollOnce(); + assert.equal(r.state.lastPollAt(), '2026-10-13T07:00:00Z'); + + // A stop stamped after the pass started, as a row changed during it can be. + api.listIssuesSince = async () => ({ rows: [] }); + api.listIssueCommentsSince = async () => ({ rows: [], resumeAt: '2026-10-14T00:00:05Z' }); + await r.poller.pollOnce(); + assert.equal(r.state.lastPollAt(), '2026-10-14T00:00:00.000Z'); +}); + +test('a first pass whose read throws keeps the window it read', async () => { + const api = new FakeGitHubApi(); + // The very first read throws, so the window must be kept before any read. + api.listIssuesSince = async () => { + throw new Error('GitHub 502 on the comment page'); + }; + const r = rig({ api, cursor: null, now: '2026-10-14T00:00:00.700Z' }); + await assert.rejects(() => r.poller.pollOnce(), /502 on the comment page/); + assert.equal(r.state.lastPollAt(), '2026-10-13T23:59:00.000Z'); +}); + +test('a stored position with milliseconds is read from its whole second', async () => { + const r = rig({ cursor: '2026-10-13T00:00:00.700Z' }); + r.state.setPollCursors({ + 'anima-research/rayforce:issues': { since: '2026-10-13T00:00:00.000Z', resumeAt: '2026-10-13T06:00:00.300Z' }, + 'anima-research/rayforce:issue_comments': { since: '2026-10-13T00:00:00.500Z' }, + }); + await r.poller.pollOnce(); + assert.deepEqual( + r.api.calls.filter((c) => String(c.method).endsWith('Since')).map((c) => c.args[1]), + ['2026-10-13T06:00:00.000Z', '2026-10-13T00:00:00.000Z', '2026-10-13T00:00:00.000Z'], + ); +}); + +test('cursors are whole seconds, so the second a pass starts in is read again', async () => { + const r = rig({ now: '2026-10-14T00:00:00.700Z' }); + await r.poller.pollOnce(); + assert.equal(r.state.lastPollAt(), '2026-10-14T00:00:00.000Z'); + assert.deepEqual(r.state.pollCursors()['anima-research/rayforce:issues'], { since: '2026-10-14T00:00:00.000Z' }); +}); + +test('a refused first pass keeps the window it read', async () => { + const api = new FakeGitHubApi(); + api.seedIssue({ number: 41, updatedAt: '2026-10-13T23:59:30.000Z' }); + const r = rig({ + api, + cursor: null, + now: '2026-10-14T00:00:00.700Z', + onEvent: () => { + throw new Error('host refused the event'); + }, + }); + + const first = await r.poller.pollOnce(); + assert.equal(first.failed, 1); + assert.equal(first.since, '2026-10-13T23:59:00.000Z', 'one interval back, in whole seconds'); + assert.equal(r.state.lastPollAt(), first.since, 'the retry asks from the same start, not one moved on by the clock'); +}); + test('the gate the state keeps is bounded by the contract POLL_DEDUP_RING', async () => { const state = new MemoryIngressState('2026-10-13T00:00:00.000Z'); for (let i = 0; i < POLL_DEDUP_RING + 5; i++) assert.equal(state.seenPollKey(`issue:${i}:t`), false); @@ -817,7 +997,7 @@ test('a transient lookup failure is retried on the next pass, after the cursor m // Pass two: the cursor has moved past the row, so `listIssuesSince` returns // nothing — the backlog is the only reason the checks are asked about again. const second = await r.poller.pollOnce(); - assert.deepEqual(await api.listIssuesSince(TEST_REPO, r.state.lastPollAt() as string), []); + assert.deepEqual((await api.listIssuesSince(TEST_REPO, r.state.lastPollAt() as string)).rows, []); assert.deepEqual( second.delivered.map((e) => e.kind), ['checks.completed'], @@ -1071,7 +1251,7 @@ test('a checks event refused long after the cursor moved past its row is still r // From here on the pull-request row is behind the cursor: the backlog is // the only thing that can produce this checks event. const cursor = r.state.lastPollAt() as string; - assert.deepEqual(await api.listIssuesSince(TEST_REPO, cursor), []); + assert.deepEqual((await api.listIssuesSince(TEST_REPO, cursor)).rows, []); // Pass two: the checks settle and the listener refuses. settled = true; diff --git a/test/smoke.test.ts b/test/smoke.test.ts index d0eff85..8f500e0 100644 --- a/test/smoke.test.ts +++ b/test/smoke.test.ts @@ -115,21 +115,21 @@ test('FakeGitHubApi reports check runs and answers since-polls', async () => { const fresh = api.seedPullRequest({ updatedAt: '2026-10-02T00:00:00.000Z' }); const comment = await api.createComment({ repo: TEST_REPO, number: old.number, kind: 'issue' }, 'late comment'); const since = '2026-10-01T12:00:00.000Z'; - const issues = await api.listIssuesSince(TEST_REPO, since); + const { rows: issues } = await api.listIssuesSince(TEST_REPO, since); assert.deepEqual(issues.map((i) => [i.number, i.isPullRequest]), [[fresh.number, true]]); assert.equal(issues[0].headSha, fresh.head.sha); - const comments = await api.listIssueCommentsSince(TEST_REPO, since); + const { rows: comments } = await api.listIssueCommentsSince(TEST_REPO, since); assert.deepEqual(comments.map((c) => [c.id, c.number]), [[comment.id, old.number]]); - assert.deepEqual(await api.listReviewCommentsSince(TEST_REPO, since), []); - assert.deepEqual(await api.listIssuesSince({ owner: 'other', name: 'repo' }, since), []); + assert.deepEqual((await api.listReviewCommentsSince(TEST_REPO, since)).rows, []); + assert.deepEqual((await api.listIssuesSince({ owner: 'other', name: 'repo' }, since)).rows, []); // An old review comment edited after the cursor is polled (reviewer finding). api.seedReviewThread({ repo: TEST_REPO, number: fresh.number, kind: 'pull' }, { comments: [{ id: 555, author: { login: 'human', isBot: false }, body: 'before', createdAt: '2026-10-01T00:00:00.000Z' }], }); - assert.deepEqual(await api.listReviewCommentsSince(TEST_REPO, since), []); + assert.deepEqual((await api.listReviewCommentsSince(TEST_REPO, since)).rows, []); await api.updateComment(TEST_REPO, 555, 'after', true); - const polledReview = await api.listReviewCommentsSince(TEST_REPO, since); + const { rows: polledReview } = await api.listReviewCommentsSince(TEST_REPO, since); assert.deepEqual(polledReview.map((c) => [c.id, c.number, c.body]), [[555, fresh.number, 'after']]); }); diff --git a/test/state.test.ts b/test/state.test.ts index 6ff1c0b..222bf7d 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -310,9 +310,66 @@ test('a state file written before amendment 2a still loads, with no heads known' assert.equal(state.lastPollAt(), '2026-10-14T00:00:00.000Z'); assert.equal(state.seenPollKey('issue:1:t0'), true); assert.deepEqual(state.snapshot().pollHeads, {}); + // Nor per-list cursors, so every list starts from `lastPollAt`. + assert.deepEqual(state.pollCursors(), {}); + assert.deepEqual(state.polledNew(), {}); assert.deepEqual(log.lines, []); }); +test('poll cursors and announced rows round-trip through a flush, and a cursor without an instant is dropped', async () => { + const file = tempFile(); + const first = GithubState.load(file, capturingLogger()); + first.setPollCursors({ + 'anima-research/rayforce:issues': { since: '2026-10-14T00:00:00.000Z', resumeAt: '2026-10-14T06:00:00Z' }, + 'anima-research/rayforce:issue_comments': { since: '2026-10-14T07:00:00.000Z' }, + }); + first.setPolledNew({ 'anima-research/rayforce:issues': { '7': '2026-10-14T01:00:00Z' } }); + await first.flush(); + const onDisk = JSON.parse(readFileSync(file, 'utf8')) as Record; + onDisk.pollCursors = { + ...(onDisk.pollCursors as object), + 'x/y:issues': { resumeAt: '2026-10-14T06:00:00Z' }, + 'x/y:issue_comments': 'w', + 'x/y:review_comments': null, + 'x/z:issues': { since: 'not a time' }, + 'x/z:issue_comments': { since: '2026-10-14T07:00:00.000Z', resumeAt: 'not a time' }, + }; + onDisk.pollNew = { + ...(onDisk.pollNew as object), + 'x/y:issues': ['a'], + 'x/z:issues': { '3': 5, '4': '2026-10-14T02:00:00Z' }, + }; + writeFileSync(file, JSON.stringify(onDisk)); + + const loaded = GithubState.load(file, capturingLogger()); + assert.deepEqual(loaded.pollCursors(), { + 'anima-research/rayforce:issues': { since: '2026-10-14T00:00:00.000Z', resumeAt: '2026-10-14T06:00:00Z' }, + 'anima-research/rayforce:issue_comments': { since: '2026-10-14T07:00:00.000Z' }, + 'x/z:issue_comments': { since: '2026-10-14T07:00:00.000Z' }, + }); + assert.deepEqual(loaded.polledNew(), { + 'anima-research/rayforce:issues': { '7': '2026-10-14T01:00:00Z' }, + 'x/y:issues': {}, + 'x/z:issues': { '4': '2026-10-14T02:00:00Z' }, + }); +}); + +test('a lastPollAt that is not an instant loads as none, says so, and keeps the rest', () => { + const file = tempFile(); + const checkpoint = { id: 'chk_1', parent: null, createdAt: '2026-10-14T00:00:00.000Z' }; + writeFileSync( + file, + JSON.stringify({ version: STATE_VERSION, checkpoints: [checkpoint], current: 'chk_1', created: {}, edited: {}, deleted: {}, lastPollAt: 'soon', pollKeys: ['issue:1:t0'] }), + ); + const log = capturingLogger(); + const state = GithubState.load(file, log); + assert.equal(state.lastPollAt(), null); + assert.equal(state.current(), 'chk_1'); + assert.equal(state.seenPollKey('issue:1:t0'), true); + assert.equal(log.lines.length, 1); + assert.match(log.lines[0], /lastPollAt that isn't a time/); +}); + test('a polled head round-trips through a flush (amendment 2a)', async () => { const file = tempFile(); const first = GithubState.load(file, capturingLogger()); diff --git a/test/webhook.test.ts b/test/webhook.test.ts index ce89d1e..1aea9c2 100644 --- a/test/webhook.test.ts +++ b/test/webhook.test.ts @@ -595,11 +595,21 @@ function pollState(cursor: string): IngressState { const keys: string[] = []; const heads = new Map(); let at = cursor; + let cursors = {}; + let fresh = {}; return { lastPollAt: () => at, setLastPollAt: (when) => { at = when; }, + pollCursors: () => cursors, + setPollCursors: (next) => { + cursors = { ...next }; + }, + polledNew: () => fresh, + setPolledNew: (next) => { + fresh = next; + }, polledHead: (channel) => heads.get(channel) ?? null, setPolledHead: (channel, sha) => { heads.set(channel, sha); @@ -774,7 +784,7 @@ test('a webhook refusal after the cursor passed the row does not cost the poll i assert.equal(first.cursorAdvanced, true); assert.equal(first.checksBacklog, 1); // From here the pull-request row is behind the cursor. - assert.deepEqual(await api.listIssuesSince(TEST_REPO, state.lastPollAt() as string), []); + assert.deepEqual((await api.listIssuesSince(TEST_REPO, state.lastPollAt() as string)).rows, []); // The checks settle, and a webhook for the sha blocks in the listener. settled = true; From 08dc1910cb992df0f6b17a24b05088632285439c Mon Sep 17 00:00:00 2001 From: Ogden-2688 Date: Wed, 21 Oct 2026 16:04:45 -0400 Subject: [PATCH 2/2] Polling fallback: end check runs at an empty page, and bring the notes to the code getCheckRuns no longer stopped at a short page, so a total_count that counts runs it can no longer serve, such as one deleted between pages, spent every page of the bound on each pass. An empty page now ends the read. A head whose runs were not all read is still never a success: it is undecided unless a run that was read failed. The warnings now say that, and say whether the count or the bound ended the read. A test now reaches the request bound inside one second, reads to the end of the page that finishes it and stops there, keeping that page's rows before the stop second and leaving the stop second's for resumeAt. No test reached that path before. The warning for a lastPollAt that isn't a time now says only lists without a cursor of their own start one interval back; a test shows the cursors survive it, and that a null or missing lastPollAt loads quietly. The held-back warning no longer logs the fallback start as if it were the cursor, and the pass's debug line names it as the fallback. The README, the deploy guide's state-file row, comments that spoke of one cursor, and the contract notes' Amendment 1 and 2a now describe the per-list cursors, pollNew, PollRead and the cursor's form. A test's error names the list it throws from. --- README.md | 4 +- changelog.d/fix-poll-paging.fixed.md | 7 ++- docs/contract-notes.md | 14 +++++- docs/deploy.md | 2 +- src/github-adapter.ts | 35 +++++++++------ src/index.ts | 2 +- src/normalize.ts | 5 ++- src/poller.ts | 16 +++---- src/state.ts | 2 +- src/types.ts | 2 +- test/githubAdapter.test.ts | 64 ++++++++++++++++++++++++++++ test/poller.test.ts | 4 +- test/state.test.ts | 18 +++++++- 13 files changed, 140 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index af420b9..dfa2ec7 100644 --- a/README.md +++ b/README.md @@ -238,10 +238,10 @@ told where Node is instead of looking. | `GITHUB_WEBHOOK_SECRET_FILE` | when `GITHUB_WEBHOOK_PORT` is set | — | Path to the webhook secret, mode 600. | | `GITHUB_WEBHOOK_PORT` | no | unset = listener off | TCP port of the webhook listener (bind it to loopback, terminate TLS in front). | | `GITHUB_WEBHOOK_PATH` | no | `/github/webhook` | Path the listener answers on. | -| `GITHUB_POLL_INTERVAL_MS` | no | `0` (off) | Polling fallback: issues and PRs, issue comments and review comments updated since the cursor, plus the checks on the heads of the open PRs seen. `0` is off; a non-zero value below `60000` is refused. Partial — it cannot see a submitted review, a review request or a push ([what polling misses](docs/deploy.md#option-3--polling-only-no-webhook-at-all)) — and it is not meant to run alongside the webhook. | +| `GITHUB_POLL_INTERVAL_MS` | no | `0` (off) | Polling fallback: issues and PRs, issue comments and review comments updated since each list's cursor, plus the checks on the heads of the open PRs seen. `0` is off; a non-zero value below `60000` is refused. Partial — it cannot see a submitted review, a review request or a push ([what polling misses](docs/deploy.md#option-3--polling-only-no-webhook-at-all)) — and it is not meant to run alongside the webhook. | | `GITHUB_REPOS` | yes | — | Comma-separated `owner/name` allow-list; a write anywhere else is refused by name. Several entries must share one owner (one App installation, one token); a second owner is a second server process. With several, every tool call names its repository. | | `GITHUB_PR_BASE_ALLOW` | no | empty = no PR target allowed | Branches a pull request may target or be retargeted to. A plain `branch` applies to every repository, `owner/name:branch` to that one: `anima-research/rayforce:dev,main`. | -| `GITHUB_STATE_FILE` | no | unset = in-memory only | Durable JSON: checkpoint lineage, rollback bookkeeping, the poll cursor. | +| `GITHUB_STATE_FILE` | no | unset = in-memory only | Durable JSON: checkpoint lineage, rollback bookkeeping, and the polling state: a cursor for each repository and list (`pollCursors`), `lastPollAt` where a list with none starts, and the rows announced as new (`pollNew`). A list with a cursor of its own starts from it, not from `lastPollAt`. | | `GITHUB_BOT_LOGIN` | yes | — | The App's bot login, e.g. `rayforce-clone[bot]`. From-self and mention detection depend on it. | | `GITHUB_MAINTAINERS` | no | empty | Comma-separated logins marked `isMaintainer` in metadata and tagged `github:actor=maintainer`. | diff --git a/changelog.d/fix-poll-paging.fixed.md b/changelog.d/fix-poll-paging.fixed.md index e951e04..1d39f42 100644 --- a/changelog.d/fix-poll-paging.fixed.md +++ b/changelog.d/fix-poll-paging.fixed.md @@ -16,5 +16,8 @@ again inside its window, after a stopped read or a refused delivery (`pollNew` in the state file). A first pass that a delivery refused keeps the window it read. -- `getCheckRuns` reads on past a short page while `total_count` says more - runs follow, so such a head is no longer left undecided. +- `getCheckRuns` reads on past a short page that isn't empty while + `total_count` says more runs follow, so such a head is no longer left + undecided. A response + without a `total_count` reads on past a full page instead of stopping at + the first. diff --git a/docs/contract-notes.md b/docs/contract-notes.md index 1f9414e..b542fdd 100644 --- a/docs/contract-notes.md +++ b/docs/contract-notes.md @@ -43,6 +43,16 @@ camel case, so a reviewer can map a contract bullet to a type by sight. `poll:::`, and asks `getCheckRuns` for the heads of the open PRs it saw, at most 20 per poll. These endpoints carry no `X-Poll-Interval`; `MIN_POLL_INTERVAL_MS` is the floor. + - Each `…Since` method returns `PollRead`, `{ rows, resumeAt? }`: + `resumeAt` is set when the request bound stopped the read, and names the + second the next read asks from. + - `StateSnapshot` gains `pollCursors` (`/:` → + `{ since, resumeAt? }`) and `pollNew` (the same key → id → `created_at`), + read and written through `StateStore.pollCursors`/`setPollCursors` and + `polledNew`/`setPolledNew`. Both fields are **additive**: a state file + without them loads with no cursors and nothing remembered, so every list + starts from `lastPollAt`, and a release before them ignores them and + reads `lastPollAt`, up to which every list has delivered. - **Amendment 2a: a poll row's kind comes from `created_at` against the cursor.** Amendment 1 left the polling ingress unable to tell an opened issue from an edited one, so every open issue row arrived as `issue.edited` @@ -70,7 +80,9 @@ camel case, so a reviewer can map a contract bullet to a type by sight. rather than not at all, which is the bounded cost of not leaving a cold start blind to what is already open. - The comparison is of instants, not strings: GitHub stamps rows without - milliseconds while the cursor carries them, and `'Z'` sorts after `'.'`. + milliseconds while a cursor is written with them (`.000Z`, and any + precision in a state file an older release wrote), and `'Z'` sorts after + `'.'`. - `StateSnapshot` gains `pollHeads` (`/#` → head sha), bounded by `POLL_HEAD_MEMORY` = 200 and recorded on a dropped row as well as on a delivered one. The field is **additive**: a state file written diff --git a/docs/deploy.md b/docs/deploy.md index 0d4dd77..fcfa022 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -15,7 +15,7 @@ user the agent runs as (`ubuntu` on a typical agent host): | `~/.config/github-mcpl/github-mcpl.env` | 600 | variable names and paths, **no secrets** | | `~/.config/github-mcpl/app-private-key.pem` | **600** | the App private key | | `~/.config/github-mcpl/webhook-secret` | **600** | the webhook secret, one line | -| `~/.config/github-mcpl/state.json` | 600 | checkpoint lineage and the poll cursor | +| `~/.config/github-mcpl/state.json` | 600 | checkpoint lineage and the polling state | The server refuses to start if either secret file is readable by group or other: `readConfig` stats both and reports the mode by name. That check is the diff --git a/src/github-adapter.ts b/src/github-adapter.ts index fddfc3b..00d6419 100644 --- a/src/github-adapter.ts +++ b/src/github-adapter.ts @@ -1051,18 +1051,21 @@ export class GitHubAdapter implements GitHubApi { * * The endpoint is paginated, and a busy commit can carry hundreds of runs: * reading one page and calling it `success` is how a failure on page two - * becomes a green light. So pages are followed until `total_count` runs are - * read, up to {@link CHECK_RUN_MAX_PAGES} pages, whatever a page's length; - * `total` is GitHub's own `total_count` rather than what we managed to read, - * and a sha whose runs did not all fit is never reported as a definite - * success — its conclusion is `null`, the same "not decided" the contract - * already uses for runs still in flight. + * becomes a green light. So pages are followed past a short one until + * `total_count` runs are read, up to {@link CHECK_RUN_MAX_PAGES} pages, or + * until a page comes back empty; `total` is GitHub's own `total_count` + * rather than what we managed to read, and a sha whose runs were not all + * read is never reported as a definite success — unless a run that was + * read failed, its conclusion is `null`, the same "not decided" the + * contract already uses for runs still in flight. */ async getCheckRuns(repo: RepoRef, sha: string): Promise { const runs: CheckRunRow[] = []; let totalCount = 0; - // Whether runs were left unread when the loop ended: then the sha is undecided. + // Whether runs were left unread when the loop ended: then the sha can't be a success. let more = false; + // Whether an empty page ended the read before `total_count` was reached. + let endedShort = false; for (let page = 1; page <= CHECK_RUN_MAX_PAGES; page++) { const batch = await this.call('getCheckRuns', async () => { @@ -1082,15 +1085,23 @@ export class GitHubAdapter implements GitHubApi { // a response without one leaves a full page as the sign of more. more = totalCount > 0 ? runs.length < totalCount : batch.rows.length === CHECK_RUN_PAGE; if (!more) break; + // An empty page is past the last one: what `total_count` still counts + // can't be read, such as a run deleted between pages. + if (batch.rows.length === 0) { + endedShort = true; + break; + } } const truncated = more; if (truncated) { - this.log('warn', 'check runs did not fit in the page bound; the sha is reported undecided', { - sha, - read: runs.length, - total: totalCount, - }); + this.log( + 'warn', + endedShort + ? 'check runs ended before total_count; the sha is reported undecided unless a run read failed' + : 'check runs did not fit in the page bound; the sha is reported undecided unless a run read failed', + { sha, read: runs.length, total: totalCount }, + ); } return summarizeChecks(sha, runs, { total: Math.max(totalCount, runs.length), truncated }); } diff --git a/src/index.ts b/src/index.ts index 12da652..87a4673 100644 --- a/src/index.ts +++ b/src/index.ts @@ -143,7 +143,7 @@ async function shutdown(running: Running, log: Logger): Promise { log.warn('webhook did not close', { error: String(err) }); } try { - // Last: the poll cursor and the checkpoint lineage must survive the exit. + // Last: the poll cursors and the checkpoint lineage must survive the exit. await running.state.flush(); } catch (err) { log.error('state flush failed', { error: String(err) }); diff --git a/src/normalize.ts b/src/normalize.ts index eca7592..5d8a440 100644 --- a/src/normalize.ts +++ b/src/normalize.ts @@ -451,8 +451,9 @@ function normalizeCheckRun(base: EventBase, payload: unknown, actor: EventActor * True when `at` is at or after `since` — both ISO-8601 instants. * * Compared as instants and not as strings: GitHub stamps a row - * `2026-10-14T12:28:38Z` while the cursor carries milliseconds - * (`2026-10-14T12:28:38.500Z`), and lexicographically `'Z'` (0x5A) sorts + * `2026-10-14T12:28:38Z` while a cursor is written with milliseconds + * (`2026-10-14T12:28:38.000Z`, or `.500Z` from an older release), and + * lexicographically `'Z'` (0x5A) sorts * *after* `'.'` (0x2E) — so a string compare calls the earlier instant the * later one. An unparseable stamp answers false, which is the fail-safe * direction: the row keeps the kind it had before amendment 2a. diff --git a/src/poller.ts b/src/poller.ts index 3b2ce11..00ff107 100644 --- a/src/poller.ts +++ b/src/poller.ts @@ -88,7 +88,7 @@ export type PollerApi = Pick< >; /** - * The durable state the poller needs: the cursor, the dedup ring and the last + * The durable state the poller needs: the cursors, the dedup ring and the last * polled head of each open pull request (amendment 2a). A structural subset of * `StateStore` (types.ts), which .7 implements over GITHUB_STATE_FILE — * `snapshot()` is in the interface so that the ring can be *read* without @@ -349,12 +349,12 @@ export class Poller { } /** - * One pass: query every allow-listed repository since the cursor, drain as + * One pass: query every allow-listed repository's lists since their cursors, drain as * much of the check backlog as the per-pass cap allows, drop what the dedup * ring already holds, deliver the rest oldest-first, record each key after * its delivery, and advance the cursors if no delivery failed. * - * A failed *lookup* does not hold the cursor back: the head stays in the + * A failed *lookup* does not hold the cursors back: the head stays in the * backlog instead, which is the thing that gets retried. A failed * *delivery* does hold it back, so its row is queried again. * @@ -435,7 +435,7 @@ export class Poller { outcome = await this.hand(candidate.event); } catch (err) { // Nothing is recorded and nothing is retired: the key stays out of - // the dedup ring, the head stays in the backlog, and the cursor is + // the dedup ring, the head stays in the backlog, and the cursors are // held back below, so the next pass tries again. result.failed += 1; this.logger.warn('poll event delivery failed; its key stays unrecorded for the next pass', { @@ -484,13 +484,13 @@ export class Poller { this.state.setPolledNew(remembered); result.cursorAdvanced = true; } else { - this.logger.warn('poll cursor held back: a delivery failed in this pass', { - since, + this.logger.warn('poll cursors held back: a delivery failed in this pass, so every list reads it again', { failed: result.failed, }); } this.logger.debug('poll pass done', { - since, + // Where a list with no cursor of its own started; the others started at their own. + fallbackSince: since, delivered: result.delivered.length, duplicates: result.duplicates, failed: result.failed, @@ -655,7 +655,7 @@ export class Poller { } // Settled — but the head stays queued until the delivery loop retires - // it, so a refused event can be retried even after the cursor has moved + // it, so a refused event can be retried even after the cursors have moved // past the pull-request row that queued this head (rule 2). this.rotateHead(key, head); const pull: ThreadRef = { repo: head.repo, number: head.number, kind: 'pull' }; diff --git a/src/state.ts b/src/state.ts index 1833a5b..e973be5 100644 --- a/src/state.ts +++ b/src/state.ts @@ -203,7 +203,7 @@ export class GithubState implements GithubStateStore { } const kept = (parsed as { lastPollAt?: unknown }).lastPollAt; if (kept !== undefined && kept !== null && snap.lastPollAt === null) { - logger.warn(`state: ${file} has a lastPollAt that isn't a time (${JSON.stringify(kept)}); polling starts one interval back`); + logger.warn(`state: ${file} has a lastPollAt that isn't a time (${JSON.stringify(kept)}); lists without a cursor of their own start one interval back`); } return new GithubState(file, snap, logger); } diff --git a/src/types.ts b/src/types.ts index cfd314f..69e1cc8 100644 --- a/src/types.ts +++ b/src/types.ts @@ -369,7 +369,7 @@ export interface PolledIssue { closedBy?: EventActor; } -/** One issue comment or review comment updated since the cursor. */ +/** One issue comment or review comment updated since its list's cursor. */ export interface PolledComment { id: number; /** The issue or pull request the comment belongs to. */ diff --git a/test/githubAdapter.test.ts b/test/githubAdapter.test.ts index fd0222a..519fff5 100644 --- a/test/githubAdapter.test.ts +++ b/test/githubAdapter.test.ts @@ -1149,6 +1149,40 @@ test('a read never stops inside a second, even past its request bound', async () ); }); +test('a read the bound finds inside a second reads to the end of the page that finishes it, and stops there', async () => { + // Rows 1–250 share second 1, which carries the read past its bound; rows 251–350 have a second each. + const r = await rig(); + const all = [ + ...Array.from({ length: 250 }, (_, i) => pagedIssue(1 + i, second(1))), + ...Array.from({ length: 100 }, (_, i) => pagedIssue(251 + i, second(2 + i))), + ]; + serveSince(r, ISSUES, all); + + const first = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); + assert.deepEqual( + r.fake.for(ISSUES).map((c) => [c.query.since, c.query.page]), + [ + ['2026-10-10T00:00:00Z', '1'], + [second(1), '1'], + [second(1), '2'], + [second(1), '3'], + ], + ); + // Page 3 finished second 1 and ended at row 300's second: that second waits for the next read. + assert.equal(first.resumeAt, second(51)); + assert.deepEqual( + first.rows.map((row) => row.id), + all.slice(0, 299).map((row) => row.id), + ); + + const next = await r.adapter.listIssuesSince(REPO, first.resumeAt as string); + assert.deepEqual( + [...first.rows, ...next.rows].map((row) => row.id), + all.map((row) => row.id), + 'the two reads together return every row once', + ); +}); + test('a Link header without rel="next" ends the read', async () => { // GitHub's last page names only prev and first. const r = await rig(); @@ -1635,6 +1669,36 @@ test('getCheckRuns without a total_count is undecided when the page bound stops assert.equal(r.fake.for(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`).length, 5); }); +test('getCheckRuns ends at an empty page as undecided, and says the count was not reached', async () => { + // total_count counts a run that is gone by the time its page is asked for. + const r = await rig(); + r.fake.on(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`, (req) => ({ + body: { total_count: 101, check_runs: Number(req.query.page) === 1 ? successRuns(0, 100) : [] }, + })); + + const summary = await r.adapter.getCheckRuns(REPO, HEAD_SHA); + assert.equal(summary.conclusion, null); + assert.equal(r.fake.for(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`).length, 2); + assert.ok(r.logs.some((l) => l.startsWith('warn check runs ended before total_count')), r.logs.join('\n')); +}); + +test('a check-run read that ended short is still a failure when a run it read failed', async () => { + const r = await rig(); + r.fake.on(`GET /repos/anima-research/rayforce/commits/${HEAD_SHA}/check-runs`, (req) => ({ + body: { + total_count: 101, + check_runs: + Number(req.query.page) === 1 + ? [...successRuns(0, 99), { name: 'lint', status: 'completed', conclusion: 'failure', html_url: 'https://ci/lint' }] + : [], + }, + })); + + const summary = await r.adapter.getCheckRuns(REPO, HEAD_SHA); + assert.equal(summary.conclusion, 'failure'); + assert.ok(r.logs.some((l) => l.includes('undecided unless a run read failed')), r.logs.join('\n')); +}); + test('summarizeChecks refuses to call a truncated read success', () => { const complete = summarizeChecks('s', successRuns(0, 3)); assert.equal(complete.conclusion, 'success'); diff --git a/test/poller.test.ts b/test/poller.test.ts index 64fa3d3..c8933a0 100644 --- a/test/poller.test.ts +++ b/test/poller.test.ts @@ -913,10 +913,10 @@ test('a first pass whose read throws keeps the window it read', async () => { const api = new FakeGitHubApi(); // The very first read throws, so the window must be kept before any read. api.listIssuesSince = async () => { - throw new Error('GitHub 502 on the comment page'); + throw new Error('GitHub 502 on the issues page'); }; const r = rig({ api, cursor: null, now: '2026-10-14T00:00:00.700Z' }); - await assert.rejects(() => r.poller.pollOnce(), /502 on the comment page/); + await assert.rejects(() => r.poller.pollOnce(), /502 on the issues page/); assert.equal(r.state.lastPollAt(), '2026-10-13T23:59:00.000Z'); }); diff --git a/test/state.test.ts b/test/state.test.ts index 222bf7d..204f7ec 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -354,22 +354,36 @@ test('poll cursors and announced rows round-trip through a flush, and a cursor w }); }); -test('a lastPollAt that is not an instant loads as none, says so, and keeps the rest', () => { +test('a lastPollAt that is not an instant loads as none, says so, and keeps the rest, cursors included', () => { const file = tempFile(); const checkpoint = { id: 'chk_1', parent: null, createdAt: '2026-10-14T00:00:00.000Z' }; writeFileSync( file, - JSON.stringify({ version: STATE_VERSION, checkpoints: [checkpoint], current: 'chk_1', created: {}, edited: {}, deleted: {}, lastPollAt: 'soon', pollKeys: ['issue:1:t0'] }), + JSON.stringify({ version: STATE_VERSION, checkpoints: [checkpoint], current: 'chk_1', created: {}, edited: {}, deleted: {}, lastPollAt: 'soon', pollKeys: ['issue:1:t0'], pollCursors: { 'x/y:issues': { since: '2026-10-14T00:00:00.000Z' } } }), ); const log = capturingLogger(); const state = GithubState.load(file, log); assert.equal(state.lastPollAt(), null); assert.equal(state.current(), 'chk_1'); assert.equal(state.seenPollKey('issue:1:t0'), true); + assert.deepEqual(state.pollCursors(), { 'x/y:issues': { since: '2026-10-14T00:00:00.000Z' } }); assert.equal(log.lines.length, 1); assert.match(log.lines[0], /lastPollAt that isn't a time/); }); +test('a lastPollAt that is null or left out loads quietly', () => { + for (const lastPollAt of [null, undefined]) { + const file = tempFile(); + writeFileSync( + file, + JSON.stringify({ version: STATE_VERSION, checkpoints: [], current: null, created: {}, edited: {}, deleted: {}, lastPollAt, pollKeys: [] }), + ); + const log = capturingLogger(); + assert.equal(GithubState.load(file, log).lastPollAt(), null); + assert.deepEqual(log.lines, [], `lastPollAt ${String(lastPollAt)}`); + } +}); + test('a polled head round-trips through a flush (amendment 2a)', async () => { const file = tempFile(); const first = GithubState.load(file, capturingLogger());