From 4ce98e7780ad8584529103cd32143982e0789874 Mon Sep 17 00:00:00 2001 From: eleith-gitea-agent Date: Tue, 27 Oct 2026 16:15:46 +0000 Subject: [PATCH] [worker] upgrade OpenWorkflow to 0.11.0 --- .changeset/openworkflow-upgrade.md | 8 + apps/web/package.json | 2 +- packages/web/jobs/package.json | 2 +- packages/web/jobs/src/index.ts | 5 +- packages/web/jobs/src/openworkflow.test.ts | 148 +++++++++++--- packages/web/jobs/src/openworkflow.ts | 88 ++++----- packages/web/worker/package.json | 2 +- packages/web/worker/src/index.test.ts | 117 +++++++++++ packages/web/worker/src/index.ts | 186 +++++++++++------- .../web/worker/src/services/context.test.ts | 10 +- packages/web/worker/src/services/context.ts | 19 +- .../web/worker/src/services/logger.test.ts | 50 ++++- packages/web/worker/src/services/logger.ts | 59 ++++-- .../src/workflows/job-progress-purge.ts | 21 +- pnpm-lock.yaml | 39 +++- 15 files changed, 542 insertions(+), 214 deletions(-) create mode 100644 .changeset/openworkflow-upgrade.md create mode 100644 packages/web/worker/src/index.test.ts diff --git a/.changeset/openworkflow-upgrade.md b/.changeset/openworkflow-upgrade.md new file mode 100644 index 00000000..86cb39b7 --- /dev/null +++ b/.changeset/openworkflow-upgrade.md @@ -0,0 +1,8 @@ +--- +"@luzzle/web": patch +"@luzzle/web.worker": patch +--- + +Upgrade OpenWorkflow to 0.11.0 across web, worker and shared jobs. Use its native +logger option and close worker-owned resources in order. Stop deleting engine job +history; preview files and application logs remain application-owned cleanup. diff --git a/apps/web/package.json b/apps/web/package.json index e30c473f..383e197a 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -90,6 +90,6 @@ "eventsource-parser": "4.1.1", "kysely": "0.28.17", "micromark": "4.0.2", - "openworkflow": "0.9.2" + "openworkflow": "0.11.0" } } diff --git a/packages/web/jobs/package.json b/packages/web/jobs/package.json index 5f833525..c90f614e 100644 --- a/packages/web/jobs/package.json +++ b/packages/web/jobs/package.json @@ -49,7 +49,7 @@ }, "dependencies": { "@luzzle/core": "workspace:*", - "openworkflow": "0.9.2" + "openworkflow": "0.11.0" }, "type": "module" } diff --git a/packages/web/jobs/src/index.ts b/packages/web/jobs/src/index.ts index c476d531..c2e56967 100644 --- a/packages/web/jobs/src/index.ts +++ b/packages/web/jobs/src/index.ts @@ -1,12 +1,13 @@ export { initOpenWorkflow, getOpenWorkflow, + getOpenWorkflowBackend, + closeOpenWorkflow, getLatestWorkflowRun, getWorkflowRun, getStepAttempts, - purgeExpiredWorkflowRuns, } from './openworkflow.js' -export type { WorkflowRunRow, StepAttemptRow } from './openworkflow.js' +export type { InitOpenWorkflowOptions, WorkflowRunRow, StepAttemptRow } from './openworkflow.js' export { jobProgressPurgeSpec, previewSpec, diff --git a/packages/web/jobs/src/openworkflow.test.ts b/packages/web/jobs/src/openworkflow.test.ts index bdac93bb..63623962 100644 --- a/packages/web/jobs/src/openworkflow.test.ts +++ b/packages/web/jobs/src/openworkflow.test.ts @@ -1,8 +1,8 @@ import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' import { BackendSqlite } from 'openworkflow/sqlite' +import type { Logger } from 'openworkflow' +import type * as WorkflowClient from './openworkflow.js' import { - initOpenWorkflow, - getOpenWorkflow, getLatestWorkflowRun, getWorkflowRun, getStepAttempts, @@ -12,15 +12,88 @@ import { } from './index.js' describe('openworkflow client initialization', () => { - test('throws before initialization', () => { - expect(() => getOpenWorkflow()).toThrow(/has not been initialized/) + let lifecycle: typeof WorkflowClient + + beforeEach(async () => { + vi.resetModules() + lifecycle = await import('./openworkflow.js') }) - test('initializes and returns singleton instance', () => { - const client = initOpenWorkflow({ dbPath: ':memory:' }) - expect(client).toBeDefined() - expect(getOpenWorkflow()).toBe(client) - expect(initOpenWorkflow({ dbPath: ':memory:' })).toBe(client) + afterEach(async () => { + await lifecycle.closeOpenWorkflow().catch(() => {}) + vi.restoreAllMocks() + }) + + test('throws before initialization', () => { + expect(() => lifecycle.getOpenWorkflow()).toThrow(/has not been initialized/) + expect(() => lifecycle.getOpenWorkflowBackend()).toThrow(/has not been initialized/) + }) + + test('owns one backend in the SDK default namespace and returns the singleton', async () => { + const client = lifecycle.initOpenWorkflow({ dbPath: ':memory:' }) + const backend = lifecycle.getOpenWorkflowBackend() + expect(lifecycle.getOpenWorkflow()).toBe(client) + expect(lifecycle.initOpenWorkflow({ dbPath: 'not-opened.sqlite' })).toBe(client) + expect(lifecycle.getOpenWorkflowBackend()).toBe(backend) + const handle = await client.runWorkflow(publishSpec, {}) + expect(handle.workflowRun.namespaceId).toBe('default') + expect((await backend.getWorkflowRun({ workflowRunId: handle.workflowRun.id }))?.id).toBe( + handle.workflowRun.id + ) + }) + + test('passes the native logger to workers without adapting its arguments', async () => { + const logger: Logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() } + const client = lifecycle.initOpenWorkflow({ dbPath: ':memory:', logger }) + const backend = lifecycle.getOpenWorkflowBackend() + const failure = new Error('poll failed') + vi.spyOn(backend, 'claimWorkflowRun').mockRejectedValue(failure) + const worker = client.newWorker({ concurrency: 1 }) + try { + await worker.start() + await vi.waitFor(() => { + expect(logger.error).toHaveBeenCalledWith(expect.any(String), failure) + }) + } finally { + await worker.stop() + } + }) + + test('shares one terminal close and never reopens the backend', async () => { + lifecycle.initOpenWorkflow({ dbPath: ':memory:' }) + const backend = lifecycle.getOpenWorkflowBackend() + const stop = vi.spyOn(backend, 'stop') + const closing = lifecycle.closeOpenWorkflow() + expect(lifecycle.closeOpenWorkflow()).toBe(closing) + expect(() => lifecycle.getOpenWorkflow()).toThrow(/closed/) + expect(() => lifecycle.getOpenWorkflowBackend()).toThrow(/closed/) + expect(() => lifecycle.initOpenWorkflow({ dbPath: ':memory:' })).toThrow(/closed/) + await closing + expect(stop).toHaveBeenCalledTimes(1) + expect(lifecycle.closeOpenWorkflow()).toBe(closing) + }) + + test('preserves a close failure without retrying or reporting success', async () => { + lifecycle.initOpenWorkflow({ dbPath: ':memory:' }) + const backend = lifecycle.getOpenWorkflowBackend() + const actualStop = backend.stop.bind(backend) + const failure = new Error('close failed') + const stop = vi.spyOn(backend, 'stop').mockRejectedValue(failure) + try { + const closing = lifecycle.closeOpenWorkflow() + await expect(closing).rejects.toBe(failure) + expect(lifecycle.closeOpenWorkflow()).toBe(closing) + expect(stop).toHaveBeenCalledTimes(1) + expect(() => lifecycle.initOpenWorkflow({ dbPath: ':memory:' })).toThrow(/closed/) + } finally { + await actualStop() + } + }) + + test('closing before initialization is harmless', async () => { + await lifecycle.closeOpenWorkflow() + expect(() => lifecycle.getOpenWorkflow()).toThrow(/has not been initialized/) + expect(lifecycle.initOpenWorkflow({ dbPath: ':memory:' })).toBeDefined() }) }) @@ -137,7 +210,7 @@ describe('SDK read helpers', () => { { name: 'PublishError', message: 'upload failed', stack: 'stack', detail: { code: 503 } }, 'legacy failure', '', - ])('preserves workflow error JSON encoding: %j', async error => { + ])('preserves workflow error JSON encoding: %j', async (error) => { const run = await createRun() await claimRun(run.id) vi.setSystemTime(new Date('2026-06-02T05:02:00.000Z')) @@ -206,20 +279,26 @@ describe('SDK read helpers', () => { await claimRun(run.id) await createStep(run.id, 'wait') const original = backend.listStepAttempts.bind(backend) - vi.spyOn(backend, 'listStepAttempts').mockImplementation(async params => { + vi.spyOn(backend, 'listStepAttempts').mockImplementation(async (params) => { const page = await original(params) - return { ...page, data: page.data.map(attempt => ({ ...attempt, startedAt: null })) } + return { ...page, data: page.data.map((attempt) => ({ ...attempt, startedAt: null })) } }) - expect(await getStepAttempts(backend, run.id)).toEqual([{ - phase: 'wait', status: 'running', started_at: null, finished_at: null, message: null, - }]) + expect(await getStepAttempts(backend, run.id)).toEqual([ + { + phase: 'wait', + status: 'running', + started_at: null, + finished_at: null, + message: null, + }, + ]) }) test.each([ { name: 'TransformError', message: 'transform failed', detail: ['cover'] }, 'legacy step failure', '', - ])('preserves step error JSON as message: %j', async error => { + ])('preserves step error JSON as message: %j', async (error) => { const run = await createRun() await claimRun(run.id) const step = await createStep(run.id, 'transform') @@ -231,13 +310,15 @@ describe('SDK read helpers', () => { error: error as unknown as RunError, }) - expect(await getStepAttempts(backend, run.id)).toEqual([{ - phase: 'transform', - status: 'failed', - started_at: initialTime.toISOString(), - finished_at: '2026-06-02T05:03:00.000Z', - message: JSON.stringify(error), - }]) + expect(await getStepAttempts(backend, run.id)).toEqual([ + { + phase: 'transform', + status: 'failed', + started_at: initialTime.toISOString(), + finished_at: '2026-06-02T05:03:00.000Z', + message: JSON.stringify(error), + }, + ]) }) test('follows every SDK page in chronological order, including timestamp ties and retries', async () => { @@ -248,24 +329,27 @@ describe('SDK read helpers', () => { // Shared timestamps also exercise the SDK cursor ID tie-breaker. vi.setSystemTime(new Date(initialTime.getTime() + Math.floor(i / 3) * 1000)) const step = await createStep(run.id, `phase-${i % 3}`) - attempts.push(await backend.completeStepAttempt({ - workflowRunId: run.id, - stepAttemptId: step.id, - workerId, - output: null, - })) + attempts.push( + await backend.completeStepAttempt({ + workflowRunId: run.id, + stepAttemptId: step.id, + workerId, + output: null, + }) + ) } const other = await createRun({ workflowName: 'Preview' }) await claimRun(other.id) await createStep(other.id, 'not-this-run') const original = backend.listStepAttempts.bind(backend) - const list = vi.spyOn(backend, 'listStepAttempts') - .mockImplementation(params => original({ ...params, limit: 2 })) + const list = vi + .spyOn(backend, 'listStepAttempts') + .mockImplementation((params) => original({ ...params, limit: 2 })) const rows = await getStepAttempts(backend, run.id) const expected = attempts .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime() || a.id.localeCompare(b.id)) - .map(attempt => ({ + .map((attempt) => ({ phase: attempt.stepName, status: 'completed', started_at: attempt.startedAt!.toISOString(), diff --git a/packages/web/jobs/src/openworkflow.ts b/packages/web/jobs/src/openworkflow.ts index 9d019362..dce51290 100644 --- a/packages/web/jobs/src/openworkflow.ts +++ b/packages/web/jobs/src/openworkflow.ts @@ -1,22 +1,45 @@ import { OpenWorkflow } from 'openworkflow' +import type { Logger } from 'openworkflow' import { BackendSqlite } from 'openworkflow/sqlite' -import type { DatabaseSync } from 'node:sqlite' let openWorkflowInstance: OpenWorkflow | null = null +let openWorkflowBackend: BackendSqlite | null = null +let closePromise: Promise | null = null export interface InitOpenWorkflowOptions { dbPath: string + logger?: Logger +} + +function assertOpen(): void { + if (closePromise) { + throw new Error('OpenWorkflow has been closed and cannot be reinitialized.') + } } export function initOpenWorkflow(opts: InitOpenWorkflowOptions): OpenWorkflow { + assertOpen() if (!openWorkflowInstance) { - const backend = BackendSqlite.connect(opts.dbPath) - openWorkflowInstance = new OpenWorkflow({ backend }) + openWorkflowBackend = BackendSqlite.connect(opts.dbPath) + openWorkflowInstance = new OpenWorkflow({ backend: openWorkflowBackend, logger: opts.logger }) } return openWorkflowInstance } +/** The caller must stop workers before closing their shared backend. */ +export function closeOpenWorkflow(): Promise { + if (!openWorkflowBackend) return Promise.resolve() + closePromise ??= openWorkflowBackend.stop() + return closePromise +} + +export function getOpenWorkflowBackend(): BackendSqlite { + getOpenWorkflow() + return openWorkflowBackend! +} + export function getOpenWorkflow(): OpenWorkflow { + assertOpen() if (!openWorkflowInstance) { throw new Error( 'OpenWorkflow client has not been initialized. Call initOpenWorkflow({ dbPath }) first.' @@ -96,58 +119,17 @@ export async function getStepAttempts( do { // Forward SDK pagination preserves chronological order across pages. const page = await backend.listStepAttempts({ workflowRunId, after }) - rows.push(...page.data.map(attempt => ({ - phase: attempt.stepName, - status: attempt.status, - started_at: attempt.startedAt?.toISOString() ?? null, - finished_at: attempt.finishedAt?.toISOString() ?? null, - message: attempt.error === null ? null : JSON.stringify(attempt.error), - }))) + rows.push( + ...page.data.map((attempt) => ({ + phase: attempt.stepName, + status: attempt.status, + started_at: attempt.startedAt?.toISOString() ?? null, + finished_at: attempt.finishedAt?.toISOString() ?? null, + message: attempt.error === null ? null : JSON.stringify(attempt.error), + })) + ) after = page.pagination.next ?? undefined } while (after) return rows } - -/** - * Purges old workflow runs and step attempts from the database in batches of 100. - * Note: OpenWorkflow will add cron and cleanup in a future release, - * so we will need to make adjustments later if/when they do. - */ -export function purgeExpiredWorkflowRuns( - db: DatabaseSync, - retentionDays: number, - limit: number = 100 -): number { - const cutoffIso = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000).toISOString() - let totalPurged = 0 - let batch: { id: string }[] = [] - - do { - const stmtSelect = db.prepare(` - SELECT id FROM workflow_runs - WHERE created_at < ? - LIMIT ? - `) - batch = stmtSelect.all(cutoffIso, limit) as { id: string }[] - - if (batch.length > 0) { - const ids = batch.map(r => r.id) - const placeholders = ids.map(() => '?').join(',') - - db.prepare(` - DELETE FROM step_attempts - WHERE workflow_run_id IN (${placeholders}) - `).run(...ids) - - db.prepare(` - DELETE FROM workflow_runs - WHERE id IN (${placeholders}) - `).run(...ids) - - totalPurged += batch.length - } - } while (batch.length === limit) - - return totalPurged -} diff --git a/packages/web/worker/package.json b/packages/web/worker/package.json index a6ad57b7..cc09d5c8 100644 --- a/packages/web/worker/package.json +++ b/packages/web/worker/package.json @@ -71,7 +71,7 @@ "micromark-util-symbol": "2.0.1", "micromark-util-types": "2.0.2", "mime-types": "3.0.2", - "openworkflow": "0.9.2", + "openworkflow": "0.11.0", "rehype-autolink-headings": "7.1.0", "rehype-pretty-code": "0.14.1", "rehype-raw": "7.0.0", diff --git a/packages/web/worker/src/index.test.ts b/packages/web/worker/src/index.test.ts new file mode 100644 index 00000000..67b0160c --- /dev/null +++ b/packages/web/worker/src/index.test.ts @@ -0,0 +1,117 @@ +import { afterEach, expect, test, vi } from 'vitest' +import { promises as fs } from 'node:fs' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { createConnection, type AddressInfo, type Socket } from 'node:net' +import { once } from 'node:events' +import { defineWorkflowSpec } from 'openworkflow' +import { BackendSqlite } from 'openworkflow/sqlite' +import { validateConfig } from '@luzzle/web.config' +import { configSchema } from '@luzzle/web.templates/config' +import type { main as startWorker } from './index.js' +import type * as Workflows from './workflows/index.js' + +const probe = defineWorkflowSpec({ name: 'ShutdownProbe' }) + +afterEach(() => { + 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() + const directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-worker-')) + const queuePath = path.join(directory, 'queue.sqlite') + const config = validateConfig( + { + storage: { root: directory }, + paths: { + config: path.join(directory, 'config.yaml'), + database: path.join(directory, 'app.sqlite'), + assets: path.join(directory, 'assets/pieces'), + }, + worker: { queue: { path: queuePath } }, + }, + {}, + { schema: configSchema } + ) + const jobs = await import('@luzzle/web.jobs') + let release!: () => void + const gate = new Promise((resolve) => { + release = resolve + }) + let entered = false + vi.doMock('./workflows/index.js', async () => { + const actual = await vi.importActual('./workflows/index.js') + return { + registerWorkflows() { + actual.registerWorkflows() + jobs.getOpenWorkflow().implementWorkflow(probe, async ({ step }) => + step.run({ name: 'active-work' }, async () => { + entered = true + await gate + return 'finished' + }) + ) + }, + } + }) + let running: Awaited> | undefined + let stalledHealth: Socket | undefined + try { + const { main } = await import('./index.js') + running = await main(config, 0) + 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') + const handle = await jobs.getOpenWorkflow().runWorkflow(probe) + await expect.poll(() => entered).toBe(true) + + const accepted = once(running.server, 'connection') + stalledHealth = createConnection(port, '127.0.0.1') + stalledHealth.on('error', () => {}) + await once(stalledHealth, 'connect') + await accepted + stalledHealth.write('GET /health HTTP/1.1\r\nHost: localhost\r\n') + let healthConnectionClosed = false + stalledHealth.once('close', () => { + healthConnectionClosed = true + }) + + const stopping = running.stop() + expect(running.stop()).toBe(stopping) + await expect.poll(() => running!.server.listening).toBe(false) + await expect.poll(() => healthConnectionClosed).toBe(true) + const { getWorkerContext } = await import('./services/context.js') + expect(await getWorkerContext().db.selectFrom('job_progress').selectAll().execute()).toEqual([]) + expect(await backend.getWorkflowRun({ workflowRunId: handle.workflowRun.id })).toMatchObject({ + status: 'running', + }) + + release() + await stopping + expect(() => jobs.getOpenWorkflow()).toThrow(/closed/) + const reopened = BackendSqlite.connect(queuePath) + try { + expect(await reopened.getWorkflowRun({ workflowRunId: handle.workflowRun.id })).toMatchObject( + { status: 'completed', output: 'finished' } + ) + expect( + (await reopened.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data + ).toHaveLength(1) + } finally { + await reopened.stop() + } + } finally { + release() + stalledHealth?.destroy() + await running?.stop() + await fs.rm(directory, { recursive: true, force: true }) + } +}, 15_000) diff --git a/packages/web/worker/src/index.ts b/packages/web/worker/src/index.ts index b020af54..40547cb4 100644 --- a/packages/web/worker/src/index.ts +++ b/packages/web/worker/src/index.ts @@ -1,83 +1,131 @@ -import { getDatabaseClient, migrate } from '@luzzle/core' -import { loadConfig } from '@luzzle/web.templates/config' -import { resolveDbPath, resolveOpenWorkflowDbPath } from './services/db.js' +import { asCoreDatabase, migrate } from '@luzzle/core' +import { loadConfig, type TemplateConfig } from '@luzzle/web.templates/config' +import { createAppDb, resolveDbPath, resolveOpenWorkflowDbPath } from './services/db.js' import { runWebMigrations } from '@luzzle/web.db' import { createHealthServer } from './services/health.js' -import { log } from './services/logger.js' -import { initOpenWorkflow } from '@luzzle/web.jobs' +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 main() { - const config = loadConfig('./config.yaml') - - const db = getDatabaseClient(resolveDbPath(config)) - - const coreResult = await migrate(db) - if (coreResult.error) { - throw new Error(`luzzle core migration failed: ${coreResult.error}`) - } - log('info', 'luzzle core migrations applied') - - const webResult = await runWebMigrations(db) - if (webResult.error) { - throw new Error(`web migration failed: ${webResult.error}`) - } - log('info', 'web migrations applied') - - const openWorkflowDb = resolveOpenWorkflowDbPath(config) - const openWorkflow = initOpenWorkflow({ dbPath: openWorkflowDb }) - log('info', 'openworkflow client configured', { openWorkflowDb }) - - registerWorkflows() - - const openWorkflowWorker = openWorkflow.newWorker() - await openWorkflowWorker.start() - log('info', 'openworkflow worker started') - - const runPurge = async () => { - try { - log('info', 'triggering openworkflow job progress purge...') - await openWorkflow.runWorkflow(jobProgressPurgeSpec, { retentionDays: PURGE_RETENTION_DAYS }) - log('info', 'openworkflow job progress purge run completed successfully') - } catch (err) { - log('error', 'openworkflow job progress purge run failed', { - error: err instanceof Error ? err.message : String(err), - }) - } - } - - // Trigger immediately on start (non-blocking) - runPurge() - - // Schedule to run every 24 hours - const purgeInterval = setInterval(runPurge, 24 * 60 * 60 * 1000) - - const shutdown = async () => { - log('info', 'shutting down openworkflow worker...') - clearInterval(purgeInterval) - await openWorkflowWorker.stop() - log('info', 'openworkflow worker stopped') - process.exit(0) - } - process.on('SIGTERM', shutdown) - process.on('SIGINT', shutdown) - - const port = Number(process.env.PORT) || DEFAULT_PORT - const server = createHealthServer() - server.listen(port, () => { - log('info', 'health server listening', { port }) +async function closeHealthServer(server: Server): Promise { + if (!server.listening) return + await new Promise((resolve, reject) => { + server.close((error) => { + if (error) reject(error) + else resolve() + }) + server.closeAllConnections() }) } +export async function main( + config: TemplateConfig = loadConfig('./config.yaml'), + port: number = Number(process.env.PORT) || DEFAULT_PORT +) { + const logger = createLogger() + 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[] = [] + try { + await closeOpenWorkflow() + } catch (error) { + errors.push(error) + } + try { + await db.destroy() + } catch (error) { + errors.push(error) + } + if (errors.length) throw new AggregateError(errors, 'Worker database cleanup failed') + logger.info('openworkflow worker stopped') + } + + function stop(): Promise { + stopPromise ??= stopResources() + return stopPromise + } + + try { + const coreResult = await migrate(asCoreDatabase(db)) + if (coreResult.error) + throw new Error('Luzzle core migration failed', { cause: coreResult.error }) + logger.info('luzzle core migrations applied') + const webResult = await runWebMigrations(db) + if (webResult.error) throw new Error('Web migration failed', { cause: webResult.error }) + logger.info('web migrations applied') + + setWorkerContext(createWorkerContext(config, db, logger)) + const openWorkflowDb = resolveOpenWorkflowDbPath(config) + const openWorkflow = initOpenWorkflow({ + dbPath: openWorkflowDb, + logger: createOpenWorkflowLogger(logger), + }) + logger.info('openworkflow client configured', { openWorkflowDb }) + registerWorkflows() + worker = openWorkflow.newWorker() + 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 + logger.info('health server listening', { port }) + return { server, stop } + } catch (error) { + try { + await stop() + } catch (cleanupError) { + throw new AggregateError([error, cleanupError], 'Worker startup and cleanup failed') + } + throw error + } +} + if (import.meta.url === `file://${process.argv[1]}`) { - main().catch((err) => { - log('error', 'worker failed to start', { - error: err instanceof Error ? err.message : String(err), + main() + .then(({ stop }) => { + const shutdown = () => { + void stop().catch(() => { + log('error', 'worker shutdown failed; resources may still be active') + process.exitCode = 1 + }) + } + process.once('SIGTERM', shutdown) + process.once('SIGINT', shutdown) + }) + .catch((error) => { + log('error', 'worker failed to start', { + error: error instanceof Error ? error.message : String(error), + }) + process.exitCode = 1 }) - process.exit(1) - }) } diff --git a/packages/web/worker/src/services/context.test.ts b/packages/web/worker/src/services/context.test.ts index 4b55f097..739e56a6 100644 --- a/packages/web/worker/src/services/context.test.ts +++ b/packages/web/worker/src/services/context.test.ts @@ -1,7 +1,7 @@ import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' import { EventEmitter } from 'node:events' import { spawn } from 'child_process' -import { createPhaseContext, type WorkerContext } from './context.js' +import { createPhaseContext, createWorkerContext, type WorkerContext } from './context.js' import { PhaseLogger } from '../core/phase-logger.js' import type { AppDatabase } from './db.js' import { RcloneClient } from './rclone.js' @@ -38,6 +38,14 @@ describe('createPhaseContext', () => { await teardownDatabase(ctx.db) }) + test('creates a context around the application database supplied by startup', () => { + const created = createWorkerContext(ctx.config, ctx.db, baseLogger) + expect(created.db).toBe(ctx.db) + expect(created.config).toBe(ctx.config) + expect(created.logger).toBeInstanceOf(PhaseLogger) + expect(created.rclone).toBeInstanceOf(RcloneClient) + }) + test('leaves contexts with a plain logger unchanged', () => { const plain = { ...ctx, logger: baseLogger } expect(createPhaseContext(plain, 'job1', 'sync')).toBe(plain) diff --git a/packages/web/worker/src/services/context.ts b/packages/web/worker/src/services/context.ts index facfb5a7..0dbb4f03 100644 --- a/packages/web/worker/src/services/context.ts +++ b/packages/web/worker/src/services/context.ts @@ -24,6 +24,15 @@ export function createPhaseContext( return { ...ctx, logger, rclone: ctx.rclone.withLogger(logger) } } +export function createWorkerContext( + config: TemplateConfig, + db: Kysely, + baseLogger: Logger = createLogger() +): WorkerContext { + const logger = new PhaseLogger(baseLogger, db) + return { config, db, logger, rclone: new RcloneClient(logger) } +} + let workerContext: WorkerContext | null = null export function setWorkerContext(ctx: WorkerContext): void { @@ -33,14 +42,6 @@ export function setWorkerContext(ctx: WorkerContext): void { export function getWorkerContext(): WorkerContext { if (workerContext) return workerContext const config = loadConfig('./config.yaml') - const baseLogger: Logger = createLogger() - const db = createAppDb(resolveDbPath(config)) - const logger = new PhaseLogger(baseLogger, db) - workerContext = { - config, - logger, - rclone: new RcloneClient(logger), - db, - } + workerContext = createWorkerContext(config, createAppDb(resolveDbPath(config))) return workerContext } diff --git a/packages/web/worker/src/services/logger.test.ts b/packages/web/worker/src/services/logger.test.ts index 5d05d156..1434acde 100644 --- a/packages/web/worker/src/services/logger.test.ts +++ b/packages/web/worker/src/services/logger.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test, vi, afterEach } from 'vitest' -import { log, createLogger } from './logger.js' +import { log, createLogger, createOpenWorkflowLogger } from './logger.js' describe('log', () => { afterEach(() => { @@ -14,7 +14,7 @@ describe('log', () => { }) return { getLine: () => line, - spy + spy, } } @@ -54,6 +54,52 @@ describe('log', () => { }) }) +describe('createOpenWorkflowLogger', () => { + function logger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + stdout: vi.fn(), + stderr: vi.fn(), + } + } + + test('routes native levels without forwarding engine messages or arguments', () => { + const base = logger() + const engine = createOpenWorkflowLogger(base) + for (const level of ['debug', 'info', 'warn', 'error'] as const) { + engine[level]('private payload', new Error('private credential'), { input: 'secret' }) + expect(base[level]).toHaveBeenCalledWith('OpenWorkflow operational log') + } + }) + + test('keeps known operational categories without error details', () => { + const base = logger() + const engine = createOpenWorkflowLogger(base) + engine.error('Worker tick failed:', new Error('private path')) + engine.warn('Worker schedule tick failed:', { secret: 'private' }) + engine.error('Heartbeat failed:', 'private error') + engine.error('Critical error during workflow execution for run private-value:', 'private error') + expect(base.error.mock.calls).toEqual([ + ['OpenWorkflow worker tick failed'], + ['OpenWorkflow heartbeat failed'], + ['OpenWorkflow execution failed'], + ]) + expect(base.warn).toHaveBeenCalledWith('OpenWorkflow schedule tick failed') + }) + + test('accepts console-style calls with no message or non-text values', () => { + const base = logger() + const engine = createOpenWorkflowLogger(base) + engine.error() + engine.info({ secret: 'private' }) + expect(base.error).toHaveBeenCalledWith('OpenWorkflow operational log') + expect(base.info).toHaveBeenCalledWith('OpenWorkflow operational log') + }) +}) + describe('createLogger', () => { afterEach(() => { vi.restoreAllMocks() diff --git a/packages/web/worker/src/services/logger.ts b/packages/web/worker/src/services/logger.ts index 6af64ac6..8d1e82af 100644 --- a/packages/web/worker/src/services/logger.ts +++ b/packages/web/worker/src/services/logger.ts @@ -1,3 +1,5 @@ +import type { Logger as OpenWorkflowLogger } from 'openworkflow' + export type LogLevel = 'debug' | 'info' | 'warn' | 'error' | 'stdout' | 'stderr' export interface Logger { @@ -9,27 +11,58 @@ export interface Logger { stderr(message: string, fields?: Record): void } -export function log( - level: LogLevel, - message: string, - fields: Record = {} -): void { +export function log(level: LogLevel, message: string, fields: Record = {}): void { const line = JSON.stringify({ ts: new Date().toISOString(), level, message, - ...fields + ...fields, }) process.stdout.write(line + '\n') } -export function createLogger(): Logger { +const engineMessages = new Map([ + ['Worker tick failed:', 'OpenWorkflow worker tick failed'], + ['Worker schedule tick failed:', 'OpenWorkflow schedule tick failed'], + ['Heartbeat failed:', 'OpenWorkflow heartbeat failed'], +]) + +function openWorkflowMessage(message: unknown): string { + if (typeof message !== 'string') return 'OpenWorkflow operational log' + if (message.startsWith('Critical error during workflow execution for run ')) { + return 'OpenWorkflow execution failed' + } + return engineMessages.get(message) ?? 'OpenWorkflow operational log' +} + +export function createOpenWorkflowLogger(logger: Logger): OpenWorkflowLogger { return { - debug(message, fields) { log('debug', message, fields) }, - info(message, fields) { log('info', message, fields) }, - warn(message, fields) { log('warn', message, fields) }, - error(message, fields) { log('error', message, fields) }, - stdout(message, fields) { log('stdout', message, fields) }, - stderr(message, fields) { log('stderr', message, fields) }, + debug: (message) => logger.debug(openWorkflowMessage(message)), + info: (message) => logger.info(openWorkflowMessage(message)), + warn: (message) => logger.warn(openWorkflowMessage(message)), + error: (message) => logger.error(openWorkflowMessage(message)), + } +} + +export function createLogger(): Logger { + return { + debug(message, fields) { + log('debug', message, fields) + }, + info(message, fields) { + log('info', message, fields) + }, + warn(message, fields) { + log('warn', message, fields) + }, + error(message, fields) { + log('error', message, fields) + }, + stdout(message, fields) { + log('stdout', message, fields) + }, + stderr(message, fields) { + log('stderr', message, fields) + }, } } diff --git a/packages/web/worker/src/workflows/job-progress-purge.ts b/packages/web/worker/src/workflows/job-progress-purge.ts index caa800ab..8847d889 100644 --- a/packages/web/worker/src/workflows/job-progress-purge.ts +++ b/packages/web/worker/src/workflows/job-progress-purge.ts @@ -1,11 +1,9 @@ -import { getOpenWorkflow, purgeExpiredWorkflowRuns } from '@luzzle/web.jobs' +import { getOpenWorkflow } 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 { promises as fs } from 'node:fs' import path from 'node:path' -import { resolveOpenWorkflowDbPath } from '../services/db.js' -import { DatabaseSync } from 'node:sqlite' export function registerJobProgressPurgeWorkflow(): void { const openWorkflow = getOpenWorkflow() @@ -15,12 +13,10 @@ export function registerJobProgressPurgeWorkflow(): void { const { db, logger, config } = ctx const retentionDays = input.retentionDays ?? 7 - // 1. Purge old entries from database const purgedJobIds = await step.run({ name: 'purge-db-entries' }, async () => { return await new JobProgress(db, retentionDays).purgeExpired() }) - // 2. Clean up preview directories on disk await step.run({ name: 'purge-disk-directories' }, async () => { for (const jobId of purgedJobIds) { const previewDir = path.join( @@ -40,21 +36,6 @@ export function registerJobProgressPurgeWorkflow(): void { } }) - // 3. Clean up OpenWorkflow database entries - await step.run({ name: 'purge-openworkflow-entries' }, async () => { - try { - const openWorkflowDbPath = resolveOpenWorkflowDbPath(config) - const openWorkflowDb = new DatabaseSync(openWorkflowDbPath) - openWorkflowDb.exec('PRAGMA journal_mode = WAL;') - openWorkflowDb.exec('PRAGMA synchronous = NORMAL;') - openWorkflowDb.exec('PRAGMA busy_timeout = 5000;') - const count = purgeExpiredWorkflowRuns(openWorkflowDb, retentionDays) - logger.info('purged openworkflow runs', { count }) - } catch (err) { - logger.error('failed to purge openworkflow runs', { error: String(err) }) - } - }) - logger.info('openworkflow purge complete', { retentionDays, purgedCount: purgedJobIds.length, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 239ccebc..e9eabd09 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -291,8 +291,8 @@ importers: specifier: 4.0.2 version: 4.0.2 openworkflow: - specifier: 0.9.2 - version: 0.9.2 + specifier: 0.11.0 + version: 0.11.0 devDependencies: '@auth/core': specifier: 0.41.1 @@ -552,8 +552,8 @@ importers: specifier: workspace:* version: link:../../../apps/core openworkflow: - specifier: 0.9.2 - version: 0.9.2 + specifier: 0.11.0 + version: 0.11.0 devDependencies: '@eslint/js': specifier: 'catalog:' @@ -865,8 +865,8 @@ importers: specifier: 3.0.2 version: 3.0.2 openworkflow: - specifier: 0.9.2 - version: 0.9.2 + specifier: 0.11.0 + version: 0.11.0 rehype-autolink-headings: specifier: 7.1.0 version: 7.1.0 @@ -2635,6 +2635,10 @@ packages: crelt@1.0.6: resolution: {integrity: sha512-VQ2MBenTq1fWZUH9DJNGti7kKv6EeAuYr3cLwxUWhIu1baTaXh4Ib5W2CqHVqib4/MqbYGJqiL3Zb8GJZr3l4g==} + cron-parser@5.10.1: + resolution: {integrity: sha512-pKRrRagItwk9rGIStcUyMxFX34x56zoX3KDf9HJ4ttwkXIK1qxK63EvXvEDmlxI9AdPJ+Y7TEKRftRlW5eSS7A==} + engines: {node: '>=18'} + cross-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -3314,6 +3318,10 @@ packages: lru-cache@10.4.3: resolution: {integrity: sha512-JNAzZcXrCt42VGLuYz0zfAzDfAvJWW6AfYlDBQyDV5DClI2m5sAmK+OIO7s59XfsRsWHp02jAJrRadPRGTt6SQ==} + luxon@3.7.2: + resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==} + engines: {node: '>=12'} + lz-string@1.5.0: resolution: {integrity: sha512-h5bgJWpxJNswbU7qCrV0tIKQCaS3blPDrqKWx+QxzuzL1zGUzij9XCWLrSLsJPu5t+eWA/ycetzYAO5IOMcWAQ==} hasBin: true @@ -3580,12 +3588,15 @@ packages: oniguruma-to-es@4.3.6: resolution: {integrity: sha512-csuQ9x3Yr0cEIs/Zgx/OEt9iBw9vqIunAPQkx19R/fiMq2oGVTgcMqO/V3Ybqefr1TBvosI6jU539ksaBULJyA==} - openworkflow@0.9.2: - resolution: {integrity: sha512-NluYomuVLf+rAMJ9ASKqw8nbPAAD+C/HSTZoJacZdelLfTpndzHvPcP8SCANg9me9zek6mao0uUOGGbrWUWhGw==} - engines: {node: '>=20'} + openworkflow@0.11.0: + resolution: {integrity: sha512-fdDfRjuNMLV+p3svZShrV5i/ZPbBS8GruWLlthk8Ezt3ueYPwVKk0mcD+knAmQ4J9RASBsY5JqiatiLtEcwgVw==} + engines: {node: '>=22.22.0'} peerDependencies: + '@opentelemetry/api': '1' postgres: ^3.4.9 peerDependenciesMeta: + '@opentelemetry/api': + optional: true postgres: optional: true @@ -6273,6 +6284,10 @@ snapshots: crelt@1.0.6: {} + cron-parser@5.10.1: + dependencies: + luxon: 3.7.2 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -7001,6 +7016,8 @@ snapshots: lru-cache@10.4.3: {} + luxon@3.7.2: {} + lz-string@1.5.0: {} magic-string@0.30.21: @@ -7503,7 +7520,9 @@ snapshots: regex: 6.1.0 regex-recursion: 6.0.2 - openworkflow@0.9.2: {} + openworkflow@0.11.0: + dependencies: + cron-parser: 5.10.1 optionator@0.9.4: dependencies: -- 2.54.0