From 848ee4faadf00f17822b872fbd004c8608a33d16 Mon Sep 17 00:00:00 2001 From: Richard R Date: Sun, 7 Jun 2026 12:08:57 -0600 Subject: [PATCH] feat(tasks): add document blob lease for safe orphan reaping and improve scheduled task robustness Introduce a document blob lease mechanism to prevent race conditions between document registration and orphaned blob cleanup. The new `document_blob_leases` table ensures that only one process can claim a document blob for mutation or deletion at a time. Update the orphan reaper to acquire a lease before deleting blobs and to re-check for ownership after acquiring the lease, avoiding accidental deletion of in-flight uploads. Enhance scheduled task infrastructure with per-task fencing tokens to prevent stale runners from overwriting newer results, enforce runtime limits with abort signals, and expose scheduler mode and minimum interval to the admin panel and API. Adjust task handlers to accept a context with abort support, and update documentation and environment variable references for the new cron secret and scheduling behavior. BREAKING CHANGE: Scheduled tasks now require a `document_blob_leases` table and updated handler signatures. Vercel deployments must set `CRON_SECRET` for scheduled maintenance. --- .env.example | 4 + docs-site/docs/configure/admin-panel.md | 11 +++ docs-site/docs/configure/tts-rate-limiting.md | 2 +- docs-site/docs/deploy/local-development.md | 4 + docs-site/docs/deploy/vercel-deployment.md | 18 +++- docs-site/docs/docker-quick-start.md | 1 + .../docs/reference/environment-variables.md | 10 ++ drizzle/postgres/0011_scheduled-tasks.sql | 7 ++ drizzle/postgres/meta/0011_snapshot.json | 39 +++++++- drizzle/postgres/meta/_journal.json | 2 +- drizzle/sqlite/0011_scheduled-tasks.sql | 7 ++ drizzle/sqlite/meta/0011_snapshot.json | 40 +++++++- drizzle/sqlite/meta/_journal.json | 2 +- src/app/api/admin/tasks/[key]/route.ts | 8 ++ src/app/api/admin/tasks/[key]/run/route.ts | 1 + src/app/api/admin/tasks/route.ts | 9 +- src/app/api/admin/tasks/tick/route.ts | 1 + .../documents/blob/upload/finalize/route.ts | 86 ++++++++-------- src/components/admin/AdminTasksPanel.tsx | 39 ++++++-- src/db/schema.ts | 1 + src/db/schema_postgres.ts | 7 ++ src/db/schema_sqlite.ts | 7 ++ src/lib/server/documents/blob-lease.ts | 75 ++++++++++++++ src/lib/server/documents/blobstore.ts | 7 ++ src/lib/server/tasks/engine.ts | 74 ++++++++++---- .../tasks/handlers/cleanup-temp-uploads.ts | 6 +- .../tasks/handlers/reap-orphaned-blobs.ts | 26 ++++- src/lib/server/tasks/scheduler.ts | 22 +++++ src/lib/server/tasks/types.ts | 9 +- .../blob-upload-finalize-docx.vitest.spec.ts | 4 + tests/unit/document-blob-lease.vitest.spec.ts | 47 +++++++++ tests/unit/reap-orphaned-blobs.vitest.spec.ts | 49 +++++++++- .../scheduled-tasks-engine.vitest.spec.ts | 72 ++++++++++++++ .../scheduled-tasks-routes.vitest.spec.ts | 98 +++++++++++++++++++ 34 files changed, 708 insertions(+), 87 deletions(-) create mode 100644 src/lib/server/documents/blob-lease.ts create mode 100644 tests/unit/document-blob-lease.vitest.spec.ts create mode 100644 tests/unit/scheduled-tasks-routes.vitest.spec.ts diff --git a/.env.example b/.env.example index 173fc65..416bbd7 100644 --- a/.env.example +++ b/.env.example @@ -29,6 +29,10 @@ AUTH_TRUSTED_ORIGINS=http://localhost:3003,http://127.0.0.1:3003 # Additional tr # (Optional) Comma-separated list of emails that are auto-promoted to admin. ADMIN_EMAILS= +# (Required for the Vercel scheduled-task cron route; Vercel sends it as a bearer token.) +# Self-hosted deployments run scheduled tasks in-process and do not require this. +# CRON_SECRET= + # (Optional) Backend DB used for server-side metadata (documents/audiobooks) and auth tables. # Defaults to SQLite at docstore/sqlite3.db when not set. # POSTGRES_URL= diff --git a/docs-site/docs/configure/admin-panel.md b/docs-site/docs/configure/admin-panel.md index 82a70c4..f239d64 100644 --- a/docs-site/docs/configure/admin-panel.md +++ b/docs-site/docs/configure/admin-panel.md @@ -117,6 +117,17 @@ At the end of the **Site features** tab, a dedicated **TTS upstream** group cont In v4 these settings are admin-only and are no longer configurable through environment variables. +## Scheduled tasks + +The **Scheduled tasks** section controls background maintenance jobs such as expired-upload cleanup, orphaned-blob reaping, and rate-limit ledger pruning. + +- Enable or disable each task, adjust its interval, or run it immediately. +- Runs use database-backed leases so multiple app instances do not normally execute the same task concurrently. +- A task that exceeds four minutes is aborted and recorded as failed. A crashed run can be reclaimed after its stale lease expires. +- Failures and the latest successful summary appear on the task card and in server logs. + +Self-hosted Node.js deployments tick the scheduler in-process once per minute. Vercel uses the authenticated `/api/admin/tasks/tick` cron route; the checked-in Vercel Hobby schedule runs once daily, so intervals shorter than one day are unavailable there. See [Vercel Deployment](../deploy/vercel-deployment#5-scheduled-maintenance-tasks). + ## Migrating off env vars In v4, runtime site features are managed by admin settings and optional JSON seed. To minimize env surface area: diff --git a/docs-site/docs/configure/tts-rate-limiting.md b/docs-site/docs/configure/tts-rate-limiting.md index 967124c..0ceac10 100644 --- a/docs-site/docs/configure/tts-rate-limiting.md +++ b/docs-site/docs/configure/tts-rate-limiting.md @@ -37,6 +37,6 @@ If a request exceeds the active limit, the TTS API returns `429` with reset meta ## Related docs - TTS/rate-limit environment variables: [Environment Variables](../reference/environment-variables#tts-provider-and-request-behavior) -- PDF parsing rate limiting (separate, compute-side throttle): [Admin Panel → Site features](./admin-panel#site-features) and [Environment Variables](../reference/environment-variables#compute-pdf-parsing-rate-limiting-runtime-settings) +- PDF parsing rate limiting (separate, compute-side throttle): [Admin Panel → Rate limiting](./admin-panel#rate-limiting) - Auth configuration: [Auth](./auth) - Provider setup: [TTS Providers](./tts-providers) diff --git a/docs-site/docs/deploy/local-development.md b/docs-site/docs/deploy/local-development.md index 3f5f0b1..dd38dcd 100644 --- a/docs-site/docs/deploy/local-development.md +++ b/docs-site/docs/deploy/local-development.md @@ -325,6 +325,10 @@ Storage configuration details are in [Object / Blob Storage](../configure/object Refer to [Database](../configure/database) for database modes. Learn about migration behavior and commands in [Migrations](../configure/migrations). +:::info Scheduled maintenance tasks +Local and self-hosted Node.js deployments start the scheduled-task loop in-process and check for due work once per minute. No `CRON_SECRET` is required unless you intentionally invoke the cron HTTP route yourself. Manage task intervals and inspect failures from **Settings → Admin → Scheduled tasks**. +::: + 4. Start the app. diff --git a/docs-site/docs/deploy/vercel-deployment.md b/docs-site/docs/deploy/vercel-deployment.md index eeedbcd..8158c10 100644 --- a/docs-site/docs/deploy/vercel-deployment.md +++ b/docs-site/docs/deploy/vercel-deployment.md @@ -36,6 +36,7 @@ S3_PREFIX=openreader BASE_URL=https://your-app.vercel.app AUTH_SECRET=... ADMIN_EMAILS=you@example.com # comma-separated; admins manage TTS + features in-app +CRON_SECRET=... # generate with: openssl rand -hex 32 # Heavy compute (required on Vercel in current releases) COMPUTE_WORKER_URL=https:// @@ -110,7 +111,15 @@ Vercel deployments do not run `scripts/openreader-entrypoint.mjs`, so automatic - Run `pnpm migrate` in a controlled environment to apply Drizzle schema migrations to your Postgres DB. - Run `pnpm migrate-fs` only when migrating legacy local filesystem data (`docstore/documents_v1`, `docstore/audiobooks_v1`) into object storage + DB rows. Fresh Vercel deployments usually do not need this. -## 5. FFmpeg packaging in Vercel functions +## 5. Scheduled maintenance tasks + +The repository configures `/api/admin/tasks/tick` as a Vercel Cron route. Set `CRON_SECRET`; requests without the matching bearer token are rejected. + +The checked-in Hobby-compatible schedule invokes the route once daily. The admin task panel therefore prevents selecting intervals shorter than one day on Vercel, even though self-hosted deployments can run tasks more frequently. + +Each due task is claimed with a database-backed lease, due tasks start independently, and individual runs are aborted and marked failed after four minutes. Review failures and run tasks manually from **Settings → Admin → Scheduled tasks**. + +## 6. FFmpeg packaging in Vercel functions `ffmpeg-static` binaries must be included in function traces. This repo already does that in `next.config.ts` via `outputFileTracingIncludes` for: @@ -124,7 +133,7 @@ Vercel deployments do not run `scripts/openreader-entrypoint.mjs`, so automatic If you change route paths or split handlers, update `outputFileTracingIncludes` accordingly. -## 6. Function memory sizing +## 7. Function memory sizing FFmpeg workloads benefit from more memory/CPU. This repo includes: @@ -140,14 +149,15 @@ FFmpeg workloads benefit from more memory/CPU. This repo includes: Adjust memory per route if your files are larger or your plan differs. -## 7. Runtime expectations and caveats +## 8. Runtime expectations and caveats - Audiobook APIs require S3 configuration; otherwise they return `503`. - For production Vercel deploys, use `POSTGRES_URL` instead of SQLite. -## 8. Smoke test after deploy +## 9. Smoke test after deploy 1. Upload and read a PDF/EPUB document. 2. Confirm sync/blob fetch works across refreshes/devices. 3. Generate at least one audiobook chapter and play/download it. 4. Verify worker-backed word highlighting and PDF parsing. +5. Open **Settings → Admin → Scheduled tasks**, run one task manually, and confirm the next daily cron invocation succeeds. diff --git a/docs-site/docs/docker-quick-start.md b/docs-site/docs/docker-quick-start.md index e9e1cc0..3495443 100644 --- a/docs-site/docs/docker-quick-start.md +++ b/docs-site/docs/docker-quick-start.md @@ -122,6 +122,7 @@ What this command enables: - `restrictUserApiKeys` controls shared-provider-only mode. For per-user BYOK, toggle it off in **Settings → Admin → Site features** or seed `runtimeConfig.restrictUserApiKeys=false` via runtime seed JSON. - Use a `/app/docstore` mount if you want data to survive container/image replacement. - Startup automatically runs DB/storage migrations via the shared entrypoint. +- Scheduled maintenance tasks run in-process and can be managed from **Settings → Admin → Scheduled tasks**; Docker/self-hosted deployments do not need `CRON_SECRET`. ::: :::warning Port `8333` Exposure diff --git a/docs-site/docs/reference/environment-variables.md b/docs-site/docs/reference/environment-variables.md index dfe985c..a280903 100644 --- a/docs-site/docs/reference/environment-variables.md +++ b/docs-site/docs/reference/environment-variables.md @@ -26,6 +26,7 @@ Runtime site features are seeded with `RUNTIME_SEED_JSON` / `RUNTIME_SEED_JSON_P | `GITHUB_CLIENT_ID` | Auth/OAuth | unset | Set with `GITHUB_CLIENT_SECRET` to enable GitHub sign-in | | `GITHUB_CLIENT_SECRET` | Auth/OAuth | unset | Set with `GITHUB_CLIENT_ID` to enable GitHub sign-in | | `ADMIN_EMAILS` | Admin | empty | Comma-separated emails auto-promoted to admin | +| `CRON_SECRET` | Scheduled tasks | unset | Required for Vercel cron invocations | | `POSTGRES_URL` | Database | unset (SQLite mode) | Set to switch metadata/auth DB to Postgres | | `USE_EMBEDDED_WEED_MINI` | Storage | `true` when unset | Set `false` to use external S3-compatible storage only | | `WEED_MINI_DIR` | Storage | `docstore/seaweedfs` | Override embedded SeaweedFS data directory | @@ -144,6 +145,15 @@ Comma-separated list of email addresses auto-promoted to admin. - Requires auth to be enabled - Admins can manage shared providers and runtime site features in-app +### CRON_SECRET + +Bearer-token secret for `GET /api/admin/tasks/tick`. + +- Required on Vercel so scheduled maintenance tasks can run from the configured Vercel Cron. +- Vercel automatically sends `Authorization: Bearer ` on cron invocations. +- Generate a strong random value, for example with `openssl rand -hex 32`. +- Self-hosted Node.js deployments run the scheduler in-process and do not require this variable. + ## Database and Object Blob Storage ### POSTGRES_URL diff --git a/drizzle/postgres/0011_scheduled-tasks.sql b/drizzle/postgres/0011_scheduled-tasks.sql index cb05f2f..3714be8 100644 --- a/drizzle/postgres/0011_scheduled-tasks.sql +++ b/drizzle/postgres/0011_scheduled-tasks.sql @@ -1,8 +1,15 @@ +CREATE TABLE "document_blob_leases" ( + "document_id" text PRIMARY KEY NOT NULL, + "lease_owner" text NOT NULL, + "lease_until_ms" bigint NOT NULL +); +--> statement-breakpoint CREATE TABLE "scheduled_tasks" ( "key" text PRIMARY KEY NOT NULL, "enabled" boolean DEFAULT true NOT NULL, "interval_ms" bigint NOT NULL, "last_status" text DEFAULT 'idle' NOT NULL, + "lease_owner" text, "last_run_at" bigint, "last_duration_ms" bigint, "last_error" text, diff --git a/drizzle/postgres/meta/0011_snapshot.json b/drizzle/postgres/meta/0011_snapshot.json index 443520f..ebf8389 100644 --- a/drizzle/postgres/meta/0011_snapshot.json +++ b/drizzle/postgres/meta/0011_snapshot.json @@ -1,5 +1,5 @@ { - "id": "afe1d934-9b20-4d50-82fc-aeebd49fe43f", + "id": "6fa9be62-3d2b-43d2-8c49-11d9c6411a33", "prevId": "1099cbe6-0c2e-4ea8-ae3c-6883c931169d", "version": "7", "dialect": "postgresql", @@ -329,6 +329,37 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.document_blob_leases": { + "name": "document_blob_leases", + "schema": "", + "columns": { + "document_id": { + "name": "document_id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "lease_owner": { + "name": "lease_owner", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "lease_until_ms": { + "name": "lease_until_ms", + "type": "bigint", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.document_previews": { "name": "document_previews", "schema": "", @@ -724,6 +755,12 @@ "notNull": true, "default": "'idle'" }, + "lease_owner": { + "name": "lease_owner", + "type": "text", + "primaryKey": false, + "notNull": false + }, "last_run_at": { "name": "last_run_at", "type": "bigint", diff --git a/drizzle/postgres/meta/_journal.json b/drizzle/postgres/meta/_journal.json index 5731fec..cfe7d84 100644 --- a/drizzle/postgres/meta/_journal.json +++ b/drizzle/postgres/meta/_journal.json @@ -82,7 +82,7 @@ { "idx": 11, "version": "7", - "when": 1780847571383, + "when": 1780851107819, "tag": "0011_scheduled-tasks", "breakpoints": true } diff --git a/drizzle/sqlite/0011_scheduled-tasks.sql b/drizzle/sqlite/0011_scheduled-tasks.sql index cca23bd..de29632 100644 --- a/drizzle/sqlite/0011_scheduled-tasks.sql +++ b/drizzle/sqlite/0011_scheduled-tasks.sql @@ -1,8 +1,15 @@ +CREATE TABLE `document_blob_leases` ( + `document_id` text PRIMARY KEY NOT NULL, + `lease_owner` text NOT NULL, + `lease_until_ms` integer NOT NULL +); +--> statement-breakpoint CREATE TABLE `scheduled_tasks` ( `key` text PRIMARY KEY NOT NULL, `enabled` integer DEFAULT true NOT NULL, `interval_ms` integer NOT NULL, `last_status` text DEFAULT 'idle' NOT NULL, + `lease_owner` text, `last_run_at` integer, `last_duration_ms` integer, `last_error` text, diff --git a/drizzle/sqlite/meta/0011_snapshot.json b/drizzle/sqlite/meta/0011_snapshot.json index 45d8f1f..66d871f 100644 --- a/drizzle/sqlite/meta/0011_snapshot.json +++ b/drizzle/sqlite/meta/0011_snapshot.json @@ -1,7 +1,7 @@ { "version": "6", "dialect": "sqlite", - "id": "4c4a203e-9b2b-4eb2-b31c-2613a7c1d377", + "id": "d063930a-ae25-4ca0-8b38-7092a16e3503", "prevId": "a32ae7d4-c074-44f0-9e90-2473949376ef", "tables": { "admin_providers": { @@ -350,6 +350,37 @@ "uniqueConstraints": {}, "checkConstraints": {} }, + "document_blob_leases": { + "name": "document_blob_leases", + "columns": { + "document_id": { + "name": "document_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "lease_owner": { + "name": "lease_owner", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "lease_until_ms": { + "name": "lease_until_ms", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, "document_previews": { "name": "document_previews", "columns": { @@ -728,6 +759,13 @@ "autoincrement": false, "default": "'idle'" }, + "lease_owner": { + "name": "lease_owner", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, "last_run_at": { "name": "last_run_at", "type": "integer", diff --git a/drizzle/sqlite/meta/_journal.json b/drizzle/sqlite/meta/_journal.json index 513ea7f..3222938 100644 --- a/drizzle/sqlite/meta/_journal.json +++ b/drizzle/sqlite/meta/_journal.json @@ -82,7 +82,7 @@ { "idx": 11, "version": "6", - "when": 1780847571062, + "when": 1780851107505, "tag": "0011_scheduled-tasks", "breakpoints": true } diff --git a/src/app/api/admin/tasks/[key]/route.ts b/src/app/api/admin/tasks/[key]/route.ts index 4ac2460..c91bc03 100644 --- a/src/app/api/admin/tasks/[key]/route.ts +++ b/src/app/api/admin/tasks/[key]/route.ts @@ -2,6 +2,7 @@ import { NextRequest, NextResponse } from 'next/server'; import { requireAdminContext } from '@/lib/server/auth/admin'; import { updateTask } from '@/lib/server/tasks/engine'; import { TASK_REGISTRY } from '@/lib/server/tasks/registry'; +import { getTaskSchedulerInfo } from '@/lib/server/tasks/scheduler'; export const dynamic = 'force-dynamic'; @@ -44,6 +45,13 @@ export async function PATCH( return NextResponse.json({ error: 'intervalMs must be a finite positive number' }, { status: 400 }); } patch.intervalMs = Math.max(1, Math.floor(payload.intervalMs)); + const minimumIntervalMs = getTaskSchedulerInfo().minimumIntervalMs; + if (patch.intervalMs < minimumIntervalMs) { + return NextResponse.json( + { error: `intervalMs must be at least ${minimumIntervalMs} for this deployment` }, + { status: 400 }, + ); + } } if (Object.keys(patch).length === 0) { return NextResponse.json({ error: 'Expected enabled or intervalMs' }, { status: 400 }); diff --git a/src/app/api/admin/tasks/[key]/run/route.ts b/src/app/api/admin/tasks/[key]/run/route.ts index 8b67a8a..f09554a 100644 --- a/src/app/api/admin/tasks/[key]/run/route.ts +++ b/src/app/api/admin/tasks/[key]/run/route.ts @@ -4,6 +4,7 @@ import { runTaskNow } from '@/lib/server/tasks/engine'; import { TASK_REGISTRY } from '@/lib/server/tasks/registry'; export const dynamic = 'force-dynamic'; +export const maxDuration = 300; export async function POST( req: NextRequest, diff --git a/src/app/api/admin/tasks/route.ts b/src/app/api/admin/tasks/route.ts index ef828f8..33dc7c3 100644 --- a/src/app/api/admin/tasks/route.ts +++ b/src/app/api/admin/tasks/route.ts @@ -1,6 +1,7 @@ import { NextRequest, NextResponse } from 'next/server'; import { requireAdminContext } from '@/lib/server/auth/admin'; import { listTasks } from '@/lib/server/tasks/engine'; +import { getTaskSchedulerInfo } from '@/lib/server/tasks/scheduler'; export const dynamic = 'force-dynamic'; @@ -8,6 +9,10 @@ export async function GET(req: NextRequest): Promise { const ctx = await requireAdminContext(req); if (ctx instanceof Response) return ctx; - const tasks = await listTasks(); - return NextResponse.json({ tasks }); + const scheduler = getTaskSchedulerInfo(); + const tasks = (await listTasks()).map((task) => ({ + ...task, + intervalMs: Math.max(task.intervalMs, scheduler.minimumIntervalMs), + })); + return NextResponse.json({ tasks, scheduler }); } diff --git a/src/app/api/admin/tasks/tick/route.ts b/src/app/api/admin/tasks/tick/route.ts index 9260281..f4d4438 100644 --- a/src/app/api/admin/tasks/tick/route.ts +++ b/src/app/api/admin/tasks/tick/route.ts @@ -2,6 +2,7 @@ import { NextRequest, NextResponse } from 'next/server'; import { runDueTasks } from '@/lib/server/tasks/engine'; export const dynamic = 'force-dynamic'; +export const maxDuration = 300; /** * Cron-driven tick for serverless (Vercel) deployments. Vercel automatically diff --git a/src/app/api/documents/blob/upload/finalize/route.ts b/src/app/api/documents/blob/upload/finalize/route.ts index 48bc74b..6944eb6 100644 --- a/src/app/api/documents/blob/upload/finalize/route.ts +++ b/src/app/api/documents/blob/upload/finalize/route.ts @@ -18,6 +18,7 @@ import { } from '@/lib/server/documents/blobstore'; import { convertDocxBufferToPdfBuffer } from '@/lib/server/documents/docx-convert'; import { registerUploadedDocument } from '@/lib/server/documents/register-upload'; +import { withDocumentBlobLease } from '@/lib/server/documents/blob-lease'; import { safeDocumentName, toDocumentTypeFromName } from '@/lib/server/documents/utils'; import { errorResponse } from '@/lib/server/errors/next-response'; import { errorToLog, serverLogger } from '@/lib/server/logger'; @@ -149,51 +150,52 @@ async function finalizeOne(input: { : input.upload.name; const documentId = createHash('sha256').update(finalizedBody).digest('hex'); - // Ensure the content-addressed blob exists. This is idempotent and race-safe - // on its own (ifNoneMatch + PreconditionFailed tolerates a concurrent finalize - // of the same content), so no lock is needed. The reaper's grace window - // protects a freshly written blob from being reaped before its row commits. - try { - await headDocumentBlob(documentId, input.namespace); - } catch (error) { - if (!isMissingBlobError(error)) throw error; - if (!isDocxUpload) { - try { - await copyTempDocumentBlobToDocument( - input.upload.token, - input.userId, - documentId, - input.namespace, - finalizedContentType, - { ifNoneMatch: true }, - ); - } catch (copyError) { - if (!isPreconditionFailed(copyError)) throw copyError; - } - } else { - try { - await putDocumentBlob( - documentId, - finalizedBody, - finalizedContentType, - input.namespace, - { ifNoneMatch: true }, - ); - } catch (putError) { - if (!isPreconditionFailed(putError)) throw putError; + const stored = await withDocumentBlobLease(documentId, async () => { + // Keep the canonical blob and ownership-row write under the same durable + // lease so the orphan reaper cannot delete between its ownership check and + // this registration. + try { + await headDocumentBlob(documentId, input.namespace); + } catch (error) { + if (!isMissingBlobError(error)) throw error; + if (!isDocxUpload) { + try { + await copyTempDocumentBlobToDocument( + input.upload.token, + input.userId, + documentId, + input.namespace, + finalizedContentType, + { ifNoneMatch: true }, + ); + } catch (copyError) { + if (!isPreconditionFailed(copyError)) throw copyError; + } + } else { + try { + await putDocumentBlob( + documentId, + finalizedBody, + finalizedContentType, + input.namespace, + { ifNoneMatch: true }, + ); + } catch (putError) { + if (!isPreconditionFailed(putError)) throw putError; + } } } - } - const canonicalHead = await headDocumentBlob(documentId, input.namespace); - const stored = await registerUploadedDocument({ - documentId, - userId: input.userId, - namespace: input.namespace, - name: finalizedName, - type: finalizedType, - size: canonicalHead.contentLength > 0 ? canonicalHead.contentLength : finalizedBody.byteLength, - lastModified: input.upload.lastModified, + const canonicalHead = await headDocumentBlob(documentId, input.namespace); + return registerUploadedDocument({ + documentId, + userId: input.userId, + namespace: input.namespace, + name: finalizedName, + type: finalizedType, + size: canonicalHead.contentLength > 0 ? canonicalHead.contentLength : finalizedBody.byteLength, + lastModified: input.upload.lastModified, + }); }); await putTempDocumentFinalizeReceipt( diff --git a/src/components/admin/AdminTasksPanel.tsx b/src/components/admin/AdminTasksPanel.tsx index ac21b87..ffcb879 100644 --- a/src/components/admin/AdminTasksPanel.tsx +++ b/src/components/admin/AdminTasksPanel.tsx @@ -23,12 +23,18 @@ interface TaskView { running: boolean; } +interface TaskSchedulerInfo { + mode: 'self-hosted' | 'vercel-cron'; + tickIntervalMs: number; + minimumIntervalMs: number; +} + const TASKS_QUERY_KEY = ['admin-tasks'] as const; -async function fetchTasks(): Promise { +async function fetchTasks(): Promise<{ tasks: TaskView[]; scheduler: TaskSchedulerInfo }> { const res = await fetch('/api/admin/tasks'); if (!res.ok) throw new Error(`HTTP ${res.status}`); - return ((await res.json()) as { tasks: TaskView[] }).tasks; + return (await res.json()) as { tasks: TaskView[]; scheduler: TaskSchedulerInfo }; } function RunningDot() { @@ -111,11 +117,18 @@ export function AdminTasksPanel() { return (
+ {data?.scheduler.mode === 'vercel-cron' && ( +

+ Vercel Hobby invokes scheduled tasks once daily. Shorter intervals are unavailable on this deployment. +

+ )}
- {(data ?? []).map((task) => ( + {(data?.tasks ?? []).map((task) => ( patchTask.mutate({ key: task.key, patch: { enabled } })} @@ -123,7 +136,7 @@ export function AdminTasksPanel() { onRun={() => runTask.mutate(task.key)} /> ))} - {data && data.length === 0 && ( + {data && data.tasks.length === 0 && (

No tasks registered.

)}
@@ -133,6 +146,8 @@ export function AdminTasksPanel() { function TaskRow({ task, + schedulerMode, + minimumIntervalMs, busy, runPending, onToggle, @@ -140,6 +155,8 @@ function TaskRow({ onRun, }: { task: TaskView; + schedulerMode: TaskSchedulerInfo['mode']; + minimumIntervalMs: number; busy: boolean; runPending: boolean; onToggle: (enabled: boolean) => void; @@ -155,7 +172,9 @@ function TaskRow({ const parsedMinutes = parseFloat(minutes); const newIntervalMs = parsedMinutes * 60000; const intervalDirty = - Number.isFinite(parsedMinutes) && parsedMinutes > 0 && newIntervalMs !== task.intervalMs; + Number.isFinite(parsedMinutes) + && newIntervalMs >= minimumIntervalMs + && newIntervalMs !== task.intervalMs; const running = task.running || runPending; @@ -187,7 +206,13 @@ function TaskRow({ {formatRelative(task.lastRunAt)} - {task.enabled && task.nextRunAt != null && ( + {task.enabled && schedulerMode === 'vercel-cron' && ( + + + next daily cron + + )} + {task.enabled && schedulerMode !== 'vercel-cron' && task.nextRunAt != null && ( next {formatRelative(task.nextRunAt)} @@ -199,7 +224,7 @@ function TaskRow({ Every 0`), ]); +export const documentBlobLeases = pgTable('document_blob_leases', { + documentId: text('document_id').primaryKey(), + leaseOwner: text('lease_owner').notNull(), + leaseUntilMs: bigint('lease_until_ms', { mode: 'number' }).notNull(), +}); + export const ttsSegmentVariants = pgTable('tts_segment_variants', { segmentId: text('segment_id').notNull(), userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }), diff --git a/src/db/schema_sqlite.ts b/src/db/schema_sqlite.ts index 2ae1c54..12f7ab7 100644 --- a/src/db/schema_sqlite.ts +++ b/src/db/schema_sqlite.ts @@ -215,6 +215,7 @@ export const scheduledTasks = sqliteTable('scheduled_tasks', { enabled: integer('enabled', { mode: 'boolean' }).notNull().default(true), intervalMs: integer('interval_ms').notNull(), lastStatus: text('last_status').notNull().default('idle'), + leaseOwner: text('lease_owner'), lastRunAt: integer('last_run_at'), lastDurationMs: integer('last_duration_ms'), lastError: text('last_error'), @@ -227,6 +228,12 @@ export const scheduledTasks = sqliteTable('scheduled_tasks', { check('scheduled_tasks_interval_ms_positive', sql`${table.intervalMs} > 0`), ]); +export const documentBlobLeases = sqliteTable('document_blob_leases', { + documentId: text('document_id').primaryKey(), + leaseOwner: text('lease_owner').notNull(), + leaseUntilMs: integer('lease_until_ms').notNull(), +}); + export const ttsSegmentVariants = sqliteTable('tts_segment_variants', { segmentId: text('segment_id').notNull(), userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }), diff --git a/src/lib/server/documents/blob-lease.ts b/src/lib/server/documents/blob-lease.ts new file mode 100644 index 0000000..1013427 --- /dev/null +++ b/src/lib/server/documents/blob-lease.ts @@ -0,0 +1,75 @@ +import { randomUUID } from 'node:crypto'; +import { and, eq, lt, or } from 'drizzle-orm'; +import { db } from '@/db'; +import { documentBlobLeases } from '@/db/schema'; + +const DEFAULT_LEASE_MS = 15 * 60 * 1000; +const RETRY_DELAY_MS = 100; + +export type DocumentBlobLease = { + owner: string; + release: () => Promise; +}; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export async function tryAcquireDocumentBlobLease( + documentId: string, + options?: { leaseMs?: number }, +): Promise { + const owner = randomUUID(); + const now = Date.now(); + const leaseUntilMs = now + (options?.leaseMs ?? DEFAULT_LEASE_MS); + + const claimed = await db + .insert(documentBlobLeases) + .values({ documentId, leaseOwner: owner, leaseUntilMs }) + .onConflictDoUpdate({ + target: documentBlobLeases.documentId, + set: { leaseOwner: owner, leaseUntilMs }, + where: or( + lt(documentBlobLeases.leaseUntilMs, now), + eq(documentBlobLeases.leaseOwner, owner), + ), + }) + .returning({ owner: documentBlobLeases.leaseOwner }); + + if (claimed[0]?.owner !== owner) return null; + + return { + owner, + release: async () => { + await db + .delete(documentBlobLeases) + .where(and( + eq(documentBlobLeases.documentId, documentId), + eq(documentBlobLeases.leaseOwner, owner), + )); + }, + }; +} + +export async function withDocumentBlobLease( + documentId: string, + fn: () => Promise, + options?: { waitMs?: number; leaseMs?: number }, +): Promise { + const deadline = Date.now() + (options?.waitMs ?? 30_000); + let lease: DocumentBlobLease | null = null; + + while (!lease && Date.now() <= deadline) { + lease = await tryAcquireDocumentBlobLease(documentId, options); + if (!lease) await sleep(RETRY_DELAY_MS); + } + if (!lease) { + throw new Error(`Timed out waiting for document blob lease: ${documentId}`); + } + + try { + return await fn(); + } finally { + await lease.release(); + } +} diff --git a/src/lib/server/documents/blobstore.ts b/src/lib/server/documents/blobstore.ts index 61ceb06..230bbed 100644 --- a/src/lib/server/documents/blobstore.ts +++ b/src/lib/server/documents/blobstore.ts @@ -607,6 +607,7 @@ export async function deleteDocumentPrefix(prefix: string): Promise { */ export async function listDocumentSourceBlobs( namespace: string | null, + options?: { signal?: AbortSignal }, ): Promise> { const cfg = getS3Config(); const client = getS3ProxyClient(); @@ -617,6 +618,7 @@ export async function listDocumentSourceBlobs( let continuationToken: string | undefined; do { + options?.signal?.throwIfAborted(); const listRes = await client.send( new ListObjectsV2Command({ Bucket: cfg.bucket, @@ -624,6 +626,7 @@ export async function listDocumentSourceBlobs( Delimiter: '/', ContinuationToken: continuationToken, }), + { abortSignal: options?.signal }, ); for (const item of listRes.Contents ?? []) { @@ -647,6 +650,7 @@ export async function listDocumentSourceBlobs( export async function deleteAllExpiredTempDocumentUploads( namespace: string | null, olderThanMs: number, + options?: { signal?: AbortSignal }, ): Promise { const cfg = getS3Config(); const client = getS3ProxyClient(); @@ -657,12 +661,14 @@ export async function deleteAllExpiredTempDocumentUploads( let deleted = 0; do { + options?.signal?.throwIfAborted(); const listRes = await client.send( new ListObjectsV2Command({ Bucket: cfg.bucket, Prefix: prefix, ContinuationToken: continuationToken, }), + { abortSignal: options?.signal }, ); const batch: string[] = []; @@ -682,6 +688,7 @@ export async function deleteAllExpiredTempDocumentUploads( Quiet: true, }, }), + { abortSignal: options?.signal }, ); if (deleteRes.Errors?.length) { throw new Error( diff --git a/src/lib/server/tasks/engine.ts b/src/lib/server/tasks/engine.ts index aecc113..93822df 100644 --- a/src/lib/server/tasks/engine.ts +++ b/src/lib/server/tasks/engine.ts @@ -1,14 +1,16 @@ +import { randomUUID } from 'node:crypto'; import { and, eq, lt, ne, or, sql } from 'drizzle-orm'; import { db } from '@/db'; import { scheduledTasks } from '@/db/schema'; import { serverLogger } from '@/lib/server/logger'; import { logDegraded } from '@/lib/server/errors/logging'; import { TASK_REGISTRY } from './registry'; -import type { TaskDef, TaskRegistry, TaskRunStatus } from './types'; +import type { TaskContext, TaskDef, TaskRegistry, TaskRunStatus } from './types'; // A task still marked 'running' after this long is assumed abandoned (process // crashed mid-run) and may be reclaimed by the next tick. const STALE_RUNNING_MS = 60 * 60 * 1000; +const DEFAULT_TASK_MAX_RUN_MS = 4 * 60 * 1000; /** * Seed a `scheduled_tasks` row for every registered task. Idempotent: existing @@ -32,14 +34,21 @@ async function ensureTaskRows(registry: TaskRegistry = TASK_REGISTRY): Promise { +async function claimTask(key: string, now: number): Promise { + const leaseOwner = randomUUID(); const claimed = await db .update(scheduledTasks) - .set({ lastStatus: 'running', runningSince: now, runRequested: false, updatedAt: now }) + .set({ + lastStatus: 'running', + leaseOwner, + runningSince: now, + runRequested: false, + updatedAt: now, + }) .where(and( eq(scheduledTasks.key, key), or( @@ -47,12 +56,13 @@ async function claimTask(key: string, now: number): Promise { lt(scheduledTasks.runningSince, now - STALE_RUNNING_MS), ), )) - .returning({ key: scheduledTasks.key }); - return claimed.length > 0; + .returning({ leaseOwner: scheduledTasks.leaseOwner }); + return claimed[0]?.leaseOwner === leaseOwner ? leaseOwner : null; } async function finishTask( key: string, + leaseOwner: string, outcome: | { status: 'ok'; startedAt: number; summary: string | null } | { status: 'error'; startedAt: number; error: unknown }, @@ -68,18 +78,42 @@ async function finishTask( lastResultJson: outcome.status === 'ok' && outcome.summary ? outcome.summary : null, // Schedule the next run off the row's (possibly user-edited) interval. nextRunAt: sql`${now} + ${scheduledTasks.intervalMs}`, + leaseOwner: null, runningSince: null, updatedAt: now, }) - .where(eq(scheduledTasks.key, key)); + .where(and( + eq(scheduledTasks.key, key), + eq(scheduledTasks.leaseOwner, leaseOwner), + )); } -async function executeTask(key: string, def: TaskDef): Promise { +function taskTimeoutError(key: string, maxRunMs: number): Error { + return new Error(`Scheduled task "${key}" exceeded its ${maxRunMs}ms runtime limit`); +} + +async function executeTask(key: string, def: TaskDef, leaseOwner: string): Promise { const startedAt = Date.now(); + const maxRunMs = def.maxRunMs ?? DEFAULT_TASK_MAX_RUN_MS; + const controller = new AbortController(); + const context: TaskContext = { + signal: controller.signal, + deadlineAt: startedAt + maxRunMs, + }; + let timeout: ReturnType | undefined; + try { - const result = await def.run(); + const result = await Promise.race([ + def.run(context), + new Promise((_resolve, reject) => { + timeout = setTimeout(() => { + controller.abort(); + reject(taskTimeoutError(key, maxRunMs)); + }, maxRunMs); + }), + ]); const summary = result && typeof result.summary === 'string' ? result.summary : null; - await finishTask(key, { status: 'ok', startedAt, summary }); + await finishTask(key, leaseOwner, { status: 'ok', startedAt, summary }); } catch (error) { logDegraded(serverLogger, { event: 'tasks.run.failed', @@ -87,7 +121,9 @@ async function executeTask(key: string, def: TaskDef): Promise { step: key, error, }); - await finishTask(key, { status: 'error', startedAt, error }); + await finishTask(key, leaseOwner, { status: 'error', startedAt, error }); + } finally { + if (timeout) clearTimeout(timeout); } } @@ -103,6 +139,7 @@ export async function runDueTasks(options?: { registry?: TaskRegistry }): Promis const now = Date.now(); const rows = await db.select().from(scheduledTasks); + const executions: Promise[] = []; for (const row of rows) { const def = registry[row.key]; if (!def) continue; // orphaned row for a task no longer in the registry @@ -112,9 +149,11 @@ export async function runDueTasks(options?: { registry?: TaskRegistry }): Promis (row.enabled && row.nextRunAt != null && now >= Number(row.nextRunAt)); if (!due) continue; - if (!(await claimTask(row.key, now))) continue; - await executeTask(row.key, def); + const leaseOwner = await claimTask(row.key, now); + if (!leaseOwner) continue; + executions.push(executeTask(row.key, def, leaseOwner)); } + await Promise.all(executions); } /** @@ -125,8 +164,9 @@ export async function runTaskNow(key: string, registry: TaskRegistry = TASK_REGI const def = registry[key]; if (!def) return false; await ensureTaskRows(registry); - if (!(await claimTask(key, Date.now()))) return false; - await executeTask(key, def); + const leaseOwner = await claimTask(key, Date.now()); + if (!leaseOwner) return false; + await executeTask(key, def, leaseOwner); return true; } diff --git a/src/lib/server/tasks/handlers/cleanup-temp-uploads.ts b/src/lib/server/tasks/handlers/cleanup-temp-uploads.ts index 1d16c1d..537607d 100644 --- a/src/lib/server/tasks/handlers/cleanup-temp-uploads.ts +++ b/src/lib/server/tasks/handlers/cleanup-temp-uploads.ts @@ -3,15 +3,15 @@ import { TEMP_DOCUMENT_UPLOAD_TTL_MS, deleteAllExpiredTempDocumentUploads, } from '@/lib/server/documents/blobstore'; -import type { TaskResult } from '../types'; +import type { TaskContext, TaskResult } from '../types'; /** Remove temporary upload objects past their TTL across all users. */ -export async function cleanupTempUploads(): Promise { +export async function cleanupTempUploads(context: TaskContext): Promise { if (!isS3Configured()) { return { summary: 'Skipped: object storage not configured', deleted: 0 }; } const cutoff = Date.now() - TEMP_DOCUMENT_UPLOAD_TTL_MS; - const deleted = await deleteAllExpiredTempDocumentUploads(null, cutoff); + const deleted = await deleteAllExpiredTempDocumentUploads(null, cutoff, { signal: context.signal }); return { summary: `Deleted ${deleted} expired upload object(s)`, deleted }; } diff --git a/src/lib/server/tasks/handlers/reap-orphaned-blobs.ts b/src/lib/server/tasks/handlers/reap-orphaned-blobs.ts index 8dd90be..a692370 100644 --- a/src/lib/server/tasks/handlers/reap-orphaned-blobs.ts +++ b/src/lib/server/tasks/handlers/reap-orphaned-blobs.ts @@ -7,7 +7,8 @@ import { deleteDocumentPreviewArtifacts } from '@/lib/server/documents/previews- import { deleteDocumentPreviewRows } from '@/lib/server/documents/previews'; import { serverLogger } from '@/lib/server/logger'; import { logDegraded } from '@/lib/server/errors/logging'; -import type { TaskResult } from '../types'; +import { tryAcquireDocumentBlobLease } from '@/lib/server/documents/blob-lease'; +import type { TaskContext, TaskResult } from '../types'; // Don't reap a blob younger than this — it may belong to an in-flight finalize // that has written the blob but not yet committed its ownership row. @@ -21,17 +22,19 @@ const OWNERSHIP_CHECK_BATCH = 200; * age past the grace window is an orphan (e.g. left by a failed inline delete) * and is safe to remove. Production data is non-namespaced. */ -export async function reapOrphanedBlobs(): Promise { +export async function reapOrphanedBlobs(context: TaskContext): Promise { if (!isS3Configured()) { return { summary: 'Skipped: object storage not configured', reaped: 0 }; } const now = Date.now(); - const blobs = await listDocumentSourceBlobs(null); + const blobs = await listDocumentSourceBlobs(null, { signal: context.signal }); const candidates = blobs.filter((blob) => now - blob.lastModifiedMs > GRACE_MS); let reaped = 0; + const failures: unknown[] = []; for (let i = 0; i < candidates.length; i += OWNERSHIP_CHECK_BATCH) { + context.signal.throwIfAborted(); const chunk = candidates.slice(i, i + OWNERSHIP_CHECK_BATCH); const ids = chunk.map((c) => c.id); const ownedRows = await db @@ -41,13 +44,24 @@ export async function reapOrphanedBlobs(): Promise { const owned = new Set(ownedRows.map((row: { id: string }) => row.id)); for (const candidate of chunk) { + context.signal.throwIfAborted(); if (owned.has(candidate.id)) continue; + const lease = await tryAcquireDocumentBlobLease(candidate.id); + if (!lease) continue; try { + const [owner] = await db + .select({ id: documents.id }) + .from(documents) + .where(inArray(documents.id, [candidate.id])) + .limit(1); + if (owner) continue; + await deleteDocumentBlob(candidate.id, null); await deleteDocumentPreviewArtifacts(candidate.id, null); await deleteDocumentPreviewRows(candidate.id, null); reaped += 1; } catch (error) { + failures.push(error); logDegraded(serverLogger, { event: 'tasks.reap_orphaned_blobs.delete_failed', msg: 'Failed to reap orphaned document storage', @@ -55,10 +69,16 @@ export async function reapOrphanedBlobs(): Promise { context: { documentId: candidate.id }, error, }); + } finally { + await lease.release(); } } } + if (failures.length > 0) { + throw new AggregateError(failures, `Failed to reap ${failures.length} orphaned blob(s)`); + } + return { summary: `Reaped ${reaped} orphaned blob(s)`, scanned: blobs.length, diff --git a/src/lib/server/tasks/scheduler.ts b/src/lib/server/tasks/scheduler.ts index 1632e85..f60a57a 100644 --- a/src/lib/server/tasks/scheduler.ts +++ b/src/lib/server/tasks/scheduler.ts @@ -3,6 +3,28 @@ import { serverLogger } from '@/lib/server/logger'; const TICK_INTERVAL_MS = 60_000; const INITIAL_DELAY_MS = 10_000; +export const VERCEL_TICK_INTERVAL_MS = 24 * 60 * 60 * 1000; + +export type TaskSchedulerInfo = { + mode: 'self-hosted' | 'vercel-cron'; + tickIntervalMs: number; + minimumIntervalMs: number; +}; + +export function getTaskSchedulerInfo(): TaskSchedulerInfo { + if (process.env.VERCEL) { + return { + mode: 'vercel-cron', + tickIntervalMs: VERCEL_TICK_INTERVAL_MS, + minimumIntervalMs: VERCEL_TICK_INTERVAL_MS, + }; + } + return { + mode: 'self-hosted', + tickIntervalMs: TICK_INTERVAL_MS, + minimumIntervalMs: TICK_INTERVAL_MS, + }; +} let started = false; diff --git a/src/lib/server/tasks/types.ts b/src/lib/server/tasks/types.ts index 6318533..3c61d7b 100644 --- a/src/lib/server/tasks/types.ts +++ b/src/lib/server/tasks/types.ts @@ -6,7 +6,12 @@ export type TaskResult = { [key: string]: unknown; }; -export type TaskHandler = () => Promise; +export type TaskContext = { + signal: AbortSignal; + deadlineAt: number; +}; + +export type TaskHandler = (context: TaskContext) => Promise; /** Static definition of a task, kept in code (the registry). */ export type TaskDef = { @@ -16,6 +21,8 @@ export type TaskDef = { description?: string; /** Default run interval in ms; the per-task row may override it. */ defaultIntervalMs: number; + /** Maximum wall-clock time before the run is marked failed and aborted. */ + maxRunMs?: number; /** The work to perform. Must be idempotent and safe to re-run. */ run: TaskHandler; }; diff --git a/tests/unit/blob-upload-finalize-docx.vitest.spec.ts b/tests/unit/blob-upload-finalize-docx.vitest.spec.ts index b818d66..fec1efc 100644 --- a/tests/unit/blob-upload-finalize-docx.vitest.spec.ts +++ b/tests/unit/blob-upload-finalize-docx.vitest.spec.ts @@ -24,6 +24,10 @@ vi.mock('@/lib/server/documents/register-upload', () => ({ registerUploadedDocument: hoisted.registerUploadedDocument, })); +vi.mock('@/lib/server/documents/blob-lease', () => ({ + withDocumentBlobLease: vi.fn(async (_documentId: string, fn: () => Promise) => fn()), +})); + vi.mock('@/lib/server/documents/docx-convert', () => ({ convertDocxBufferToPdfBuffer: hoisted.convertDocxBufferToPdfBuffer, })); diff --git a/tests/unit/document-blob-lease.vitest.spec.ts b/tests/unit/document-blob-lease.vitest.spec.ts new file mode 100644 index 0000000..6923409 --- /dev/null +++ b/tests/unit/document-blob-lease.vitest.spec.ts @@ -0,0 +1,47 @@ +import { beforeEach, describe, expect, test, vi } from 'vitest'; +import Database from 'better-sqlite3'; +import { drizzle } from 'drizzle-orm/better-sqlite3'; +import * as sqliteSchema from '../../src/db/schema_sqlite'; + +const holder = vi.hoisted(() => ({ db: null as unknown as ReturnType })); +vi.mock('@/db', () => ({ + get db() { + return holder.db; + }, +})); + +import { tryAcquireDocumentBlobLease } from '../../src/lib/server/documents/blob-lease'; + +beforeEach(() => { + const sqlite = new Database(':memory:'); + sqlite.exec(`CREATE TABLE document_blob_leases ( + document_id text PRIMARY KEY NOT NULL, + lease_owner text NOT NULL, + lease_until_ms integer NOT NULL + );`); + holder.db = drizzle(sqlite, { schema: sqliteSchema }); +}); + +describe('document blob lease', () => { + test('allows only one owner until the lease is released', async () => { + const first = await tryAcquireDocumentBlobLease('doc-1'); + const blocked = await tryAcquireDocumentBlobLease('doc-1'); + + expect(first).not.toBeNull(); + expect(blocked).toBeNull(); + + await first?.release(); + await expect(tryAcquireDocumentBlobLease('doc-1')).resolves.not.toBeNull(); + }); + + test('allows an expired lease to be reclaimed', async () => { + const first = await tryAcquireDocumentBlobLease('doc-1', { leaseMs: 1 }); + expect(first).not.toBeNull(); + + await new Promise((resolve) => setTimeout(resolve, 2)); + const replacement = await tryAcquireDocumentBlobLease('doc-1'); + + expect(replacement).not.toBeNull(); + expect(replacement?.owner).not.toBe(first?.owner); + }); +}); diff --git a/tests/unit/reap-orphaned-blobs.vitest.spec.ts b/tests/unit/reap-orphaned-blobs.vitest.spec.ts index b6d865e..667a242 100644 --- a/tests/unit/reap-orphaned-blobs.vitest.spec.ts +++ b/tests/unit/reap-orphaned-blobs.vitest.spec.ts @@ -4,6 +4,8 @@ const mocks = vi.hoisted(() => ({ list: vi.fn(), del: vi.fn(async () => undefined), owned: [] as Array<{ id: string }>, + ownedAfterLease: [] as Array<{ id: string }>, + leaseRelease: vi.fn(async () => undefined), })); vi.mock('@/lib/server/storage/s3', () => ({ isS3Configured: () => true })); @@ -17,21 +19,41 @@ vi.mock('@/lib/server/documents/previews-blobstore', () => ({ vi.mock('@/lib/server/documents/previews', () => ({ deleteDocumentPreviewRows: vi.fn(async () => undefined), })); +vi.mock('@/lib/server/documents/blob-lease', () => ({ + tryAcquireDocumentBlobLease: vi.fn(async () => ({ + owner: 'lease-owner', + release: mocks.leaseRelease, + })), +})); +vi.mock('@/lib/server/errors/logging', () => ({ logDegraded: vi.fn() })); vi.mock('@/db', () => ({ db: { - select: () => ({ from: () => ({ where: () => Promise.resolve(mocks.owned) }) }), + select: () => ({ + from: () => ({ + where: () => { + const rows = mocks.owned; + mocks.owned = mocks.ownedAfterLease; + return Object.assign(Promise.resolve(rows), { + limit: () => Promise.resolve(rows.slice(0, 1)), + }); + }, + }), + }), }, })); import { reapOrphanedBlobs } from '../../src/lib/server/tasks/handlers/reap-orphaned-blobs'; const TWO_HOURS = 2 * 60 * 60 * 1000; +const context = () => ({ signal: new AbortController().signal, deadlineAt: Date.now() + 60_000 }); beforeEach(() => { mocks.list.mockReset(); mocks.del.mockReset(); mocks.del.mockResolvedValue(undefined); mocks.owned = []; + mocks.ownedAfterLease = []; + mocks.leaseRelease.mockClear(); }); describe('reap-orphaned-blobs', () => { @@ -44,7 +66,7 @@ describe('reap-orphaned-blobs', () => { ]); mocks.owned = [{ id: 'owned-old' }]; - const result = await reapOrphanedBlobs(); + const result = await reapOrphanedBlobs(context()); expect(mocks.del).toHaveBeenCalledTimes(1); expect(mocks.del).toHaveBeenCalledWith('orphan-old', null); @@ -58,9 +80,30 @@ describe('reap-orphaned-blobs', () => { mocks.list.mockResolvedValue([{ id: 'kept', lastModifiedMs: now - TWO_HOURS }]); mocks.owned = [{ id: 'kept' }]; - const result = await reapOrphanedBlobs(); + const result = await reapOrphanedBlobs(context()); expect(mocks.del).not.toHaveBeenCalled(); expect(result.reaped).toBe(0); }); + + test('does not delete when an owner appears before the post-lease recheck', async () => { + const now = Date.now(); + mocks.list.mockResolvedValue([{ id: 'claimed-during-run', lastModifiedMs: now - TWO_HOURS }]); + mocks.ownedAfterLease = [{ id: 'claimed-during-run' }]; + + const result = await reapOrphanedBlobs(context()); + + expect(mocks.del).not.toHaveBeenCalled(); + expect(result.reaped).toBe(0); + expect(mocks.leaseRelease).toHaveBeenCalledTimes(1); + }); + + test('fails the task when an orphan deletion fails', async () => { + const now = Date.now(); + mocks.list.mockResolvedValue([{ id: 'failed-orphan', lastModifiedMs: now - TWO_HOURS }]); + mocks.del.mockRejectedValue(new Error('storage failed')); + + await expect(reapOrphanedBlobs(context())).rejects.toThrow('Failed to reap 1'); + expect(mocks.leaseRelease).toHaveBeenCalledTimes(1); + }); }); diff --git a/tests/unit/scheduled-tasks-engine.vitest.spec.ts b/tests/unit/scheduled-tasks-engine.vitest.spec.ts index 737fdf8..715bac4 100644 --- a/tests/unit/scheduled-tasks-engine.vitest.spec.ts +++ b/tests/unit/scheduled-tasks-engine.vitest.spec.ts @@ -26,6 +26,7 @@ const CREATE_TABLE = `CREATE TABLE scheduled_tasks ( enabled integer DEFAULT true NOT NULL, interval_ms integer NOT NULL, last_status text DEFAULT 'idle' NOT NULL, + lease_owner text, last_run_at integer, last_duration_ms integer, last_error text, @@ -151,6 +152,77 @@ describe('scheduled task engine', () => { expect((await readRow()).runRequested).toBe(true); }); + test('does not let a stale runner overwrite its replacement run', async () => { + let resolveFirst!: (value: { summary: string }) => void; + const firstResult = new Promise<{ summary: string }>((resolve) => { + resolveFirst = resolve; + }); + const handler = vi.fn() + .mockImplementationOnce(() => firstResult) + .mockResolvedValueOnce({ summary: 'replacement' }); + await seedRow({ nextRunAt: Date.now() - 1 }); + + let now = Date.now(); + const nowSpy = vi.spyOn(Date, 'now').mockImplementation(() => now); + const firstRun = runDueTasks({ registry: registryWith(handler) }); + await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1)); + + now += 2 * 60 * 60 * 1000; + await runDueTasks({ registry: registryWith(handler) }); + expect((await readRow()).lastResultJson).toBe('replacement'); + + resolveFirst({ summary: 'stale original' }); + await firstRun; + expect((await readRow()).lastResultJson).toBe('replacement'); + nowSpy.mockRestore(); + }); + + test('starts independent due tasks concurrently', async () => { + const started: string[] = []; + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const registry: TaskRegistry = { + first: { + name: 'First', + defaultIntervalMs: 1000, + run: async () => { + started.push('first'); + await gate; + }, + }, + second: { + name: 'Second', + defaultIntervalMs: 1000, + run: async () => { + started.push('second'); + await gate; + }, + }, + }; + + const running = runDueTasks({ registry }); + await vi.waitFor(() => expect(started).toEqual(['first', 'second'])); + release(); + await running; + }); + + test('aborts and records an error when a task exceeds its runtime limit', async () => { + const handler = vi.fn(async ({ signal }: { signal: AbortSignal }) => { + await new Promise((resolve) => signal.addEventListener('abort', () => resolve(), { once: true })); + }); + await seedRow({ nextRunAt: Date.now() - 1 }); + const registry: TaskRegistry = { + [KEY]: { name: 'Timed task', defaultIntervalMs: 1000, maxRunMs: 5, run: handler }, + }; + + await runDueTasks({ registry }); + + expect(handler).toHaveBeenCalledTimes(1); + expect((await readRow()).lastError).toContain('runtime limit'); + }); + test('rejects non-positive intervals at the database boundary', async () => { await expect(seedRow({ intervalMs: 0 })).rejects.toThrow(); }); diff --git a/tests/unit/scheduled-tasks-routes.vitest.spec.ts b/tests/unit/scheduled-tasks-routes.vitest.spec.ts new file mode 100644 index 0000000..cfb418f --- /dev/null +++ b/tests/unit/scheduled-tasks-routes.vitest.spec.ts @@ -0,0 +1,98 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { NextRequest } from 'next/server'; + +const mocks = vi.hoisted(() => ({ + requireAdminContext: vi.fn(), + listTasks: vi.fn(), + runDueTasks: vi.fn(), + updateTask: vi.fn(), +})); + +vi.mock('@/lib/server/auth/admin', () => ({ + requireAdminContext: mocks.requireAdminContext, +})); +vi.mock('@/lib/server/tasks/engine', () => ({ + listTasks: mocks.listTasks, + runDueTasks: mocks.runDueTasks, + updateTask: mocks.updateTask, +})); + +describe('scheduled task routes', () => { + const originalCronSecret = process.env.CRON_SECRET; + const originalVercel = process.env.VERCEL; + + beforeEach(() => { + mocks.requireAdminContext.mockReset(); + mocks.requireAdminContext.mockResolvedValue({ userId: 'admin-1' }); + mocks.listTasks.mockReset(); + mocks.listTasks.mockResolvedValue([]); + mocks.runDueTasks.mockReset(); + mocks.runDueTasks.mockResolvedValue(undefined); + mocks.updateTask.mockReset(); + mocks.updateTask.mockResolvedValue(undefined); + delete process.env.CRON_SECRET; + delete process.env.VERCEL; + }); + + afterEach(() => { + if (originalCronSecret === undefined) delete process.env.CRON_SECRET; + else process.env.CRON_SECRET = originalCronSecret; + if (originalVercel === undefined) delete process.env.VERCEL; + else process.env.VERCEL = originalVercel; + }); + + test('cron tick requires the configured bearer secret', async () => { + const { GET } = await import('../../src/app/api/admin/tasks/tick/route'); + + const unconfigured = await GET(new NextRequest('http://localhost/api/admin/tasks/tick')); + expect(unconfigured.status).toBe(503); + + process.env.CRON_SECRET = 'cron-secret'; + const unauthorized = await GET(new NextRequest('http://localhost/api/admin/tasks/tick')); + expect(unauthorized.status).toBe(401); + + const authorized = await GET(new NextRequest('http://localhost/api/admin/tasks/tick', { + headers: { authorization: 'Bearer cron-secret' }, + })); + expect(authorized.status).toBe(200); + expect(mocks.runDueTasks).toHaveBeenCalledTimes(1); + }); + + test('admin task list enforces admin authorization', async () => { + const denied = new Response('Forbidden', { status: 403 }); + mocks.requireAdminContext.mockResolvedValue(denied); + const { GET } = await import('../../src/app/api/admin/tasks/route'); + + const response = await GET(new NextRequest('http://localhost/api/admin/tasks')); + + expect(response).toBe(denied); + expect(mocks.listTasks).not.toHaveBeenCalled(); + }); + + test('reports daily Vercel cadence and rejects shorter intervals', async () => { + process.env.VERCEL = '1'; + mocks.listTasks.mockResolvedValue([{ + key: 'cleanup-temp-uploads', + intervalMs: 60 * 60 * 1000, + }]); + const { GET } = await import('../../src/app/api/admin/tasks/route'); + const listResponse = await GET(new NextRequest('http://localhost/api/admin/tasks')); + await expect(listResponse.json()).resolves.toMatchObject({ + tasks: [{ intervalMs: 86_400_000 }], + scheduler: { + mode: 'vercel-cron', + minimumIntervalMs: 86_400_000, + }, + }); + + const { PATCH } = await import('../../src/app/api/admin/tasks/[key]/route'); + const patchResponse = await PATCH(new NextRequest('http://localhost/api/admin/tasks/cleanup-temp-uploads', { + method: 'PATCH', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ intervalMs: 60 * 60 * 1000 }), + }), { params: Promise.resolve({ key: 'cleanup-temp-uploads' }) }); + + expect(patchResponse.status).toBe(400); + expect(mocks.updateTask).not.toHaveBeenCalled(); + }); +});