From 2fb36a24c7f1cdb5ac51cee021bedf50b3d8244a Mon Sep 17 00:00:00 2001 From: Richard R Date: Sun, 7 Jun 2026 09:54:18 -0600 Subject: [PATCH] refactor(tasks): enforce positive interval and improve scheduled task updates Add database-level check constraints to ensure scheduled task intervals are always positive for both Postgres and SQLite. Update admin API and UI to support sub-minute intervals and stricter validation. Refactor scheduled task update logic to upsert rows, ensuring tasks can be updated even if not yet present in the database. Improve temporary upload cleanup to delete in paginated batches. Enhance tests to cover new constraints and update behaviors. --- drizzle/postgres/0011_scheduled-tasks.sql | 3 +- drizzle/postgres/meta/0011_snapshot.json | 9 +++- drizzle/postgres/meta/_journal.json | 2 +- drizzle/sqlite/0011_scheduled-tasks.sql | 3 +- drizzle/sqlite/meta/0011_snapshot.json | 9 +++- drizzle/sqlite/meta/_journal.json | 2 +- src/app/api/admin/tasks/[key]/route.ts | 25 ++++++++-- src/app/api/admin/tasks/[key]/run/route.ts | 2 +- src/components/admin/AdminTasksPanel.tsx | 14 +++--- src/db/schema_postgres.ts | 6 ++- src/db/schema_sqlite.ts | 6 ++- src/lib/server/documents/blobstore.ts | 42 +++++++++-------- src/lib/server/tasks/engine.ts | 24 ++++++++-- src/lib/server/user/claim-data.ts | 23 ++++++++-- .../scheduled-tasks-engine.vitest.spec.ts | 38 ++++++++++++++- .../storage-prefix-cleanup.vitest.spec.ts | 46 ++++++++++++++++++- 16 files changed, 198 insertions(+), 56 deletions(-) diff --git a/drizzle/postgres/0011_scheduled-tasks.sql b/drizzle/postgres/0011_scheduled-tasks.sql index 8eef94b..cb05f2f 100644 --- a/drizzle/postgres/0011_scheduled-tasks.sql +++ b/drizzle/postgres/0011_scheduled-tasks.sql @@ -10,5 +10,6 @@ CREATE TABLE "scheduled_tasks" ( "next_run_at" bigint, "run_requested" boolean DEFAULT false NOT NULL, "running_since" bigint, - "updated_at" bigint DEFAULT (extract(epoch from now()) * 1000)::bigint NOT NULL + "updated_at" bigint DEFAULT (extract(epoch from now()) * 1000)::bigint NOT NULL, + CONSTRAINT "scheduled_tasks_interval_ms_positive" CHECK ("scheduled_tasks"."interval_ms" > 0) ); diff --git a/drizzle/postgres/meta/0011_snapshot.json b/drizzle/postgres/meta/0011_snapshot.json index 33c6003..443520f 100644 --- a/drizzle/postgres/meta/0011_snapshot.json +++ b/drizzle/postgres/meta/0011_snapshot.json @@ -1,5 +1,5 @@ { - "id": "195ce074-f013-420d-9238-30c32f38365e", + "id": "afe1d934-9b20-4d50-82fc-aeebd49fe43f", "prevId": "1099cbe6-0c2e-4ea8-ae3c-6883c931169d", "version": "7", "dialect": "postgresql", @@ -780,7 +780,12 @@ "compositePrimaryKeys": {}, "uniqueConstraints": {}, "policies": {}, - "checkConstraints": {}, + "checkConstraints": { + "scheduled_tasks_interval_ms_positive": { + "name": "scheduled_tasks_interval_ms_positive", + "value": "\"scheduled_tasks\".\"interval_ms\" > 0" + } + }, "isRLSEnabled": false }, "public.tts_segment_entries": { diff --git a/drizzle/postgres/meta/_journal.json b/drizzle/postgres/meta/_journal.json index 488ccd2..5731fec 100644 --- a/drizzle/postgres/meta/_journal.json +++ b/drizzle/postgres/meta/_journal.json @@ -82,7 +82,7 @@ { "idx": 11, "version": "7", - "when": 1780825886154, + "when": 1780847571383, "tag": "0011_scheduled-tasks", "breakpoints": true } diff --git a/drizzle/sqlite/0011_scheduled-tasks.sql b/drizzle/sqlite/0011_scheduled-tasks.sql index af6466c..cca23bd 100644 --- a/drizzle/sqlite/0011_scheduled-tasks.sql +++ b/drizzle/sqlite/0011_scheduled-tasks.sql @@ -10,5 +10,6 @@ CREATE TABLE `scheduled_tasks` ( `next_run_at` integer, `run_requested` integer DEFAULT false NOT NULL, `running_since` integer, - `updated_at` integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL + `updated_at` integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL, + CONSTRAINT "scheduled_tasks_interval_ms_positive" CHECK("scheduled_tasks"."interval_ms" > 0) ); diff --git a/drizzle/sqlite/meta/0011_snapshot.json b/drizzle/sqlite/meta/0011_snapshot.json index 66b1549..45d8f1f 100644 --- a/drizzle/sqlite/meta/0011_snapshot.json +++ b/drizzle/sqlite/meta/0011_snapshot.json @@ -1,7 +1,7 @@ { "version": "6", "dialect": "sqlite", - "id": "504118c7-02f7-4fc5-806d-4a369e415173", + "id": "4c4a203e-9b2b-4eb2-b31c-2613a7c1d377", "prevId": "a32ae7d4-c074-44f0-9e90-2473949376ef", "tables": { "admin_providers": { @@ -791,7 +791,12 @@ "foreignKeys": {}, "compositePrimaryKeys": {}, "uniqueConstraints": {}, - "checkConstraints": {} + "checkConstraints": { + "scheduled_tasks_interval_ms_positive": { + "name": "scheduled_tasks_interval_ms_positive", + "value": "\"scheduled_tasks\".\"interval_ms\" > 0" + } + } }, "tts_segment_entries": { "name": "tts_segment_entries", diff --git a/drizzle/sqlite/meta/_journal.json b/drizzle/sqlite/meta/_journal.json index 5546859..513ea7f 100644 --- a/drizzle/sqlite/meta/_journal.json +++ b/drizzle/sqlite/meta/_journal.json @@ -82,7 +82,7 @@ { "idx": 11, "version": "6", - "when": 1780825885684, + "when": 1780847571062, "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 cb302f9..4ac2460 100644 --- a/src/app/api/admin/tasks/[key]/route.ts +++ b/src/app/api/admin/tasks/[key]/route.ts @@ -13,7 +13,7 @@ export async function PATCH( if (ctx instanceof Response) return ctx; const { key } = await params; - if (!(key in TASK_REGISTRY)) { + if (!Object.hasOwn(TASK_REGISTRY, key)) { return NextResponse.json({ error: 'Unknown task' }, { status: 404 }); } @@ -27,11 +27,26 @@ export async function PATCH( return NextResponse.json({ error: 'Expected JSON object' }, { status: 400 }); } - const { enabled, intervalMs } = body as { enabled?: unknown; intervalMs?: unknown }; + const payload = body as { enabled?: unknown; intervalMs?: unknown }; const patch: { enabled?: boolean; intervalMs?: number } = {}; - if (typeof enabled === 'boolean') patch.enabled = enabled; - if (typeof intervalMs === 'number' && Number.isFinite(intervalMs) && intervalMs > 0) { - patch.intervalMs = Math.floor(intervalMs); + if (Object.hasOwn(payload, 'enabled')) { + if (typeof payload.enabled !== 'boolean') { + return NextResponse.json({ error: 'enabled must be a boolean' }, { status: 400 }); + } + patch.enabled = payload.enabled; + } + if (Object.hasOwn(payload, 'intervalMs')) { + if ( + typeof payload.intervalMs !== 'number' + || !Number.isFinite(payload.intervalMs) + || payload.intervalMs <= 0 + ) { + return NextResponse.json({ error: 'intervalMs must be a finite positive number' }, { status: 400 }); + } + patch.intervalMs = Math.max(1, Math.floor(payload.intervalMs)); + } + if (Object.keys(patch).length === 0) { + return NextResponse.json({ error: 'Expected enabled or intervalMs' }, { status: 400 }); } await updateTask(key, patch); diff --git a/src/app/api/admin/tasks/[key]/run/route.ts b/src/app/api/admin/tasks/[key]/run/route.ts index 5c37fc2..8b67a8a 100644 --- a/src/app/api/admin/tasks/[key]/run/route.ts +++ b/src/app/api/admin/tasks/[key]/run/route.ts @@ -13,7 +13,7 @@ export async function POST( if (ctx instanceof Response) return ctx; const { key } = await params; - if (!(key in TASK_REGISTRY)) { + if (!Object.hasOwn(TASK_REGISTRY, key)) { return NextResponse.json({ error: 'Unknown task' }, { status: 404 }); } diff --git a/src/components/admin/AdminTasksPanel.tsx b/src/components/admin/AdminTasksPanel.tsx index 3affd25..e5a232c 100644 --- a/src/components/admin/AdminTasksPanel.tsx +++ b/src/components/admin/AdminTasksPanel.tsx @@ -141,15 +141,16 @@ function TaskRow({ onSaveInterval: (intervalMs: number) => void; onRun: () => void; }) { - const [minutes, setMinutes] = useState(String(Math.round(task.intervalMs / 60000))); + const [minutes, setMinutes] = useState(String(task.intervalMs / 60000)); useEffect(() => { - setMinutes(String(Math.round(task.intervalMs / 60000))); + setMinutes(String(task.intervalMs / 60000)); }, [task.intervalMs]); - const parsedMinutes = Number(minutes); + const parsedMinutes = parseFloat(minutes); + const newIntervalMs = parsedMinutes * 60000; const intervalDirty = - Number.isFinite(parsedMinutes) && parsedMinutes > 0 && parsedMinutes * 60000 !== task.intervalMs; + Number.isFinite(parsedMinutes) && parsedMinutes > 0 && newIntervalMs !== task.intervalMs; return (
@@ -177,7 +178,8 @@ function TaskRow({ Every onSaveInterval(Math.floor(parsedMinutes) * 60000)} + onClick={() => onSaveInterval(newIntervalMs)} disabled={busy} > Save diff --git a/src/db/schema_postgres.ts b/src/db/schema_postgres.ts index 5171903..d771d87 100644 --- a/src/db/schema_postgres.ts +++ b/src/db/schema_postgres.ts @@ -1,5 +1,5 @@ import { sql } from 'drizzle-orm'; -import { pgTable, text, integer, real, date, bigint, boolean, primaryKey, index, jsonb, foreignKey } from 'drizzle-orm/pg-core'; +import { pgTable, text, integer, real, date, bigint, boolean, primaryKey, index, jsonb, foreignKey, check } from 'drizzle-orm/pg-core'; import { user } from './schema_auth_postgres'; const PG_NOW_MS = sql`(extract(epoch from now()) * 1000)::bigint`; @@ -223,7 +223,9 @@ export const scheduledTasks = pgTable('scheduled_tasks', { runRequested: boolean('run_requested').notNull().default(false), runningSince: bigint('running_since', { mode: 'number' }), updatedAt: bigint('updated_at', { mode: 'number' }).notNull().default(PG_NOW_MS), -}); +}, (table) => [ + check('scheduled_tasks_interval_ms_positive', sql`${table.intervalMs} > 0`), +]); export const ttsSegmentVariants = pgTable('tts_segment_variants', { segmentId: text('segment_id').notNull(), diff --git a/src/db/schema_sqlite.ts b/src/db/schema_sqlite.ts index 78f0588..2ae1c54 100644 --- a/src/db/schema_sqlite.ts +++ b/src/db/schema_sqlite.ts @@ -1,4 +1,4 @@ -import { sqliteTable, text, integer, real, primaryKey, index, foreignKey } from 'drizzle-orm/sqlite-core'; +import { sqliteTable, text, integer, real, primaryKey, index, foreignKey, check } from 'drizzle-orm/sqlite-core'; import { sql } from 'drizzle-orm'; import { user } from './schema_auth_sqlite'; @@ -223,7 +223,9 @@ export const scheduledTasks = sqliteTable('scheduled_tasks', { runRequested: integer('run_requested', { mode: 'boolean' }).notNull().default(false), runningSince: integer('running_since'), updatedAt: integer('updated_at').notNull().default(SQLITE_NOW_MS), -}); +}, (table) => [ + check('scheduled_tasks_interval_ms_positive', sql`${table.intervalMs} > 0`), +]); export const ttsSegmentVariants = sqliteTable('tts_segment_variants', { segmentId: text('segment_id').notNull(), diff --git a/src/lib/server/documents/blobstore.ts b/src/lib/server/documents/blobstore.ts index 438f3b8..61ceb06 100644 --- a/src/lib/server/documents/blobstore.ts +++ b/src/lib/server/documents/blobstore.ts @@ -654,7 +654,7 @@ export async function deleteAllExpiredTempDocumentUploads( const nsSegment = ns ? `ns/${ns}/` : ''; const prefix = `${cfg.prefix}/document_uploads_temp_v1/${nsSegment}`; let continuationToken: string | undefined; - const keys: string[] = []; + let deleted = 0; do { const listRes = await client.send( @@ -665,33 +665,35 @@ export async function deleteAllExpiredTempDocumentUploads( }), ); + const batch: string[] = []; for (const item of listRes.Contents ?? []) { const key = item.Key; const lastModified = item.LastModified?.getTime() ?? 0; if (!key || lastModified <= 0 || lastModified >= olderThanMs) continue; - keys.push(key); + batch.push(key); + } + + if (batch.length > 0) { + const deleteRes = await client.send( + new DeleteObjectsCommand({ + Bucket: cfg.bucket, + Delete: { + Objects: batch.map((Key) => ({ Key })), + Quiet: true, + }, + }), + ); + if (deleteRes.Errors?.length) { + throw new Error( + `Failed to delete temporary uploads from bucket "${cfg.bucket}" ` + + `(keys: ${JSON.stringify(batch)}, errors: ${JSON.stringify(deleteRes.Errors)})`, + ); + } + deleted += batch.length; } continuationToken = listRes.IsTruncated ? listRes.NextContinuationToken : undefined; } while (continuationToken); - if (keys.length === 0) return 0; - - let deleted = 0; - for (let i = 0; i < keys.length; i += 1000) { - const batch = keys.slice(i, i + 1000); - const deleteRes = await client.send( - new DeleteObjectsCommand({ - Bucket: cfg.bucket, - Delete: { - Objects: batch.map((Key) => ({ Key })), - Quiet: true, - }, - }), - ); - deleted += deleteRes.Deleted?.length ?? batch.length - (deleteRes.Errors?.length ?? 0); - } - return deleted; } - diff --git a/src/lib/server/tasks/engine.ts b/src/lib/server/tasks/engine.ts index ac198ee..aecc113 100644 --- a/src/lib/server/tasks/engine.ts +++ b/src/lib/server/tasks/engine.ts @@ -69,7 +69,6 @@ async function finishTask( // Schedule the next run off the row's (possibly user-edited) interval. nextRunAt: sql`${now} + ${scheduledTasks.intervalMs}`, runningSince: null, - runRequested: false, updatedAt: now, }) .where(eq(scheduledTasks.key, key)); @@ -136,12 +135,29 @@ export async function updateTask( key: string, patch: { enabled?: boolean; intervalMs?: number }, ): Promise { - const set: Record = { updatedAt: Date.now() }; + const def = Object.hasOwn(TASK_REGISTRY, key) ? TASK_REGISTRY[key] : undefined; + if (!def) return; + + const now = Date.now(); + const set: Record = { updatedAt: now }; if (typeof patch.enabled === 'boolean') set.enabled = patch.enabled; if (typeof patch.intervalMs === 'number' && Number.isFinite(patch.intervalMs) && patch.intervalMs > 0) { - set.intervalMs = Math.floor(patch.intervalMs); + set.intervalMs = Math.max(1, Math.floor(patch.intervalMs)); } - await db.update(scheduledTasks).set(set).where(eq(scheduledTasks.key, key)); + await db + .insert(scheduledTasks) + .values({ + key, + enabled: typeof set.enabled === 'boolean' ? set.enabled : true, + intervalMs: typeof set.intervalMs === 'number' ? set.intervalMs : def.defaultIntervalMs, + lastStatus: 'idle', + nextRunAt: now, + updatedAt: now, + }) + .onConflictDoUpdate({ + target: scheduledTasks.key, + set, + }); } export type TaskView = { diff --git a/src/lib/server/user/claim-data.ts b/src/lib/server/user/claim-data.ts index 28167fd..47a52c9 100644 --- a/src/lib/server/user/claim-data.ts +++ b/src/lib/server/user/claim-data.ts @@ -499,15 +499,28 @@ export async function transferUserDocumentSettings( .select() .from(documentSettings) .where(eq(documentSettings.userId, fromUserId)); - for (const row of rows) { - const [existing] = await database - .select({ clientUpdatedAtMs: documentSettings.clientUpdatedAtMs }) + const documentIds = rows.map((row: { documentId: string }) => row.documentId); + const existingRows = documentIds.length > 0 + ? await database + .select({ + documentId: documentSettings.documentId, + clientUpdatedAtMs: documentSettings.clientUpdatedAtMs, + }) .from(documentSettings) .where(and( eq(documentSettings.userId, toUserId), - eq(documentSettings.documentId, row.documentId), + inArray(documentSettings.documentId, documentIds), )) - .limit(1); + : []; + const existingByDocumentId = new Map( + existingRows.map((row: { documentId: string; clientUpdatedAtMs: number | null }) => [ + row.documentId, + row, + ] as const), + ); + + for (const row of rows) { + const existing = existingByDocumentId.get(row.documentId); if (existing && Number(existing.clientUpdatedAtMs ?? 0) >= Number(row.clientUpdatedAtMs ?? 0)) { continue; } diff --git a/tests/unit/scheduled-tasks-engine.vitest.spec.ts b/tests/unit/scheduled-tasks-engine.vitest.spec.ts index c7587de..737fdf8 100644 --- a/tests/unit/scheduled-tasks-engine.vitest.spec.ts +++ b/tests/unit/scheduled-tasks-engine.vitest.spec.ts @@ -16,7 +16,7 @@ vi.mock('@/db', () => ({ // Keep the error-path test from printing the expected failure to the console. vi.mock('@/lib/server/errors/logging', () => ({ logDegraded: vi.fn() })); -import { runDueTasks } from '../../src/lib/server/tasks/engine'; +import { runDueTasks, updateTask } from '../../src/lib/server/tasks/engine'; import type { TaskRegistry } from '../../src/lib/server/tasks/types'; const tasks = sqliteSchema.scheduledTasks; @@ -33,7 +33,8 @@ const CREATE_TABLE = `CREATE TABLE scheduled_tasks ( next_run_at integer, run_requested integer DEFAULT false NOT NULL, running_since integer, - updated_at integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL + updated_at integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL, + CONSTRAINT scheduled_tasks_interval_ms_positive CHECK(interval_ms > 0) );`; const KEY = 'test-task'; @@ -134,4 +135,37 @@ describe('scheduled task engine', () => { expect(handler).toHaveBeenCalledTimes(1); }); + + test('preserves a manual rerun requested while a task is running', async () => { + const handler = vi.fn(async () => { + await holder.db + .update(tasks) + .set({ runRequested: true }) + .where(eq(tasks.key, KEY)); + }); + await seedRow({ nextRunAt: Date.now() - 1 }); + + await runDueTasks({ registry: registryWith(handler) }); + + expect(handler).toHaveBeenCalledTimes(1); + expect((await readRow()).runRequested).toBe(true); + }); + + test('rejects non-positive intervals at the database boundary', async () => { + await expect(seedRow({ intervalMs: 0 })).rejects.toThrow(); + }); + + test('updates a registered task even when its row has not been seeded', async () => { + await updateTask('prune-job-events', { enabled: false, intervalMs: 12_345 }); + + const [row] = await holder.db + .select() + .from(tasks) + .where(eq(tasks.key, 'prune-job-events')); + expect(row).toEqual(expect.objectContaining({ + enabled: false, + intervalMs: 12_345, + lastStatus: 'idle', + })); + }); }); diff --git a/tests/unit/storage-prefix-cleanup.vitest.spec.ts b/tests/unit/storage-prefix-cleanup.vitest.spec.ts index 6267527..7afb7de 100644 --- a/tests/unit/storage-prefix-cleanup.vitest.spec.ts +++ b/tests/unit/storage-prefix-cleanup.vitest.spec.ts @@ -11,7 +11,11 @@ vi.mock('@/lib/server/storage/s3', () => ({ })); import { deleteAudiobookPrefix } from '../../src/lib/server/audiobooks/blobstore'; -import { deleteDocumentBlob, deleteDocumentPrefix } from '../../src/lib/server/documents/blobstore'; +import { + deleteAllExpiredTempDocumentUploads, + deleteDocumentBlob, + deleteDocumentPrefix, +} from '../../src/lib/server/documents/blobstore'; describe('storage prefix cleanup', () => { beforeEach(() => { @@ -85,4 +89,44 @@ describe('storage prefix cleanup', () => { && command.input.Key === `openreader-test/documents_v1/${documentId}`); expect(sourceDelete).toBeUndefined(); }); + + test('deletes expired temporary uploads page by page', async () => { + const old = new Date(Date.now() - 60_000); + mocks.send + .mockResolvedValueOnce({ + Contents: [{ Key: 'temp/a', LastModified: old }], + IsTruncated: true, + NextContinuationToken: 'next', + }) + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({ + Contents: [{ Key: 'temp/b', LastModified: old }], + IsTruncated: false, + }) + .mockResolvedValueOnce({}); + + await expect(deleteAllExpiredTempDocumentUploads(null, Date.now())).resolves.toBe(2); + + expect(mocks.send.mock.calls.map(([command]) => command.constructor.name)).toEqual([ + 'ListObjectsV2Command', + 'DeleteObjectsCommand', + 'ListObjectsV2Command', + 'DeleteObjectsCommand', + ]); + }); + + test('temporary upload cleanup fails on per-object storage errors', async () => { + const old = new Date(Date.now() - 60_000); + mocks.send + .mockResolvedValueOnce({ + Contents: [{ Key: 'temp/a', LastModified: old }], + IsTruncated: false, + }) + .mockResolvedValueOnce({ + Errors: [{ Key: 'temp/a', Code: 'AccessDenied' }], + }); + + await expect(deleteAllExpiredTempDocumentUploads(null, Date.now())) + .rejects.toThrow('test-bucket'); + }); });