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(); + }); +});