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 new file mode 100644 index 0000000..1d39f42 --- /dev/null +++ b/changelog.d/fix-poll-paging.fixed.md @@ -0,0 +1,23 @@ +- 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 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 5ddd932..b542fdd 100644 --- a/docs/contract-notes.md +++ b/docs/contract-notes.md @@ -33,23 +33,43 @@ 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 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` 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 @@ -60,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 1c4d991..00d6419 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,15 +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 to {@link CHECK_RUN_MAX_PAGES}, - * `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 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 () => { @@ -1065,16 +1081,27 @@ 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; + // 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 = 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, - 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 }); } @@ -1350,24 +1377,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 +1428,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 +1500,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/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 718cd47..5d8a440 100644 --- a/src/normalize.ts +++ b/src/normalize.ts @@ -451,13 +451,14 @@ 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. */ -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 +469,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 +523,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..00ff107 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'; @@ -81,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 @@ -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 { @@ -300,29 +349,40 @@ 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 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 + * 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. * * 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)) { @@ -369,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', { @@ -402,16 +468,29 @@ 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', { - 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, @@ -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; } @@ -550,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 09aab53..e973be5 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)}); lists without a cursor of their own start 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..69e1cc8 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; @@ -368,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. */ @@ -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..519fff5 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,361 @@ 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'); + const ids = new Set(rows.map((row) => row.id)); + assert.deepEqual( + 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'); +}); - const rows = await r.adapter.listIssuesSince(REPO, '2026-10-10T00:00:00Z'); - assert.equal(rows.length, POLL_MAX_ENTRIES); +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('GET /repos/anima-research/rayforce/issues').map((c) => c.query.page), - ['1', '2', '3'], + 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 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(); + 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 +1398,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 +1420,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 +1625,80 @@ 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('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/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..c8933a0 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 issues page'); + }; + const r = rig({ api, cursor: null, now: '2026-10-14T00:00:00.700Z' }); + await assert.rejects(() => r.poller.pollOnce(), /502 on the issues 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..204f7ec 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -310,9 +310,80 @@ 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, 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'], 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()); 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;