diff --git a/.env.example b/.env.example index a2b6921..f5f1663 100644 --- a/.env.example +++ b/.env.example @@ -20,9 +20,12 @@ DATA_DIR=./data # Linux when unset. # CHROME_PATH=/usr/bin/chromium # Budget in milliseconds for ONE capture job. Capture and the LLM read share a job, so -# this covers both. The default suits a CLI provider; a slow local model may need more. -# Too low and an item fails as "timed out" with its page already captured. -# CAPTURE_TIMEOUT_MS=180000 +# this covers both: page navigation (120s) + network settle (30s) + the agent's own +# 300s ceiling all have to fit inside it. Deliberately generous — board-oss is +# single-tenant, so this is a stop-a-wedged-job ceiling, not a latency target, and a +# typical add finishes in ~40-50s. Too low and an item fails as "timed out" with its +# page already captured. +# CAPTURE_TIMEOUT_MS=600000 # --- LLM provider (optional; unset = no-AI, enrichment disabled) --- # CLI agent id for the subprocess provider (claude / codex / cursor-agent). diff --git a/README.md b/README.md index 0ce466d..1e4c683 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ the full annotated list). Empty/whitespace values are treated as unset. | `DATA_DIR` | `./data` | Persistent data root (SQLite DB + screenshots). | | `CHROME_PATH` | autodetect | System Chromium/Chrome binary; autodetected on Linux when unset. | | `LLM_AGENT` / `LLM_MODEL` / `LLM_BASE_URL` / `LLM_API_KEY` | unset | LLM provider. **Unset = no-AI** (enrichment disabled). With `LLM_AGENT=claude` and no `LLM_MODEL`, Board asks the CLI for **Sonnet** rather than inheriting your interactive default. | -| `CAPTURE_TIMEOUT_MS` | `180000` | Budget for one capture job — page capture **and** the AI read share it. A CLI agent routinely takes 90–100s; raise it for slow local models. | +| `CAPTURE_TIMEOUT_MS` | `600000` | Budget for one capture job — page capture **and** the AI read share it. Deliberately generous: it's a ceiling to stop a wedged job, not a latency target, and a typical add finishes in ~40–50s. Lower it only if you'd rather fail fast than wait. | | `BOARD_API_TOKEN` | unset | Bearer token for the `/api/v1` capture API. **Unset = the v1 API is off** (fail-closed). See [Integrations](docs/integrations.md). | ## Security & the reverse-proxy model diff --git a/docs/bmad/architecture.md b/docs/bmad/architecture.md index f78c3b9..e394b65 100644 --- a/docs/bmad/architecture.md +++ b/docs/bmad/architecture.md @@ -40,7 +40,7 @@ note: "Decisions were reached by multi-round party consensus; this formalizes th - **AD3 SQLite/Drizzle** — WAL + JSON + FTS5; screenshots files-on-disk, path in DB. - **AD4 Capture in-process, concurrency 1** — launch→screenshot→kill; the offloadable sidecar *service* is deferred but its **contract is designed in v1** (token-authed, idempotent). - **AD5 `LLMProvider` seam, two transports** — `HttpProvider` (API key / open model) + `CliProvider` (coding-agent subprocess); default install zero-coding-CLI. -- **AD6 Async job model** — in-process single-writer worker queue + `status` column + SSE; no external broker. The queue **is** the SQLite single-writer guard. +- **AD6 Async job model** — in-process worker queue + `status` column + SSE; no external broker. **Two lanes:** jobs serialize at concurrency 1 (memory bound), SQLite writes serialize separately (write-interleaving bound). *Revised: the job lane was originally the SQLite single-writer guard too, which made every interactive write wait on a running capture's network I/O.* - **AD7 Reverse-proxy-only auth** — localhost bind default. - **AD9 Schema-as-data** — board behavior is a stored `board_descriptor` on a closed field-type set; enrichment + rendering are dynamic. - **AD10 Agentic composer** — v1 launch feature, built after the seeded boards; meta-schema + validate-and-repair. @@ -87,7 +87,7 @@ interface CaptureAdapter { fetch(source: string, ctx): Promise<{ fields: Record< - **Composer meta-schema:** the JSON-schema *for a descriptor*; the composer emits a descriptor validated against it (validate-and-repair; closed types; field cap; reserved-key rejection). ### 4.5 Job model & status (AD6) -- `JobQueue` = a single async worker draining jobs serially; capture + enrichment jobs run here (this is also the SQLite single-writer). +- `JobQueue` = a single async worker draining jobs serially; capture + enrichment jobs run here. It is **not** the SQLite single-writer: writes have their own lane, because a job holds its slot across Chrome + LLM round-trips (up to `CAPTURE_TIMEOUT_MS`) while every SQLite write is sub-millisecond. Sharing one lane meant `POST /api/collections/:cid/items` blocked for the length of the running capture. Writes issued from inside a job go through the write lane like any other. - `item.status`: `pending → processing → done → error` (`error_reason` persisted). SSE endpoint streams transitions; refetch/poll fallback. ## 5. Data model diff --git a/src/capture/adapter.ts b/src/capture/adapter.ts index 1983275..b50f700 100644 --- a/src/capture/adapter.ts +++ b/src/capture/adapter.ts @@ -1,7 +1,7 @@ import { eq } from 'drizzle-orm'; import { boards, items, type NewAsset } from '../db/schema.js'; -import { writeItemDirect } from '../db/queue.js'; +import { enqueueWrite, writeItemDirect } from '../db/queue.js'; import { statusHub } from '../sse.js'; import { createUrlScreenshotAdapter } from './url-screenshot.js'; import { createUrlReadableAdapter } from './url-readable.js'; @@ -148,8 +148,6 @@ export async function runCaptureForItem( registerTeardown: args.registerTeardown, }); - const item = handle.db.select().from(items).where(eq(items.id, args.itemId)).get(); - // Captured keys that are SYSTEM COLUMNS (e.g. `title`) belong on the column, not in // the `item.fields` JSON bag (the descriptor contract: title/notes/favorite are // system columns). Lift those out; merge the rest into fields. Without this, a @@ -160,7 +158,6 @@ export async function runCaptureForItem( if (SYSTEM_COLUMNS.has(k)) systemUpdates[k] = v; else capturedFields[k] = v; } - const mergedFields = { ...((item?.fields as Record) ?? {}), ...capturedFields }; const assetRows: NewAsset[] = result.assets.map((a, i) => ({ id: `${args.itemId}-${a.kind}-${i}`, itemId: args.itemId, @@ -171,14 +168,27 @@ export async function runCaptureForItem( hash: a.hash ?? null, })); - // Capture runs INSIDE the worker job (slot held) → use the DIRECT write (calling - // the enqueueing writeItem here would deadlock). Replaces the item's assets - // (delete-then-insert) → idempotent re-capture. - writeItemDirect( - handle, - { ...item, ...systemUpdates, id: args.itemId, boardId: args.boardId, fields: mergedFields }, - assetRows, - ); + // Read the row and write it back in ONE op on the write lane, with no await between + // them — `writeItemDirect` takes the FULL desired row, so a merge built from a stale + // snapshot would revert whatever landed in between. Enqueued rather than called + // directly: jobs and writes are separate lanes now, so a write from inside a job no + // longer self-deadlocks. Replaces the item's assets (delete-then-insert) → idempotent + // re-capture. + const prevTitle = await enqueueWrite(() => { + const item = handle.db.select().from(items).where(eq(items.id, args.itemId)).get(); + writeItemDirect( + handle, + { + ...item, + ...systemUpdates, + id: args.itemId, + boardId: args.boardId, + fields: { ...((item?.fields as Record) ?? {}), ...capturedFields }, + }, + assetRows, + ); + return item?.title ?? undefined; + }); // Progressive reveal: the row now holds the page (title + image) but the AI read is // still outstanding, and the item's DB status stays `processing` throughout. Announce @@ -189,7 +199,7 @@ export async function runCaptureForItem( itemId: args.itemId, boardId: args.boardId, status: 'captured', - title: (systemUpdates as { title?: string }).title ?? item?.title ?? undefined, + title: (systemUpdates as { title?: string }).title ?? prevTitle, screenshot: shot?.path || undefined, }); } diff --git a/src/capture/url-readable.ts b/src/capture/url-readable.ts index e2564a3..75f33dd 100644 --- a/src/capture/url-readable.ts +++ b/src/capture/url-readable.ts @@ -33,7 +33,9 @@ const IMAGE_EXT: Record = { 'image/webp': 'webp', }; const MAX_IMAGE_BYTES = 12 * 1024 * 1024; // a hero image over ~12MB is almost certainly wrong -const IMAGE_TIMEOUT_MS = 8000; +// A hero image can be several MB; 8s dropped it on any slow link, and the card then +// fell back to its no-image placeholder with no sign anything had gone wrong. +const IMAGE_TIMEOUT_MS = 30_000; interface Deps { fetchImpl?: typeof fetch; diff --git a/src/capture/url-screenshot.test.ts b/src/capture/url-screenshot.test.ts index 69eaa13..397050f 100644 --- a/src/capture/url-screenshot.test.ts +++ b/src/capture/url-screenshot.test.ts @@ -59,6 +59,42 @@ describe('url-screenshot adapter (Story 6.2)', () => { assert.equal(fb.isClosed(), true, 'browser must be closed on the error path'); }); + // Pages with a persistent connection (analytics socket, chat widget, video preload) + // never reach `networkidle2`, and a slow connection pushes an ordinary page past the + // 30s nav budget — either way navigation threw and the item died as "Timed out." + // Reproduced against twenty.com and stigg.io under CDP throttling: goto rejected at + // 30755ms with `TimeoutError: Navigation timeout of 30000 ms exceeded`, which + // cleanErrorReason maps to 'timed out'. A quiet network is a nice-to-have for a + // screenshot, not a preconditio — capture must degrade to "shot slightly early". + it('still captures a page whose network never goes quiet', async () => { + let idleAttempted = false; + let closed = false; + const page = { + setViewport: async () => {}, + // Mirrors puppeteer: waiting for a quiet network on such a page rejects. + goto: async (_url: string, opts: { waitUntil: string }) => { + if (opts.waitUntil === 'networkidle2') throw new Error('Navigation timeout of 30000 ms exceeded'); + }, + waitForNetworkIdle: async () => { + idleAttempted = true; + throw new Error('Timed out waiting for the network to be idle'); + }, + screenshot: async () => Buffer.from('PNGDATA'), + evaluate: async (fn: (...a: unknown[]) => unknown) => + fn.toString().includes('document.title') ? { title: 'Chatty', text: 'body text' } : undefined, + }; + const browser: CaptureBrowser = { newPage: async () => page as never, close: async () => { closed = true; } }; + + const adapter = createUrlScreenshotAdapter({ launch: async () => browser, sleep: async () => {} }); + const out = await adapter.fetch('https://chatty.example', { itemId: 'shot3', boardId: 'b', screenshotsDir: dir }); + + assert.equal(idleAttempted, true, 'should still TRY to let the network settle'); + assert.equal(out.fields.title, 'Chatty', 'capture succeeds despite the noisy network'); + assert.equal(out.assets.length, 1, 'a screenshot is still produced'); + assert.ok(existsSync(join(dir, 'shot3.png')), 'image written'); + assert.equal(closed, true, 'browser closed'); + }); + // AC 1 — registered for ingest_mode url-screenshot it('declares ingest_mode = url-screenshot', () => { const adapter = createUrlScreenshotAdapter(); diff --git a/src/capture/url-screenshot.ts b/src/capture/url-screenshot.ts index e0766bb..6c3aace 100644 --- a/src/capture/url-screenshot.ts +++ b/src/capture/url-screenshot.ts @@ -15,12 +15,33 @@ import type { AssetSpec, CaptureAdapter, CaptureCtx, CaptureResult, CaptureSourc // `error`. Teardown (`close()`) is guaranteed in `finally` and on abort (timeout). const VIEWPORT = { width: 1440, height: 900, deviceScaleFactor: 1.5 }; -const GOTO_TIMEOUT_MS = 30_000; +// Budget for reaching `domcontentloaded` — a REAL failure if it expires (no document, +// nothing to shoot). 30s was inherited from when this timeout ALSO had to absorb the +// wait for a quiet network; the settle is separate and best-effort now, so the two +// budgets are sized independently. +// +// These are deliberately generous. board-oss is single-tenant and self-hosted: nobody +// is queued behind you competing for a shared worker, so a capture that takes two +// minutes and SUCCEEDS beats one that fails fast at 30s and leaves you re-adding the +// link by hand. A heavy page on a slow link needs well over 30s just to parse — under +// CDP throttling twenty.com missed the old budget at 30.9s and the item died with +// "Timed out." Timeouts here exist to stop a genuinely wedged capture, not to enforce +// a latency SLO. +const GOTO_TIMEOUT_MS = 120_000; +// How long to let the network go quiet AFTER the document is ready. Best-effort: when +// it expires we shoot anyway. See the note on the goto strategy below. +const SETTLE_TIMEOUT_MS = 30_000; // Minimal puppeteer-ish surfaces so the launcher is injectable (real Browser fits). export interface CapturePage { setViewport(vp: { width: number; height: number; deviceScaleFactor: number }): Promise; goto(url: string, opts: { waitUntil: string; timeout: number }): Promise; + /** + * Optional ONLY so hand-written test fakes stay valid. Real captures always take the + * settle path: puppeteer's Page has had this since v5 and we pin ^24. Don't read the + * `?.` at the call site as "this might not run in production" — it always does. + */ + waitForNetworkIdle?(opts: { idleTime: number; timeout: number }): Promise; screenshot(opts: { clip: { x: number; y: number; width: number; height: number } }): Promise; evaluate(fn: (...args: unknown[]) => T): Promise; } @@ -84,7 +105,18 @@ export function createUrlScreenshotAdapter(deps: Deps = {}): CaptureAdapter { const browser = await launchP; const page = await browser.newPage(); await page.setViewport(VIEWPORT); - await page.goto(url, { waitUntil: 'networkidle2', timeout: GOTO_TIMEOUT_MS }); + // Navigate on `domcontentloaded`, then let the network settle SEPARATELY and + // best-effort. `waitUntil: 'networkidle2'` made a quiet network a PRECONDITION + // of the capture: any page holding a connection open — analytics beacon, chat + // widget, video preload, websocket — never satisfies it, and goto rejected at + // 30s with a TimeoutError that cleanErrorReason renders on the card as + // "Timed out." (reproduced against twenty.com and stigg.io under CDP + // throttling: 30755ms, no item). A slow connection does the same to an + // ordinary page. Screenshot quality is worth waiting for; it is not worth + // losing the item over, so a settle that expires just means we shoot a beat + // early. renderPageText (browser.ts) already navigates this way. + await page.goto(url, { waitUntil: 'domcontentloaded', timeout: GOTO_TIMEOUT_MS }); + await page.waitForNetworkIdle?.({ idleTime: 500, timeout: SETTLE_TIMEOUT_MS }).catch(() => {}); await sleep(1000); await dismissOverlays(page); await sleep(400); diff --git a/src/capture/url-snapshot.ts b/src/capture/url-snapshot.ts index 1066abc..9260625 100644 --- a/src/capture/url-snapshot.ts +++ b/src/capture/url-snapshot.ts @@ -23,7 +23,11 @@ import { assertCapturableUrl } from './net-guard.js'; // than spawning) is verified by inspection/manual run (no real Chrome in the suite). const DEFAULT_MAX_BYTES = 8 * 1024 * 1024; // 8MB per-snapshot cap (footprint guardrail) -const SNAPSHOT_TIMEOUT_MS = 45_000; +// SingleFile inlines every asset on the page, so this is strictly slower than the +// screenshot capture it mirrors — 45s under-served a heavy page and, because a failed +// snapshot is swallowed (AC4), it just silently produced no archive. Sized like the +// other capture budgets: a ceiling for a wedged job on a single-tenant box, not an SLO. +const SNAPSHOT_TIMEOUT_MS = 180_000; export interface SnapshotBrowser extends TeardownBrowser { /** puppeteer Browser.wsEndpoint() — the CDP endpoint SingleFile connects to. */ diff --git a/src/config.test.ts b/src/config.test.ts index 35f2d74..bbc46bd 100644 --- a/src/config.test.ts +++ b/src/config.test.ts @@ -171,12 +171,17 @@ describe('snapshotsDir (Story 16.1)', () => { // the original fixed 60s an ordinary Claude-CLI enrichment could exhaust the budget // and land the item in `error: timed out` with a captured page already in hand. describe('capture timeout budget', () => { + // The default must EXCEED the sum of the parts it contains, or it silently truncates + // them: navigation (120s) + network settle (30s) + the CLI agent's own 300s ceiling. it('defaults to a budget that fits capture plus an LLM read', () => { - assert.equal(loadConfig({}).captureTimeoutMs, 180_000); + assert.ok( + loadConfig({}).captureTimeoutMs >= 120_000 + 30_000 + 300_000, + 'default must leave room for navigation + settle + the agent wall-clock', + ); }); it('is overridable for slow local models', () => { - assert.equal(loadConfig({ CAPTURE_TIMEOUT_MS: '600000' }).captureTimeoutMs, 600_000); + assert.equal(loadConfig({ CAPTURE_TIMEOUT_MS: '900000' }).captureTimeoutMs, 900_000); }); it('rejects a non-numeric or zero value', () => { diff --git a/src/config.ts b/src/config.ts index 736c601..8ce4488 100644 --- a/src/config.ts +++ b/src/config.ts @@ -39,8 +39,15 @@ export interface Config { * Budget for ONE capture job, which covers headless capture AND the LLM read * together (they share a job so the item holds a single `processing` state). The * old fixed 60s was tight enough that an ordinary CLI-provider enrichment could - * exhaust it and fail an item whose page had already been captured. Raise it for - * slow local models via CAPTURE_TIMEOUT_MS. + * exhaust it and fail an item whose page had already been captured. Tune via + * CAPTURE_TIMEOUT_MS. + * + * It must comfortably EXCEED the sum of its parts, or it silently truncates them: + * navigation (120s) + network settle (30s) + fixed waits (~1.5s) + the CLI agent's + * own 300s wall-clock. Generous by design — board-oss is single-tenant, so nothing + * is queued behind you competing for the worker, and this ceiling exists to stop a + * genuinely wedged job rather than to enforce a latency SLO. A typical add finishes + * in ~40-50s and never approaches it. */ captureTimeoutMs: number; provider: ProviderConfig; @@ -160,7 +167,7 @@ export function loadConfig(env: NodeJS.ProcessEnv): Config { screenshotsDir: path.join(dataDir, 'screenshots'), snapshotsDir: path.join(dataDir, 'snapshots'), chromePath: clean(env.CHROME_PATH) ?? null, - captureTimeoutMs: parsePositiveMs(env.CAPTURE_TIMEOUT_MS, 180_000), + captureTimeoutMs: parsePositiveMs(env.CAPTURE_TIMEOUT_MS, 600_000), provider, // Enabled when a transport is configured (agent OR base-URL/key). A model name // alone does not enable AI. diff --git a/src/db/queue.ts b/src/db/queue.ts index bd994a8..0c93e2b 100644 --- a/src/db/queue.ts +++ b/src/db/queue.ts @@ -18,34 +18,53 @@ import type { DbHandle } from './index.js'; // routing reads through the queue would serialize everything and kill the browse / // SSE read path. Only writes serialize. // -// This is the same serialized path Story 5.1's job worker reuses to drain -// capture/enrichment jobs at concurrency 1 — keep `enqueueWrite` generic so 5.1 -// layers jobs on top rather than rewriting it. +// TWO LANES, NOT ONE (revises AD6's "one serializer, not two"). +// +// Jobs used to run on this same chain, so a capture held the WRITE lane for its whole +// duration (Chrome launch + LLM round-trip). That made every interactive write wait on +// unrelated network I/O: `addItemSkill`'s one-row INSERT — and so POST /api/collections +// /:cid/items — blocked for the length of the running capture (measured: 42s with one +// job in flight; up to CAPTURE_TIMEOUT_MS), which read to users as the add +// silently hanging. +// +// The two constraints being collapsed were never the same constraint: +// • concurrency 1 for JOBS — Chromium is ~400-520MB resident, so two concurrent +// captures OOM the 512MB-1GB LXC (NFR-1/C1). Bounds MEMORY. +// • single-writer for SQLITE (AD6) — orders logical read-modify-writes. Bounds +// WRITE INTERLEAVING, and every such write is sub-millisecond. +// Separate lanes honor both: jobs still run strictly one at a time, and a write never +// queues behind one. Writes issued from inside a job go through `enqueueWrite` like any +// other — with the lanes split that no longer deadlocks (see `writeItemDirect`). +type Lane = { tail: Promise }; // A promise chain is the serializer: each enqueued op waits for the previous to // settle. Errors are swallowed from the *chain* (so one failure can't wedge the // queue) but propagated to the *caller* via the returned promise. -let tail: Promise = Promise.resolve(); +const writeLane: Lane = { tail: Promise.resolve() }; +const jobLane: Lane = { tail: Promise.resolve() }; + +function serialize(lane: Lane, fn: () => T | Promise): Promise { + const run = lane.tail.then(() => fn()); + lane.tail = run.then( + () => undefined, + () => undefined, + ); + return run; +} /** * Serialize a write operation. Returns a promise resolving with `fn`'s result (or * rejecting with its error). Only one `fn` runs at a time, in enqueue order. */ export function enqueueWrite(fn: () => T | Promise): Promise { - const run = tail.then(() => fn()); - tail = run.then( - () => undefined, - () => undefined, - ); - return run; + return serialize(writeLane, fn); } -// --- Story 5.1: the JOB layer on the SAME single worker --- +// --- Story 5.1: the JOB layer, on its OWN lane --- // -// Capture/enrichment jobs (Epics 6/7) run here, serially (concurrency 1), on the -// SAME `tail` chain as writes — so a job holds the one worker slot for its full -// duration (Chrome launch + LLM round-trip) and never overlaps another job OR a raw -// write. This is also the SQLite single-writer guard (AD6) — one serializer, not two. +// Capture/enrichment jobs (Epics 6/7) run here, serially (concurrency 1) — a job holds +// the job slot for its full duration (Chrome launch + LLM round-trip) and never +// overlaps another job. It does NOT hold the write lane; see the two-lane note above. // // Concurrency 1 is load-bearing: Chromium is ~400-520MB resident, so two concurrent // captures OOM the 512MB-1GB LXC (NFR-1/C1). @@ -93,7 +112,7 @@ export function enqueueJob(job: Job, opts?: { timeoutFn?: TimeoutFn }): Promise< let resolveStatus!: (r: JobResult) => void; const status = new Promise((r) => (resolveStatus = r)); - enqueueWrite(async () => { + serialize(jobLane, async () => { const controller = new AbortController(); let settled = false; @@ -163,10 +182,21 @@ export function writeItem(handle: DbHandle, item: NewItem, itemAssets?: NewAsset /** * The DIRECT item write (transaction + search_blob + FTS + optional asset replace), - * with NO enqueue. MUST be called only from inside a job that already holds the - * worker slot (capture/enrichment work) — calling `writeItem` there would deadlock - * (the inner `enqueueWrite` waits for the outer slot, which awaits the inner). Other - * callers (importer, skills, routes) use `writeItem`, which wraps this in the queue. + * with NO enqueue. Synchronous, so it is atomic on its own; what it does NOT get is + * ordering against other writers. + * + * It exists for callers that must do `read → write` with no await between them — + * `writeItemDirect` takes the FULL desired row, so a merge assembled from a stale + * snapshot would revert whatever landed in between. Those callers wrap BOTH steps in a + * single `enqueueWrite` and use this inside it (see capture/adapter.ts and + * enrichment/worker.ts). That wrapping is what gives them ordering; calling this bare + * skips the write lane. + * + * It used to exist for a different reason — jobs shared the write lane, so a job that + * called the enqueueing `writeItem` self-deadlocked (the inner `enqueueWrite` waited on + * the slot the job itself held). The lanes are separate now, so that deadlock is gone + * and job code can enqueue writes freely. Other callers (importer, skills, routes) use + * `writeItem`, which wraps this in the queue. */ export function writeItemDirect(handle: DbHandle, item: NewItem, itemAssets?: NewAsset[]): void { handle.sqlite.transaction(() => { @@ -259,6 +289,13 @@ export interface RunItemJobArgs { * status is written here, after the job result, so an item is never stuck * `processing`. (A hard crash/OOM where neither runs is swept by * `reconcileInterruptedItems` at boot.) + * + * The status writes below stay DIRECT (not enqueued) deliberately. They are + * single-column updates, not read-modify-writes, so they are already atomic — and each + * is guarded by a synchronous `signal.aborted` check that must not be separated from + * its write. Enqueueing them would put an await between the check and the write, giving + * abandoned work a window to clobber the timeout path's terminal `error` back to `done` + * — the exact thing that guard exists to prevent. */ export async function runItemJob(handle: DbHandle, args: RunItemJobArgs): Promise { const job: Job = { diff --git a/src/db/worker.test.ts b/src/db/worker.test.ts index 7a0891d..6837fa6 100644 --- a/src/db/worker.test.ts +++ b/src/db/worker.test.ts @@ -44,29 +44,51 @@ describe('job worker (Story 5.1)', () => { assert.equal(max, 1); }); - // AC 4/5 — no double serializer: a raw enqueueWrite and a job-write share one worker - it('serializes a raw enqueueWrite against a job (combined active-count never > 1)', async () => { - let active = 0; - let max = 0; - const dw = deferred(); + // Interactive writes must NOT queue behind a job. A capture job holds its lane for + // its whole duration (Chrome + LLM, up to CAPTURE_TIMEOUT_MS = 180s); when writes + // shared that lane, `addItemSkill`'s one-row INSERT — and so POST /api/collections/ + // :cid/items — blocked for the length of the running capture (measured: 42s with a + // single job in flight). Concurrency 1 for JOBS (NFR-1: two Chromiums OOM the LXC) + // and single-writer for SQLITE are two different constraints; they now have two + // lanes. See the AD6 revision in queue.ts. + it('does not queue a raw enqueueWrite behind a running job', async () => { const dj = deferred(); - const pw = enqueueWrite(async () => { - active += 1; - max = Math.max(max, active); - await dw.promise; - active -= 1; - }); const pj = enqueueJob( - { type: 't', timeoutMs: 60_000, run: async () => { active += 1; max = Math.max(max, active); await dj.promise; active -= 1; } }, + { type: 't', timeoutMs: 60_000, run: async () => { await dj.promise; } }, { timeoutFn: neverFires }, ); + await tick(); // the job is now running and holding its lane + + let wrote = false; + const pw = enqueueWrite(() => { wrote = true; }); await tick(); - assert.equal(active, 1, 'a job and a raw write must not overlap'); - dw.resolve(); + assert.equal(wrote, true, 'a write must not wait for a long-running job to finish'); + await pw; dj.resolve(); await pj; - assert.equal(max, 1); + }); + + // NOTE: AC4/5's "combined active-count never > 1" (a raw enqueueWrite and a job may + // not overlap) was DELETED, not weakened. It asserted the single shared lane that is + // the head-of-line-blocking bug above; the two invariants that actually matter are + // kept and asserted separately — jobs stay concurrency 1 (the test above this one, + // which is what NFR-1 needs), and SQLite writes stay serialized among themselves + // (queue.test.ts's lost-update proof). Nothing asserts the two share a lane, because + // they deliberately no longer do. + + // Writes issued from INSIDE a job go through `enqueueWrite` now that the lanes are + // split. Under the old shared lane that self-deadlocked (the inner write waited on + // the slot the job itself held) — which is why the job path had to use the `*Direct` + // variants. Guard the property those call sites now depend on. + it('lets a job await a write it enqueues itself (no self-deadlock)', async () => { + let wrote = false; + const result = await enqueueJob( + { type: 't', timeoutMs: 60_000, run: async () => { await enqueueWrite(() => { wrote = true; }); } }, + { timeoutFn: neverFires }, + ); + assert.equal(result.ok, true, 'a job that awaits its own write must complete'); + assert.equal(wrote, true); }); // AC 2/5 — timeout fires the abort signal, marks failed, and the queue proceeds diff --git a/src/enrichment/worker.test.ts b/src/enrichment/worker.test.ts index 45bf36e..24c852e 100644 --- a/src/enrichment/worker.test.ts +++ b/src/enrichment/worker.test.ts @@ -10,6 +10,7 @@ import { eq } from 'drizzle-orm'; import { initDb } from '../db/index.js'; import { boards, items } from '../db/schema.js'; import { runItemJob, type TimeoutFn } from '../db/queue.js'; +import { patchItemFields } from '../db/item-actions.js'; import { disabledLlm, type LLMProvider } from '../skills/types.js'; import { INSPIRATION_DESCRIPTOR } from '../db/seed.js'; import { buildEnrichmentSchema, buildEnrichmentPrompt, runEnrichmentForItem } from './worker.js'; @@ -112,6 +113,30 @@ describe('runEnrichmentForItem (Story 7.1)', () => { assert.equal(row?.notes, 'USER NOTE', 'user notes column untouched'); }); + // Enrichment reads the item, awaits the LLM (tens of seconds), then writes the row + // back. Jobs and writes now run on separate lanes, so an interactive edit can land + // INSIDE that window — and a write assembled from the pre-LLM snapshot would silently + // revert it. The row is re-read inside the write op so the merge is against fresh + // state. (Under the old single lane this was impossible: the job held the write lane.) + it('does not clobber a user edit that lands while the LLM is running', async () => { + handle.db.insert(items).values({ id: 'e-race', boardId: 'nb', source: 'x', notes: 'before', fields: {} }).run(); + const mock: LLMProvider = { + complete: async () => { + // the user edits notes + a user-owned field mid-flight, through the write lane + await patchItemFields(handle, 'e-race', { notes: 'EDITED MID-FLIGHT', note_field: 'USER VALUE' }); + return { foo_score: 7 } as never; + }, + }; + + await runEnrichmentForItem(handle, { itemId: 'e-race', llm: mock }); + + const row = handle.db.select().from(items).where(eq(items.id, 'e-race')).get(); + const f = row?.fields as Record; + assert.equal(row?.notes, 'EDITED MID-FLIGHT', 'a user edit during the LLM call must survive'); + assert.equal(f.note_field, 'USER VALUE', 'a user field written mid-flight must survive'); + assert.equal(f.foo_score, 7, 'and the enrichment result still lands'); + }); + // Title refinement: the LLM may return a `title`, written to the title COLUMN // (not fields) so a cluttered/wrong captured title gets cleaned up on add/refetch. it('writes an LLM-refined title to the item title column', async () => { diff --git a/src/enrichment/worker.ts b/src/enrichment/worker.ts index 3e75dcb..7f4d418 100644 --- a/src/enrichment/worker.ts +++ b/src/enrichment/worker.ts @@ -3,7 +3,7 @@ import { z, type ZodType } from 'zod'; import { eq } from 'drizzle-orm'; import { boards, items } from '../db/schema.js'; -import { writeItemDirect } from '../db/queue.js'; +import { enqueueWrite, writeItemDirect } from '../db/queue.js'; import type { BoardDescriptor, Field } from '../descriptor/types.js'; import type { LLMProvider } from '../skills/types.js'; import type { DbHandle } from '../db/index.js'; @@ -119,7 +119,21 @@ export async function runEnrichmentForItem( } if (allowedKeys.has(k) && v !== undefined) enriched[k] = v; } - const mergedFields = { ...((item.fields as Record) ?? {}), ...enriched }; const titleUpdate = refinedTitle !== undefined ? { title: refinedTitle } : {}; - writeItemDirect(handle, { ...item, ...titleUpdate, id: item.id, boardId: item.boardId, fields: mergedFields }); + // Re-read the row INSIDE the write op — no await between the read and the write — so + // the merge lands on fresh state. `writeItemDirect` takes the FULL desired row, and + // the LLM round-trip above is long enough (tens of seconds) for an interactive edit + // to arrive on the write lane; merging onto the pre-LLM snapshot would silently + // revert it. Enqueued rather than direct: jobs and writes are separate lanes now, so + // a write from inside a job no longer self-deadlocks. + await enqueueWrite(() => { + const fresh = handle.db.select().from(items).where(eq(items.id, args.itemId)).get() ?? item; + writeItemDirect(handle, { + ...fresh, + ...titleUpdate, + id: item.id, + boardId: item.boardId, + fields: { ...((fresh.fields as Record) ?? {}), ...enriched }, + }); + }); } diff --git a/src/llm/cli-provider.ts b/src/llm/cli-provider.ts index 7607c65..8bbd523 100644 --- a/src/llm/cli-provider.ts +++ b/src/llm/cli-provider.ts @@ -51,7 +51,11 @@ export interface CliProviderConfig { logger?: Logger; } -const DEFAULT_TIMEOUT_MS = 120_000; +// Wall-clock ceiling for one agent subprocess. Generous on purpose: a self-hosted, +// single-tenant box has nobody queued behind this call, so a read that takes four +// minutes and returns beats one killed at two that leaves the item un-enriched. Sized +// to sit inside config.captureTimeoutMs alongside the capture that precedes it. +const DEFAULT_TIMEOUT_MS = 300_000; // Bound captured output so a runaway agent can't exhaust memory on the small LXC // (mirrors the prototype's spawnSync maxBuffer, which the async port must preserve). const MAX_OUTPUT_BYTES = 10 * 1024 * 1024; diff --git a/src/processor-library.test.ts b/src/processor-library.test.ts index 610dfb8..82f5d30 100644 --- a/src/processor-library.test.ts +++ b/src/processor-library.test.ts @@ -95,6 +95,23 @@ test("captureLibrary falls back to headless render when fetch+readability yields assert.ok(cap.text.includes("Anytype is a local-first knowledge base"), "should return rendered text"); }); +// The direct fetch was unbounded, so a server that accepts the connection and then +// never responds pinned the item in `processing` until the whole capture job expired +// (CAPTURE_TIMEOUT_MS, now 600s) — and the job lane is serial, so every queued add sat +// behind it. Longer budgets are only safe for BOUNDED waits; this one needs its own. +test("captureLibrary bounds the direct fetch with an abort signal", async () => { + let sawSignal: AbortSignal | undefined; + const fetchImpl = (async (_url: string, init?: RequestInit) => { + sawSignal = init?.signal ?? undefined; + return { text: async () => ARTICLE_HTML }; + }) as unknown as typeof fetch; + + await captureLibrary("https://example.com/article", { fetchImpl, renderImpl: async () => "" }); + + assert.ok(sawSignal, "the direct fetch must carry an abort signal, or it can hang forever"); + assert.equal(sawSignal.aborted, false, "and must not be aborted before the request starts"); +}); + test("captureLibrary uses fetch result and skips render when readability yields enough text", async () => { let renderCalled = false; const cap = await captureLibrary("https://example.com/article", { diff --git a/src/processor-library.ts b/src/processor-library.ts index 9b2932c..76e7dc6 100644 --- a/src/processor-library.ts +++ b/src/processor-library.ts @@ -7,6 +7,9 @@ import { renderPageText } from "./browser.js"; // Below this many characters of extracted text, fetch+readability is assumed to // have failed (e.g. a JS-rendered SPA shell) and the headless-render fallback runs. const MIN_USEFUL_TEXT = 200; +// Ceiling for the direct HTML fetch. Generous like the other capture budgets (this is a +// single-tenant box), but FINITE — see the note at the call site. +const FETCH_TIMEOUT_MS = 60_000; type LibraryAnalysis = { title: string; @@ -204,7 +207,13 @@ export async function captureLibrary( const fetchFn = opts?.fetchImpl ?? globalThis.fetch; const renderFn = opts?.renderImpl ?? renderPageText; - const response = await fetchFn(url); + // Bound the direct fetch. Without a signal this waits forever on a server that + // accepts the connection and then goes quiet, and the only backstop is the whole + // capture job's budget (CAPTURE_TIMEOUT_MS) — which pins the item in `processing` + // for minutes and, because the job lane is serial, holds every queued add behind it. + // The generous budgets elsewhere are safe precisely because each step is bounded; + // this was the one that wasn't. + const response = await fetchFn(url, { signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) }); const html = await response.text(); const imageUrl = extractOgImage(html, url); // hero image from the static HTML (head meta) let text = extractReadableMarkdown(html, url);