diff --git a/.changeset/native-worker-maintenance.md b/.changeset/native-worker-maintenance.md new file mode 100644 index 00000000..85ccdf63 --- /dev/null +++ b/.changeset/native-worker-maintenance.md @@ -0,0 +1,8 @@ +--- +"@luzzle/web.worker": patch +--- + +Schedule preview-data cleanup with OpenWorkflow cron and durable jitter near +02:00 UTC instead of a startup run and process-local timer. Remove expired preview +files before their application progress/log rows, retain active jobs and engine +history, and remove the unused duplicate cleanup step. diff --git a/docs/web/deployment.md b/docs/web/deployment.md index e907c2e1..b21b658c 100644 --- a/docs/web/deployment.md +++ b/docs/web/deployment.md @@ -67,6 +67,13 @@ To update, back up your archive and deployment data, change `LUZZLE_VERSION` if pinned, then run `docker compose pull` and `docker compose up -d`. [Republish affected pieces](templates.md#apply-your-choices) after OG changes. +## Preview cleanup + +The worker cleans expired preview files and app logs daily, near **02:00 UTC +±12 minutes**. It waits through OpenWorkflow, so other jobs can run during the +random delay. There is no extra cleanup run at startup; downtime or a busy worker +can delay it. Finished-job data is kept for seven days. Workflow history stays. + ## Optional remote sync Put your rclone configuration in `rclone/rclone.conf`. Set `sync.archive` for diff --git a/packages/web/jobs/src/specs.ts b/packages/web/jobs/src/specs.ts index 52b379c4..cfddab1b 100644 --- a/packages/web/jobs/src/specs.ts +++ b/packages/web/jobs/src/specs.ts @@ -17,6 +17,14 @@ export const jobProgressPurgeSpec = defineWorkflowSpec< JobProgressPurgeResult >({ name: 'JobProgressPurge', + schedules: [ + { + id: 'luzzle:job-progress-purge', + cron: '48 1 * * *', + timezone: 'UTC', + input: { retentionDays: 7 }, + }, + ], }) export const previewSpec = defineWorkflowSpec({ diff --git a/packages/web/worker/src/core/job-progress.test.ts b/packages/web/worker/src/core/job-progress.test.ts index 22ad68b1..b8d59215 100644 --- a/packages/web/worker/src/core/job-progress.test.ts +++ b/packages/web/worker/src/core/job-progress.test.ts @@ -1,119 +1,113 @@ -/* eslint-disable @typescript-eslint/no-explicit-any */ import { describe, it, expect, beforeEach, afterEach } from 'vitest' -import { setupDatabase, teardownDatabase } from '../../test/db.js' +import { createAppDb, runWebMigrations } from '@luzzle/web.db' +import { sql } from 'kysely' import { JobProgress } from './job-progress.js' -import type { Kysely } from 'kysely' -import type { WebDatabase } from '../services/db.js' describe('JobProgress', () => { - let testDb: Kysely + let testDb: ReturnType let jobProgress: JobProgress beforeEach(async () => { - testDb = await setupDatabase() - jobProgress = new JobProgress(testDb, 2) + testDb = createAppDb(':memory:') + expect((await runWebMigrations(testDb)).error).toBeUndefined() + jobProgress = new JobProgress(testDb) }) afterEach(async () => { - await teardownDatabase(testDb) + await testDb.destroy() }) - it('should start a phase', async () => { + it('starts and restarts a phase', async () => { await jobProgress.start('42', 'archive.sync') - - const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute() + await jobProgress.complete('42', 'archive.sync', 'done') + await jobProgress.start('42', 'archive.sync') + const rows = await testDb.selectFrom('job_progress').selectAll().execute() expect(rows).toHaveLength(1) expect(rows[0]).toMatchObject({ job_id: '42', phase: 'archive.sync', status: 'running', message: null, - finished_at: null + finished_at: null, }) expect(rows[0].started_at).toBeGreaterThan(0) }) - it('should complete a phase', async () => { + it('completes a phase', async () => { await jobProgress.start('42', 'archive.sync') await jobProgress.complete('42', 'archive.sync', 'completed successfully') - - const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute() - expect(rows).toHaveLength(1) - expect(rows[0]).toMatchObject({ - job_id: '42', - phase: 'archive.sync', - status: 'completed', - message: 'completed successfully' - }) - expect(rows[0].finished_at).toBeGreaterThan(0) + const row = await testDb.selectFrom('job_progress').selectAll().executeTakeFirstOrThrow() + expect(row).toMatchObject({ status: 'completed', message: 'completed successfully' }) + expect(row.finished_at).toBeGreaterThan(0) }) - it('should skip a phase', async () => { + it('skips a phase', async () => { await jobProgress.start('42', 'archive.sync') await jobProgress.skip('42', 'archive.sync', 'no remote configured') - - const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute() - expect(rows).toHaveLength(1) - expect(rows[0]).toMatchObject({ - job_id: '42', - phase: 'archive.sync', - status: 'skipped', - message: 'no remote configured' - }) - expect(rows[0].finished_at).toBeGreaterThan(0) + const row = await testDb.selectFrom('job_progress').selectAll().executeTakeFirstOrThrow() + expect(row).toMatchObject({ status: 'skipped', message: 'no remote configured' }) + expect(row.finished_at).toBeGreaterThan(0) }) - it('should fail a phase with an Error', async () => { + it.each([new Error('network error'), 'network error'])('fails a phase: %s', async (error) => { await jobProgress.start('42', 'archive.sync') - await jobProgress.fail('42', 'archive.sync', new Error('network error')) - - const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute() - expect(rows).toHaveLength(1) - expect(rows[0]).toMatchObject({ - job_id: '42', - phase: 'archive.sync', - status: 'failed', - message: 'network error' - }) - expect(rows[0].finished_at).toBeGreaterThan(0) + await jobProgress.fail('42', 'archive.sync', error) + const row = await testDb.selectFrom('job_progress').selectAll().executeTakeFirstOrThrow() + expect(row).toMatchObject({ status: 'failed', message: 'network error' }) + expect(row.finished_at).toBeGreaterThan(0) }) - it('should fail a phase with a string', async () => { - await jobProgress.start('42', 'archive.sync') - await jobProgress.fail('42', 'archive.sync', 'string error') + async function log(jobId: string) { + await testDb + .insertInto('job_progress_logs') + .values({ + job_id: jobId, + phase: 'render', + line_number: 1, + ts: Date.now(), + level: 'info', + message: 'log', + }) + .execute() + } - const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute() - expect(rows).toHaveLength(1) - expect(rows[0]).toMatchObject({ - job_id: '42', - phase: 'archive.sync', - status: 'failed', - message: 'string error' - }) - expect(rows[0].finished_at).toBeGreaterThan(0) + it('lists the distinct union of progress and log IDs, including logs without progress', async () => { + await jobProgress.start('both', 'one') + await jobProgress.start('both', 'two') + await jobProgress.start('progress-only', 'one') + await log('both') + await log('logs-only') + expect((await jobProgress.listJobIds()).sort()).toEqual(['both', 'logs-only', 'progress-only']) }) - it('should purge old records based on started_at', async () => { - // Insert an old record - await testDb.insertInto('job_progress' as any).values({ - job_id: '1', - phase: 'old.phase', - status: 'completed', - started_at: Date.now() - 3 * 24 * 60 * 60 * 1000, // 3 days old - }).execute() - - // Insert a recent record - await testDb.insertInto('job_progress' as any).values({ - job_id: '2', - phase: 'new.phase', - status: 'completed', - started_at: Date.now() - 1 * 24 * 60 * 60 * 1000, // 1 day old - }).execute() - - await jobProgress.purgeExpired() - - const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute() - expect(rows).toHaveLength(1) - expect(rows[0].job_id).toBe('2') + it('purges only selected IDs in bounded chunks, idempotently', async () => { + const ids = Array.from({ length: 205 }, (_, index) => `job-${index}`) + for (const id of [...ids, 'keep']) { + await jobProgress.start(id, 'render') + await log(id) + } + await jobProgress.purgeJobs(ids) + await jobProgress.purgeJobs(ids) + await jobProgress.purgeJobs([]) + expect(await jobProgress.listJobIds()).toEqual(['keep']) + expect(await testDb.selectFrom('job_progress').select('job_id').execute()).toEqual([ + { job_id: 'keep' }, + ]) + expect(await testDb.selectFrom('job_progress_logs').select('job_id').execute()).toEqual([ + { job_id: 'keep' }, + ]) }) -}) \ No newline at end of file + + it('rolls back logs and earlier chunks and propagates database failures', async () => { + const ids = Array.from({ length: 101 }, (_, index) => `job-${index}`) + for (const id of ids) { + await jobProgress.start(id, 'render') + await log(id) + } + await sql`CREATE TRIGGER prevent_purge BEFORE DELETE ON job_progress + WHEN OLD.job_id = 'job-100' BEGIN SELECT RAISE(FAIL, 'delete failed'); END`.execute(testDb) + await expect(jobProgress.purgeJobs(ids)).rejects.toThrow('delete failed') + expect(await testDb.selectFrom('job_progress').selectAll().execute()).toHaveLength(101) + expect(await testDb.selectFrom('job_progress_logs').selectAll().execute()).toHaveLength(101) + }) +}) diff --git a/packages/web/worker/src/core/job-progress.ts b/packages/web/worker/src/core/job-progress.ts index fe4fced9..1cefe65b 100644 --- a/packages/web/worker/src/core/job-progress.ts +++ b/packages/web/worker/src/core/job-progress.ts @@ -3,47 +3,29 @@ import type { AppDatabase } from '../services/db.js' export class JobProgress { private readonly db: Kysely - private readonly retentionDays: number - constructor(db: Kysely, retentionDays: number = 7) { + constructor(db: Kysely) { this.db = db - this.retentionDays = retentionDays } - async purgeExpiredBatch(limit: number): Promise { - try { - const cutoffMs = Date.now() - this.retentionDays * 24 * 60 * 60 * 1000 + async listJobIds(): Promise { + const rows = await this.db + .selectFrom('job_progress') + .select('job_id') + .union(this.db.selectFrom('job_progress_logs').select('job_id')) + .execute() + return rows.map((row) => row.job_id) + } - const oldJobs = await this.db - .selectFrom('job_progress') - .select('job_id') - .distinct() - .where('started_at', '<', cutoffMs) - .limit(limit) - .execute() - - const oldJobIds = oldJobs.map((row) => row.job_id) - - if (oldJobIds.length > 0) { - await this.db.deleteFrom('job_progress_logs').where('job_id', 'in', oldJobIds).execute() - await this.db.deleteFrom('job_progress').where('job_id', 'in', oldJobIds).execute() + async purgeJobs(jobIds: string[]): Promise { + if (jobIds.length === 0) return + await this.db.transaction().execute(async (trx) => { + for (let offset = 0; offset < jobIds.length; offset += 100) { + const batch = jobIds.slice(offset, offset + 100) + await trx.deleteFrom('job_progress_logs').where('job_id', 'in', batch).execute() + await trx.deleteFrom('job_progress').where('job_id', 'in', batch).execute() } - - return oldJobIds - } catch (err) { - console.error('JobProgress purgeExpiredBatch failed:', err) - return [] - } - } - - async purgeExpired(): Promise { - const allPurged: string[] = [] - let batch: string[] = [] - do { - batch = await this.purgeExpiredBatch(100) - allPurged.push(...batch) - } while (batch.length === 100) - return allPurged + }) } async start(jobId: string, phase: string): Promise { diff --git a/packages/web/worker/src/index.test.ts b/packages/web/worker/src/index.test.ts index 67b0160c..353c00fe 100644 --- a/packages/web/worker/src/index.test.ts +++ b/packages/web/worker/src/index.test.ts @@ -14,12 +14,15 @@ import type * as Workflows from './workflows/index.js' const probe = defineWorkflowSpec({ name: 'ShutdownProbe' }) afterEach(() => { + vi.useRealTimers() vi.doUnmock('./workflows/index.js') vi.resetModules() }) test('starts the native worker and health server, then lets active work finish before closing databases', async () => { vi.resetModules() + vi.useFakeTimers({ toFake: ['Date'] }) + vi.setSystemTime(new Date('2026-06-02T12:00:00.000Z')) const directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-worker-')) const queuePath = path.join(directory, 'queue.sqlite') const config = validateConfig( @@ -64,12 +67,12 @@ test('starts the native worker and health server, then lets active work finish b const { port } = running.server.address() as AddressInfo expect((await fetch(`http://127.0.0.1:${port}/health`)).status).toBe(200) const backend = jobs.getOpenWorkflowBackend() - await expect - .poll( - async () => - (await backend.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data[0]?.status - ) - .toBe('completed') + expect(await jobs.getOpenWorkflow().schedules.get('luzzle:job-progress-purge')).toMatchObject({ + cron: '0 48 1 * * *', + timezone: 'UTC', + nextRunAt: new Date('2026-06-03T01:48:00.000Z'), + }) + expect((await backend.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data).toEqual([]) const handle = await jobs.getOpenWorkflow().runWorkflow(probe) await expect.poll(() => entered).toBe(true) @@ -104,7 +107,7 @@ test('starts the native worker and health server, then lets active work finish b ) expect( (await reopened.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data - ).toHaveLength(1) + ).toHaveLength(0) } finally { await reopened.stop() } diff --git a/packages/web/worker/src/index.ts b/packages/web/worker/src/index.ts index 40547cb4..60d4d26f 100644 --- a/packages/web/worker/src/index.ts +++ b/packages/web/worker/src/index.ts @@ -6,14 +6,12 @@ import { createHealthServer } from './services/health.js' import { createLogger, createOpenWorkflowLogger, log } from './services/logger.js' import { createWorkerContext, setWorkerContext } from './services/context.js' import { closeOpenWorkflow, initOpenWorkflow } from '@luzzle/web.jobs' -import { jobProgressPurgeSpec } from '@luzzle/web.jobs/specs' import { registerWorkflows } from './workflows/index.js' import { once } from 'node:events' import type { Server } from 'node:http' import type { Worker } from 'openworkflow' const DEFAULT_PORT = 9000 -const PURGE_RETENTION_DAYS = 7 async function closeHealthServer(server: Server): Promise { if (!server.listening) return @@ -34,11 +32,9 @@ export async function main( const db = createAppDb(resolveDbPath(config)) const server = createHealthServer() let worker: Worker | undefined - let purgeInterval: ReturnType | undefined let stopPromise: Promise | undefined async function stopResources(): Promise { - clearInterval(purgeInterval) await closeHealthServer(server) await worker?.stop() const errors: unknown[] = [] @@ -82,19 +78,6 @@ export async function main( await worker.start() logger.info('openworkflow worker started') - async function enqueuePurge(): Promise { - try { - const run = await openWorkflow.runWorkflow(jobProgressPurgeSpec, { - retentionDays: PURGE_RETENTION_DAYS, - }) - logger.info('openworkflow job progress purge enqueued', { jobId: run.workflowRun.id }) - } catch { - logger.error('failed to enqueue openworkflow job progress purge') - } - } - - void enqueuePurge() - purgeInterval = setInterval(enqueuePurge, 24 * 60 * 60 * 1000) const listening = once(server, 'listening') server.listen(port) await listening diff --git a/packages/web/worker/src/steps/job-progress-purge.test.ts b/packages/web/worker/src/steps/job-progress-purge.test.ts deleted file mode 100644 index 3a4c017e..00000000 --- a/packages/web/worker/src/steps/job-progress-purge.test.ts +++ /dev/null @@ -1,57 +0,0 @@ -import { describe, test, expect, vi } from 'vitest' -import { jobProgressPurgeStep } from './job-progress-purge.js' -import { JobProgress } from '../core/job-progress.js' -import type { WorkerContext } from '../services/context.js' -import { promises as fs } from 'node:fs' - -const mockPurgeExpired = vi.fn().mockResolvedValue([]) - -vi.mock('../core/job-progress.js', () => ({ - JobProgress: vi.fn().mockImplementation(() => ({ - purgeExpired: mockPurgeExpired, - })), -})) - -function makeCtx(): WorkerContext { - return { - config: { paths: { assets: '/assets/pieces' } } as unknown as WorkerContext['config'], - logger: { - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - stdout: vi.fn(), - stderr: vi.fn(), - }, - rclone: {} as WorkerContext['rclone'], - db: {} as WorkerContext['db'], - } -} - -describe('jobProgressPurgeStep', () => { - test('uses default retentionDays=7', async () => { - mockPurgeExpired.mockResolvedValueOnce([]) - const result = await jobProgressPurgeStep.run({}, makeCtx()) - expect(result.status).toBe('completed') - expect(JobProgress).toHaveBeenCalledWith(expect.anything(), 7) - }) - - test('honors explicit retentionDays', async () => { - mockPurgeExpired.mockResolvedValueOnce([]) - await jobProgressPurgeStep.run({ retentionDays: 14 }, makeCtx()) - expect(JobProgress).toHaveBeenCalledWith(expect.anything(), 14) - }) - - test('deletes cached preview folders for purged job IDs', async () => { - mockPurgeExpired.mockResolvedValueOnce([101, 102]) - const rmSpy = vi.spyOn(fs, 'rm').mockResolvedValue(undefined) - - await jobProgressPurgeStep.run({}, makeCtx()) - - expect(rmSpy).toHaveBeenCalledTimes(2) - expect(rmSpy).toHaveBeenNthCalledWith(1, '/assets/previews/101', { recursive: true, force: true }) - expect(rmSpy).toHaveBeenNthCalledWith(2, '/assets/previews/102', { recursive: true, force: true }) - - rmSpy.mockRestore() - }) -}) diff --git a/packages/web/worker/src/steps/job-progress-purge.ts b/packages/web/worker/src/steps/job-progress-purge.ts deleted file mode 100644 index 65f57cb5..00000000 --- a/packages/web/worker/src/steps/job-progress-purge.ts +++ /dev/null @@ -1,30 +0,0 @@ -import { completed, type Step, type StepResult } from '../core/step.js' -import { JobProgress } from '../core/job-progress.js' -import { promises as fs } from 'node:fs' -import path from 'node:path' - -export interface JobProgressPurgeInput { - retentionDays?: number -} - -export const jobProgressPurgeStep: Step = { - name: 'job_progress.purge', - async run(input, ctx): Promise> { - const { db, logger, config } = ctx - const retentionDays = input.retentionDays ?? 7 - const purgedJobIds = await new JobProgress(db, retentionDays).purgeExpired() - - for (const jobId of purgedJobIds) { - const previewDir = path.join(path.dirname(config.paths.assets), 'previews', jobId.toString()) - try { - await fs.rm(previewDir, { recursive: true, force: true }) - logger.info('purged preview assets directory', { jobId, previewDir }) - } catch (err) { - logger.warn('failed to purge preview assets directory', { jobId, error: String(err) }) - } - } - - logger.info('job_progress purge complete', { retentionDays, purgedCount: purgedJobIds.length }) - return completed(undefined) - }, -} diff --git a/packages/web/worker/src/workflows/job-progress-purge.test.ts b/packages/web/worker/src/workflows/job-progress-purge.test.ts index d88bfc9c..5da64862 100644 --- a/packages/web/worker/src/workflows/job-progress-purge.test.ts +++ b/packages/web/worker/src/workflows/job-progress-purge.test.ts @@ -1,27 +1,335 @@ -import { describe, test, expect, beforeEach } from 'vitest' -import { registerJobProgressPurgeWorkflow } from './job-progress-purge.js' -import { initOpenWorkflow, getOpenWorkflow } from '@luzzle/web.jobs' +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { OpenWorkflow, defineWorkflowSpec, type Worker } from 'openworkflow' +import { BackendSqlite } from 'openworkflow/sqlite' +import { createAppDb, runWebMigrations } from '@luzzle/web.db' +import { getOpenWorkflow, getOpenWorkflowBackend } from '@luzzle/web.jobs' import { jobProgressPurgeSpec } from '@luzzle/web.jobs/specs' +import { promises as fs } from 'node:fs' +import type * as Crypto from 'node:crypto' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { validateConfig } from '@luzzle/web.config' +import { getWorkerContext, type WorkerContext } from '../services/context.js' +import { JobProgress } from '../core/job-progress.js' +import { makeLogger } from '../../test/logger.js' +import { registerJobProgressPurgeWorkflow } from './job-progress-purge.js' -describe('workflows/job-progress-purge', () => { - beforeEach(() => { - initOpenWorkflow({ dbPath: ':memory:' }) +const { jitter } = vi.hoisted(() => ({ jitter: vi.fn(() => 0) })) +vi.mock('node:crypto', async (original) => ({ + ...(await original()), + randomInt: jitter, +})) +vi.mock('@luzzle/web.jobs', () => ({ getOpenWorkflow: vi.fn(), getOpenWorkflowBackend: vi.fn() })) +vi.mock('../services/context.js', () => ({ getWorkerContext: vi.fn() })) +const day = 86_400_000 +const now = new Date('2026-06-02T12:00:00Z') +const candidateSpec = defineWorkflowSpec({ name: 'PreviewFixture' }) + +describe('native preview maintenance', () => { + let directory: string, previews: string + let backend: BackendSqlite, client: OpenWorkflow, worker: Worker + let ctx: WorkerContext, progress: JobProgress + const getRun = (workflowRunId: string) => backend.getWorkflowRun({ workflowRunId }) + const steps = async (workflowRunId: string) => + (await backend.listStepAttempts({ workflowRunId })).data + const preview = (id: string) => path.join(previews, id) + + function connect() { + backend = BackendSqlite.connect(path.join(directory, 'workflows.sqlite')) + client = new OpenWorkflow({ backend, logger: makeLogger() }) + vi.mocked(getOpenWorkflow).mockReturnValue(client) + vi.mocked(getOpenWorkflowBackend).mockReturnValue(backend) + registerJobProgressPurgeWorkflow() + worker = client.newWorker({ concurrency: 1 }) + } + beforeEach(async () => { + vi.useFakeTimers({ toFake: ['Date'] }) + vi.setSystemTime(now) + jitter.mockReset().mockReturnValue(0) + directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-maintenance-')) + previews = path.join(directory, 'previews') + const db = createAppDb(path.join(directory, 'app.sqlite')) + expect((await runWebMigrations(db)).error).toBeUndefined() + const config = validateConfig( + { + storage: { root: directory }, + paths: { assets: path.join(directory, 'pieces') }, + }, + {} + ) + ctx = { db, config, logger: makeLogger(), rclone: {} as WorkerContext['rclone'] } + vi.mocked(getWorkerContext).mockReturnValue(ctx) + progress = new JobProgress(db) + connect() + }) + afterEach(async () => { + await worker.stop() + vi.restoreAllMocks() + await backend.stop() + await ctx.db.destroy() + await fs.rm(directory, { recursive: true, force: true }) + vi.useRealTimers() + }) + async function tick() { + expect(await worker.tick()).toBe(1) + await worker.stop() + } + async function seedPreviewData(id: string, rows: 'both' | 'logs' | 'none' = 'both') { + if (rows === 'both') await progress.start(id, 'render') + if (rows !== 'none') + await ctx.db + .insertInto('job_progress_logs') + .values({ + job_id: id, + phase: 'render', + line_number: 1, + ts: Date.now(), + level: 'info', + message: 'rendered', + }) + .execute() + await fs.mkdir(preview(id), { recursive: true }) + await fs.writeFile(path.join(preview(id), 'preview.bin'), 'preview output') + } + async function createCandidateRun( + status = 'completed', + finishedAgeMs = 8 * day, + rows: 'both' | 'logs' | 'none' = 'both' + ) { + vi.setSystemTime(now.getTime() - 20 * day) + const options = status === 'pending' ? { availableAt: new Date(now.getTime() + day) } : {} + const { workflowRun: run } = await client.runWorkflow(candidateSpec, {}, options) + const owned = { workflowRunId: run.id, workerId: 'fixture' } + if (status !== 'pending') { + const claim = await backend.claimWorkflowRun({ + workerId: 'fixture', + leaseDurationMs: 40 * day, + }) + expect(claim?.id).toBe(run.id) + const step = await backend.createStepAttempt({ + ...owned, + stepName: 'render', + kind: 'function', + config: {}, + context: null, + }) + await backend.completeStepAttempt({ ...owned, stepAttemptId: step.id, output: 'rendered' }) + vi.setSystemTime(now.getTime() - finishedAgeMs) + if (status === 'completed') await backend.completeWorkflowRun({ ...owned, output: 'ok' }) + if (status === 'failed') + await backend.failWorkflowRun({ + ...owned, + error: { message: 'failed' }, + retryPolicy: { + maximumAttempts: 1, + initialInterval: '1s', + maximumInterval: '1s', + backoffCoefficient: 1, + }, + }) + if (status === 'canceled') await backend.cancelWorkflowRun({ workflowRunId: run.id }) + } + vi.setSystemTime(now) + await seedPreviewData(run.id, rows) + return run.id + } + async function purge() { + const handle = await client.runWorkflow(jobProgressPurgeSpec, {}) + await tick() + return handle.workflowRun.id + } + + test('removes aged terminal app data and directories without deleting SDK history', async () => { + const expired = [ + await createCandidateRun(), + await createCandidateRun('failed', 8 * day, 'logs'), + await createCandidateRun('canceled', 8 * day, 'none'), + ] + const kept = [ + await createCandidateRun('completed', day), + await createCandidateRun('completed', 7 * day), + await createCandidateRun('running'), + await createCandidateRun('pending'), + 'unknown', + ] + await seedPreviewData('unknown') + await progress.start('../outside', 'render') + await fs.writeFile(path.join(directory, 'outside'), 'keep') + const history = await Promise.all( + expired.map(async (id) => [await getRun(id), await steps(id)]) + ) + await purge() + for (let index = 0; index < expired.length; index++) { + const id = expired[index] + await expect(fs.stat(preview(id))).rejects.toMatchObject({ code: 'ENOENT' }) + expect(await progress.listJobIds()).not.toContain(id) + expect([await getRun(id), await steps(id)]).toEqual(history[index]) + } + for (const id of kept) { + expect(await fs.readFile(path.join(preview(id), 'preview.bin'), 'utf8')).toBe( + 'preview output' + ) + expect(await progress.listJobIds()).toContain(id) + } + expect(await fs.readFile(path.join(directory, 'outside'), 'utf8')).toBe('keep') }) - test('registers workflow implementation successfully', () => { - registerJobProgressPurgeWorkflow() - const ow = getOpenWorkflow() + test('honors an explicit retention window and rejects a non-positive window without deleting data', async () => { + const id = await createCandidateRun() + const extended = await client.runWorkflow(jobProgressPurgeSpec, { retentionDays: 14 }) + await tick() + expect((await getRun(extended.workflowRun.id))?.status).toBe('completed') + expect(await fs.stat(preview(id))).toBeDefined() + expect(await progress.listJobIds()).toContain(id) + const invalid = await client.runWorkflow(jobProgressPurgeSpec, { retentionDays: 0 }) + await tick() + expect((await getRun(invalid.workflowRun.id))?.status).toBe('failed') + expect(await fs.stat(preview(id))).toBeDefined() + expect(await progress.listJobIds()).toContain(id) + }) - // Verify the workflow is correctly registered under its name - const registered = ( - ow as unknown as { - registry: { - get: (name: string) => { spec: { name: string }; fn: (...args: unknown[]) => unknown } - } - } - ).registry.get(jobProgressPurgeSpec.name) - expect(registered).toBeDefined() - expect(registered.spec.name).toBe(jobProgressPurgeSpec.name) - expect(typeof registered.fn).toBe('function') + test('file failure preserves rows and retries saved paths and IDs, even after deleting files', async () => { + const id = await createCandidateRun() + const recent = await createCandidateRun('completed', day) + const list = vi.spyOn(JobProgress.prototype, 'listJobIds') + const rm = fs.rm + vi.spyOn(fs, 'rm').mockImplementationOnce(async (target, options) => { + await rm(target, options) + throw new Error('file deletion interrupted') + }) + const purgeId = await purge() + await expect(fs.stat(preview(id))).rejects.toMatchObject({ code: 'ENOENT' }) + expect( + await ctx.db + .selectFrom('job_progress_logs') + .select('job_id') + .where('job_id', '=', id) + .execute() + ).toHaveLength(1) + expect( + await ctx.db.selectFrom('job_progress').select('job_id').where('job_id', '=', id).execute() + ).toHaveLength(1) + const attempts = await steps(purgeId) + expect(attempts.find((step) => step.stepName === 'plan-cleanup')?.output).toEqual({ + cutoffMs: now.getTime() - 7 * day, + jobIds: [id], + previewDirs: [preview(id)], + }) + expect(attempts.find((step) => step.stepName === 'purge-preview-directories')?.status).toBe( + 'failed' + ) + ctx.config.paths.assets = path.join(directory, 'different-root', 'pieces') + vi.setSystemTime(now.getTime() + 10 * day) + await tick() + expect(list).toHaveBeenCalledTimes(1) + expect(jitter).toHaveBeenCalledTimes(1) + expect(await progress.listJobIds()).toEqual([recent]) + expect(await fs.stat(preview(recent))).toBeDefined() + expect((await getRun(purgeId))?.status).toBe('completed') + }) + + test.each(['before', 'after'])( + 'retries app deletion %s commit after files are gone', + async (boundary) => { + const id = await createCandidateRun() + const original = JobProgress.prototype.purgeJobs + const remove = vi + .spyOn(JobProgress.prototype, 'purgeJobs') + .mockImplementationOnce(async function (this: JobProgress, ids) { + if (boundary === 'after') await original.call(this, ids) + throw new Error('app deletion interrupted') + }) + const files = vi.spyOn(fs, 'rm') + const list = vi.spyOn(JobProgress.prototype, 'listJobIds') + const purgeId = await purge() + await expect(fs.stat(preview(id))).rejects.toMatchObject({ code: 'ENOENT' }) + expect(await progress.listJobIds()).toEqual(boundary === 'before' ? [id] : []) + list.mockClear() + vi.setSystemTime(now.getTime() + day) + await tick() + expect(list).not.toHaveBeenCalled() + expect(remove.mock.calls).toEqual([[[id]], [[id]]]) + expect(files).toHaveBeenCalledTimes(1) + expect(await progress.listJobIds()).toEqual([]) + expect((await getRun(purgeId))?.status).toBe('completed') + } + ) + + test('propagates discovery errors but treats a missing previews root as empty', async () => { + const id = await createCandidateRun() + await fs.rm(previews, { recursive: true }) + const list = vi.spyOn(fs, 'readdir').mockRejectedValueOnce(new Error('directory unavailable')) + const purgeId = await purge() + expect(await progress.listJobIds()).toEqual([id]) + vi.setSystemTime(now.getTime() + day) + await tick() + expect(list).toHaveBeenCalledTimes(2) + expect(await progress.listJobIds()).toEqual([]) + expect((await getRun(purgeId))?.status).toBe('completed') + }) + + test('persists jitter, releases the only slot during native sleep, and resumes the saved choice', async () => { + jitter.mockReturnValueOnce(1440).mockReturnValueOnce(0) + const first = await purge() + expect(await getRun(first)).toMatchObject({ + status: 'running', + workerId: null, + availableAt: new Date(now.getTime() + 1440_000), + }) + const attempts = await steps(first) + expect(attempts.find((step) => step.stepName === 'choose-cleanup-jitter')).toMatchObject({ + status: 'completed', + output: 1440, + }) + expect(attempts.find((step) => step.stepName === 'cleanup-jitter')).toMatchObject({ + kind: 'sleep', + }) + await backend.stop() + connect() + const second = await purge() + expect((await getRun(second))?.status).toBe('completed') + vi.setSystemTime(now.getTime() + 1440_000) + await tick() + expect((await getRun(first))?.status).toBe('completed') + expect(jitter.mock.calls).toEqual([ + [0, 1441], + [0, 1441], + ]) + }) + + test('persists cron and pause across restart and coalesces missed occurrences', async () => { + const scheduleId = 'luzzle:job-progress-purge' + await worker.start() + await worker.stop() + const saved = await client.schedules.get(scheduleId) + expect(saved).toMatchObject({ + cron: '0 48 1 * * *', + timezone: 'UTC', + input: { retentionDays: 7 }, + nextRunAt: new Date('2026-06-03T01:48:00Z'), + }) + await backend.stop() + connect() + await worker.start() + await worker.stop() + expect(await client.schedules.get(scheduleId)).toEqual(saved) + await client.schedules.pause(scheduleId) + await backend.stop() + connect() + await worker.start() + await worker.stop() + expect((await client.schedules.get(scheduleId))?.nextRunAt).toBeNull() + await client.schedules.resume(scheduleId) + vi.setSystemTime(new Date('2026-06-06T12:00:00Z')) + expect(await backend.enqueueScheduledRuns({ before: new Date(), limit: 100 })).toBe(1) + expect(await backend.enqueueScheduledRuns({ before: new Date(), limit: 100 })).toBe(0) + const runs = (await backend.listWorkflowRuns({ workflowName: jobProgressPurgeSpec.name })).data + expect(runs).toHaveLength(1) + expect(runs[0].workflowScheduleId).toBe(scheduleId) + await tick() + expect((await getRun(runs[0].id))?.status).toBe('completed') + expect((await client.schedules.get(scheduleId))?.nextRunAt).toEqual( + new Date('2026-06-07T01:48:00Z') + ) }) }) diff --git a/packages/web/worker/src/workflows/job-progress-purge.ts b/packages/web/worker/src/workflows/job-progress-purge.ts index 8847d889..9205f9e8 100644 --- a/packages/web/worker/src/workflows/job-progress-purge.ts +++ b/packages/web/worker/src/workflows/job-progress-purge.ts @@ -1,46 +1,67 @@ -import { getOpenWorkflow } from '@luzzle/web.jobs' +import { getOpenWorkflow, getOpenWorkflowBackend } from '@luzzle/web.jobs' import { jobProgressPurgeSpec } from '@luzzle/web.jobs/specs' -import { getWorkerContext } from '../services/context.js' -import { JobProgress } from '../core/job-progress.js' +import { randomInt } from 'node:crypto' import { promises as fs } from 'node:fs' import path from 'node:path' +import { getWorkerContext } from '../services/context.js' +import { JobProgress } from '../core/job-progress.js' export function registerJobProgressPurgeWorkflow(): void { const openWorkflow = getOpenWorkflow() openWorkflow.implementWorkflow(jobProgressPurgeSpec, async ({ input, step }) => { - const ctx = getWorkerContext() - const { db, logger, config } = ctx const retentionDays = input.retentionDays ?? 7 + if (!Number.isFinite(retentionDays) || retentionDays <= 0) { + throw new Error('Cleanup retentionDays must be positive and finite') + } + const delaySeconds = await step.run({ name: 'choose-cleanup-jitter' }, () => randomInt(0, 1441)) + if (delaySeconds > 0) await step.sleep('cleanup-jitter', `${delaySeconds}s`) - const purgedJobIds = await step.run({ name: 'purge-db-entries' }, async () => { - return await new JobProgress(db, retentionDays).purgeExpired() + const { db, logger, config } = getWorkerContext() + const progress = new JobProgress(db) + const plan = await step.run({ name: 'plan-cleanup' }, async () => { + const cutoffMs = Date.now() - retentionDays * 24 * 60 * 60 * 1000 + const previewsRoot = path.resolve(path.dirname(config.paths.assets), 'previews') + let previewNames: string[] + try { + previewNames = (await fs.readdir(previewsRoot, { withFileTypes: true })) + .filter((entry) => entry.isDirectory()) + .map((entry) => entry.name) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error + previewNames = [] + } + const directories = new Set(previewNames) + const candidates = new Set([...(await progress.listJobIds()), ...previewNames]) + const jobIds: string[] = [] + const previewDirs: string[] = [] + const backend = getOpenWorkflowBackend() + for (const jobId of candidates) { + if (!/^[a-zA-Z0-9_-]+$/.test(jobId)) continue + const previewDir = path.resolve(previewsRoot, jobId) + if (path.dirname(previewDir) !== previewsRoot) continue + const run = await backend.getWorkflowRun({ workflowRunId: jobId }) + if (!run) continue + if (!['completed', 'failed', 'canceled', 'succeeded'].includes(run.status)) continue + if (!run.finishedAt) continue + const isExpired = run.finishedAt.getTime() < cutoffMs + if (!isExpired) continue + jobIds.push(jobId) + if (directories.has(jobId)) previewDirs.push(previewDir) + } + return { cutoffMs, jobIds, previewDirs } }) - await step.run({ name: 'purge-disk-directories' }, async () => { - for (const jobId of purgedJobIds) { - const previewDir = path.join( - path.dirname(config.paths.assets), - 'previews', - jobId.toString() - ) - try { - await fs.rm(previewDir, { recursive: true, force: true }) - logger.info('purged preview assets directory', { jobId, previewDir }) - } catch (err) { - logger.warn('failed to purge preview assets directory', { - jobId, - error: String(err), - }) - } + await step.run({ name: 'purge-preview-directories' }, async () => { + for (const previewDir of plan.previewDirs) { + await fs.rm(previewDir, { recursive: true, force: true }) } }) - - logger.info('openworkflow purge complete', { - retentionDays, - purgedCount: purgedJobIds.length, + await step.run({ name: 'purge-app-rows' }, async () => { + await progress.purgeJobs(plan.jobIds) }) + logger.info('openworkflow purge complete', { purgedCount: plan.jobIds.length }) return 'ok' }) }