danvics/openreader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(tasks): add document blob lease for safe orphan reaping and improve scheduled task robustness

Browse source 
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.
This commit is contained in:
Richard R 2026-06-07 12:08:57 -06:00
parent fdda0df301
commit 848ee4faad
34 changed files with 708 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.env.example
View file

@ -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=

11
docs-site/docs/configure/admin-panel.md
View file

@ -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:

2
docs-site/docs/configure/tts-rate-limiting.md
View file

@ -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)

4
docs-site/docs/deploy/local-development.md
View file

@ -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.
<Tabs groupId="local-run-mode">

18
docs-site/docs/deploy/vercel-deployment.md
View file

@ -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://<railway-worker-domain>
@ -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.

1
docs-site/docs/docker-quick-start.md
View file

@ -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

10
docs-site/docs/reference/environment-variables.md
View file

@ -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 <CRON_SECRET>` 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

7
drizzle/postgres/0011_scheduled-tasks.sql
View file

@ -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,

39
drizzle/postgres/meta/0011_snapshot.json
View file

@ -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",

2
drizzle/postgres/meta/_journal.json
View file

@ -82,7 +82,7 @@
{
"idx": 11,
"version": "7",
"when": 1780847571383,
"when": 1780851107819,
"tag": "0011_scheduled-tasks",
"breakpoints": true
}

7
drizzle/sqlite/0011_scheduled-tasks.sql
View file

@ -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,

40
drizzle/sqlite/meta/0011_snapshot.json
View file

@ -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",

2
drizzle/sqlite/meta/_journal.json
View file

@ -82,7 +82,7 @@
{
"idx": 11,
"version": "6",
"when": 1780847571062,
"when": 1780851107505,
"tag": "0011_scheduled-tasks",
"breakpoints": true
}

8
src/app/api/admin/tasks/[key]/route.ts
View file

@ -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 });

1
src/app/api/admin/tasks/[key]/run/route.ts
View file

@ -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,

9
src/app/api/admin/tasks/route.ts
View file

@ -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<Response> {
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 });
}

1
src/app/api/admin/tasks/tick/route.ts
View file

@ -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

86
src/app/api/documents/blob/upload/finalize/route.ts
View file

@ -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(

39
src/components/admin/AdminTasksPanel.tsx
View file

@ -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<TaskView[]> {
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 (
<Section title="Scheduled tasks" subtitle="Background maintenance jobs. Run them on demand or adjust their schedule.">
{data?.scheduler.mode === 'vercel-cron' && (
<p className="mb-2 text-xs text-soft">
Vercel Hobby invokes scheduled tasks once daily. Shorter intervals are unavailable on this deployment.
</p>
)}
<div className="space-y-2">
{(data ?? []).map((task) => (
{(data?.tasks ?? []).map((task) => (
<TaskRow
key={task.key}
task={task}
schedulerMode={data?.scheduler.mode ?? 'self-hosted'}
minimumIntervalMs={data?.scheduler.minimumIntervalMs ?? 60_000}
busy={patchTask.isPending || runTask.isPending}
runPending={runTask.isPending && runTask.variables === task.key}
onToggle={(enabled) => 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 && (
<p className="text-sm text-soft">No tasks registered.</p>
)}
</div>
@ -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({
<ClockIcon className="size-3 text-faint" />
{formatRelative(task.lastRunAt)}
</span>
{task.enabled && task.nextRunAt != null && (
{task.enabled && schedulerMode === 'vercel-cron' && (
<span className="inline-flex items-center gap-1 text-faint">
<RefreshIcon className="size-3" />
next daily cron
</span>
)}
{task.enabled && schedulerMode !== 'vercel-cron' && task.nextRunAt != null && (
<span className="inline-flex items-center gap-1 text-faint">
<RefreshIcon className="size-3" />
next {formatRelative(task.nextRunAt)}
@ -199,7 +224,7 @@ function TaskRow({
<span>Every</span>
<Input
type="number"
min={0.001}
min={minimumIntervalMs / 60000}
step="any"
controlSize="sm"
className="w-14 text-center"

1
src/db/schema.ts
View file

@ -20,3 +20,4 @@ export const ttsSegmentVariants = usePostgres ? postgresSchema.ttsSegmentVariant
export const adminProviders = usePostgres ? postgresSchema.adminProviders : sqliteSchema.adminProviders;
export const adminSettings = usePostgres ? postgresSchema.adminSettings : sqliteSchema.adminSettings;
export const scheduledTasks = usePostgres ? postgresSchema.scheduledTasks : sqliteSchema.scheduledTasks;
export const documentBlobLeases = usePostgres ? postgresSchema.documentBlobLeases : sqliteSchema.documentBlobLeases;

7
src/db/schema_postgres.ts
View file

@ -215,6 +215,7 @@ export const scheduledTasks = pgTable('scheduled_tasks', {
enabled: boolean('enabled').notNull().default(true),
intervalMs: bigint('interval_ms', { mode: 'number' }).notNull(),
lastStatus: text('last_status').notNull().default('idle'),
leaseOwner: text('lease_owner'),
lastRunAt: bigint('last_run_at', { mode: 'number' }),
lastDurationMs: bigint('last_duration_ms', { mode: 'number' }),
lastError: text('last_error'),
@ -227,6 +228,12 @@ export const scheduledTasks = pgTable('scheduled_tasks', {
check('scheduled_tasks_interval_ms_positive', sql`${table.intervalMs} > 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' }),

7
src/db/schema_sqlite.ts
View file

@ -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' }),

75
src/lib/server/documents/blob-lease.ts Normal file
View file

@ -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<void>;
};
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
export async function tryAcquireDocumentBlobLease(
documentId: string,
options?: { leaseMs?: number },
): Promise<DocumentBlobLease | null> {
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<T>(
documentId: string,
fn: () => Promise<T>,
options?: { waitMs?: number; leaseMs?: number },
): Promise<T> {
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();
}
}

7
src/lib/server/documents/blobstore.ts
View file

@ -607,6 +607,7 @@ export async function deleteDocumentPrefix(prefix: string): Promise<number> {
*/
export async function listDocumentSourceBlobs(
namespace: string | null,
options?: { signal?: AbortSignal },
): Promise<Array<{ id: string; lastModifiedMs: number }>> {
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<number> {
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(

74
src/lib/server/tasks/engine.ts
View file

@ -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<v
/**
* Atomically claim a task for execution. Returns true only if this caller won
* the claim, so the same task never runs concurrently (across instances too):
* the single UPDATE flips status to 'running' iff it is not already running
* (or its running marker is stale).
* the claim. The single UPDATE flips status to 'running' iff it is not already
* running (or its running marker is stale). A unique owner token fences final
* state updates if a stale runner later completes after being replaced.
*/
async function claimTask(key: string, now: number): Promise<boolean> {
async function claimTask(key: string, now: number): Promise<string | null> {
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<boolean> {
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<void> {
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<void> {
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<typeof setTimeout> | undefined;
try {
const result = await def.run();
const result = await Promise.race([
def.run(context),
new Promise<never>((_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<void> {
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<void>[] = [];
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;
}

6
src/lib/server/tasks/handlers/cleanup-temp-uploads.ts
View file

@ -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<TaskResult> {
export async function cleanupTempUploads(context: TaskContext): Promise<TaskResult> {
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 };
}

26
src/lib/server/tasks/handlers/reap-orphaned-blobs.ts
View file

@ -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<TaskResult> {
export async function reapOrphanedBlobs(context: TaskContext): Promise<TaskResult> {
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<TaskResult> {
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<TaskResult> {
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,

22
src/lib/server/tasks/scheduler.ts
View file

@ -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;

9
src/lib/server/tasks/types.ts
View file

@ -6,7 +6,12 @@ export type TaskResult = {
[key: string]: unknown;
};
export type TaskHandler = () => Promise<TaskResult | void>;
export type TaskContext = {
signal: AbortSignal;
deadlineAt: number;
};
export type TaskHandler = (context: TaskContext) => Promise<TaskResult | void>;
/** 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;
};

4
tests/unit/blob-upload-finalize-docx.vitest.spec.ts
View file

@ -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<unknown>) => fn()),
}));
vi.mock('@/lib/server/documents/docx-convert', () => ({
convertDocxBufferToPdfBuffer: hoisted.convertDocxBufferToPdfBuffer,
}));

47
tests/unit/document-blob-lease.vitest.spec.ts Normal file
View file

@ -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<typeof drizzle> }));
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);
});
});

49
tests/unit/reap-orphaned-blobs.vitest.spec.ts
View file

@ -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);
});
});

72
tests/unit/scheduled-tasks-engine.vitest.spec.ts
View file

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

98
tests/unit/scheduled-tasks-routes.vitest.spec.ts Normal file
View file

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