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