diff --git a/apple/whistlegraph/TURNS.md b/apple/whistlegraph/TURNS.md index af42e37ca0..c6fcef5681 100644 --- a/apple/whistlegraph/TURNS.md +++ b/apple/whistlegraph/TURNS.md @@ -78,11 +78,15 @@ reach for `document`. The worker shims those to no-ops the way the conformance worker does. Pieces that depend on them still paint; what they cannot do headlessly, they do on the phone when the version arrives. -Fallback: a pool of headless Chrome tabs loading -`https://aesthetic.computer/wipe?…preview=walkieware`, which is exactly how -the mime feed was verified on 2026-10-09. It is proven and slower; it stays -as the escape hatch for a piece the isolate cannot run, decided per piece by -a `needsBrowser` flag the checks can set. +Built first, 2026-10-10: a pool of warm headless Chrome tabs +(`lith/whistlegraph-render.mjs`) loading +`https://aesthetic.computer/wipe?…preview=walkieware`, the runtime the phone +paints with. A tab comes up in ~2 s; a render is ~3 s including four frames +over 2.2 s; the same tab rasterizes chalk. A throw before the first paint +means no painted event (`unverified-render`); a throw after it arrives as a +warn-level console line with a stack, which the worker reads as a runtime +error, as the phone does. The graph.mjs isolate above is the lighter path and +can replace the pool per piece later without touching callers. ## Scale @@ -101,9 +105,9 @@ A turn is mostly waiting. One worker process holds many turns at once; the CPU-bound part is a second or two per candidate. Budget per worker: - `TURN_CONCURRENCY` model streams in flight (start at 24). -- `RENDER_CONCURRENCY` isolates at once (start at cores − 1); renders queue - inside the worker. -- Memory: ~60 MB per isolate while it runs, ~5 MB per idle turn. +- `RENDER_CONCURRENCY` renderers at once (Chrome tabs now, isolates later; + start at 2 per 4 GB); renders queue inside the worker. +- Memory: ~150 MB per warm Chrome tab, ~5 MB per idle turn. So **100 people iterating at once** is roughly 100 turns in flight, 100 × (2–4 renders of ~1 s) = a few hundred render-seconds per minute. Two workers on a diff --git a/lith/whistlegraph-render.live.mjs b/lith/whistlegraph-render.live.mjs new file mode 100644 index 0000000000..4f99930723 --- /dev/null +++ b/lith/whistlegraph-render.live.mjs @@ -0,0 +1,19 @@ +// Live check for the render pool: paints a real piece from the thread store +// and a broken one, prints what the runtime said. node lith/whistlegraph-render.live.mjs [code] +import {RenderPool} from './whistlegraph-render.mjs'; +import {readFileSync, writeFileSync} from 'node:fs'; +const token = JSON.parse(readFileSync(`${process.env.HOME}/.ac-token`, 'utf8')).access_token; +const code = process.argv[2] || 'wgLodaf'; +const thread = await (await fetch(`https://aesthetic.computer/api/whistlegraph?code=${code}`, {headers: {Authorization: `Bearer ${token}`}})).json(); +const head = thread.ledger.versions.find(v => v.id === thread.ledger.head); +const pool = new RenderPool({size: 1, log: (...a) => console.log(...a)}); +const t0 = Date.now(); await pool.start(); console.log('pool up in', Date.now() - t0, 'ms'); +const t1 = Date.now(); const good = await pool.render({source: head.source, renderID: 1}); +console.log('good logs sample', good.logs.slice(0, 3)); +console.log(`${code} v${head.id}:`, 'rendered', good.rendered, 'frames', good.frames.length, 'span', good.frames.at(-1)?.atMs, 'ms', 'logs', good.logs.length, 'in', Date.now() - t1, 'ms', good.error || ''); +if (good.frames[0]) { const b = Buffer.from(good.frames[0].png, 'base64'); writeFileSync('/tmp/wg-render-frame0.png', b); console.log('frame0', good.frames[0].width + 'x' + good.frames[0].height, 'declared; png IHDR', b.readUInt32BE(16) + 'x' + b.readUInt32BE(20), 'bytes', b.length, 'base64 chars', good.frames[0].png.length); } +const bad = await pool.render({source: 'export function paint({wipe}) { wipe("red"); throw Error("boom"); }', renderID: 2}, {timeoutMs: 8000}); +console.log('broken piece:', 'rendered', bad.rendered, 'error', bad.error || '', 'logs', bad.logs.length, 'levels', [...new Set(bad.logs.map(l => l.level))], 'first texts', bad.logs.filter(l => l.text).slice(0, 3).map(l => l.level + ':' + l.text.slice(0, 80))); +const chalk = await pool.chalkImage({schema: 'whistlegraph-drawing/v1', id: '9DF1CAA5-B135-4B26-8BA4-91533A8D08BC', revision: 1, aspect: 1.333, strokes: [[[100, 100, 0], [500, 500, 50], [900, 100, 100]]]}); +console.log('chalk png bytes', Math.round(chalk.source.data.length * 3 / 4)); +await pool.close(); diff --git a/lith/whistlegraph-render.mjs b/lith/whistlegraph-render.mjs new file mode 100644 index 0000000000..d84220e1e6 --- /dev/null +++ b/lith/whistlegraph-render.mjs @@ -0,0 +1,128 @@ +// Pixels for turns that run off the phone (apple/whistlegraph/TURNS.md, slice 3). +// +// A pool of warm headless Chrome tabs, each holding the same AC runtime the +// phone paints with (/wipe?…preview=walkieware). A render hands a tab a +// candidate's source, waits for the runtime's own painted/invalidated event +// for that exact source hash, and takes four frames over two seconds — the +// evidence shape the picture review already expects. The same tab also +// rasterizes chalk strokes to the PNG the model sees. +// +// This is the proven renderer; a graph.mjs worker-thread isolate is the +// lighter one and can replace it per piece later without touching callers. +import puppeteer from 'puppeteer'; +import {existsSync} from 'node:fs'; +import {normalizeDrawing} from '../apple/whistlegraph/Resources/Web/drawing-input.mjs'; + +const PREVIEW = '/wipe?noauth=true&noplot=true&nogap=true&nolabel=true&preview=walkieware'; +const CHROME = '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome'; +const sleep = ms => new Promise(r => setTimeout(r, ms)); +const PAGE_SCRIPT = `(() => { + window.acFORCE_NOGAP = true; window.acPACK_DENSITY = 2; window.acAutoDensityOverride = true; + const state = window.__wg = {ready: false, events: [], sessionID: 'wg-worker', revision: 0}; + window.addEventListener('aesel-preview', e => { const d = e.detail; if (!d || d.sessionID !== state.sessionID || d.revision !== state.revision) return; state.events.push({...d, atMs: performance.now()}); }); + window.__wgRender = async (source, renderID) => { + const current = ++state.revision; state.events = []; + const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(source)); + const hash = [...new Uint8Array(digest)].map(b => b.toString(16).padStart(2, '0')).join(''); + window.AC?.startAudio?.(); + window.acSEND({type: 'dropped:piece', content: {name: 'whistlegraph-preview', source, search: 'noauth=true&noplot=true&nogap=true&nolabel=true', isKidLisp: false, + aeselPreview: {sessionID: state.sessionID, revision: current, sourceHash: hash, requestID: renderID}}}); + return hash; + }; + const poll = setInterval(() => { if (window.preloaded && window.acSEND) { clearInterval(poll); state.ready = true; } }, 100); +})();`; + +export class RenderPool { + #browser = null; #tabs = []; #waiting = []; #site; #size; #viewport; #frameSize; #log; + // The runtime paints at the phone's proportions; the frames the reviewer + // sees are scaled to fit its 384-pixel limit (deviceScaleFactor < 1). + constructor({site = 'https://aesthetic.computer', size = 2, viewport = {width: 390, height: 520}, log = () => {}} = {}) { + const scale = Math.min(1, 384 / Math.max(viewport.width, viewport.height)); + this.#site = site; this.#size = size; this.#log = log; + this.#viewport = {...viewport, deviceScaleFactor: scale}; + this.#frameSize = {width: Math.round(viewport.width * scale), height: Math.round(viewport.height * scale)}; + } + async start() { + let shell = null; try { shell = puppeteer.executablePath({headless: 'shell'}); } catch {} + const executablePath = process.env.CHROME_PATH || (shell && existsSync(shell) ? shell : existsSync(CHROME) ? CHROME : undefined); + this.#browser = await puppeteer.launch({headless: shell && !process.env.CHROME_PATH ? 'shell' : true, executablePath, + args: ['--no-sandbox', '--disable-dev-shm-usage', '--autoplay-policy=no-user-gesture-required', '--mute-audio', `--window-size=${this.#viewport.width},${this.#viewport.height}`]}); + for (let i = 0; i < this.#size; i++) this.#tabs.push(await this.#open(i)); + this.#log('render pool ready', `${this.#size} tabs`, executablePath || 'bundled chrome'); + } + async #open(index) { + const page = await this.#browser.newPage(); + await page.setViewport(this.#viewport); + await page.evaluateOnNewDocument(PAGE_SCRIPT); + await page.goto(this.#site + PREVIEW, {waitUntil: 'domcontentloaded', timeout: 90_000}); + return {index, page, busy: false, renders: 0}; + } + async #acquire() { + for (;;) { + const tab = this.#tabs.find(t => !t.busy); + if (tab) { tab.busy = true; return tab; } + await new Promise(r => this.#waiting.push(r)); + } + } + async #release(tab, broken = false) { + try { + if (broken || ++tab.renders >= 40) { await tab.page.close().catch(() => {}); const fresh = await this.#open(tab.index); Object.assign(tab, fresh, {renders: 0}); } + } catch (error) { this.#log('tab recycle failed', error.message); } + tab.busy = false; this.#waiting.shift()?.(); + } + // {rendered, sourceHash, renderID, logs, frames, error} — frames only when it painted. + async render({source, renderID}, {timeoutMs = 20_000, spanMs = 2200, frames = 4} = {}) { + const tab = await this.#acquire(); let broken = false; + try { + await tab.page.waitForFunction('window.__wg && window.__wg.ready', {timeout: 90_000}); + const hash = await tab.page.evaluate((s, id) => window.__wgRender(s, id), source, renderID); + const logs = []; let painted = false, invalidated = false; + const drain = async () => { for (const e of await tab.page.evaluate(() => window.__wg.events.splice(0))) { + if (e.sourceHash !== hash) continue; + if (e.kind === 'painted') painted = true; + if (e.kind === 'invalidated') invalidated = true; + if (e.kind === 'console') logs.push({level: e.event?.level || 'log', text: String(e.event?.message || '').slice(0, 500)}); + } }; + const started = Date.now(); + while (Date.now() - started < timeoutMs) { await drain(); if (painted || invalidated) break; await sleep(100); } + const out = {rendered: painted && !invalidated, sourceHash: hash, renderID, logs, frames: []}; + if (!painted && !invalidated) out.error = `Nothing painted within ${timeoutMs / 1000} s`; + if (out.rendered) { + const first = Date.now(); + for (let i = 0; i < frames; i++) { + const at = Math.round(i * spanMs / (frames - 1)); + while (Date.now() - first < at) await sleep(20); + const png = await tab.page.screenshot({encoding: 'base64', type: 'png'}); + out.frames.push({png, atMs: Date.now() - first, width: this.#frameSize.width, height: this.#frameSize.height}); + } + await drain(); if (invalidated) { out.rendered = false; out.error = 'The piece broke while being watched'; } + } + return out; + } catch (error) { broken = true; return {rendered: false, sourceHash: null, renderID, logs: [], frames: [], error: error.message}; } + finally { await this.#release(tab, broken); } + } + // The chalk as the model sees it: the app's drawingImage, run in a tab's canvas. + async chalkImage(drawing) { + const value = normalizeDrawing(drawing); + const tab = await this.#acquire(); + try { + const data = await tab.page.evaluate(drawing => { + const canvas = document.createElement('canvas'), side = 768; + canvas.width = Math.round(side * Math.min(1, drawing.aspect)); canvas.height = Math.round(side / Math.max(1, drawing.aspect)); + const ctx = canvas.getContext('2d'); + ctx.fillStyle = '#ffffff'; ctx.fillRect(0, 0, canvas.width, canvas.height); + ctx.strokeStyle = ctx.fillStyle = '#202020'; ctx.lineWidth = 3; ctx.lineCap = ctx.lineJoin = 'round'; + const point = p => [p[0] / 1000 * canvas.width, p[1] / 1000 * canvas.height]; + for (const stroke of drawing.strokes) { + ctx.beginPath(); + if (stroke.every(p => p[0] === stroke[0][0] && p[1] === stroke[0][1])) { const [x, y] = point(stroke[0]); ctx.arc(x, y, 1.5, 0, Math.PI * 2); ctx.fill(); } + else { stroke.forEach((p, i) => ctx[i ? 'lineTo' : 'moveTo'](...point(p))); ctx.stroke(); } + } + return canvas.toDataURL('image/png').split(',')[1]; + }, value); + if (!data || data.length > 700_000) throw Error('Could not encode chalk image'); + return {type: 'image', source: {type: 'base64', media_type: 'image/png', data}}; + } finally { await this.#release(tab); } + } + async close() { await this.#browser?.close().catch(() => {}); this.#browser = null; this.#tabs = []; } +} diff --git a/lith/whistlegraph-worker.mjs b/lith/whistlegraph-worker.mjs index a2e4ca3750..6038f728d4 100644 --- a/lith/whistlegraph-worker.mjs +++ b/lith/whistlegraph-worker.mjs @@ -16,14 +16,18 @@ import {tmpdir, hostname} from 'node:os'; import {mongoTurnQueue} from '../system/backend/whistlegraph-turns.mjs'; import {mongoWhistlegraphStore, sourceHash} from '../system/backend/whistlegraph.mjs'; import {AcServer} from '../aesel/src/ac-server.mjs'; -import {sourceChecks, compileEditContract, runEditExperiment} from '../aesel/src/edit-contract.mjs'; +import {sourceChecks, compileEditContract, runEditExperiment, validateCandidate} from '../aesel/src/edit-contract.mjs'; import {selectedBranch} from '../apple/whistlegraph/Resources/Web/branch-context.mjs'; import {GENERATION_INSTRUCTIONS, generationProfile} from '../apple/whistlegraph/Resources/Web/generation-policy.mjs'; import {WARE_INSTRUCTIONS} from '../apple/whistlegraph/Resources/Web/wares.mjs'; +import {inferenceRequest} from '../apple/whistlegraph/Resources/Web/inference-input.mjs'; +import {reviewVisualResult} from '../apple/whistlegraph/Resources/Web/visual-review.mjs'; +import {RenderPool} from './whistlegraph-render.mjs'; const SITE = process.env.AC_SITE || 'https://aesthetic.computer'; const WORKER = process.env.WORKER_NAME || `${hostname()}:${process.pid}`; const CONCURRENCY = Math.max(1, Number(process.env.TURN_CONCURRENCY) || 4); +const RENDERERS = Math.max(1, Number(process.env.RENDER_CONCURRENCY) || 2); const HEARTBEAT_MS = 30_000; const BASE_PIECE = 'export function paint({wipe}) { wipe("black"); }'; const log = (...args) => console.log(new Date().toISOString(), `[${WORKER}]`, ...args); @@ -40,8 +44,11 @@ function workerFetch(owner, settings) { } // One turn: the same contract the phone compiles, the same model loop, the -// same single repair. Returns {source, findings, repairs, completed, error}. -export async function runTurn({job, thread, onCheckpoint = () => {}, signal}) { +// same single repair, the same picture review — on a render pool instead of +// the phone's preview. Returns {source, findings, repairs, notes, acceptance, checks}. +const REPAIR_NOTE = (findings) => '\n\nREPAIR THIS CANDIDATE ONCE. Preserve the edit contract; fix only these checks, inspect the resulting preview, and do not claim visual acceptance:\n' + JSON.stringify(findings); +const runtimeLevel = l => l.level === 'warn' && /^(?:Type|Reference|Range|Syntax)?Error\b|\bat paint \(/.test(l.text) ? 'error' : l.level; +export async function runTurn({job, thread, pool, onCheckpoint = () => {}, signal, log = () => {}}) { const head = thread.ledger?.versions.find(v => v.id === thread.ledger.head); if (!head || head.id !== job.baseVersion || sourceHash(head.source) !== job.baseHash) throw Object.assign(Error('The piece moved on before this turn ran'), {code: 'moved'}); const cwd = mkdtempSync(join(tmpdir(), 'wg-turn-')); @@ -49,11 +56,14 @@ export async function runTurn({job, thread, onCheckpoint = () => {}, signal}) { try { let source = head.source || BASE_PIECE; writeFileSync(file, source.endsWith('\n') ? source : source + '\n'); + const chalk = job.request.drawing && pool ? await pool.chalkImage(job.request.drawing).catch(error => { log('chalk image failed', error.message); return null; }) : null; const prompt = compileEditContract({request: job.request.text, ...selectedBranch(thread.ledger), source}) + '\nCurrent preview: 390 × 520 CSS points. Compose for this shape using screen.width and screen.height; keep subjects within the canvas and remain responsive when it resizes.'; - let completed = false, error = '', server = null, checkpointAt = 0; + const settingsFor = repair => generationProfile('', {repair, model: job.model || '', image: !!job.request.drawing}); + let completed = false, error = '', server = null, checkpointAt = 0, renders = 0, lastPainted = null, repairs = 0; + const checks = [], notes = []; const generate = async (task, repair) => { - const settings = generationProfile('', {repair, model: job.model || '', image: !!job.request.drawing}); + const settings = settingsFor(repair); server = new AcServer({cwd, piece: {file, checkpoint: async () => { source = readFileSync(file, 'utf8'); const now = Date.now(); if (now - checkpointAt > 5000) { checkpointAt = now; await onCheckpoint(source); } }}, token: async () => 'worker', fetch: workerFetch(job.owner, settings), site: SITE, preview: false, frameCapture: false, model: settings.model, rounds: settings.rounds, outputContinuations: settings.outputContinuations, reasoning: settings.reasoning, thinking: settings.thinking, @@ -63,43 +73,85 @@ export async function runTurn({job, thread, onCheckpoint = () => {}, signal}) { if (method === 'turn/completed') { completed = !params.turn.error && params.turn.status === 'completed'; error = params.turn.error ? String(params.turn.error.message || params.turn.error) : ''; } }); signal?.addEventListener('abort', () => server?.interrupt(), {once: true}); - await server.startTurn(task); + await server.startTurn(chalk ? [{type: 'text', text: task}, chalk] : task); source = readFileSync(file, 'utf8'); return completed; }; - const inspect = async () => { const findings = sourceChecks(source); return {passed: findings.length === 0, findings, sourceHash: sourceHash(source)}; }; - const result = await runEditExperiment({prompt, generate, inspect, cancelled: () => signal?.aborted === true, onRepair: () => log('repairing', job.code)}); + // The paint check: the candidate's own painted event for its exact hash, + // plus the runtime's console, judged by the same validator the phone uses. + const inspect = async () => { + const hash = sourceHash(source); + if (!pool) { const findings = sourceChecks(source); return {passed: findings.length === 0, findings, sourceHash: hash, proof: null}; } + const proof = await pool.render({source, renderID: ++renders}); + proof.logs = proof.logs.map(l => ({...l, level: runtimeLevel(l)})); + const validation = validateCandidate(source, {rendered: proof.rendered, sourceHash: proof.sourceHash, logs: proof.logs}, hash); + if (proof.rendered && validation.passed) lastPainted = {source, proof}; + log('paint check', job.code, proof.rendered ? 'painted' : 'not painted', validation.findings.map(f => f.code).join(',') || 'clean'); + return {...validation, proof}; + }; + const result = await runEditExperiment({prompt, generate, inspect, cancelled: () => signal?.aborted === true, onRepair: () => { repairs = 1; log('repairing', job.code); }}); if (result.cancelled) throw Object.assign(Error('Stopped'), {code: 'cancelled'}); - if (!result.completed) throw Error(error || 'The model did not finish'); + if (!result.completed) throw Object.assign(Error(error || 'The model did not finish'), {draft: source !== head.source ? source : null}); if (source.trim() === (head.source || '').trim()) throw Error('No change to the piece'); - return {source, findings: result.validation?.findings || [], repairs: result.repairs, completed: true}; + let candidate = source, proof = result.validation?.proof || null, findings = result.validation?.findings || []; + if (!result.validation?.passed) { + // A repair that broke the picture does not cost the work: the last candidate that painted is kept, with a note. + if (lastPainted && lastPainted.source !== candidate) { candidate = lastPainted.source; proof = lastPainted.proof; findings = []; notes.push('The last repair did not paint; kept the last picture that did.'); } + else throw Object.assign(Error('Nothing painted: ' + findings.map(f => f.code).join(', ')), {draft: source}); + } + if (!pool) return {source: candidate, findings, repairs, notes, acceptance: 'unreviewed', checks: ['source-only']}; + // The picture review, as on the phone: one repair if none was spent, the + // reviewer being unavailable is not a verdict, and findings left after the + // repair are notes on the saved version rather than a veto. + const review = (src, prf) => reviewVisualResult({evidence: {sourceHash: prf.sourceHash, renderID: prf.renderID, frames: prf.frames}, sourceHash: prf.sourceHash, renderID: prf.renderID, source: src, + request: inferenceRequest(job.request.text), history: selectedBranch(thread.ledger), drawing: chalk, model: settingsFor(false).model, token: 'worker', fetch: workerFetch(job.owner, settingsFor(false)), signal}); + let acceptance = 'unreviewed'; + try { + let verdict = await review(candidate, proof); checks.push(verdict.passed ? 'visual-pass' : 'visual-fail'); + log('picture review', job.code, verdict.passed ? 'passed' : 'findings: ' + verdict.findings.join('; ').slice(0, 200)); + if (!verdict.passed && repairs === 0 && candidate === source) { + repairs = 1; log('repairing picture', job.code); + const before = {source: candidate, proof}; + const ok = await generate(prompt + REPAIR_NOTE(verdict.findings), true); + const again = ok ? await inspect() : null; + if (again?.passed) { candidate = source; proof = again.proof; verdict = await review(candidate, proof); checks.push(verdict.passed ? 'visual-pass' : 'visual-fail'); } + else { candidate = before.source; proof = before.proof; notes.push('The repair did not paint; kept the picture before it.'); } + } + acceptance = verdict.passed ? 'reviewed' : 'reviewed-with-notes'; + if (!verdict.passed) notes.push(...verdict.findings.slice(0, 4)); + } catch (reviewError) { + if (signal?.aborted) throw reviewError; + checks.push('visual-unreviewed'); notes.push('Review unavailable: ' + String(reviewError.message || reviewError).slice(0, 160)); + log('picture review unavailable', job.code, reviewError.message); + } + return {source: candidate, findings, repairs, notes, acceptance, checks}; } finally { rmSync(cwd, {recursive: true, force: true}); } } // Append the version to the thread, as the phone's commit does. store.save // refuses if the revision moved, so a race with the phone cannot fork. -export async function commitVersion(store, job, thread, source, notes) { +export async function commitVersion(store, job, thread, source, notes, {acceptance = 'unreviewed', checks = []} = {}) { const ledger = thread.ledger; const id = Math.max(...ledger.versions.map(v => v.id)) + 1; const version = {id, parent: ledger.head, source, request: job.request.text, createdAt: new Date().toISOString(), layers: 0}; const saved = await store.save(job.owner, thread._id, thread.revision, {...ledger, head: id, versions: [...ledger.versions, version]}); if (!saved) throw Object.assign(Error('The piece moved on while the turn ran'), {code: 'moved'}); - return {versionID: id, revision: saved.revision, sourceHash: sourceHash(source), notes, acceptance: 'unreviewed'}; + return {versionID: id, revision: saved.revision, sourceHash: sourceHash(source), notes, acceptance, checks}; } -async function work(queue, store, job) { +async function work(queue, store, job, pool) { let beat = setInterval(() => queue.heartbeat(job._id, WORKER).catch(() => {}), HEARTBEAT_MS); const started = Date.now(); try { const thread = await store.read(job.owner, job.code); if (!thread) throw Object.assign(Error('Thread unavailable'), {code: 'moved'}); - const turn = await runTurn({job, thread, onCheckpoint: source => queue.heartbeat(job._id, WORKER, source)}); + const turn = await runTurn({job, thread, pool, onCheckpoint: source => queue.heartbeat(job._id, WORKER, source), log}); const fresh = await store.read(job.owner, job.code); - const result = await commitVersion(store, job, fresh, turn.source, turn.findings.map(f => f.code)); + const result = await commitVersion(store, job, fresh, turn.source, [...turn.findings.map(f => f.code), ...turn.notes], {acceptance: turn.acceptance, checks: turn.checks}); await queue.complete(job._id, WORKER, {...result, repairs: turn.repairs, elapsedMs: Date.now() - started}); - log('done', job.code, 'v' + result.versionID, `${Date.now() - started} ms`, turn.repairs ? 'after a repair' : ''); + log('done', job.code, 'v' + result.versionID, turn.acceptance, `${Date.now() - started} ms`, turn.repairs ? 'after a repair' : ''); } catch (error) { - const draft = job.checkpoint || null; + const draft = error.draft || job.checkpoint || null; await queue.fail(job._id, WORKER, error.message, {draft}); log('failed', job.code, error.message); } finally { clearInterval(beat); } @@ -112,19 +164,22 @@ export async function main() { const db = client.db(process.env.MONGODB_NAME); const queue = mongoTurnQueue(db.collection('walkieware-turns')); const store = mongoWhistlegraphStore(db.collection('walkieware-threads')); + const pool = new RenderPool({site: SITE, size: RENDERERS, log}); + await pool.start(); let running = 0, stopping = false; const stop = () => { stopping = true; log('stopping after current turns'); }; process.on('SIGTERM', stop); process.on('SIGINT', stop); - log('ready', `concurrency ${CONCURRENCY}`, SITE); + log('ready', `turns ${CONCURRENCY}, renderers ${RENDERERS}`, SITE); while (!stopping) { if (running >= CONCURRENCY) { await new Promise(r => setTimeout(r, 500)); continue; } let job = null; try { job = await queue.claim(WORKER); } catch (error) { log('claim failed', error.message); } if (!job) { await new Promise(r => setTimeout(r, 2000)); continue; } running++; log('claimed', job.code, job._id, `attempt ${job.attempts}`); - work(queue, store, job).finally(() => running--); + work(queue, store, job, pool).finally(() => running--); } while (running > 0) await new Promise(r => setTimeout(r, 500)); + await pool.close(); await client.close(); } if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) main().catch(error => { console.error(error); process.exit(1); });