Artifact versioning on changed content (no write-once conflict), lease only queued runs, infeasible briefs never retried, worker client retries transient control-plane errors, graceful worker restart, low-downtime control-plane deploy script
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
6 changed files +84 −19
modified
apps/web/next.config.ts
+1 −0
@@ -30,6 +30,7 @@ const securityHeaders = [ | ||
| 30 | 30 | ]; |
| 31 | 31 | |
| 32 | 32 | const nextConfig: NextConfig = { |
| 33 | + distDir: process.env.NEXT_DIST_DIR || ".next", | |
| 33 | 34 | reactStrictMode: true, |
| 34 | 35 | poweredByHeader: false, |
| 35 | 36 | serverExternalPackages: ["pg", "pino", "@node-rs/argon2", "pdfjs-dist", "pdf-lib", "drizzle-orm"], |
modified
packages/core/src/workers.ts
+29 −17
@@ -177,7 +177,7 @@ export async function leaseJob(db: Db, worker: WorkerRow, profiles: string[]) { | ||
| 177 | 177 | .from(researchRuns) |
| 178 | 178 | .where( |
| 179 | 179 | and( |
| 180 | − inArray(researchRuns.status, ["queued", "retrying"]), | |
| 180 | + eq(researchRuns.status, "queued"), | |
| 181 | 181 | inArray(researchRuns.computeProfile, profiles.length ? profiles : ["standard"]), |
| 182 | 182 | ), |
| 183 | 183 | ) |
@@ -416,22 +416,24 @@ export async function ingestArtifact( | ||
| 416 | 416 | if (body.length > 512 * 1024 * 1024) throw invalid("artifact exceeds 512 MiB"); |
| 417 | 417 | const lease = await assertLease(db, runId, worker, fencingToken); |
| 418 | 418 | const [attempt] = await db.select().from(runAttempts).where(eq(runAttempts.id, lease.attemptId)).limit(1); |
| 419 | − const key = runArtifactKey(runId, attempt!.attemptNo, desc.path); | |
| 419 | + // existing versions of this path (any attempt): identical content is idempotent, changed content gets a new version | |
| 420 | + const [prev] = await db | |
| 421 | + .select({ version: artifacts.version, sha256: artifacts.sha256, id: artifacts.id }) | |
| 422 | + .from(artifacts) | |
| 423 | + .where(and(eq(artifacts.runId, runId), eq(artifacts.path, desc.path))) | |
| 424 | + .orderBy(descOrder(artifacts.version)) | |
| 425 | + .limit(1); | |
| 426 | + if (prev && prev.sha256 === desc.sha256) | |
| 427 | + return { ok: true as const, artifactId: prev.id, sha256: prev.sha256, bytes: body.length }; | |
| 428 | + const version = (prev?.version ?? 0) + 1; | |
| 429 | + const key = runArtifactKey( | |
| 430 | + runId, | |
| 431 | + attempt!.attemptNo, | |
| 432 | + version === 1 ? desc.path : `${desc.path}.v${version}`, | |
| 433 | + ); | |
| 420 | 434 | const stored = await storage.put(key, body, desc.sha256); |
| 421 | 435 | return db.transaction(async (tx) => { |
| 422 | 436 | await assertLease(tx, runId, worker, fencingToken); |
| 423 | − const [prev] = await tx | |
| 424 | − .select({ version: artifacts.version, sha256: artifacts.sha256, id: artifacts.id }) | |
| 425 | − .from(artifacts) | |
| 426 | − .where(and(eq(artifacts.runId, runId), eq(artifacts.path, desc.path))) | |
| 427 | − .orderBy(descOrder(artifacts.version)) | |
| 428 | − .limit(1); | |
| 429 | − if (prev && prev.sha256 === stored.sha256) | |
| 430 | − return { ok: true as const, artifactId: prev.id, sha256: stored.sha256, bytes: stored.bytes }; | |
| 431 | − const version = (prev?.version ?? 0) + 1; | |
| 432 | − const versionedKey = | |
| 433 | − version === 1 ? key : runArtifactKey(runId, attempt!.attemptNo, `${desc.path}.v${version}`); | |
| 434 | − if (version > 1) await storage.put(versionedKey, body, desc.sha256); | |
| 435 | 437 | const [row] = await tx |
| 436 | 438 | .insert(artifacts) |
| 437 | 439 | .values({ |
@@ -444,11 +446,20 @@ export async function ingestArtifact( | ||
| 444 | 446 | sha256: stored.sha256, |
| 445 | 447 | bytes: stored.bytes, |
| 446 | 448 | contentType: desc.contentType, |
| 447 | − storageKey: versionedKey, | |
| 449 | + storageKey: key, | |
| 448 | 450 | synthetic: desc.synthetic, |
| 449 | 451 | }) |
| 452 | + .onConflictDoNothing() | |
| 450 | 453 | .returning({ id: artifacts.id }); |
| 451 | − return { ok: true as const, artifactId: row!.id, sha256: stored.sha256, bytes: stored.bytes }; | |
| 454 | + if (!row) { | |
| 455 | + const [again] = await tx | |
| 456 | + .select({ id: artifacts.id }) | |
| 457 | + .from(artifacts) | |
| 458 | + .where(and(eq(artifacts.runId, runId), eq(artifacts.path, desc.path), eq(artifacts.version, version))) | |
| 459 | + .limit(1); | |
| 460 | + return { ok: true as const, artifactId: again!.id, sha256: stored.sha256, bytes: stored.bytes }; | |
| 461 | + } | |
| 462 | + return { ok: true as const, artifactId: row.id, sha256: stored.sha256, bytes: stored.bytes }; | |
| 452 | 463 | }); |
| 453 | 464 | } |
| 454 | 465 | |
@@ -507,7 +518,8 @@ export async function completeAttempt( | ||
| 507 | 518 | await transition(tx, runId, "cancelled", { ...base, reason: body.error ?? "cancelled" }); |
| 508 | 519 | } else { |
| 509 | 520 | const reason = `${body.outcome}${body.error ? `: ${body.error}` : ""}`; |
| 510 | − if (body.outcome === "budget_exceeded" || run.currentAttemptNo >= run.maxAttempts) { | |
| 521 | + const noRetry = /^(INFEASIBLE|POLICY)\b/.test(body.error ?? "") || /INFEASIBLE:/.test(body.error ?? ""); | |
| 522 | + if (body.outcome === "budget_exceeded" || noRetry || run.currentAttemptNo >= run.maxAttempts) { | |
| 511 | 523 | await transition(tx, runId, "failed", { ...base, reason }); |
| 512 | 524 | } else { |
| 513 | 525 | await transition(tx, runId, "retrying", { |
modified
packages/maclustr-adapter/src/index.ts
+26 −0
@@ -30,6 +30,7 @@ export interface WorkerClientOptions { | ||
| 30 | 30 | fetchImpl?: typeof fetch; |
| 31 | 31 | timeoutMs?: number; |
| 32 | 32 | userAgent?: string; |
| 33 | + retryAttempts?: number; | |
| 33 | 34 | } |
| 34 | 35 | |
| 35 | 36 | /** Worker-side HTTP client for the versioned MacLustr worker protocol (outbound-only). */ |
@@ -39,6 +40,7 @@ export class WorkerClient { | ||
| 39 | 40 | private readonly fetchImpl: typeof fetch; |
| 40 | 41 | private readonly timeoutMs: number; |
| 41 | 42 | private readonly userAgent: string; |
| 43 | + private readonly retryAttempts: number; | |
| 42 | 44 | |
| 43 | 45 | constructor(opts: WorkerClientOptions) { |
| 44 | 46 | this.base = opts.baseUrl.replace(/\/+$/, ""); |
@@ -46,6 +48,7 @@ export class WorkerClient { | ||
| 46 | 48 | this.fetchImpl = opts.fetchImpl ?? fetch; |
| 47 | 49 | this.timeoutMs = opts.timeoutMs ?? 30_000; |
| 48 | 50 | this.userAgent = opts.userAgent ?? `autoscholar-worker/protocol-${WORKER_PROTOCOL_VERSION}`; |
| 51 | + this.retryAttempts = opts.retryAttempts ?? 8; | |
| 49 | 52 | } |
| 50 | 53 | |
| 51 | 54 | setToken(token: string): void { |
@@ -58,6 +61,29 @@ export class WorkerClient { | ||
| 58 | 61 | body?: unknown, |
| 59 | 62 | schema?: z.ZodType<T>, |
| 60 | 63 | raw?: { buffer: Buffer; contentType: string }, |
| 64 | + ): Promise<T> { | |
| 65 | + // transient failures (network, 5xx) are retried with backoff so a short control-plane restart never kills a run | |
| 66 | + const maxAttempts = this.retryAttempts; | |
| 67 | + let lastErr: unknown; | |
| 68 | + for (let attempt = 1; attempt <= maxAttempts; attempt++) { | |
| 69 | + try { | |
| 70 | + return await this.callOnce(method, path, body, schema, raw); | |
| 71 | + } catch (e) { | |
| 72 | + lastErr = e; | |
| 73 | + const transient = !(e instanceof ControlPlaneError) || e.status >= 500; | |
| 74 | + if (!transient || attempt === maxAttempts) throw e; | |
| 75 | + await new Promise((r) => setTimeout(r, Math.min(15_000, 1000 * 2 ** (attempt - 1)))); | |
| 76 | + } | |
| 77 | + } | |
| 78 | + throw lastErr; | |
| 79 | + } | |
| 80 | + | |
| 81 | + private async callOnce<T>( | |
| 82 | + method: string, | |
| 83 | + path: string, | |
| 84 | + body?: unknown, | |
| 85 | + schema?: z.ZodType<T>, | |
| 86 | + raw?: { buffer: Buffer; contentType: string }, | |
| 61 | 87 | ): Promise<T> { |
| 62 | 88 | const ctrl = new AbortController(); |
| 63 | 89 | const t = setTimeout(() => ctrl.abort(), this.timeoutMs); |
added
scripts/deploy-control-plane.sh
+15 −0
@@ -0,0 +1,15 @@ | ||
| 1 | +#!/bin/bash | |
| 2 | +# Zero-rebuild-downtime update of the control plane on the node (run ON M3U96b from ~/apps/autoscholar): | |
| 3 | +# git pull → install → build packages/services → migrate → build web into .next-new → swap → pm2 restart (≈5 s gap) | |
| 4 | +set -euo pipefail | |
| 5 | +export CI=true PATH=/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin | |
| 6 | +cd "$(dirname "$0")/.." | |
| 7 | +git pull -q origin main && git log --oneline -1 | |
| 8 | +pnpm install --frozen-lockfile 2>&1 | tail -1 | |
| 9 | +pnpm -r --filter "./packages/**" --filter "./services/**" build 2>&1 | grep -E "error|Failed" && exit 1 || true | |
| 10 | +DBURL="${DATABASE_URL:-postgresql://autoscholar:$(cat ~/apps/autoscholar-data/.pgpass-autoscholar)@127.0.0.1:5432/autoscholar}" | |
| 11 | +(cd packages/db && DATABASE_URL="$DBURL" pnpm migrate 2>&1 | tail -1) | |
| 12 | +(cd apps/web && rm -rf .next-new && NEXT_DIST_DIR=.next-new APP_BASE_URL=https://autoscholar.ai pnpm build 2>&1 | grep -E "error|✓ Compiled|Failed") | |
| 13 | +(cd apps/web && rm -rf .next-old && { [ -d .next ] && mv .next .next-old || true; } && mv .next-new .next && rm -rf .next-old) | |
| 14 | +pm2 restart autoscholar autoscholar-scheduler >/dev/null | |
| 15 | +sleep 4; curl -s -o /dev/null -w "local health %{http_code}\n" http://127.0.0.1:8470/api/public/topics | |
modified
services/worker/src/main.ts
+9 −1
@@ -281,6 +281,13 @@ class Worker { | ||
| 281 | 281 | if (!this.cfg.apiKey) throw new Error("ANTHROPIC_API_KEY not configured on this worker"); |
| 282 | 282 | outcome = await runResearch(runnerJob, host); |
| 283 | 283 | } |
| 284 | + if (state.restart && outcome.outcome === "cancelled") { | |
| 285 | + outcome = { | |
| 286 | + ...outcome, | |
| 287 | + outcome: "failed", | |
| 288 | + error: "WORKER_RESTART: worker daemon restarted; resume from checkpoint", | |
| 289 | + }; | |
| 290 | + } | |
| 284 | 291 | } catch (e) { |
| 285 | 292 | outcome = { |
| 286 | 293 | outcome: "failed", |
@@ -315,7 +322,8 @@ class Worker { | ||
| 315 | 322 | |
| 316 | 323 | async stop() { |
| 317 | 324 | this.stopping = true; |
| 318 | − for (const r of this.running.values()) r.cancel = true; | |
| 325 | + // graceful restart: abort sessions (isCancelled() → true via `stopping`) but report a retryable failure, not a cancel | |
| 326 | + for (const r of this.running.values()) r.restart = true; | |
| 319 | 327 | const t0 = Date.now(); |
| 320 | 328 | while (this.running.size > 0 && Date.now() - t0 < 60_000) await new Promise((r) => setTimeout(r, 500)); |
| 321 | 329 | } |
modified
services/worker/src/node-meta.ts
+4 −1
@@ -20,7 +20,10 @@ function tryExec(cmd: string, args: string[]): string | undefined { | ||
| 20 | 20 | } |
| 21 | 21 | |
| 22 | 22 | export function sdkVersion(): string { |
| 23 | − for (const p of [new URL("../node_modules/@anthropic-ai/claude-agent-sdk/package.json", import.meta.url), new URL("../../../node_modules/@anthropic-ai/claude-agent-sdk/package.json", import.meta.url)]) { | |
| 23 | + for (const p of [ | |
| 24 | + new URL("../node_modules/@anthropic-ai/claude-agent-sdk/package.json", import.meta.url), | |
| 25 | + new URL("../../../node_modules/@anthropic-ai/claude-agent-sdk/package.json", import.meta.url), | |
| 26 | + ]) { | |
| 24 | 27 | try { |
| 25 | 28 | return (JSON.parse(readFileSync(p, "utf8")) as { version: string }).version; |
| 26 | 29 | } catch { |
| 27 | 30 | |