From fcb35a8145383a83e6b31e196e01991bc5085b7f Mon Sep 17 00:00:00 2001 From: Ben Senescu <44480372+bensenescu@users.noreply.github.com> Date: Tue, 25 Aug 2026 23:12:26 -0400 Subject: [PATCH] fix: move the site-audit engine to a dedicated open-seo-audit worker (stops audit OOMs) (#530) --- alchemy.run.ts | 105 ++++++++++-- docs/SELF_HOSTING_CLOUDFLARE.md | 6 +- docs/SELF_HOSTING_CLOUDFLARE_LEGACY.md | 8 +- knip.jsonc | 2 + package.json | 4 +- src/audit-worker.ts | 33 ++++ src/env.d.ts | 10 +- src/server.ts | 7 +- .../features/audit/services/AuditService.ts | 8 +- src/server/gdpr/storage-erasure.ts | 5 +- src/server/lib/audit/lighthouse.ts | 44 +++-- src/server/lib/dataforseo/core.ts | 24 +++ src/server/lib/dataforseo/lighthouse.ts | 97 +++++++---- .../workflows/siteAuditWorkflowPhases.test.ts | 97 ++++++----- .../workflows/siteAuditWorkflowPhases.ts | 157 +++++++++--------- vite-plugin-lean-worker-bundle.ts | 12 +- vite.config.ts | 10 +- worker-configuration.d.ts | 4 +- wrangler.audit.jsonc | 83 +++++++++ wrangler.jsonc | 34 +++- 20 files changed, 538 insertions(+), 212 deletions(-) create mode 100644 src/audit-worker.ts create mode 100644 wrangler.audit.jsonc diff --git a/alchemy.run.ts b/alchemy.run.ts index 3ea11ff..4e16731 100644 --- a/alchemy.run.ts +++ b/alchemy.run.ts @@ -61,6 +61,9 @@ const wrangler = z binding: z.string(), name: z.string(), class_name: z.string(), + // Set when the workflow class lives in the open-seo-audit aux + // worker; alchemy substitutes the stage-suffixed worker name below. + script_name: z.string().optional(), }), ), }) @@ -352,6 +355,68 @@ export default Alchemy.Stack( workersSubdomain, ); + // Created once and bound into BOTH workers — they share the same + // D1/KV/R2 (and prod Hyperdrive). OAUTH_KV stays app-worker-only. + const resources = makeResources(stage); + const prodHyperdrive = prod ? makeHyperdrive() : undefined; + + // Aux worker: the site-audit engine (src/audit-worker.ts) — the + // SiteAuditWorkflow orchestrator and the per-audit AuditScratchpad DO. + // Its memory spikes (multi-MB Lighthouse payloads, in-flight HTML + // batches) OOMed the app worker's near-limit baseline heap. Deployed + // BEFORE the app worker so the app's cross-script workflow/DO bindings + // always have a target. Takes no direct traffic (url off). + const auditWorker = yield* Cloudflare.Worker("open-seo-audit", { + name: `${workerName(stage)}-audit`, + main: "./dist/open_seo_audit/index.js", + bundle: false, + url: false, + compatibility: { + date: wrangler.compatibility_date, + flags: wrangler.compatibility_flags, + }, + // Audit workflow steps parse and persist batches of HTML — the same + // CPU allowance the app worker used to carry for them. Configurable + // CPU limits are a paid-plan feature; self-host deploys + // (cloudflare_access) may run on the free plan, which rejects them. + ...(authMode === "cloudflare_access" + ? {} + : { limits: { cpuMs: 300_000 } }), + observability: { + enabled: wrangler.observability?.enabled ?? true, + traces: { enabled: wrangler.observability?.traces?.enabled ?? false }, + }, + env: { + DB: resources.DB, + KV: resources.KV, + R2: resources.R2, + // Deliberately NOT ...dataEnv: this worker crawls and parses + // attacker-influenced HTML, so it gets only the secrets its code + // path reads — DataForSEO (Lighthouse), Autumn (metering), PostHog + // (capture). No auth/OAuth/Loops/Turnstile secrets. + DATAFORSEO_API_KEY: dataEnv.DATAFORSEO_API_KEY, + AUTUMN_SECRET_KEY: dataEnv.AUTUMN_SECRET_KEY, + POSTHOG_PUBLIC_KEY: dataEnv.POSTHOG_PUBLIC_KEY, + POSTHOG_HOST: dataEnv.POSTHOG_HOST, + AUTH_MODE: authMode, + DATABASE_PROVIDER: databaseProvider || "d1", + ...(prodHyperdrive ? { HYPERDRIVE: prodHyperdrive } : {}), + // This worker is the code home of the scratchpad DO and the + // site-audit workflow; the app worker binds to both cross-script. + AUDIT_SCRATCHPAD: Cloudflare.DurableObject("AUDIT_SCRATCHPAD", { + className: "AuditScratchpad", + }), + // The name must stay in exact sync with wrangler.jsonc's + // "site-audit-workflow" entry (and the app env mapping below): the + // alchemy resource id IS this name, so a drift orphans the live + // registration and deletes it instead of repointing it. + SITE_AUDIT_WORKFLOW: Cloudflare.Workflow( + prod ? "site-audit-workflow" : `site-audit-workflow-${stage}`, + { className: "SiteAuditWorkflow" }, + ), + }, + }).pipe(Alchemy.RemovalPolicy.retain(prod)); + const app = yield* Cloudflare.Worker("open-seo", { name: workerName(stage), // Prod serves the real domains; the zone is inferred from the hostname. @@ -369,12 +434,12 @@ export default Alchemy.Stack( date: wrangler.compatibility_date, flags: wrangler.compatibility_flags, }, - // Site audits parse and persist batches of HTML inside Workflow steps. - // Paid Workers permit up to five minutes; keep headroom for unusually - // link-heavy sites after bounding page bodies and bulk-writing links. - // Configurable CPU limits are a paid-plan feature, and self-host - // deploys (cloudflare_access) may run on the free plan — which rejects - // them — so those get the plan default instead. + // Site audits moved to the open-seo-audit worker, but RankCheckWorkflow + // still parses SERP batches here — keep the CPU allowance until that + // workflow's per-tick CPU is measured or it moves too. Configurable CPU + // limits are a paid-plan feature, and self-host deploys + // (cloudflare_access) may run on the free plan — which rejects them — + // so those get the plan default instead. ...(authMode === "cloudflare_access" ? {} : { limits: { cpuMs: 300_000 } }), @@ -387,7 +452,7 @@ export default Alchemy.Stack( // Scheduled rank checks — src/server.ts `scheduled` handler. crons: wrangler.triggers.crons, env: { - ...makeResources(stage), + ...resources, ...dataEnv, AUTH_MODE: authMode, DATABASE_PROVIDER: databaseProvider || "d1", @@ -396,12 +461,17 @@ export default Alchemy.Stack( POLICY_AUD: access.policyAud, // Prod-only: pooled Postgres via the existing Hyperdrive config. - ...(prod ? { HYPERDRIVE: makeHyperdrive() } : {}), + ...(prodHyperdrive ? { HYPERDRIVE: prodHyperdrive } : {}), - // Durable Objects (chat agents + the per-audit crawl scratchpad). - // Alchemy backs new DO classes with SQLite storage, which all of - // them require; the `migrations` array in wrangler.jsonc only - // applies to the wrangler/workerd surfaces (local dev, Docker). + // Service binding to the audit worker's AuditEngine entrypoint + // (cancel + GDPR erasure of scratchpad state; env key = binding + // name). + AUDIT_ENGINE: auditWorker, + + // Durable Objects (the chat agents; the audit scratchpad lives + // privately in the open-seo-audit worker). Alchemy backs new DO + // classes with SQLite storage; wrangler.jsonc's `migrations` only + // apply to the wrangler/workerd surfaces. ...Object.fromEntries( wrangler.durable_objects.bindings.map((binding) => [ binding.name, @@ -415,13 +485,20 @@ export default Alchemy.Stack( // workers). Workflow names are ACCOUNT-scoped: prod owns the // unsuffixed names; previews carry the stage suffix so concurrent // stages can't repoint each other's workflows (registration is a - // PUT-as-upsert on the name). + // PUT-as-upsert on the name). Entries carrying a script_name bind + // cross-script to the audit worker's workflow instead of + // registering a class of this worker. ...Object.fromEntries( wrangler.workflows.map((workflow) => [ workflow.binding, Cloudflare.Workflow( prod ? workflow.name : `${workflow.name}-${stage}`, - { className: workflow.class_name }, + { + className: workflow.class_name, + scriptName: workflow.script_name + ? auditWorker.workerName + : undefined, + }, ), ]), ), diff --git a/docs/SELF_HOSTING_CLOUDFLARE.md b/docs/SELF_HOSTING_CLOUDFLARE.md index 2f9fd20..ebc4b4d 100644 --- a/docs/SELF_HOSTING_CLOUDFLARE.md +++ b/docs/SELF_HOSTING_CLOUDFLARE.md @@ -56,7 +56,7 @@ cp .env.selfhost.example .env.selfhost pnpm deploy:selfhost --yes ``` -This provisions the D1 database, KV namespaces, and R2 bucket, applies the database migrations, deploys the Worker, and creates the Cloudflare Access application protecting it (allowing exactly `ACCESS_ALLOWED_EMAILS`). If the account has no Zero Trust team yet, one is created for you, named after your workers.dev subdomain. +This provisions the D1 database, KV namespaces, and R2 bucket, applies the database migrations, deploys the Workers, and creates the Cloudflare Access application protecting it (allowing exactly `ACCESS_ALLOWED_EMAILS`). If the account has no Zero Trust team yet, one is created for you, named after your workers.dev subdomain. To manage the Access application yourself instead, set `TEAM_DOMAIN` (`https://your-team.cloudflareaccess.com`) and `POLICY_AUD` (the application's audience tag) in `.env.selfhost` — the deploy then provisions no Access resources. @@ -86,7 +86,7 @@ Everyone allowed through Cloudflare Access works in one shared workspace and see - Login fails: re-check `ACCESS_ALLOWED_EMAILS` in `.env.selfhost` and redeploy. - `https:///api/health` reports runtime configuration checks and database status. -- For server errors, open the Worker `Logs` or run `pnpm exec wrangler tail`. +- For server errors, open the Worker `Logs` or run `pnpm exec wrangler tail`. Site audits run in a separate worker: `pnpm exec wrangler tail open-seo-selfhost-audit`. ## Tearing it down @@ -94,7 +94,7 @@ Everyone allowed through Cloudflare Access works in one shared workspace and see pnpm alchemy destroy --env-file .env.selfhost --stage selfhost ``` -This deletes the Worker, the stage-suffixed D1/KV/R2 resources (including your data), and the Access application. +This deletes the Workers, the stage-suffixed D1/KV/R2 resources (including your data), and the Access application. ## Next steps diff --git a/docs/SELF_HOSTING_CLOUDFLARE_LEGACY.md b/docs/SELF_HOSTING_CLOUDFLARE_LEGACY.md index 36b98b2..ab7074e 100644 --- a/docs/SELF_HOSTING_CLOUDFLARE_LEGACY.md +++ b/docs/SELF_HOSTING_CLOUDFLARE_LEGACY.md @@ -33,6 +33,12 @@ pnpm install pnpm run deploy ``` +`pnpm run deploy` also deploys a second worker, `open-seo-audit`, which runs site audits. Copy your `DB`, `KV`, and `R2` bindings from `wrangler.jsonc` into `wrangler.audit.jsonc` (it needs no `OAUTH_KV`) — the deploy fails on ids that don't exist in your account. Then set its DataForSEO key once, or every Lighthouse check in an audit fails: + +```bash +pnpm exec wrangler secret put DATAFORSEO_API_KEY --name open-seo-audit +``` + ## Giving teammates access 1. Open Cloudflare Zero Trust. @@ -59,7 +65,7 @@ Replace `open-seo` with your bucket name if you changed it. **Login fails or OpenSEO doesn't load.** Re-check, on your Worker under `Settings`: - `Domains & Routes`: `Cloudflare Access` is enabled for the `workers.dev` route. -- `Variables & Secrets`: `TEAM_DOMAIN` (for example `https://your-team.cloudflareaccess.com`), `POLICY_AUD` (the Access application audience tag), and `DATAFORSEO_API_KEY` are set. +- `Variables & Secrets`: `TEAM_DOMAIN` (for example `https://your-team.cloudflareaccess.com`), `POLICY_AUD` (the Access application audience tag), and `DATAFORSEO_API_KEY` are set. The `open-seo-audit` worker needs `DATAFORSEO_API_KEY` too. - Manual Wrangler deployments: the binding IDs in `wrangler.jsonc` match your resources. `https:///api/health` reports runtime configuration checks and database status. For server errors, open the Worker `Logs` or run `pnpm exec wrangler tail`. diff --git a/knip.jsonc b/knip.jsonc index a01b853..ddb7097 100644 --- a/knip.jsonc +++ b/knip.jsonc @@ -5,6 +5,8 @@ "alchemy.preview-access.run.ts", // Detect Tanstack Start Routes "cli-auth.ts", + // Site-audit aux worker entry (wrangler.audit.jsonc) + "src/audit-worker.ts", "src/start.ts", "src/router.tsx", "src/routes/**/*.ts", diff --git a/package.json b/package.json index 30955f1..5457cb0 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,7 @@ "lint": "oxlint . --type-aware", "lint:fix": "oxlint . --type-aware --fix", "preview": "npm run build && vite preview --port 3001", - "deploy": "npm run db:migrate:prod && npm run build && wrangler deploy", + "deploy": "npm run db:migrate:prod && npm run build && wrangler deploy -c dist/open_seo_audit/wrangler.json && wrangler deploy", "deploy:selfhost": "node scripts/selfhost-deploy-preflight.mjs && vite build --mode selfhost && tsc --noEmit && pnpm alchemy deploy --env-file .env.selfhost --stage selfhost", "deploy:postgres": "npm run db:migrate:pg && npm run build && pnpm alchemy deploy --env-file .env.production --stage hosted-prod --adopt", "deploy:preview": "vite build --mode preview && pnpm alchemy deploy --env-file .env.preview", @@ -22,7 +22,7 @@ "alchemy": "NODE_OPTIONS=\"$NODE_OPTIONS --experimental-strip-types\" alchemy", "destroy:preview": "pnpm alchemy destroy --env-file .env.preview", "sourcemaps:upload": "POSTHOG_SOURCEMAPS=true NODE_OPTIONS=--max-old-space-size=8192 npm run build && pnpm dlx @posthog/cli sourcemap inject --directory ./dist-sourcemaps && pnpm dlx @posthog/cli sourcemap upload --directory ./dist-sourcemaps", - "cf-typegen": "wrangler types", + "cf-typegen": "wrangler types --config wrangler.jsonc --config wrangler.audit.jsonc", "types:check": "tsc --noEmit", "format:check": "prettier --check .", "format:write": "prettier . --write", diff --git a/src/audit-worker.ts b/src/audit-worker.ts new file mode 100644 index 0000000..23e54c8 --- /dev/null +++ b/src/audit-worker.ts @@ -0,0 +1,33 @@ +// Auxiliary worker "open-seo-audit": hosts the entire site-audit engine — the +// SiteAuditWorkflow orchestrator (crawl, Lighthouse, finalize) and the +// per-audit AuditScratchpad DO — so its memory spikes (multi-MB Lighthouse +// payloads, in-flight HTML batches) land on a small-baseline isolate instead +// of the app worker's near-limit one. The app worker starts audits via the +// cross-script SITE_AUDIT_WORKFLOW binding and reads results from the shared +// DB/KV. +// +// Keep this entry's eager graph lean: autumn-js and the page analyzer must +// stay behind their existing lazy boundaries +// (vite-plugin-lean-worker-bundle.ts asserts this at build time). +import { WorkerEntrypoint } from "cloudflare:workers"; +import { getAuditScratchpad } from "@/server/features/audit/AuditScratchpad"; + +export { SiteAuditWorkflow } from "./server/workflows/SiteAuditWorkflow"; +export { AuditScratchpad } from "./server/features/audit/AuditScratchpad"; + +// The scratchpad DO is private to this worker: the Cloudflare API refuses an +// upload that deletes a class while any binding still references its name, so +// the app worker could not keep a cross-script binding through the cutover. +// The app's two control needs (audit delete/cancel, GDPR erasure) go through +// this entrypoint instead. Nothing routes to fetch — the app worker owns all +// traffic. +export default class AuditEngine extends WorkerEntrypoint { + override fetch(): Response { + return new Response("Not found", { status: 404 }); + } + + /** Wipe an audit's crawl scratch state (storage + alarm). Idempotent. */ + async destroyScratchpad(auditId: string): Promise { + await getAuditScratchpad(auditId).destroy(); + } +} diff --git a/src/env.d.ts b/src/env.d.ts index c0199f1..b052b65 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -13,9 +13,17 @@ declare namespace Cloudflare { SAM_CHAT: DurableObjectNamespace; // Durable Object holding per-audit crawl scratch state (frontier, link - // edges, page mirror). Untyped here; getAuditScratchpad narrows the stub. + // edges, page mirror). Bound ONLY in the open-seo-audit aux worker; + // untyped here — getAuditScratchpad narrows the stub. AUDIT_SCRATCHPAD: DurableObjectNamespace; + // Service binding to the audit worker's AuditEngine entrypoint (cancel + + // GDPR erasure of scratchpad state). The inline import() is required: a + // top-level import would turn this ambient file into a module and break + // the global augmentation. + // oxlint-disable-next-line typescript-eslint/consistent-type-imports + AUDIT_ENGINE: Service; + AUTH_MODE?: "cloudflare_access" | "local_noauth" | "hosted"; BYPASS_EMAIL_VERIFICATION?: string; TEAM_DOMAIN?: string; diff --git a/src/server.ts b/src/server.ts index 00c828d..a832e4d 100644 --- a/src/server.ts +++ b/src/server.ts @@ -178,15 +178,14 @@ function handleFetch( return appFetch(request); } -// Export Workflow classes as named exports -export { SiteAuditWorkflow } from "./server/workflows/SiteAuditWorkflow"; +// Export Workflow classes as named exports. SiteAuditWorkflow and the +// AuditScratchpad DO live in the open-seo-audit aux worker +// (src/audit-worker.ts); this worker reaches them via cross-script bindings. export { RankCheckWorkflow } from "./server/workflows/RankCheckWorkflow"; // Durable Object class for the onboarding strategy chat (Agents SDK). export { OnboardingChatAgent } from "./server/features/onboarding/OnboardingChatAgent"; // Durable Object class for the SAM in-app agent (Agents SDK). export { SamChatAgent } from "./server/features/sam/SamChatAgent"; -// Durable Object class for the per-audit crawl scratchpad. -export { AuditScratchpad } from "./server/features/audit/AuditScratchpad"; // Daily OAuth KV garbage collection; must match a trigger in wrangler.jsonc. const MCP_OAUTH_PURGE_CRON = "17 3 * * *"; diff --git a/src/server/features/audit/services/AuditService.ts b/src/server/features/audit/services/AuditService.ts index a66b3d6..e880a20 100644 --- a/src/server/features/audit/services/AuditService.ts +++ b/src/server/features/audit/services/AuditService.ts @@ -5,7 +5,6 @@ import { type BillingCustomerContext, } from "@/server/billing/subscription"; import { AuditRepository } from "@/server/features/audit/repositories/AuditRepository"; -import { getAuditScratchpad } from "@/server/features/audit/AuditScratchpad"; import { AUDIT_LIMITS, clampAuditMaxPages, @@ -259,10 +258,11 @@ async function remove(auditId: string, projectId: string) { } await AuditRepository.deleteAuditForProject(auditId, projectId); - // Best-effort: drop the crawl scratchpad DO with the audit. A missed - // destroy self-cleans via the DO's 7-day alarm. + // Best-effort: drop the crawl scratchpad DO with the audit (it lives in + // the open-seo-audit worker, behind the AuditEngine RPC). A missed destroy + // self-cleans via the DO's 7-day alarm. try { - await getAuditScratchpad(auditId).destroy(); + await env.AUDIT_ENGINE.destroyScratchpad(auditId); } catch (error) { console.warn(`Failed to destroy audit scratchpad ${auditId}:`, error); } diff --git a/src/server/gdpr/storage-erasure.ts b/src/server/gdpr/storage-erasure.ts index e9005fd..336cb9d 100644 --- a/src/server/gdpr/storage-erasure.ts +++ b/src/server/gdpr/storage-erasure.ts @@ -1,5 +1,4 @@ import { getAuth } from "@/lib/auth"; -import { getAuditScratchpad } from "@/server/features/audit/AuditScratchpad"; import type { OnboardingChatAgent } from "@/server/features/onboarding/OnboardingChatAgent"; import type { SamChatAgent } from "@/server/features/sam/SamChatAgent"; import { captureServerError } from "@/server/lib/posthog"; @@ -207,7 +206,9 @@ async function eraseStorage(env: Env, payload: GdprStorageErasurePayload) { .destroyForErasure(); } for (const auditId of payload.auditIds) { - await getAuditScratchpad(auditId).destroyForErasure(); + // The scratchpad DO lives in the open-seo-audit worker; destroy is the + // same full wipe destroyForErasure performs. + await env.AUDIT_ENGINE.destroyScratchpad(auditId); await env.KV.delete(`audit-progress:${auditId}`); } // The autumn:customer-ensured KV markers are deliberately left alone: they diff --git a/src/server/lib/audit/lighthouse.ts b/src/server/lib/audit/lighthouse.ts index 264b55e..834e920 100644 --- a/src/server/lib/audit/lighthouse.ts +++ b/src/server/lib/audit/lighthouse.ts @@ -22,6 +22,32 @@ type LighthouseFetchResult = { payloadJson: string | null; }; +/** A check that produced no payload — provider error, or a failed fetch step. */ +export function failedLighthouseFetch( + url: string, + pageId: string, + strategy: "mobile" | "desktop", + errorMessage: string, +): LighthouseFetchResult { + return { + result: { + url, + pageId, + strategy, + performanceScore: null, + accessibilityScore: null, + bestPracticesScore: null, + seoScore: null, + lcpMs: null, + cls: null, + inpMs: null, + ttfbMs: null, + errorMessage, + }, + payloadJson: null, + }; +} + export async function fetchLighthouseResult( url: string, pageId: string, @@ -51,23 +77,7 @@ export async function fetchLighthouseResult( } catch (error) { const failed = error instanceof Error ? error : new Error(String(error)); console.error(`Lighthouse failed for ${url}:`, failed.message); - return { - result: { - url, - pageId, - strategy, - performanceScore: null, - accessibilityScore: null, - bestPracticesScore: null, - seoScore: null, - lcpMs: null, - cls: null, - inpMs: null, - ttfbMs: null, - errorMessage: failed.message, - }, - payloadJson: null, - }; + return failedLighthouseFetch(url, pageId, strategy, failed.message); } } diff --git a/src/server/lib/dataforseo/core.ts b/src/server/lib/dataforseo/core.ts index 6aa38c9..63061d9 100644 --- a/src/server/lib/dataforseo/core.ts +++ b/src/server/lib/dataforseo/core.ts @@ -167,6 +167,30 @@ export function dataforseoPost< return requestDataforseo("POST", path, tasks, options); } +/** + * POST like {@link dataforseoPost} but return the UN-CONSUMED Response + * (headers only) — the same auth, retry policy, and HTTP-error ladder apply. + * For the one endpoint whose body is too big to read eagerly: Lighthouse + * gates its multi-MB body reads behind a parse lock in the audit worker, and + * passes its own `signal` so the timeout can be cleared once headers arrive. + */ +export function dataforseoPostResponse( + path: string, + tasks: unknown[], + options: DataforseoRequestOptions & { signal?: AbortSignal } = {}, +): Promise { + const doFetch = createAuthenticatedFetch( + options.classify, + options.maxServerErrorRetries, + ); + return doFetch(`${API_BASE}${path}`, { + method: "POST", + headers: { Accept: "application/json", "Content-Type": "application/json" }, + body: JSON.stringify(tasks), + signal: options.signal, + }); +} + /** GET a DataForSEO endpoint (task_get collection, appendix/locations data). */ export function dataforseoGet< TTask extends DataforseoTaskLike = DataforseoTaskLike, diff --git a/src/server/lib/dataforseo/lighthouse.ts b/src/server/lib/dataforseo/lighthouse.ts index 9a92521..7653f4a 100644 --- a/src/server/lib/dataforseo/lighthouse.ts +++ b/src/server/lib/dataforseo/lighthouse.ts @@ -1,45 +1,80 @@ +import { dataforseoPostResponse } from "@/server/lib/dataforseo/core"; +import { + assertOk, + buildTaskBilling, + DataforseoChargedTaskError, + type DataforseoApiResponse, + type DataforseoResponseLike, + type DataforseoTaskLike, +} from "@/server/lib/dataforseo/envelope"; import { parseDataforseoLighthousePayload, requestCategories, type LighthouseStrategy, } from "@/server/lib/dataforseoLighthousePayload"; import type { StoredLighthousePayload } from "@/server/lib/lighthouseStoredPayload"; -import { dataforseoPost } from "@/server/lib/dataforseo/core"; -import { - assertOk, - buildTaskBilling, - DataforseoChargedTaskError, - type DataforseoApiResponse, -} from "@/server/lib/dataforseo/envelope"; + +const LIGHTHOUSE_PATH = "/v3/on_page/lighthouse/live/json"; +const REQUEST_TIMEOUT_MS = 60_000; + +// One payload read+parse at a time per isolate. This module runs in the +// open-seo-audit worker, and the raw Lighthouse payload (1-10MB, held several +// times over while parsing) is the operation that OOMed the main worker; +// concurrent checks bursting onto one isolate could do the same here. The +// DataForSEO fetches themselves stay concurrent — parsing (well under a +// second each) is cheap against a 30-60s fetch, and workerd streams un-read +// response bodies, so queued siblings don't buffer. +let parseChain: Promise = Promise.resolve(); +function withParseLock(fn: () => Promise): Promise { + const run = parseChain.then(fn, fn); + parseChain = run.catch(() => {}); + return run; +} export async function fetchLighthouseResult(input: { url: string; strategy: LighthouseStrategy; }): Promise> { - const response = await dataforseoPost( - "/v3/on_page/lighthouse/live/json", - [ - { - url: input.url, - for_mobile: input.strategy === "mobile", - categories: [...requestCategories], - }, - ], - // Billed, non-idempotent POST: a 5xx does not prove the provider skipped - // the charge, so never replay it. - { maxServerErrorRetries: 0 }, - ); - - // Build the metering envelope before parsing. The provider has already - // charged a successful task, so a malformed payload must carry its billing - // metadata out to the metered client instead of looking retryable. - const task = assertOk(response); - const billing = buildTaskBilling(task); + // Billed, non-idempotent POST: a 5xx does not prove the provider skipped + // the charge, so never replay it. The response is taken un-consumed (unlike + // dataforseoPost) so the multi-MB body read happens inside the parse lock, + // and the timeout is cleared once headers arrive — an armed signal would + // otherwise cover a body read queued behind the lock past 60s and abort an + // already-billed call unmetered. + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS); + let response: Response; try { - const data = parseDataforseoLighthousePayload(response, input); - return { data, billing }; - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - throw new DataforseoChargedTaskError(message, billing); + response = await dataforseoPostResponse( + LIGHTHOUSE_PATH, + [ + { + url: input.url, + for_mobile: input.strategy === "mobile", + categories: [...requestCategories], + }, + ], + { maxServerErrorRetries: 0, signal: controller.signal }, + ); + } finally { + clearTimeout(timeout); } + + return withParseLock(async () => { + const body = + await response.json>(); + // Build the metering envelope before parsing. The provider has already + // charged a successful task, so a malformed payload must carry its + // billing metadata out to the metered client instead of looking + // retryable. + const task = assertOk(body); + const billing = buildTaskBilling(task); + try { + const data = parseDataforseoLighthousePayload(body, input); + return { data, billing }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new DataforseoChargedTaskError(message, billing); + } + }); } diff --git a/src/server/workflows/siteAuditWorkflowPhases.test.ts b/src/server/workflows/siteAuditWorkflowPhases.test.ts index abe3dc4..48209b3 100644 --- a/src/server/workflows/siteAuditWorkflowPhases.test.ts +++ b/src/server/workflows/siteAuditWorkflowPhases.test.ts @@ -1,4 +1,5 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { LighthouseResult } from "@/server/lib/audit/types"; const { fetchLighthouseResultMock, @@ -18,11 +19,19 @@ const { updateAuditProgressMock: vi.fn(), })); -vi.mock("@/server/lib/audit/lighthouse", () => ({ - fetchLighthouseResult: fetchLighthouseResultMock, - selectLighthouseSample: selectLighthouseSampleMock, - storeLighthouseResult: storeLighthouseResultMock, -})); +// The real module reaches cloudflare:workers through r2.ts at import time. +vi.mock("cloudflare:workers", () => ({ env: {} })); +// failedLighthouseFetch stays real — the failure-path test asserts on the +// rows it builds. +vi.mock("@/server/lib/audit/lighthouse", async (importOriginal) => { + const actual = await importOriginal>(); + return { + ...actual, + fetchLighthouseResult: fetchLighthouseResultMock, + selectLighthouseSample: selectLighthouseSampleMock, + storeLighthouseResult: storeLighthouseResultMock, + }; +}); vi.mock("@/server/features/audit/repositories/AuditRepository", () => ({ AuditRepository: { getPagesForAudit: getPagesForAuditMock, @@ -49,6 +58,19 @@ vi.mock("@/server/workflows/pgStep", () => ({ pgStep: pgStepMock })); import { runLighthousePhase } from "@/server/workflows/siteAuditWorkflowPhases"; +const PHASE_PARAMS = { + auditId: "audit-1", + workflowInstanceId: "workflow-1", + billingCustomer: { + userId: "user-1", + userEmail: "test@example.com", + organizationId: "org-1", + }, + projectId: "project-1", + startUrl: "https://example.com/", + config: { maxPages: 50, lighthouseStrategy: "auto" as const }, +}; + describe("runLighthousePhase", () => { beforeEach(() => { vi.clearAllMocks(); @@ -87,7 +109,7 @@ describe("runLighthousePhase", () => { if (name === "lighthouse-fetch-1") { fetchRetryLimit = config.retries?.limit; } - if (name === "lighthouse-persist-1") { + if (name === "lighthouse-persist-chunk-1") { persistenceRetryLimit = config.retries?.limit; persistenceAttempts += 1; try { @@ -109,18 +131,7 @@ describe("runLighthousePhase", () => { // pgStep is mocked above, so the opaque WorkflowStep object is never read. // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion - await runLighthousePhase({} as never, { - auditId: "audit-1", - workflowInstanceId: "workflow-1", - billingCustomer: { - userId: "user-1", - userEmail: "test@example.com", - organizationId: "org-1", - }, - projectId: "project-1", - startUrl: "https://example.com/", - config: { maxPages: 50, lighthouseStrategy: "auto" }, - }); + await runLighthousePhase({} as never, PHASE_PARAMS); expect(fetchLighthouseResultMock).toHaveBeenCalledTimes(2); expect(storeLighthouseResultMock).toHaveBeenCalledTimes(4); @@ -135,7 +146,15 @@ describe("runLighthousePhase", () => { ); }); - it("does not replay paid calls for a cached legacy batch", async () => { + it("persists sibling results when one fetch step fails", async () => { + getPagesForAuditMock.mockResolvedValue([ + { id: "page-1", url: "https://example.com/", statusCode: 200 }, + { id: "page-2", url: "https://example.com/about", statusCode: 200 }, + ]); + selectLighthouseSampleMock.mockReturnValue([ + "https://example.com/", + "https://example.com/about", + ]); pgStepMock.mockImplementation( async ( _step: unknown, @@ -143,8 +162,8 @@ describe("runLighthousePhase", () => { _config: unknown, callback: () => Promise, ) => { - if (name === "lighthouse-batch-1") { - return { completed: 2, failed: 0 }; + if (name === "lighthouse-fetch-2") { + throw new Error("step timed out"); } return callback(); }, @@ -152,25 +171,25 @@ describe("runLighthousePhase", () => { // pgStep is mocked above, so the opaque WorkflowStep object is never read. // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion - await runLighthousePhase({} as never, { - auditId: "audit-1", - workflowInstanceId: "workflow-1", - billingCustomer: { - userId: "user-1", - userEmail: "test@example.com", - organizationId: "org-1", - }, - projectId: "project-1", - startUrl: "https://example.com/", - config: { maxPages: 50, lighthouseStrategy: "auto" }, - }); + await runLighthousePhase({} as never, PHASE_PARAMS); - expect(pgStepMock).toHaveBeenCalledWith( - {}, - "lighthouse-batch-1", - expect.anything(), - expect.any(Function), + // Only the surviving URL's pair was fetched; the failed step's checks + // still land as errorMessage rows in the same insert. + expect(fetchLighthouseResultMock).toHaveBeenCalledTimes(2); + expect(insertLighthouseResultsMock).toHaveBeenCalledTimes(1); + // vi.fn() mock calls are untyped; the mocked storeLighthouseResult above + // passes fetch results through as rows. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + const inserted = insertLighthouseResultsMock.mock + .calls[0]?.[1] as LighthouseResult[]; + expect(inserted).toHaveLength(4); + expect( + inserted.filter((row) => row.errorMessage === "step timed out"), + ).toHaveLength(2); + expect(updateAuditProgressMock).toHaveBeenLastCalledWith( + "audit-1", + "workflow-1", + { lighthouseCompleted: 2, lighthouseFailed: 2 }, ); - expect(fetchLighthouseResultMock).not.toHaveBeenCalled(); }); }); diff --git a/src/server/workflows/siteAuditWorkflowPhases.ts b/src/server/workflows/siteAuditWorkflowPhases.ts index f0fd0a7..a9a46a3 100644 --- a/src/server/workflows/siteAuditWorkflowPhases.ts +++ b/src/server/workflows/siteAuditWorkflowPhases.ts @@ -2,6 +2,7 @@ import type { WorkflowStep } from "cloudflare:workers"; import type { BillingCustomerContext } from "@/server/billing/subscription"; import { discoverUrls, parseRobotsTxt } from "@/server/lib/audit/discovery"; import { + failedLighthouseFetch, fetchLighthouseResult, selectLighthouseSample, storeLighthouseResult, @@ -32,14 +33,15 @@ import { MULTIPAGE_CHECKS_STEP, } from "@/server/workflows/auditStepConfigs"; -const LEGACY_LIGHTHOUSE_URL_BATCH_SIZE = 10; +/** + * URLs fetched concurrently per wave. Each URL runs mobile + desktop, so one + * wave holds up to 10 paid DataForSEO calls in flight; the aux worker's parse + * lock serializes the memory-heavy payload parsing behind them. + */ +const LIGHTHOUSE_URL_CONCURRENCY = 5; /** Frontier seeds per scratchpad RPC call. */ const SEED_RPC_BATCH = 2_000; -type LighthouseBatchBoundary = - | { schema: "retry-safe-v2" } - | { completed: number; failed: number }; - type AuditPhasesParams = { auditId: string; workflowInstanceId: string; @@ -197,87 +199,80 @@ export async function runLighthousePhase( let completedChecks = 0; let failedChecks = 0; for ( - let batchStart = 0; - batchStart < lighthouseWork.length; - batchStart += LEGACY_LIGHTHOUSE_URL_BATCH_SIZE + let chunkStart = 0; + chunkStart < lighthouseWork.length; + chunkStart += LIGHTHOUSE_URL_CONCURRENCY ) { - const batchIndex = Math.floor( - batchStart / LEGACY_LIGHTHOUSE_URL_BATCH_SIZE, + const chunk = lighthouseWork.slice( + chunkStart, + chunkStart + LIGHTHOUSE_URL_CONCURRENCY, ); - const boundary = await pgStep( + + // The paid calls are checkpointed separately from all storage. With + // Workflow retries disabled, a later R2/DB/progress failure cannot replay + // DataForSEO. One URL groups its mobile + desktop checks into one compact + // checkpoint. allSettled, not all: a rejected step must not orphan the + // sibling paid calls mid-flight — their checkpoints complete and persist + // below either way. + const settled = await Promise.allSettled( + chunk.map(({ url, pageId }, chunkOffset) => + pgStep( + step, + `lighthouse-fetch-${chunkStart + chunkOffset + 1}`, + LIGHTHOUSE_FETCH_STEP, + () => + Promise.all([ + fetchLighthouseResult(url, pageId, "mobile", billingCustomer), + fetchLighthouseResult(url, pageId, "desktop", billingCustomer), + ]), + ), + ), + ); + const fetched = settled.flatMap((outcome, chunkOffset) => { + if (outcome.status === "fulfilled") return outcome.value; + // Step timeout or engine failure — provider errors never reject here + // (the audit-layer fetch converts them into errorMessage results). + const { url, pageId } = chunk[chunkOffset]; + const message = + outcome.reason instanceof Error + ? outcome.reason.message + : String(outcome.reason); + return (["mobile", "desktop"] as const).map((strategy) => + failedLighthouseFetch(url, pageId, strategy, message), + ); + }); + + const chunkIndex = Math.floor(chunkStart / LIGHTHOUSE_URL_CONCURRENCY) + 1; + const priorCompleted = completedChecks; + const priorFailed = failedChecks; + const counts = await pgStep( step, - `lighthouse-batch-${batchIndex + 1}`, - DB_STEP, - async (): Promise => ({ - schema: "retry-safe-v2", - }), + `lighthouse-persist-chunk-${chunkIndex}`, + LIGHTHOUSE_PERSIST_STEP, + async () => { + const results = await Promise.all( + fetched.map((result) => + storeLighthouseResult({ + projectId, + auditId, + fetched: result, + }), + ), + ); + await AuditRepository.insertLighthouseResults(auditId, results); + + const failed = results.filter((result) => result.errorMessage).length; + const completed = results.length - failed; + await AuditRepository.updateAuditProgress(auditId, workflowInstanceId, { + lighthouseCompleted: priorCompleted + completed, + lighthouseFailed: priorFailed + failed, + }); + return { completed, failed }; + }, ); - // Older deployments used this checkpoint name for a complete paid batch. - // If that cached shape replays under current code, all results and progress - // are already persisted; skip the batch instead of buying it again. - if ("completed" in boundary) { - completedChecks += boundary.completed; - failedChecks += boundary.failed; - continue; - } - - const batch = lighthouseWork.slice( - batchStart, - batchStart + LEGACY_LIGHTHOUSE_URL_BATCH_SIZE, - ); - for (const [batchOffset, { url, pageId }] of batch.entries()) { - const index = batchStart + batchOffset; - // The paid calls are checkpointed separately from all storage. With - // Workflow retries disabled, a later R2/DB/progress failure cannot replay - // DataForSEO. One URL groups its mobile + desktop checks into one compact - // checkpoint rather than returning a whole Lighthouse batch. - const fetched = await pgStep( - step, - `lighthouse-fetch-${index + 1}`, - LIGHTHOUSE_FETCH_STEP, - () => - Promise.all([ - fetchLighthouseResult(url, pageId, "mobile", billingCustomer), - fetchLighthouseResult(url, pageId, "desktop", billingCustomer), - ]), - ); - - const priorCompleted = completedChecks; - const priorFailed = failedChecks; - const counts = await pgStep( - step, - `lighthouse-persist-${index + 1}`, - LIGHTHOUSE_PERSIST_STEP, - async () => { - const results = await Promise.all( - fetched.map((result) => - storeLighthouseResult({ - projectId, - auditId, - fetched: result, - }), - ), - ); - await AuditRepository.insertLighthouseResults(auditId, results); - - const failed = results.filter((result) => result.errorMessage).length; - const completed = results.length - failed; - await AuditRepository.updateAuditProgress( - auditId, - workflowInstanceId, - { - lighthouseCompleted: priorCompleted + completed, - lighthouseFailed: priorFailed + failed, - }, - ); - return { completed, failed }; - }, - ); - - completedChecks += counts.completed; - failedChecks += counts.failed; - } + completedChecks += counts.completed; + failedChecks += counts.failed; } } diff --git a/vite-plugin-lean-worker-bundle.ts b/vite-plugin-lean-worker-bundle.ts index 3b965aa..a3e3929 100644 --- a/vite-plugin-lean-worker-bundle.ts +++ b/vite-plugin-lean-worker-bundle.ts @@ -10,8 +10,8 @@ const WORKERS_AI_PROVIDER_STUB = fileURLToPath( new URL("./src/server/lib/workers-ai-provider-stub.ts", import.meta.url), ); /** - * Dependencies that must never be reachable from the worker's eager startup - * module graph. The 128 MB isolate limit is shared by everything evaluated at + * Dependencies that must never be reachable from the workers' eager startup + * module graphs. The 128 MB isolate limit is shared by everything evaluated at * startup (production OOM bursts trace back to baseline heap, not leaks), so * each of these is either loaded lazily behind a dynamic import or stubbed * out. `generateBundle` below fails the build if one sneaks back in via a @@ -107,9 +107,13 @@ export function leanWorkerBundle(): Plugin { } }, generateBundle(_options, bundle) { - // Only the worker build matters for isolate memory; the client bundle + // Only the worker builds matter for isolate memory; the client bundle // never contains these packages (and the zod swap applies everywhere). - if (this.environment.name !== "ssr") return; + // "ssr" is the main worker; "open_seo_audit" is the site-audit aux + // worker, which must stay lean for the same reason it exists. + if (!["ssr", "open_seo_audit"].includes(this.environment.name)) { + return; + } // Bundle keys are chunk fileNames already. const chunkAt = (fileName: string) => { diff --git a/vite.config.ts b/vite.config.ts index c5221eb..5bb4cc1 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -52,7 +52,15 @@ export default defineConfig(({ mode }) => { }, }) : null, - cloudflare({ inspectorPort: false, viteEnvironment: { name: "ssr" } }), + cloudflare({ + inspectorPort: false, + viteEnvironment: { name: "ssr" }, + // The site-audit aux worker builds to dist/open_seo_audit/ and runs + // beside the main worker in dev and preview, with the app's + // cross-script SITE_AUDIT_WORKFLOW / AUDIT_SCRATCHPAD bindings + // resolved against it. + auxiliaryWorkers: [{ configPath: "./wrangler.audit.jsonc" }], + }), tsConfigPaths(), tanstackStart(), viteReact(), diff --git a/worker-configuration.d.ts b/worker-configuration.d.ts index 146e210..9ce19fd 100644 --- a/worker-configuration.d.ts +++ b/worker-configuration.d.ts @@ -20,7 +20,9 @@ declare namespace Cloudflare { LOOPS_TRANSACTIONAL_RESET_PASSWORD_ID: string; POSTHOG_HOST: string; POSTHOG_PUBLIC_KEY: string; - SITE_AUDIT_WORKFLOW: Workflow[0]['payload']>; + // Hand-patched: the class moved to src/audit-worker.ts and a full regen pulls + // unrelated runtime-type drift (see cf-typegen); keep this in sync until then. + SITE_AUDIT_WORKFLOW: Workflow[0]['payload']>; RANK_CHECK_WORKFLOW: Workflow[0]['payload']>; } } diff --git a/wrangler.audit.jsonc b/wrangler.audit.jsonc new file mode 100644 index 0000000..e769db0 --- /dev/null +++ b/wrangler.audit.jsonc @@ -0,0 +1,83 @@ +{ + "$schema": "node_modules/wrangler/config-schema.json", + // Auxiliary worker: the site-audit engine (src/audit-worker.ts) — the + // SiteAuditWorkflow orchestrator plus the per-audit AuditScratchpad DO. + // Must live at the repo root: wrangler resolves .env.local / .dev.vars + // relative to this file's directory, so a nested config would never see + // local secrets. + "name": "open-seo-audit", + "main": "src/audit-worker.ts", + // Only cross-script bindings from the app worker reach this worker. + "workers_dev": false, + // Match the main worker's runtime exactly. + "compatibility_date": "2025-09-02", + "compatibility_flags": ["nodejs_compat", "global_fetch_strictly_public"], + "observability": { + "enabled": true, + "traces": { + "enabled": true, + }, + }, + // This worker is the code home of the site-audit workflow; the app worker + // binds to it cross-script via script_name (see wrangler.jsonc). + "workflows": [ + { + "name": "site-audit-workflow", + "binding": "SITE_AUDIT_WORKFLOW", + "class_name": "SiteAuditWorkflow", + }, + ], + // Per-audit crawl scratchpad: frontier, link edges, and a slim page mirror + // in the DO's SQLite. Destroyed at finalize; self-cleans via alarm if the + // audit dies. The app worker binds cross-script for cancel + GDPR erasure. + "durable_objects": { + "bindings": [ + { + "name": "AUDIT_SCRATCHPAD", + "class_name": "AuditScratchpad", + }, + ], + }, + "migrations": [ + { + "tag": "v1", + "new_sqlite_classes": ["AuditScratchpad"], + }, + ], + // This config serves local dev and Docker self-host only; Alchemy + // deployments provision the audit worker from alchemy.run.ts and never read + // these ids. Mirror any DB/KV/R2 resource-id edits you make in + // wrangler.jsonc into this file too — both workers share the same D1/KV/R2 + // (no OAUTH_KV here; only the app worker runs OAuth), and miniflare derives + // local storage filenames from these ids. + "kv_namespaces": [ + { + "binding": "KV", + "id": "4abd52f3f2c549ac83cc2cb4ceec8620", + }, + ], + "d1_databases": [ + { + "binding": "DB", + "database_name": "open-seo", + "database_id": "37bee90a-e1aa-404f-b01e-b0d1d479bda1", + "migrations_dir": "drizzle", + }, + ], + // Postgres scale path — same rules as wrangler.jsonc: only local Postgres + // dev reads this block. Uncomment together with the block in wrangler.jsonc + // per docs/LOCAL_POSTGRES.md. + // "hyperdrive": [ + // { + // "binding": "HYPERDRIVE", + // "id": "9d64ccfb559f44449ce52a143912f898", + // "localConnectionString": "postgres://openseo:openseo@localhost:5433/openseo", + // }, + // ], + "r2_buckets": [ + { + "bucket_name": "open-seo", + "binding": "R2", + }, + ], +} diff --git a/wrangler.jsonc b/wrangler.jsonc index 6897419..50a5349 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -20,10 +20,16 @@ "mode": "smart", }, "workflows": [ + // The site-audit engine lives in the open-seo-audit aux worker + // (wrangler.audit.jsonc) — its memory spikes (multi-MB Lighthouse + // payloads, in-flight HTML) OOMed this worker's near-limit baseline + // heap. This is a cross-script binding for starting audits and reading + // instance status. { "name": "site-audit-workflow", "binding": "SITE_AUDIT_WORKFLOW", "class_name": "SiteAuditWorkflow", + "script_name": "open-seo-audit", }, { "name": "rank-check-workflow", @@ -48,15 +54,19 @@ "name": "SAM_CHAT", "class_name": "SamChatAgent", }, - // Per-audit crawl scratchpad: frontier, link edges, and a slim page - // mirror in the DO's SQLite. Destroyed at finalize; self-cleans via - // alarm if the audit dies. - { - "name": "AUDIT_SCRATCHPAD", - "class_name": "AuditScratchpad", - }, ], }, + // The site-audit engine lives in the open-seo-audit aux worker + // (wrangler.audit.jsonc), including the scratchpad DO — the DO cannot be + // bound cross-script here (the Cloudflare API rejects deleting the class + // below while any binding still references its name). Cancel + GDPR + // erasure go through this service binding's AuditEngine entrypoint. + "services": [ + { + "binding": "AUDIT_ENGINE", + "service": "open-seo-audit", + }, + ], "migrations": [ { "tag": "v1", @@ -70,6 +80,16 @@ "tag": "v3", "new_sqlite_classes": ["AuditScratchpad"], }, + // AuditScratchpad moved to the open-seo-audit worker (no binding may + // still reference the class name in the same upload — see the services + // comment above). Deleting it here is safe: scratchpad state is + // disposable by design (destroyed at finalize, 7-day cleanup alarm) — + // in-flight audits at the cutover fail once and the stale-audit + // reconciler reaps them. + { + "tag": "v4", + "deleted_classes": ["AuditScratchpad"], + }, ], "triggers": { // Every 5 min: rank checks + stale-audit reconcile. Daily: OAuth KV GC.