| 1 | import type { JobCompletion, JobSpec, LogLine, StepReport } from "@diffinery/shared"; |
| 2 | |
| 3 | export class HttpError extends Error { |
| 4 | constructor( |
| 5 | public status: number, |
| 6 | message: string, |
| 7 | public body?: string, |
| 8 | ) { |
| 9 | super(message); |
| 10 | } |
| 11 | } |
| 12 | |
| 13 | const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); |
| 14 | |
| 15 | export interface RequestOptions { |
| 16 | method?: string; |
| 17 | /** JSON body, or a factory producing a raw body (re-invoked on retry). */ |
| 18 | json?: unknown; |
| 19 | body?: () => Promise<{ body: NonNullable<RequestInit["body"]>; length?: number; contentType?: string }>; |
| 20 | retries?: number; |
| 21 | /** Return the response for these statuses instead of throwing. */ |
| 22 | allow?: number[]; |
| 23 | headers?: Record<string, string>; |
| 24 | } |
| 25 | |
| 26 | /** |
| 27 | * Client for `${apiUrl}/runner/jobs/${jobId}` endpoints. When `base` is null |
| 28 | * (offline `--spec` mode) all calls are no-ops. |
| 29 | */ |
| 30 | export class RunnerApi { |
| 31 | constructor( |
| 32 | public readonly base: string | null, |
| 33 | private readonly token: string, |
| 34 | private readonly opts: { minBackoffMs?: number; maxBackoffMs?: number } = {}, |
| 35 | ) {} |
| 36 | |
| 37 | get online(): boolean { |
| 38 | return this.base !== null; |
| 39 | } |
| 40 | |
| 41 | url(path: string): string { |
| 42 | return `${this.base}${path}`; |
| 43 | } |
| 44 | |
| 45 | authHeaders(): Record<string, string> { |
| 46 | return { authorization: `Bearer ${this.token}` }; |
| 47 | } |
| 48 | |
| 49 | async request(path: string, o: RequestOptions = {}): Promise<Response> { |
| 50 | if (!this.base) throw new Error("runner API not configured"); |
| 51 | return this.requestUrl(this.url(path), o); |
| 52 | } |
| 53 | |
| 54 | async requestUrl(url: string, o: RequestOptions = {}, withAuth = true): Promise<Response> { |
| 55 | const retries = o.retries ?? 6; |
| 56 | const min = this.opts.minBackoffMs ?? 500; |
| 57 | const max = this.opts.maxBackoffMs ?? 15000; |
| 58 | let lastErr: unknown; |
| 59 | for (let attempt = 0; attempt <= retries; attempt++) { |
| 60 | if (attempt > 0) await sleep(Math.min(max, min * 2 ** (attempt - 1))); |
| 61 | try { |
| 62 | const headers: Record<string, string> = { ...(withAuth ? this.authHeaders() : {}), ...(o.headers ?? {}) }; |
| 63 | const init: RequestInit & { duplex?: string } = { method: o.method ?? "GET", headers }; |
| 64 | if (o.json !== undefined) { |
| 65 | headers["content-type"] = "application/json"; |
| 66 | init.body = JSON.stringify(o.json); |
| 67 | } else if (o.body) { |
| 68 | const b = await o.body(); |
| 69 | init.body = b.body; |
| 70 | if (b.contentType) headers["content-type"] = b.contentType; |
| 71 | if (b.length !== undefined) headers["content-length"] = String(b.length); |
| 72 | init.duplex = "half"; |
| 73 | } |
| 74 | const res = await fetch(url, init); |
| 75 | if (res.ok || o.allow?.includes(res.status)) return res; |
| 76 | const text = await res.text().catch(() => ""); |
| 77 | const err = new HttpError(res.status, `${init.method} ${redactUrl(url)} -> ${res.status} ${text.slice(0, 300)}`, text); |
| 78 | if (res.status === 408 || res.status === 429 || res.status >= 500) { |
| 79 | lastErr = err; |
| 80 | continue; |
| 81 | } |
| 82 | throw err; |
| 83 | } catch (e) { |
| 84 | if (e instanceof HttpError && !(e.status === 408 || e.status === 429 || e.status >= 500)) throw e; |
| 85 | lastErr = e; |
| 86 | } |
| 87 | } |
| 88 | throw lastErr instanceof Error ? lastErr : new Error(String(lastErr)); |
| 89 | } |
| 90 | |
| 91 | async getSpec(): Promise<JobSpec> { |
| 92 | const res = await this.request("", { retries: 8 }); |
| 93 | return (await res.json()) as JobSpec; |
| 94 | } |
| 95 | |
| 96 | async postLogs(lines: LogLine[]): Promise<void> { |
| 97 | if (!this.base) return; |
| 98 | await this.request("/logs", { method: "POST", json: { lines }, retries: 0 }); |
| 99 | } |
| 100 | |
| 101 | async postStep(index: number, report: StepReport): Promise<void> { |
| 102 | if (!this.base) return; |
| 103 | await this.request(`/steps/${index}`, { method: "POST", json: report }); |
| 104 | } |
| 105 | |
| 106 | async complete(c: JobCompletion): Promise<void> { |
| 107 | if (!this.base) return; |
| 108 | await this.request("/complete", { method: "POST", json: c, retries: 10 }); |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | function redactUrl(u: string): string { |
| 113 | return u.replace(/([?&](?:token|sig|signature)=)[^&]+/gi, "$1***"); |
| 114 | } |
| 115 | |
| 116 | /** |
| 117 | * Batches log lines and ships them every `intervalMs` or when `maxLines` |
| 118 | * accumulate. Lines are never dropped: failed batches are retried with |
| 119 | * exponential backoff and kept in order. |
| 120 | */ |
| 121 | export class LogBatcher { |
| 122 | private queue: LogLine[] = []; |
| 123 | private timer: NodeJS.Timeout | null = null; |
| 124 | private inflight: Promise<void> | null = null; |
| 125 | private failures = 0; |
| 126 | private nextAttemptAt = 0; |
| 127 | private closed = false; |
| 128 | |
| 129 | constructor( |
| 130 | private readonly sink: (lines: LogLine[]) => Promise<void>, |
| 131 | private readonly opts: { intervalMs?: number; maxLines?: number; maxBatch?: number; maxBackoffMs?: number; onError?: (e: unknown) => void } = {}, |
| 132 | ) { |
| 133 | this.timer = setInterval(() => void this.tick(), this.opts.intervalMs ?? 500); |
| 134 | this.timer.unref?.(); |
| 135 | } |
| 136 | |
| 137 | push(line: LogLine): void { |
| 138 | this.queue.push(line); |
| 139 | if (this.queue.length >= (this.opts.maxLines ?? 200)) void this.tick(); |
| 140 | } |
| 141 | |
| 142 | get pending(): number { |
| 143 | return this.queue.length; |
| 144 | } |
| 145 | |
| 146 | private tick(): Promise<void> { |
| 147 | if (this.inflight) return this.inflight; |
| 148 | if (this.queue.length === 0) return Promise.resolve(); |
| 149 | if (Date.now() < this.nextAttemptAt) return Promise.resolve(); |
| 150 | const batch = this.queue.slice(0, this.opts.maxBatch ?? 1000); |
| 151 | this.inflight = this.sink(batch) |
| 152 | .then(() => { |
| 153 | this.queue.splice(0, batch.length); |
| 154 | this.failures = 0; |
| 155 | this.nextAttemptAt = 0; |
| 156 | }) |
| 157 | .catch((e) => { |
| 158 | this.failures++; |
| 159 | this.nextAttemptAt = Date.now() + Math.min(this.opts.maxBackoffMs ?? 30000, 250 * 2 ** this.failures); |
| 160 | this.opts.onError?.(e); |
| 161 | }) |
| 162 | .finally(() => { |
| 163 | this.inflight = null; |
| 164 | }); |
| 165 | return this.inflight; |
| 166 | } |
| 167 | |
| 168 | /** Flushes everything, retrying until `deadlineMs` elapses. */ |
| 169 | async flush(deadlineMs = 120000): Promise<boolean> { |
| 170 | const until = Date.now() + deadlineMs; |
| 171 | while (this.queue.length > 0 || this.inflight) { |
| 172 | if (Date.now() > until) return false; |
| 173 | if (this.inflight) { |
| 174 | await this.inflight; |
| 175 | continue; |
| 176 | } |
| 177 | const wait = this.nextAttemptAt - Date.now(); |
| 178 | if (wait > 0) await sleep(Math.min(wait, Math.max(0, until - Date.now()))); |
| 179 | await this.tick(); |
| 180 | } |
| 181 | return true; |
| 182 | } |
| 183 | |
| 184 | async close(deadlineMs?: number): Promise<boolean> { |
| 185 | if (this.closed) return true; |
| 186 | this.closed = true; |
| 187 | if (this.timer) clearInterval(this.timer); |
| 188 | return this.flush(deadlineMs); |
| 189 | } |
| 190 | } |