From 4aa72a6a7c1d628f8061092ec33084bc3dc29002 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 13 Sep 2026 06:56:06 +0000 Subject: [PATCH] gh-pulse: throttle the hour (0.36.0) A run died this morning with GitHub 403, rate limit exceeded, ten repos into the events step, and the spinner kept ticking after the error because the queue went on draining in the background. Four range scans and a daily run in one hour had spent the whole 5,000-call budget the token shares with gh and everything else, and the client's answer to a 403 was to wait at most two minutes, four times, then throw. The GitHub client is now the only place the hour is spent, and it behaves. GH_PULSE_RESERVE calls (default 500) are never spent: down to the reserve, or rate limited anyway, a scan pauses until the hour resets, on one shared pause for every in-flight request, for as long as the headers say, and then carries on. Four requests in flight and 100ms between starts keep it out of the secondary limits; retry-after is honoured. Every reply is cached by ETag under /cache and offered back with If-None-Match, and GitHub does not charge for a 304, so an unchanged page is free; since= is asked from the top of the hour so the URL repeats and a rescan hits the cache, with the exact cutoff applied to the items. Once a task has failed nothing queued behind it starts, and show exits the process when the screen closes, so a crash or a quit no longer spends the rest of the hour in the background. One daily run at a time, through /run.lock. The budget is read off the first real reply's headers, not /rate_limit, which reported a fresh hour for this token while real replies carried 4,695 left. Each scan logs what it has to spend and closes with what was charged, what came from cache and what is left. Measured live on one repo for the last hour: a cold scan charged 20 calls, the warm rescan 0, with 26 answered from cache. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012nnVu9u7HeHEVRoCwKcQM3 --- bin/gh-pulse.ts | 12 +- package.json | 2 +- src/gh-pulse.ts | 349 +++++++++++++++++++++++++++++++---- test/gh-pulse-client.test.ts | 265 ++++++++++++++++++++++++++ 4 files changed, 588 insertions(+), 40 deletions(-) create mode 100644 test/gh-pulse-client.test.ts 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); + }); +});