From e7cb495d914f547e848ba563974f7abbd1520624 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 16:20:44 +0000 Subject: [PATCH] fix(ads): call the c0mpute image gateway over node:http Live 2026-11-08: Bun 1.4.4's fetch gave up on a hero after ~6 min ('The operation timed out.') despite timeout:false and a 14-min AbortSignal. node:http with req.setTimeout(14 min) waits as long as we say. Test drives postLong against a real local server. Co-Authored-By: Claude Opus 5.5 --- lib/ads/heroBackfill.ts | 64 ++++++++++++++++++++++++--------- tests/ads-hero-backfill.test.ts | 50 ++++++++++++++++++++------ 2 files changed, 87 insertions(+), 27 deletions(-) diff --git a/lib/ads/heroBackfill.ts b/lib/ads/heroBackfill.ts index ccbe2e9..a8ef78a 100644 --- a/lib/ads/heroBackfill.ts +++ b/lib/ads/heroBackfill.ts @@ -12,6 +12,8 @@ // this is the free tier that runs when nothing else drew one. import type { SupabaseClient } from "@supabase/supabase-js"; +import { request as httpRequest } from "node:http"; +import { request as httpsRequest } from "node:https"; import { buildPrompt, uploadHero } from "./heroImage"; type Env = Record; @@ -44,27 +46,57 @@ export class C0mputeImageError extends Error { } } +type Poster = ( + url: string, + token: string, + body: string, + timeoutMs: number, +) => Promise<{ status: number; body: Buffer }>; + +/** + * POST over node:http. Not fetch: Bun 1.4's fetch gave up on a c0mpute image + * after ~6 minutes even with `timeout: false` (seen live 2026-11-08), and + * Node's fetch stops waiting for headers at 300 s. This waits `timeoutMs`. + */ +export const postLong: Poster = (url, token, body, timeoutMs) => + new Promise((resolve, reject) => { + const u = new URL(url); + const send = u.protocol === "https:" ? httpsRequest : httpRequest; + const req = send( + u, + { + method: "POST", + headers: { + authorization: `Bearer ${token}`, + "content-type": "application/json", + "content-length": Buffer.byteLength(body), + }, + }, + (res) => { + const chunks: Buffer[] = []; + res.on("data", (c: Buffer) => chunks.push(c)); + res.on("end", () => resolve({ status: res.statusCode ?? 0, body: Buffer.concat(chunks) })); + res.on("error", reject); + }, + ); + req.setTimeout(timeoutMs, () => req.destroy(new Error(`c0mpute image: no answer in ${Math.round(timeoutMs / 1000)}s`))); + req.on("error", reject); + req.end(body); + }); + /** One 768x512 (3:2, the hero's shape) image from a c0mpute CPU worker. */ export async function generateC0mputeImage( prompt: string, target: C0mputeImageTarget, - fetchImpl: typeof fetch = fetch, + post: Poster = postLong, ): Promise { - const res = await fetchImpl(target.url, { - method: "POST", - headers: { authorization: `Bearer ${target.token}`, "content-type": "application/json" }, - body: JSON.stringify({ prompt: prompt.slice(0, 2000), width: 768, height: 512 }), - signal: AbortSignal.timeout(GENERATE_TIMEOUT_MS), - // The worker runs on Bun, whose fetch has its own default timeout; ours - // above is the one that should apply (a CPU image can pass 5 minutes). - // Node's fetch ignores this key. - timeout: false, - } as RequestInit); - if (!res.ok) { - const detail = (await res.text().catch(() => "")).slice(0, 200); + const payload = JSON.stringify({ prompt: prompt.slice(0, 2000), width: 768, height: 512 }); + const res = await post(target.url, target.token, payload, GENERATE_TIMEOUT_MS); + if (res.status !== 200) { + const detail = res.body.toString("utf8").slice(0, 200); throw new C0mputeImageError(`c0mpute image ${res.status}: ${detail}`, res.status); } - const bytes = Buffer.from(await res.arrayBuffer()); + const bytes = res.body; if (bytes.length < 1000 || !bytes.subarray(0, 8).equals(PNG_MAGIC)) { throw new C0mputeImageError("c0mpute image: not a PNG", 502); } @@ -109,7 +141,7 @@ export type HeroBackfillResult = { */ export async function backfillAdHeroes( supabase: SupabaseClient, - opts: { env?: Env; now?: number; fetchImpl?: typeof fetch } = {}, + opts: { env?: Env; now?: number; post?: Poster } = {}, ): Promise { const target = c0mputeImageTarget(opts.env); if (!target) return { skipped: "not_configured" }; @@ -146,7 +178,7 @@ export async function backfillAdHeroes( let bytes: Buffer; try { - bytes = await generateC0mputeImage(prompt, target, opts.fetchImpl); + bytes = await generateC0mputeImage(prompt, target, opts.post); } catch (err) { const status = err instanceof C0mputeImageError ? err.status : 0; // 503 = no worker free / none online: back off the whole sweep, not this campaign. diff --git a/tests/ads-hero-backfill.test.ts b/tests/ads-hero-backfill.test.ts index d272d71..4c278c4 100644 --- a/tests/ads-hero-backfill.test.ts +++ b/tests/ads-hero-backfill.test.ts @@ -73,35 +73,38 @@ describe("ad hero backfill on c0mpute", () => { it("draws one hero and writes it to the campaign's image-less creatives", async () => { const { client, updates } = fakeSupabase([row]); uploadHero.mockResolvedValue("https://cdn/heroes/1.webp"); - const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { - const body = JSON.parse(String(init.body)); + const post = vi.fn(async (url: string, token: string, raw: string, timeoutMs: number) => { + expect(url).toBe("http://gw:8787/v1/images"); + expect(token).toBe(ENV.C0MPUTE_SPEECH_TOKEN); + expect(timeoutMs).toBe(14 * 60 * 1000); + const body = JSON.parse(raw); expect(body).toMatchObject({ width: 768, height: 512 }); expect(body.prompt).toContain("Ship faster"); expect(body.prompt).toContain("example.dev"); - return new Response(PNG, { status: 200, headers: { "content-type": "image/png" } }); + return { status: 200, body: PNG }; }); - const r = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never }); + const r = await backfillAdHeroes(client, { env: ENV, post }); expect(r).toEqual({ campaignId: "c1", heroUrl: "https://cdn/heroes/1.webp", creativesUpdated: 2 }); expect(updates).toEqual([{ values: { image_url: "https://cdn/heroes/1.webp" }, campaign: "c1" }]); }); it("backs the whole sweep off when no worker is free (503)", async () => { const { client } = fakeSupabase([row]); - const fetchImpl = vi.fn(async () => new Response("no worker", { status: 503 })); + const post = vi.fn(async () => ({ status: 503, body: Buffer.from("no worker") })); const now = 1_000_000; - const first = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now }); + const first = await backfillAdHeroes(client, { env: ENV, post, now }); expect(first.error).toMatch(/503/); - const second = await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now: now + 60_000 }); + const second = await backfillAdHeroes(client, { env: ENV, post, now: now + 60_000 }); expect(second).toEqual({ skipped: "network_backoff" }); - expect(fetchImpl).toHaveBeenCalledTimes(1); + expect(post).toHaveBeenCalledTimes(1); }); it("skips a campaign that failed on its own, and rejects non-PNG bytes", async () => { const { client } = fakeSupabase([row]); - const fetchImpl = vi.fn(async () => new Response("", { status: 200 })); + const post = vi.fn(async () => ({ status: 200, body: Buffer.from("") })); const now = 5_000_000; - expect((await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now })).error).toMatch(/not a PNG/); - expect(await backfillAdHeroes(client, { env: ENV, fetchImpl: fetchImpl as never, now: now + 1 })).toEqual({ + expect((await backfillAdHeroes(client, { env: ENV, post, now })).error).toMatch(/not a PNG/); + expect(await backfillAdHeroes(client, { env: ENV, post, now: now + 1 })).toEqual({ skipped: "nothing_to_do", }); }); @@ -111,3 +114,28 @@ describe("ad hero backfill on c0mpute", () => { expect(await backfillAdHeroes(client, { env: {} })).toEqual({ skipped: "not_configured" }); }); }); + +import { createServer } from "node:http"; +import { postLong } from "@/lib/ads/heroBackfill"; + +describe("postLong", () => { + it("POSTs to a real server and returns the body", async () => { + const server = createServer((req, res) => { + let b = ""; + req.on("data", (c) => (b += c)); + req.on("end", () => { + res.writeHead(200, { "x-auth": String(req.headers.authorization) }); + res.end(`ok:${b}`); + }); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", () => r())); + const { port } = server.address() as { port: number }; + try { + const r = await postLong(`http://127.0.0.1:${port}/v1/images`, "tok", '{"a":1}', 5000); + expect(r.status).toBe(200); + expect(r.body.toString()).toBe('ok:{"a":1}'); + } finally { + server.close(); + } + }); +});