diff --git a/bin/gh-pulse.ts b/bin/gh-pulse.ts index fbaf42f..bfa0743 100755 --- a/bin/gh-pulse.ts +++ b/bin/gh-pulse.ts @@ -29,6 +29,14 @@ * * Mail goes through Resend: RESEND_API_KEY from the environment, else from * ~/.config/logicsrc/shell.env (the cron case). A crash sends a FAILED mail. + * + * The hour: GitHub allows 5000 calls an hour on the token, shared with `gh` + * and everything else that uses it. A scan paces its requests, caches replies + * by ETag (an unchanged page comes back as a free 304) and keeps + * GH_PULSE_RESERVE calls (default 500) unspent for everyone else; down to the + * reserve, or rate limited anyway, it pauses until the hour resets and then + * carries on. One daily run at a time: a second refuses while /run.lock + * names a live process. */ import { readFileSync } from 'node:fs'; @@ -141,7 +149,9 @@ export async function main(argv: readonly string[]): Promise { if (verb === 'show') { const { showTui } = await import('../src/gh-pulse-tui.ts'); const startKey: RangeKey | undefined = spec && spec.key !== 'custom' ? spec.key : undefined; - return showTui(dir, { range: startKey, loadRange: (key, progress) => loadRange(rangeSpec(key, deps.now()), progress) }); + const code = await showTui(dir, { range: startKey, loadRange: (key, progress) => loadRange(rangeSpec(key, deps.now()), progress) }); + // A range scan still in flight would keep the process alive after the screen is gone, spending the hour on a report nobody reads. + process.exit(code); } if (verb === 'open' || verb === 'text' || verb === 'json') { if (spec) { diff --git a/package.json b/package.json index 4e84bd5..578c388 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@profullstack/cli-tools", - "version": "0.35.0", + "version": "0.36.0", "private": true, "description": "Local command-line tools, in TypeScript, exposed on PATH.", "type": "module", diff --git a/src/gh-pulse.ts b/src/gh-pulse.ts index fc42ed2..766cf74 100644 --- a/src/gh-pulse.ts +++ b/src/gh-pulse.ts @@ -28,7 +28,8 @@ */ import { execFileSync, spawnSync } from 'node:child_process'; -import { existsSync, mkdirSync, readdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { createHash } from 'node:crypto'; +import { existsSync, mkdirSync, readdirSync, readFileSync, renameSync, statSync, unlinkSync, utimesSync, writeFileSync } from 'node:fs'; import { createRequire } from 'node:module'; import { homedir, hostname } from 'node:os'; import { join } from 'node:path'; @@ -574,30 +575,96 @@ export function writeSnapshot(dir: string, snap: Snapshot): string { // ---------------------------------------------------------------- github +export interface GitHubOptions { + /** Requests in flight at once. Four keeps well clear of GitHub's concurrency limit. */ + concurrency?: number; + /** Minimum gap between two request starts. 100ms is at most 600 a minute, under the 900-points-a-minute secondary limit. */ + minIntervalMs?: number; + /** + * Calls of the hour that are never spent, so `gh` and everything else on the + * same token keep working. When the hour is down to the reserve the scan + * pauses until it resets and then carries on. + */ + reserve?: number; + /** ETag cache directory. GitHub does not charge for a 304, so an unchanged page is free. Null disables the cache. */ + cacheDir?: string | null; + /** Where a pause is announced: the spinner line, or the log. */ + onWait?: (line: string) => void; + now?: () => number; + sleep?: (ms: number) => Promise; +} + +export const DEFAULT_RESERVE = 500; +export const cacheDir = (dataDir: string): string => join(dataDir, 'cache'); + +/** A cached reply: the ETag to offer next time and the body to reuse when GitHub says nothing changed. */ +interface CachedReply { url: string; etag: string; link: string; status: number; data: unknown; at: string } + +const hhmm = (epochMs: number): string => `${new Date(epochMs).toISOString().slice(11, 16)} UTC`; + +/** + * The GitHub client every scan goes through, and the only place the hour's + * budget is spent. It paces itself (a few in flight, a gap between starts), + * reads the rate-limit headers on every reply, and when the hour is down to + * the reserve it stops and waits for the reset instead of running into a 403. + * A 403 or 429 that arrives anyway is waited out for as long as GitHub says, + * on one shared pause, never on a capped guess. Replies are cached by ETag so + * a page that has not changed comes back as a free 304. Once any task has + * failed nothing queued behind it starts, because a crashed scan that keeps + * draining its queue in the background is how an hour disappears. + */ export class GitHub { - readonly calls = { made: 0, retries: 0 }; + /** made: requests sent. cached: answered 304, free. counted: what the hour was charged. */ + readonly calls = { made: 0, retries: 0, cached: 0, counted: 0 }; remaining: number | null = null; + /** When the hour resets, epoch ms, from the last reply. */ + resetAt: number | null = null; private readonly queue: (() => void)[] = []; private active = 0; + private aborted: Error | null = null; + private lastStart = Number.NEGATIVE_INFINITY; + private spacer: Promise = Promise.resolve(); + private pause: Promise | null = null; // Explicit fields rather than constructor parameter properties: Node's type // stripping accepts only syntax it can erase, and this is the one construct // in the repo it refuses (see gh.ts). private readonly token: string; private readonly fetchImpl: typeof fetch; private readonly concurrency: number; - - constructor(token: string, fetchImpl: typeof fetch, concurrency = 6) { + private readonly minIntervalMs: number; + private readonly reserve: number; + private readonly cacheDir: string | null; + private readonly onWait: (line: string) => void; + private readonly now: () => number; + private readonly sleep: (ms: number) => Promise; + + constructor(token: string, fetchImpl: typeof fetch, options: GitHubOptions | number = {}) { + const o = typeof options === 'number' ? { concurrency: options } : options; this.token = token; this.fetchImpl = fetchImpl; - this.concurrency = concurrency; + this.concurrency = o.concurrency ?? 4; + this.minIntervalMs = o.minIntervalMs ?? 100; + this.reserve = o.reserve ?? DEFAULT_RESERVE; + this.cacheDir = o.cacheDir ?? null; + this.onWait = o.onWait ?? (() => {}); + this.now = o.now ?? (() => Date.now()); + this.sleep = o.sleep ?? ((ms) => new Promise((s) => setTimeout(s, ms))); } - /** Run `fn` when a slot is free; six at a time keeps clear of the abuse limits. */ + /** Run `fn` when a slot is free. After a failure, queued tasks are rejected instead of started. */ limit(fn: () => Promise): Promise { return new Promise((resolve, reject) => { const start = () => { + if (this.aborted) { + reject(this.aborted); + this.queue.shift()?.(); + return; + } this.active += 1; - fn().then(resolve, reject).finally(() => { + fn().then(resolve, (error: unknown) => { + this.aborted ??= error instanceof Error ? error : new Error(String(error)); + reject(error); + }).finally(() => { this.active -= 1; this.queue.shift()?.(); }); @@ -607,30 +674,55 @@ export class GitHub { }); } + /** + * One line for the log: what the hour has left and when it turns, as the + * last reply's headers said. Not from /rate_limit: that endpoint reported a + * fresh hour for this token while every real reply carried the true count. + */ + budgetLine(): string { + if (this.remaining === null) return 'GitHub: rate limit unknown'; + return `GitHub: ${this.remaining} calls left this hour${this.resetAt ? `, resets ${hhmm(this.resetAt)}` : ''}, reserve ${this.reserve}`; + } + async get(pathOrUrl: string, options: { accept?: string; allow?: number[] } = {}): Promise<{ data: T | null; link: string; status: number }> { const url = pathOrUrl.startsWith('http') ? pathOrUrl : `https://api.github.com${pathOrUrl}`; + const accept = options.accept ?? 'application/vnd.github+json'; + const cached = this.readCache(url, accept); for (let attempt = 0; ; attempt += 1) { + if (this.aborted) throw this.aborted; + await this.pace(); this.calls.made += 1; - const r = await this.fetchImpl(url, { - headers: { - authorization: `Bearer ${this.token}`, - accept: options.accept ?? 'application/vnd.github+json', - 'x-github-api-version': '2022-11-28', - 'user-agent': 'gh-pulse (cli-tools)', - }, - }); - const rem = r.headers.get('x-ratelimit-remaining'); - if (rem !== null) this.remaining = Number(rem); - if (r.ok) return { data: (await r.json()) as T, link: r.headers.get('link') ?? '', status: r.status }; + const headers = this.headers(accept); + if (cached) headers['if-none-match'] = cached.etag; + const r = await this.fetchImpl(url, { headers }); + this.readLimits(r.headers); + if (r.status === 304 && cached) { + this.calls.cached += 1; + this.touchCache(url, accept); + return { data: cached.data as T, link: cached.link, status: cached.status }; + } + this.calls.counted += 1; + if (r.ok) { + const data = (await r.json()) as T; + const link = r.headers.get('link') ?? ''; + const etag = r.headers.get('etag'); + if (etag) this.writeCache({ url, etag, link, status: r.status, data, at: new Date(this.now()).toISOString() }, accept); + return { data, link, status: r.status }; + } if (options.allow?.includes(r.status)) return { data: null, link: '', status: r.status }; - const retryable = r.status === 429 || r.status === 502 || r.status === 503 || - (r.status === 403 && (r.headers.get('retry-after') !== null || rem === '0')); - if (retryable && attempt < 4) { + const retryAfter = Number(r.headers.get('retry-after') ?? 0) * 1000; + const limited = r.status === 429 || (r.status === 403 && (retryAfter > 0 || this.remaining === 0)); + const flaky = r.status === 502 || r.status === 503 || r.status === 504; + if ((limited || flaky) && attempt < 4) { this.calls.retries += 1; - let wait = Number(r.headers.get('retry-after') ?? 0) * 1000; - if (!wait && rem === '0') wait = Math.max(0, Number(r.headers.get('x-ratelimit-reset')) * 1000 - Date.now()) + 1000; - if (!wait) wait = 2000 * (attempt + 1); - await new Promise((s) => setTimeout(s, Math.min(wait, 120000))); + if (!limited) await this.sleep(2000 * (attempt + 1)); + else if (retryAfter > 0) { + this.onWait(`GitHub asked for a ${Math.ceil(retryAfter / 1000)}s pause`); + await this.sleep(retryAfter); + } else { + this.remaining = 0; + await this.holdUntilReset(); + } continue; } const body = await r.text().catch(() => ''); @@ -649,6 +741,162 @@ export class GitHub { } return out; } + + private headers(accept: string): Record { + return { + authorization: `Bearer ${this.token}`, + accept, + 'x-github-api-version': '2022-11-28', + 'user-agent': 'gh-pulse (cli-tools)', + }; + } + + private readLimits(headers: Headers): void { + const rem = headers.get('x-ratelimit-remaining'); + if (rem !== null && rem !== '' && Number.isFinite(Number(rem))) this.remaining = Number(rem); + const reset = headers.get('x-ratelimit-reset'); + if (reset !== null && reset !== '' && Number.isFinite(Number(reset))) this.resetAt = Number(reset) * 1000; + } + + /** + * Before a request starts: wait out a pause in progress, start one when the + * hour is down to the reserve, then keep the minimum gap from the previous + * start. Starts are serialised so concurrent callers space out; the slot + * itself is not held while waiting. + */ + private async pace(): Promise { + const previous = this.spacer; + let release!: () => void; + this.spacer = new Promise((r) => { release = r; }); + try { + await previous; + if (this.pause) await this.pause; + if (this.remaining !== null && this.remaining <= this.reserve && this.resetAt !== null && this.resetAt > this.now()) { + await this.holdUntilReset(); + } + const gap = this.lastStart + this.minIntervalMs - this.now(); + if (gap > 0) await this.sleep(gap); + this.lastStart = this.now(); + } finally { + release(); + } + } + + /** One shared pause until the hour resets; every caller waits on the same promise. */ + private holdUntilReset(): Promise { + if (!this.pause) { + // Without a reset time (no header on the reply) a minute is the guess, and the next reply corrects it. + const until = (this.resetAt ?? this.now() + 60000) + 1000; + const ms = Math.max(0, until - this.now()); + this.onWait(`rate limit: ${this.remaining ?? 0} of the hour left, keeping ${this.reserve} in reserve; resuming ${hhmm(until)}, in ${Math.max(1, Math.ceil(ms / 60000))} min`); + this.pause = this.sleep(ms).then(() => { + this.pause = null; + this.remaining = null; + this.resetAt = null; + }); + } + return this.pause; + } + + private cacheFile(url: string, accept: string): string | null { + if (!this.cacheDir) return null; + return join(this.cacheDir, `${createHash('sha1').update(`${accept}\n${url}`).digest('hex')}.json.gz`); + } + + private readCache(url: string, accept: string): CachedReply | null { + const file = this.cacheFile(url, accept); + if (!file || !existsSync(file)) return null; + try { + const reply = JSON.parse(gunzipSync(readFileSync(file)).toString('utf8')) as CachedReply; + return reply.url === url && typeof reply.etag === 'string' ? reply : null; + } catch { + return null; + } + } + + private writeCache(reply: CachedReply, accept: string): void { + const file = this.cacheFile(reply.url, accept); + if (!file || !this.cacheDir) return; + try { + mkdirSync(this.cacheDir, { recursive: true }); + // Written whole then renamed, so a run killed mid-write leaves no torn entry behind. + const tmp = `${file}.${process.pid}.tmp`; + writeFileSync(tmp, gzipSync(JSON.stringify(reply))); + renameSync(tmp, file); + } catch { + // A cache that cannot be written is only a cache. + } + } + + private touchCache(url: string, accept: string): void { + const file = this.cacheFile(url, accept); + if (!file) return; + try { + const t = new Date(this.now()); + utimesSync(file, t, t); + } catch { + // Pruning will take it a little early; nothing is lost. + } + } +} + +/** Drop cache entries GitHub has not been asked about for `maxAgeMs`: a deleted repo, a page from a range nobody scans. Returns how many went. */ +export function pruneCache(dir: string, now: number, maxAgeMs = 30 * 86400000): number { + if (!existsSync(dir)) return 0; + let removed = 0; + for (const f of readdirSync(dir)) { + if (!f.endsWith('.json.gz')) continue; + const p = join(dir, f); + try { + if (now - statSync(p).mtimeMs > maxAgeMs) { unlinkSync(p); removed += 1; } + } catch { + // Raced with another run; it is gone either way. + } + } + return removed; +} + +/** + * The client options every scan uses: the ETag cache under the data dir, the + * reserve from GH_PULSE_RESERVE (calls of the hour never spent, default 500), + * pauses announced on the progress line. + */ +export function clientOptions(dataDir: string, onWait: (line: string) => void, env: NodeJS.ProcessEnv = process.env): GitHubOptions { + const reserve = Number(env['GH_PULSE_RESERVE'] ?? DEFAULT_RESERVE); + return { cacheDir: cacheDir(dataDir), reserve: Number.isFinite(reserve) && reserve >= 0 ? reserve : DEFAULT_RESERVE, onWait }; +} + +/** + * One daily run at a time. The lock names the pid; a lock left by a process + * that no longer exists is stale and taken over. Two runs at once would spend + * the hour twice and then fight over the baseline. + */ +export function acquireRunLock(dataDir: string, pid = process.pid, alive: (pid: number) => boolean = isAlive): () => void { + mkdirSync(dataDir, { recursive: true }); + const file = join(dataDir, 'run.lock'); + if (existsSync(file)) { + const holder = Number(readFileSync(file, 'utf8').trim()); + if (Number.isFinite(holder) && holder !== pid && alive(holder)) { + throw new Error(`another gh-pulse run is in progress (pid ${holder}); wait for it, or remove ${file} if it is not`); + } + } + writeFileSync(file, `${pid}\n`); + return () => { + try { + if (readFileSync(file, 'utf8').trim() === String(pid)) unlinkSync(file); + } catch { + // Already gone. + } + }; +} + +function isAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === 'EPERM'; + } } export const nextLink = (link: string): string | undefined => /<([^>]+)>;\s*rel="next"/.exec(link)?.[1]; @@ -977,11 +1225,18 @@ export async function sendFailure(error: unknown, to: string, from: string, key: interface StarEntry { starred_at: string; user: { login: string } } interface ForkEntry { full_name: string; created_at: string } -interface CommitEntry { author?: { login?: string } | null; commit?: { author?: { name?: string } } } +interface CommitEntry { author?: { login?: string } | null; commit?: { author?: { name?: string; date?: string }; committer?: { date?: string } } } interface PullEntry { number: number; title: string; created_at: string; updated_at: string; merged_at: string | null; closed_at: string | null; user?: { login?: string }; html_url: string } interface IssueEntry { number: number; title: string; created_at: string; closed_at: string | null; user?: { login?: string }; html_url: string; pull_request?: unknown } interface ReleaseEntry { tag_name: string; published_at: string | null; created_at: string; html_url: string } +/** + * `since=` for a URL, floored to the hour. The same URL for an hour means a + * page that has not changed is a free 304 on every rescan; the exact cutoff is + * still applied to the items, so nothing from the extra minutes is counted. + */ +export const sinceParam = (cutoffMs: number): string => new Date(Math.floor(cutoffMs / 3600000) * 3600000).toISOString(); + /** Walk pages newest-first until `stop` says the rest is older than we need, or the budget is spent. Returns whether the walk was cut short. */ async function walk(gh: GitHub, path: string, max: number, stop: (item: T) => boolean, options: { accept?: string; allow?: number[] } = {}): Promise<{ items: T[]; partial: boolean }> { const items: T[] = []; @@ -1000,7 +1255,7 @@ async function walk(gh: GitHub, path: string, max: number, stop: (item: T) => /** Commits, PRs, issues and releases since `cutoffMs`, written into `m`. */ export async function collectEvents(gh: GitHub, name: string, cutoffMs: number, budget: PageBudget, m: Movement): Promise { - const cutoffIso = new Date(cutoffMs).toISOString(); + const cutoffIso = sinceParam(cutoffMs); const older = (iso: string | null | undefined): boolean => !!iso && Date.parse(iso) < cutoffMs; const [commits, pulls, issues, releases] = await Promise.all([ walk(gh, `/repos/${name}/commits?since=${cutoffIso}&per_page=100`, budget.commits, () => false, { allow: [409, 404] }), @@ -1008,8 +1263,10 @@ export async function collectEvents(gh: GitHub, name: string, cutoffMs: number, walk(gh, `/repos/${name}/issues?state=all&since=${cutoffIso}&sort=updated&per_page=100`, budget.issues, () => false, { allow: [404] }), walk(gh, `/repos/${name}/releases?per_page=100`, budget.releases, (r) => older(r.published_at ?? r.created_at), { allow: [404] }), ]); - m.commits = commits.items.length; - m.authors = [...new Set(commits.items.map((c) => c.author?.login ?? c.commit?.author?.name).filter((a): a is string => !!a))]; + // The URL asked from the top of the hour; the commits from before the exact cutoff are dropped here. + const recent = commits.items.filter((c) => { const d = c.commit?.committer?.date ?? c.commit?.author?.date; return !d || Date.parse(d) >= cutoffMs; }); + m.commits = recent.length; + m.authors = [...new Set(recent.map((c) => c.author?.login ?? c.commit?.author?.name).filter((a): a is string => !!a))]; for (const pr of pulls.items) { const ref: Ref = { n: pr.number, t: pr.title, u: pr.user?.login ?? '', url: pr.html_url }; if (Date.parse(pr.created_at) >= cutoffMs) { m.prOpened += 1; m.openedPrs.push(ref); } @@ -1070,8 +1327,18 @@ export function svgToPng(svg: string): Buffer { } export async function run(opt: RunOptions, deps: RunDeps = defaultDeps()): Promise { - const gh = new GitHub(deps.token(), deps.fetch); + const say = opt.progress ?? deps.log; + const gh = new GitHub(deps.token(), deps.fetch, clientOptions(opt.dataDir, say)); const now = deps.now(); + const unlock = acquireRunLock(opt.dataDir); + try { + return await runLocked(opt, deps, gh, now); + } finally { + unlock(); + } +} + +async function runLocked(opt: RunOptions, deps: RunDeps, gh: GitHub, now: Date): Promise { const prev = latestSnapshot(opt.dataDir, deps.log); const base = baselineFor(prev, now); const cutoff = base ? new Date(base.at) : new Date(now.getTime() - opt.hours * 3600000); @@ -1080,7 +1347,7 @@ export async function run(opt: RunOptions, deps: RunDeps = defaultDeps()): Promi const prevRepos = new Map(Object.entries(base?.repos ?? {})); const me = (await gh.get<{ login: string; followers: number }>('/user')).data!; - deps.log(`${me.login}, window since ${cutoffIso}${base ? ' (previous snapshot)' : ' (clock)'}`); + deps.log(`${me.login}, window since ${cutoffIso}${base ? ' (previous snapshot)' : ' (clock)'}; ${gh.budgetLine()}`); // 1. every repo the token can see, minus forks let repos = await gh.all('/user/repos?affiliation=owner,organization_member&per_page=100&sort=pushed', { max: 30 }); @@ -1239,8 +1506,14 @@ export async function run(opt: RunOptions, deps: RunDeps = defaultDeps()): Promi snapshotFile = writeSnapshot(opt.dataDir, snap); deps.log(`snapshot ${snapshotFile}`); } - deps.log(`${gh.calls.made} API calls (${gh.calls.retries} retries), ${gh.remaining} remaining this hour`); - return { subject, movers: movers.length, repos: repos.length, sent, snapshot: snapshotFile, html, calls: gh.calls.made }; + const pruned = pruneCache(cacheDir(opt.dataDir), now.getTime()); + deps.log(`${callsLine(gh)}${pruned ? `, ${pruned} stale cache entries dropped` : ''}`); + return { subject, movers: movers.length, repos: repos.length, sent, snapshot: snapshotFile, html, calls: gh.calls.counted }; +} + +/** The closing line of a scan: what the hour was charged, what came free, what is left. */ +export function callsLine(gh: GitHub): string { + return `${gh.calls.counted} API calls charged (${gh.calls.cached} answered from cache, ${gh.calls.retries} retries), ${gh.remaining ?? 'unknown'} left this hour`; } // ---------------------------------------------------------------- range scans @@ -1264,16 +1537,16 @@ export interface RangeOptions { } export async function rangeScan(opt: RangeOptions, deps: RunDeps = defaultDeps()): Promise<{ report: ReportJson; html: string; subject: string; sent: boolean; calls: number }> { - const gh = new GitHub(deps.token(), deps.fetch); + const say = opt.progress ?? deps.log; + const gh = new GitHub(deps.token(), deps.fetch, clientOptions(opt.dataDir, say)); const now = deps.now(); const { spec } = opt; const cutoffMs = spec.since.getTime(); const days = rangeDays(spec, now); const budget = pageBudget(spec.key, days); - const say = opt.progress ?? deps.log; const me = (await gh.get<{ login: string; followers: number }>('/user')).data!; - say(`${me.login}, ${spec.label}${spec.key === 'all' ? '' : ` (since ${spec.since.toISOString()})`}`); + say(`${me.login}, ${spec.label}${spec.key === 'all' ? '' : ` (since ${spec.since.toISOString()})`}; ${gh.budgetLine()}`); let repos = await gh.all('/user/repos?affiliation=owner,organization_member&per_page=100&sort=pushed', { max: 30 }); repos = repos.filter((r) => !r.fork); @@ -1418,8 +1691,8 @@ export async function rangeScan(opt: RangeOptions, deps: RunDeps = defaultDeps() sent = true; say(`sent to ${opt.to}: ${subject}`); } - say(`${gh.calls.made} API calls (${gh.calls.retries} retries), ${gh.remaining} remaining this hour`); - return { report, html, subject, sent, calls: gh.calls.made }; + say(callsLine(gh)); + return { report, html, subject, sent, calls: gh.calls.counted }; } /** The cached range report, if one exists and is younger than `maxAgeMs`. */ diff --git a/test/gh-pulse-client.test.ts b/test/gh-pulse-client.test.ts new file mode 100644 index 0000000..0e80097 --- /dev/null +++ b/test/gh-pulse-client.test.ts @@ -0,0 +1,265 @@ +import { describe, expect, it } from 'vitest'; +import { existsSync, mkdtempSync, readdirSync, utimesSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { + DEFAULT_RESERVE, + GitHub, + acquireRunLock, + cacheDir, + clientOptions, + collectEvents, + emptyMovement, + pageBudget, + pruneCache, + sinceParam, +} from '../src/gh-pulse.ts'; + +/** A GitHub reply; the headers are what the client reads on every response. */ +function reply(status: number, body: unknown = null, headers: Record = {}): Response { + return new Response(status === 304 ? null : JSON.stringify(body), { status, headers: { 'content-type': 'application/json', ...headers } }); +} + +/** A fake clock: `sleep` advances it instead of waiting, and records every wait. */ +function clock(start = 1_000_000_000_000) { + let t = start; + const sleeps: number[] = []; + return { now: () => t, sleep: async (ms: number) => { sleeps.push(ms); t += ms; }, sleeps }; +} + +interface Call { url: string; headers: Record; at: number } + +/** A fetch that answers from a script, recording every request with the clock time it started. */ +function scripted(c: ReturnType, answers: ((call: Call) => Response)[]) { + const calls: Call[] = []; + const fetchImpl = (async (input: string | URL | Request, init?: RequestInit) => { + const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url; + const headers = Object.fromEntries(Object.entries((init?.headers ?? {}) as Record).map(([k, v]) => [k.toLowerCase(), v])); + const call = { url, headers, at: c.now() }; + calls.push(call); + const next = answers.shift(); + if (!next) throw new Error(`no scripted reply for ${url}`); + return next(call); + }) as unknown as typeof fetch; + return { fetchImpl, calls }; +} + +const HOUR = 3600_000; + +describe('GitHub client pacing', () => { + it('keeps the minimum gap between request starts', async () => { + const c = clock(); + const { fetchImpl, calls } = scripted(c, [() => reply(200, { a: 1 }), () => reply(200, { a: 2 }), () => reply(200, { a: 3 })]); + const gh = new GitHub('t', fetchImpl, { minIntervalMs: 100, now: c.now, sleep: c.sleep }); + await Promise.all([gh.get('/a'), gh.get('/b'), gh.get('/c')]); + // Three starts, two gaps of the minimum: the first goes at once, each later one waits for its gap. + // (The fake clock jumps inside sleep, so the recorded fetch times are not the thing to assert on.) + expect(c.sleeps).toEqual([100, 100]); + expect(calls).toHaveLength(3); + expect(gh.calls).toEqual({ made: 3, retries: 0, cached: 0, counted: 3 }); + }); + + it('pauses until the hour resets once it is down to the reserve, then goes on', async () => { + const c = clock(); + const reset = String(Math.floor((c.now() + 30 * 60_000) / 1000)); + const waits: string[] = []; + const { fetchImpl, calls } = scripted(c, [ + () => reply(200, [1], { 'x-ratelimit-remaining': '500', 'x-ratelimit-reset': reset }), + () => reply(200, [2], { 'x-ratelimit-remaining': '4999', 'x-ratelimit-reset': String(Number(reset) + 3600) }), + ]); + const gh = new GitHub('t', fetchImpl, { reserve: 500, minIntervalMs: 0, now: c.now, sleep: c.sleep, onWait: (l) => waits.push(l) }); + const first = await gh.get('/one'); + expect(first.data).toEqual([1]); + expect(gh.remaining).toBe(500); + const second = await gh.get('/two'); + expect(second.data).toEqual([2]); + // The second request started after the reset, not before it. + expect(calls[1]!.at).toBeGreaterThanOrEqual(Number(reset) * 1000); + expect(c.sleeps.some((ms) => ms >= 30 * 60_000)).toBe(true); + expect(waits).toHaveLength(1); + expect(waits[0]).toMatch(/500 of the hour left/); + expect(waits[0]).toMatch(/reserve/); + expect(gh.calls.counted).toBe(2); + }); + + it('does not pause while the hour has more than the reserve', async () => { + const c = clock(); + const reset = String(Math.floor((c.now() + HOUR) / 1000)); + const { fetchImpl } = scripted(c, [() => reply(200, 1, { 'x-ratelimit-remaining': '501', 'x-ratelimit-reset': reset }), () => reply(200, 2)]); + const gh = new GitHub('t', fetchImpl, { reserve: 500, minIntervalMs: 0, now: c.now, sleep: c.sleep }); + await gh.get('/a'); + await gh.get('/b'); + expect(c.sleeps).toEqual([]); + }); + + it('waits out a 403 with the hour spent for as long as GitHub says, on one shared pause', async () => { + const c = clock(); + const reset = String(Math.floor((c.now() + 50 * 60_000) / 1000)); + const waits: string[] = []; + const limited = () => reply(403, { message: 'API rate limit exceeded' }, { 'x-ratelimit-remaining': '0', 'x-ratelimit-reset': reset }); + const { fetchImpl, calls } = scripted(c, [ + limited, limited, + () => reply(200, 'a', { 'x-ratelimit-remaining': '4999' }), + () => reply(200, 'b', { 'x-ratelimit-remaining': '4998' }), + ]); + const gh = new GitHub('t', fetchImpl, { minIntervalMs: 0, now: c.now, sleep: c.sleep, onWait: (l) => waits.push(l) }); + const [a, b] = await Promise.all([gh.get('/a'), gh.get('/b')]); + expect([a.data, b.data].sort()).toEqual(['a', 'b']); + expect(gh.calls.retries).toBe(2); + // One pause for both, not one each: one long sleep, and both retries start after the reset. + expect(c.sleeps.filter((ms) => ms >= 50 * 60_000)).toHaveLength(1); + expect(calls.slice(2).every((k) => k.at >= Number(reset) * 1000)).toBe(true); + expect(waits).toHaveLength(1); + expect(waits[0]).toMatch(/resuming \d\d:\d\d UTC, in 5[01] min/); + }); + + it('honours retry-after on a secondary limit', async () => { + const c = clock(); + const { fetchImpl } = scripted(c, [() => reply(403, { message: 'secondary rate limit' }, { 'retry-after': '30' }), () => reply(200, 'ok')]); + const gh = new GitHub('t', fetchImpl, { minIntervalMs: 0, now: c.now, sleep: c.sleep }); + expect((await gh.get('/x')).data).toBe('ok'); + expect(c.sleeps).toEqual([30_000]); + }); + + it('gives up on a 403 that is not about the rate limit', async () => { + const c = clock(); + const { fetchImpl } = scripted(c, [() => reply(403, { message: 'Resource not accessible' }, { 'x-ratelimit-remaining': '4000' })]); + const gh = new GitHub('t', fetchImpl, { minIntervalMs: 0, now: c.now, sleep: c.sleep }); + await expect(gh.get('/x')).rejects.toThrow(/GitHub 403/); + expect(c.sleeps).toEqual([]); + }); +}); + +describe('GitHub client cache', () => { + it('offers the ETag next time and takes a 304 from the cache without charging the hour', async () => { + const c = clock(); + const dir = mkdtempSync(join(tmpdir(), 'gh-pulse-cache-')); + const { fetchImpl, calls } = scripted(c, [ + () => reply(200, { views: [1, 2] }, { etag: 'W/"abc"', link: '; rel="next"' }), + () => reply(304), + ]); + const gh = new GitHub('t', fetchImpl, { cacheDir: dir, minIntervalMs: 0, now: c.now, sleep: c.sleep }); + const first = await gh.get('/repos/o/r/traffic/views'); + expect(calls[0]!.headers['if-none-match']).toBeUndefined(); + expect(readdirSync(dir)).toHaveLength(1); + const second = await gh.get('/repos/o/r/traffic/views'); + expect(calls[1]!.headers['if-none-match']).toBe('W/"abc"'); + expect(second).toEqual(first); + expect(second.link).toContain('rel="next"'); + expect(gh.calls).toEqual({ made: 2, retries: 0, cached: 1, counted: 1 }); + }); + + it('keeps entries apart by accept header, since the star listing is a different body', async () => { + const c = clock(); + const dir = mkdtempSync(join(tmpdir(), 'gh-pulse-cache-')); + const { fetchImpl, calls } = scripted(c, [() => reply(200, ['plain'], { etag: '"p"' }), () => reply(200, ['starred'], { etag: '"s"' })]); + const gh = new GitHub('t', fetchImpl, { cacheDir: dir, minIntervalMs: 0, now: c.now, sleep: c.sleep }); + await gh.get('/repos/o/r/stargazers'); + const starred = await gh.get('/repos/o/r/stargazers', { accept: 'application/vnd.github.star+json' }); + expect(calls[1]!.headers['if-none-match']).toBeUndefined(); + expect(starred.data).toEqual(['starred']); + expect(readdirSync(dir)).toHaveLength(2); + }); + + it('runs without a cache directory', async () => { + const c = clock(); + const { fetchImpl, calls } = scripted(c, [() => reply(200, 1, { etag: '"x"' }), () => reply(200, 2, { etag: '"x"' })]); + const gh = new GitHub('t', fetchImpl, { minIntervalMs: 0, now: c.now, sleep: c.sleep }); + await gh.get('/a'); + await gh.get('/a'); + expect(calls[1]!.headers['if-none-match']).toBeUndefined(); + expect(gh.calls.cached).toBe(0); + }); + + it('prunes entries nobody has asked about for a month and leaves the rest', () => { + const dir = mkdtempSync(join(tmpdir(), 'gh-pulse-cache-')); + const now = Date.now(); + writeFileSync(join(dir, 'old.json.gz'), 'x'); + writeFileSync(join(dir, 'fresh.json.gz'), 'x'); + writeFileSync(join(dir, 'other.txt'), 'x'); + const old = new Date(now - 40 * 86400000); + utimesSync(join(dir, 'old.json.gz'), old, old); + utimesSync(join(dir, 'other.txt'), old, old); + expect(pruneCache(dir, now)).toBe(1); + expect(readdirSync(dir).sort()).toEqual(['fresh.json.gz', 'other.txt']); + expect(pruneCache(join(dir, 'missing'), now)).toBe(0); + }); + + it('reads the reserve from the environment and puts the cache under the data dir', () => { + expect(clientOptions('/d', () => {}, {})).toMatchObject({ cacheDir: cacheDir('/d'), reserve: DEFAULT_RESERVE }); + expect(clientOptions('/d', () => {}, { GH_PULSE_RESERVE: '1200' }).reserve).toBe(1200); + expect(clientOptions('/d', () => {}, { GH_PULSE_RESERVE: 'lots' }).reserve).toBe(DEFAULT_RESERVE); + expect(clientOptions('/d', () => {}, { GH_PULSE_RESERVE: '-5' }).reserve).toBe(DEFAULT_RESERVE); + }); +}); + +describe('GitHub client failure', () => { + it('starts nothing queued behind a failed task, and refuses further requests', async () => { + const c = clock(); + const { fetchImpl } = scripted(c, [() => reply(500, { message: 'boom' }), () => reply(200, 'never')]); + const gh = new GitHub('t', fetchImpl, { concurrency: 1, minIntervalMs: 0, now: c.now, sleep: c.sleep }); + let ran = 0; + const results = await Promise.allSettled([ + gh.limit(() => gh.get('/first')), + gh.limit(async () => { ran += 1; return gh.get('/second'); }), + gh.limit(async () => { ran += 1; return 'third'; }), + ]); + expect(results.map((r) => r.status)).toEqual(['rejected', 'rejected', 'rejected']); + expect(ran).toBe(0); + expect(gh.calls.made).toBe(1); + await expect(gh.get('/later')).rejects.toThrow(/GitHub 500/); + }); + + it('reads the budget off the reply headers, and says so when it has not seen one', async () => { + const c = clock(); + const reset = Math.floor((c.now() + HOUR) / 1000); + const { fetchImpl } = scripted(c, [() => reply(200, { login: 'u' }, { 'x-ratelimit-remaining': '4321', 'x-ratelimit-reset': String(reset) })]); + const gh = new GitHub('t', fetchImpl, { minIntervalMs: 0, now: c.now, sleep: c.sleep }); + expect(gh.budgetLine()).toBe('GitHub: rate limit unknown'); + await gh.get('/user'); + expect(gh.budgetLine()).toMatch(/^GitHub: 4321 calls left this hour, resets \d\d:\d\d UTC, reserve 500$/); + }); +}); + +describe('events since the cutoff', () => { + it('asks from the top of the hour so the URL repeats, and drops the commits from before the exact cutoff', async () => { + const cutoff = Date.parse('2026-09-13T03:05:36.694Z'); + expect(sinceParam(cutoff)).toBe('2026-09-13T03:00:00.000Z'); + const c = clock(); + const commit = (date: string, name: string) => ({ author: { login: name }, commit: { author: { name, date }, committer: { date } } }); + const { fetchImpl, calls } = scripted(c, [ + (call) => { + if (call.url.includes('/commits')) { + expect(call.url).toContain('since=2026-09-13T03:00:00.000Z'); + return reply(200, [commit('2026-09-13T04:00:00Z', 'late'), commit('2026-09-13T03:02:00Z', 'early')]); + } + return reply(200, []); + }, + () => reply(200, []), + () => reply(200, []), + () => reply(200, []), + ]); + const gh = new GitHub('t', fetchImpl, { minIntervalMs: 0, now: c.now, sleep: c.sleep }); + const m = emptyMovement(); + await collectEvents(gh, 'o/r', cutoff, pageBudget('daily'), m); + expect(calls.some((k) => k.url.includes('/commits?since=2026-09-13T03:00:00.000Z'))).toBe(true); + expect(m.commits).toBe(1); + expect(m.authors).toEqual(['late']); + }); +}); + +describe('run lock', () => { + it('refuses a second run while the first is alive, and takes over a stale lock', () => { + const dir = mkdtempSync(join(tmpdir(), 'gh-pulse-lock-')); + const release = acquireRunLock(dir, 100, () => true); + expect(existsSync(join(dir, 'run.lock'))).toBe(true); + expect(() => acquireRunLock(dir, 200, () => true)).toThrow(/pid 100/); + // The holder died: the next run takes the lock over. + const release2 = acquireRunLock(dir, 200, () => false); + release(); // the old releaser must not remove the new holder's lock + expect(existsSync(join(dir, 'run.lock'))).toBe(true); + release2(); + expect(existsSync(join(dir, 'run.lock'))).toBe(false); + }); +});