From e553dcc3843bbe0f855f9e4f4cec7b00ff300c74 Mon Sep 17 00:00:00 2001 From: Sascha Date: Fri, 10 Jul 2026 10:47:10 +0200 Subject: [PATCH] fix: make download queue recovery durable --- sake/drizzle/0023_queue_task_recovery.sql | 3 + sake/drizzle/meta/_journal.json | 7 + .../application/composition/downloads.ts | 22 ++- .../lib/server/infrastructure/db/schema.ts | 2 + .../infrastructure/queue/downloadQueue.ts | 129 +++++++++--------- .../infrastructure/queue/persistence.ts | 45 +++++- .../repositories/QueueJobRepository.ts | 89 ++++++++---- sake/tests/queue/queuePersistence.test.ts | 30 +++- 8 files changed, 226 insertions(+), 101 deletions(-) create mode 100644 sake/drizzle/0023_queue_task_recovery.sql diff --git a/sake/drizzle/0023_queue_task_recovery.sql b/sake/drizzle/0023_queue_task_recovery.sql new file mode 100644 index 0000000..6edc389 --- /dev/null +++ b/sake/drizzle/0023_queue_task_recovery.sql @@ -0,0 +1,3 @@ +ALTER TABLE `QueueJobs` ADD `task_type` text; +--> statement-breakpoint +ALTER TABLE `QueueJobs` ADD `task_payload` text; diff --git a/sake/drizzle/meta/_journal.json b/sake/drizzle/meta/_journal.json index a29087a..1cf4b1f 100644 --- a/sake/drizzle/meta/_journal.json +++ b/sake/drizzle/meta/_journal.json @@ -162,6 +162,13 @@ "when": 1782627929457, "tag": "0022_hardcover_progress_sync", "breakpoints": true + }, + { + "idx": 23, + "version": "7", + "when": 1782627929458, + "tag": "0023_queue_task_recovery", + "breakpoints": true } ] } diff --git a/sake/src/lib/server/application/composition/downloads.ts b/sake/src/lib/server/application/composition/downloads.ts index 865775d..7a9ccb6 100644 --- a/sake/src/lib/server/application/composition/downloads.ts +++ b/sake/src/lib/server/application/composition/downloads.ts @@ -1,5 +1,6 @@ import { DavUploadServiceFactory } from '$lib/server/infrastructure/factories/DavUploadServiceFactory'; -import { downloadQueue } from '$lib/server/infrastructure/queue/downloadQueue'; +import { DownloadQueue } from '$lib/server/infrastructure/queue/downloadQueue'; +import { QueueJobRepository } from '$lib/server/infrastructure/repositories/QueueJobRepository'; import { DownloadBookUseCase } from '$lib/server/application/use-cases/DownloadBookUseCase'; import { QueueDownloadUseCase } from '$lib/server/application/use-cases/QueueDownloadUseCase'; import { QueueSearchBookUseCase } from '$lib/server/application/use-cases/QueueSearchBookUseCase'; @@ -31,6 +32,7 @@ import { zlibraryClient } from './foundation'; import { externalBookMetadataService } from './providers'; +import { downloadSearchBookUseCase } from './search'; export const downloadBookUseCase = new DownloadBookUseCase( zlibraryClient, @@ -41,12 +43,6 @@ export const downloadBookUseCase = new DownloadBookUseCase( undefined, externalBookMetadataService ); -export const queueDownloadUseCase = new QueueDownloadUseCase(downloadQueue); -export const queueSearchBookUseCase = new QueueSearchBookUseCase(downloadQueue); -export const getQueueStatusUseCase = new GetQueueStatusUseCase( - downloadQueue, - hardcoverProgressSyncJobRepository -); export const getNewBooksForDeviceUseCase = new GetNewBooksForDeviceUseCase(bookRepository); export const confirmDownloadUseCase = new ConfirmDownloadUseCase(deviceDownloadRepository); export const removeDeviceDownloadUseCase = new RemoveDeviceDownloadUseCase(deviceDownloadRepository); @@ -80,6 +76,18 @@ export const putLibraryFileUseCase = new PutLibraryFileUseCase( undefined, externalBookMetadataService ); +export const downloadQueue = new DownloadQueue( + new QueueJobRepository(), + downloadBookUseCase, + downloadSearchBookUseCase, + putLibraryFileUseCase +); +export const queueDownloadUseCase = new QueueDownloadUseCase(downloadQueue); +export const queueSearchBookUseCase = new QueueSearchBookUseCase(downloadQueue); +export const getQueueStatusUseCase = new GetQueueStatusUseCase( + downloadQueue, + hardcoverProgressSyncJobRepository +); export const exportDeviceLibraryBookUseCase = new ExportDeviceLibraryBookUseCase( bookRepository, deviceDownloadRepository, diff --git a/sake/src/lib/server/infrastructure/db/schema.ts b/sake/src/lib/server/infrastructure/db/schema.ts index d106d82..8620db1 100644 --- a/sake/src/lib/server/infrastructure/db/schema.ts +++ b/sake/src/lib/server/infrastructure/db/schema.ts @@ -239,6 +239,8 @@ export const queueJobs = sqliteTable( year: integer('year'), userId: text('user_id').notNull(), userKey: text('user_key').notNull(), + taskType: text('task_type', { enum: ['zlibrary', 'provider-import'] }), + taskPayload: text('task_payload'), status: text('status', { enum: ['queued', 'processing', 'completed', 'failed'] }).notNull(), diff --git a/sake/src/lib/server/infrastructure/queue/downloadQueue.ts b/sake/src/lib/server/infrastructure/queue/downloadQueue.ts index 3455cb1..e56f59b 100644 --- a/sake/src/lib/server/infrastructure/queue/downloadQueue.ts +++ b/sake/src/lib/server/infrastructure/queue/downloadQueue.ts @@ -1,25 +1,19 @@ import { DownloadBookUseCase } from '$lib/server/application/use-cases/DownloadBookUseCase'; import { DownloadSearchBookUseCase } from '$lib/server/application/use-cases/DownloadSearchBookUseCase'; import { PutLibraryFileUseCase } from '$lib/server/application/use-cases/PutLibraryFileUseCase'; -import { ManagedBookCoverService } from '$lib/server/application/services/ManagedBookCoverService'; -import { ZLibraryClient } from '$lib/server/infrastructure/clients/ZLibraryClient'; import { createChildLogger, toLogError } from '$lib/server/infrastructure/logging/logger'; import { QueueJobRepository } from '$lib/server/infrastructure/repositories/QueueJobRepository'; import type { QueueJobRecord } from '$lib/server/infrastructure/repositories/QueueJobRepository'; -import { BookRepository } from '$lib/server/infrastructure/repositories/BookRepository'; -import { S3Storage } from '$lib/server/infrastructure/storage/S3Storage'; -import { DavUploadServiceFactory } from '$lib/server/infrastructure/factories/DavUploadServiceFactory'; import { RECOVERY_REQUEUE_REQUIRED_ERROR, - PERSISTED_QUEUE_USER_KEY + parseQueueTask, + serializeQueueTask } from '$lib/server/infrastructure/queue/persistence'; -import { createLazySingleton } from '$lib/server/utils/createLazySingleton'; import type { + DownloadQueueTaskInput, SearchImportQueueTaskInput, ZLibraryQueueTaskInput } from '$lib/server/application/ports/DownloadQueuePort'; -import { createSearchProviders } from '$lib/server/infrastructure/search-providers/searchProviderFactory'; -import { SEARCH_PROVIDER_IDS } from '$lib/types/Search/Provider'; import { randomUUID } from 'node:crypto'; interface BaseQueuedDownload { @@ -78,48 +72,24 @@ export interface QueuedDownloadSnapshot { finishedAt?: string; } -class DownloadQueue { - private queue: QueuedDownload[] = []; +export class DownloadQueue { private isProcessing = false; private readonly defaultMaxAttempts = 3; private readonly terminalRetentionMs = 90 * 24 * 60 * 60 * 1000; private readonly queueLogger = createChildLogger({ component: 'downloadQueue' }); - private readonly queueJobRepository = new QueueJobRepository(); private isInitialized = false; private initializePromise: Promise | null = null; - private readonly storage = new S3Storage(); - private readonly bookRepository = new BookRepository(); - private readonly managedBookCoverService = new ManagedBookCoverService(this.storage); - private readonly zlibraryClient = new ZLibraryClient('https://1lib.sk'); - - private readonly downloadBookUseCase = new DownloadBookUseCase( - this.zlibraryClient, - this.bookRepository, - this.storage, - () => DavUploadServiceFactory.createS3(), - this.managedBookCoverService - ); - private readonly downloadSearchBookUseCase = new DownloadSearchBookUseCase( - createSearchProviders([...SEARCH_PROVIDER_IDS], { - zlibrary: this.zlibraryClient - }) - ); - private readonly putLibraryFileUseCase = new PutLibraryFileUseCase( - this.storage, - this.bookRepository, - this.managedBookCoverService - ); + private readonly volatileZLibraryCredentials = new Map(); + + constructor( + private readonly queueJobRepository: QueueJobRepository, + private readonly downloadBookUseCase: DownloadBookUseCase, + private readonly downloadSearchBookUseCase: DownloadSearchBookUseCase, + private readonly putLibraryFileUseCase: PutLibraryFileUseCase + ) {} async enqueue( - task: - | Omit< - QueuedZLibraryDownload, - 'id' | 'status' | 'attempts' | 'maxAttempts' | 'createdAt' | 'updatedAt' | 'finishedAt' - > - | Omit< - QueuedSearchImportDownload, - 'id' | 'status' | 'attempts' | 'maxAttempts' | 'createdAt' | 'updatedAt' | 'finishedAt' - > + task: DownloadQueueTaskInput ): Promise { await this.ensureInitialized(); @@ -136,7 +106,9 @@ class DownloadQueue { updatedAt: now }; - this.queue.push(queuedTask); + if (task.source === 'zlibrary') { + this.volatileZLibraryCredentials.set(id, { userId: task.userId, userKey: task.userKey }); + } await this.queueJobRepository.create(this.toQueueJobRecord(queuedTask)); this.queueLogger.info( { event: 'queue.task.enqueued', taskId: id, bookId: task.bookId, title: task.title }, @@ -185,21 +157,17 @@ class DownloadQueue { private async initialize(): Promise { try { const nowIso = new Date().toISOString(); - const failedRecoveredJobs = await this.queueJobRepository.failActiveJobsAfterRestart( - RECOVERY_REQUEUE_REQUIRED_ERROR, - nowIso, - nowIso - ); + const failedRecoveredJobs = await this.queueJobRepository.recoverActiveJobsAfterRestart(nowIso, nowIso); await this.cleanupOldTerminalJobs(); this.isInitialized = true; if (failedRecoveredJobs > 0) { this.queueLogger.warn( { - event: 'queue.recovery.credentials_missing', + event: 'queue.recovery.non_resumable', failedRecoveredJobs }, - 'Marked queued jobs as failed because queue state cannot resume after restart' + 'Marked queue jobs without resumable payloads as failed after restart' ); } } catch (error: unknown) { @@ -216,13 +184,17 @@ class DownloadQueue { this.isProcessing = true; try { while (true) { - const task = this.queue.find((candidate) => candidate.status === 'queued'); - if (!task) { - break; - } + const record = await this.queueJobRepository.claimNextQueued(); + if (!record) break; + const task = this.toQueuedDownload(record); + if (!task) { + const now = new Date().toISOString(); + await this.queueJobRepository.updateFailed(record.id, record.attempts, RECOVERY_REQUEUE_REQUIRED_ERROR, now, now); + continue; + } - task.status = 'processing'; - task.updatedAt = new Date(); + task.status = 'processing'; + task.updatedAt = new Date(); this.queueLogger.info( { event: 'queue.task.processing', taskId: task.id, bookId: task.bookId, title: task.title }, 'Processing queue task' @@ -270,9 +242,7 @@ class DownloadQueue { }, 'Queue task failed' ); - } finally { - this.queue = this.queue.filter((candidate) => candidate.id !== task.id); - } + } } await this.cleanupOldTerminalJobs(); } finally { @@ -356,7 +326,7 @@ class DownloadQueue { year: task.year ?? undefined, downloadToDevice: false }, - credentials: { + credentials: this.volatileZLibraryCredentials.get(task.id) ?? { userId: task.userId, userKey: task.userKey } @@ -458,7 +428,9 @@ class DownloadQueue { language: task.language, year: task.year, userId: task.userId, - userKey: PERSISTED_QUEUE_USER_KEY, + userKey: '', + taskType: task.source, + taskPayload: serializeQueueTask(task), status: task.status, attempts: task.attempts, maxAttempts: task.maxAttempts, @@ -469,11 +441,40 @@ class DownloadQueue { }; } + private toQueuedDownload(record: QueueJobRecord): QueuedDownload | null { + const task = parseQueueTask(record.taskPayload); + if (!task || task.source !== record.taskType || task.bookId !== record.bookId) { + return null; + } + + if (task.source === 'zlibrary') { + const credentials = this.volatileZLibraryCredentials.get(record.id); + if (!credentials) return null; + return { + ...task, + ...record, + source: 'zlibrary', + ...credentials, + createdAt: new Date(record.createdAt), + updatedAt: new Date(record.updatedAt), + finishedAt: record.finishedAt ? new Date(record.finishedAt) : undefined + }; + } + + return { + ...task, + ...record, + source: 'provider-import', + userKey: '', + createdAt: new Date(record.createdAt), + updatedAt: new Date(record.updatedAt), + finishedAt: record.finishedAt ? new Date(record.finishedAt) : undefined + }; + } + private toArrayBuffer(data: Uint8Array): ArrayBuffer { const copy = new Uint8Array(data.byteLength); copy.set(data); return copy.buffer; } } - -export const downloadQueue = createLazySingleton(() => new DownloadQueue()); diff --git a/sake/src/lib/server/infrastructure/queue/persistence.ts b/sake/src/lib/server/infrastructure/queue/persistence.ts index 788cc10..9baa194 100644 --- a/sake/src/lib/server/infrastructure/queue/persistence.ts +++ b/sake/src/lib/server/infrastructure/queue/persistence.ts @@ -1,7 +1,12 @@ export const PERSISTED_QUEUE_USER_KEY = ''; +export const QUEUE_TASK_PAYLOAD_VERSION = 1; + export const RECOVERY_REQUEUE_REQUIRED_ERROR = - 'Queued job could not resume after restart because transient queue data is not persisted. Requeue the download.'; + 'Queued job could not resume after restart because its task payload is missing or invalid. Requeue the download.'; + +export const RECOVERY_ZLIBRARY_CREDENTIALS_MISSING_ERROR = + 'Z-Library queue job could not resume after restart because its credentials are intentionally not persisted. Requeue the download.'; export function sanitizePersistedQueueJob(job: T): T { return { @@ -9,3 +14,41 @@ export function sanitizePersistedQueueJob(job: T) userKey: PERSISTED_QUEUE_USER_KEY }; } + +export interface PersistedQueueTaskPayload { + version: typeof QUEUE_TASK_PAYLOAD_VERSION; + task: T; +} + +export function serializeQueueTask(task: T): string { + const { userKey: _userKey, ...safeTask } = task; + const payload: PersistedQueueTaskPayload> = { + version: QUEUE_TASK_PAYLOAD_VERSION, + task: safeTask + }; + return JSON.stringify(payload); +} + +export function parseQueueTask(payload: string | null | undefined): T | null { + if (!payload) { + return null; + } + + try { + const parsed: unknown = JSON.parse(payload); + if ( + typeof parsed !== 'object' || + parsed === null || + !('version' in parsed) || + parsed.version !== QUEUE_TASK_PAYLOAD_VERSION || + !('task' in parsed) || + typeof parsed.task !== 'object' || + parsed.task === null + ) { + return null; + } + return parsed.task as T; + } catch { + return null; + } +} diff --git a/sake/src/lib/server/infrastructure/repositories/QueueJobRepository.ts b/sake/src/lib/server/infrastructure/repositories/QueueJobRepository.ts index a561dc3..129b382 100644 --- a/sake/src/lib/server/infrastructure/repositories/QueueJobRepository.ts +++ b/sake/src/lib/server/infrastructure/repositories/QueueJobRepository.ts @@ -3,7 +3,9 @@ import { queueJobs } from '$lib/server/infrastructure/db/schema'; import { createChildLogger } from '$lib/server/infrastructure/logging/logger'; import { PERSISTED_QUEUE_USER_KEY, - sanitizePersistedQueueJob + sanitizePersistedQueueJob, + RECOVERY_REQUEUE_REQUIRED_ERROR, + RECOVERY_ZLIBRARY_CREDENTIALS_MISSING_ERROR } from '$lib/server/infrastructure/queue/persistence'; import { and, desc, eq, inArray, sql } from 'drizzle-orm'; @@ -30,6 +32,8 @@ export interface QueueJobRecord { year: number | null; userId: string; userKey: string; + taskType: 'zlibrary' | 'provider-import' | null; + taskPayload: string | null; status: QueueJobStatus; attempts: number; maxAttempts: number; @@ -61,6 +65,8 @@ function mapQueueJobRow(row: typeof queueJobs.$inferSelect): QueueJobRecord { year: row.year ?? null, userId: row.userId, userKey: row.userKey, + taskType: row.taskType ?? null, + taskPayload: row.taskPayload ?? null, status: row.status, attempts: row.attempts, maxAttempts: row.maxAttempts, @@ -97,6 +103,8 @@ export class QueueJobRepository { year: persistedJob.year, userId: persistedJob.userId, userKey: persistedJob.userKey, + taskType: persistedJob.taskType, + taskPayload: persistedJob.taskPayload, status: persistedJob.status, attempts: persistedJob.attempts, maxAttempts: persistedJob.maxAttempts, @@ -122,7 +130,27 @@ export class QueueJobRepository { updatedAt, finishedAt: null }) - .where(eq(queueJobs.id, id)); + .where(eq(queueJobs.id, id)); + } + + async claimNextQueued(): Promise { + const [candidate] = await drizzleDb + .select() + .from(queueJobs) + .where(eq(queueJobs.status, 'queued')) + .orderBy(queueJobs.createdAt) + .limit(1); + if (!candidate) { + return null; + } + + const now = new Date().toISOString(); + const [claimed] = await drizzleDb + .update(queueJobs) + .set({ status: 'processing', updatedAt: now, error: null, finishedAt: null }) + .where(and(eq(queueJobs.id, candidate.id), eq(queueJobs.status, 'queued'))) + .returning(); + return claimed ? mapQueueJobRow(claimed) : null; } async updateCompleted(id: string, attempts: number, updatedAt: string, finishedAt: string): Promise { @@ -159,37 +187,42 @@ export class QueueJobRepository { .where(eq(queueJobs.id, id)); } - async failActiveJobsAfterRestart( - error: string, - updatedAt: string, - finishedAt: string - ): Promise { - const activeStatuses: QueueJobStatus[] = ['queued', 'processing']; - const condition = inArray(queueJobs.status, activeStatuses); + async recoverActiveJobsAfterRestart(updatedAt: string, finishedAt: string): Promise { + const resumableCondition = and( + inArray(queueJobs.status, ['queued', 'processing']), + sql`${queueJobs.taskPayload} is not null`, + sql`${queueJobs.taskType} = 'provider-import'` + ); + await drizzleDb + .update(queueJobs) + .set({ status: 'queued', updatedAt, error: null, finishedAt: null }) + .where(resumableCondition); + + const nonResumableCondition = and( + inArray(queueJobs.status, ['queued', 'processing']), + sql`(${queueJobs.taskPayload} is null or ${queueJobs.taskType} = 'zlibrary')` + ); const [countRow] = await drizzleDb .select({ count: sql`count(*)` }) .from(queueJobs) - .where(condition); + .where(nonResumableCondition); const count = Number(countRow?.count ?? 0); - if (count === 0) { - return 0; + if (count > 0) { + await drizzleDb + .update(queueJobs) + .set({ + status: 'failed', + error: sql`case when ${queueJobs.taskType} = 'zlibrary' then ${RECOVERY_ZLIBRARY_CREDENTIALS_MISSING_ERROR} else ${RECOVERY_REQUEUE_REQUIRED_ERROR} end`, + userKey: PERSISTED_QUEUE_USER_KEY, + updatedAt, + finishedAt + }) + .where(nonResumableCondition); + this.repoLogger.warn( + { event: 'queue_job.recovery.failed', count, updatedAt }, + 'Marked queue jobs without resumable payloads as failed' + ); } - - await drizzleDb - .update(queueJobs) - .set({ - status: 'failed', - error, - userKey: PERSISTED_QUEUE_USER_KEY, - updatedAt, - finishedAt - }) - .where(condition); - - this.repoLogger.warn( - { event: 'queue_job.recovery.failed', count, updatedAt }, - 'Marked queued jobs as failed because queue state cannot resume after restart' - ); return count; } diff --git a/sake/tests/queue/queuePersistence.test.ts b/sake/tests/queue/queuePersistence.test.ts index 2f7506c..f0b267e 100644 --- a/sake/tests/queue/queuePersistence.test.ts +++ b/sake/tests/queue/queuePersistence.test.ts @@ -2,8 +2,10 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; import { PERSISTED_QUEUE_USER_KEY, + parseQueueTask, RECOVERY_REQUEUE_REQUIRED_ERROR, - sanitizePersistedQueueJob + sanitizePersistedQueueJob, + serializeQueueTask } from '$lib/server/infrastructure/queue/persistence'; describe('queue persistence', () => { @@ -23,4 +25,30 @@ describe('queue persistence', () => { assert.match(RECOVERY_REQUEUE_REQUIRED_ERROR, /requeue/i); assert.match(RECOVERY_REQUEUE_REQUIRED_ERROR, /restart/i); }); + + test('versioned task payloads round-trip without credentials', () => { + const payload = serializeQueueTask({ + source: 'provider-import' as const, + userKey: 'not-persisted', + userId: 'user-1', + bookId: 'book-1', + title: 'A book', + extension: 'epub' + }); + + assert.equal(payload.includes('not-persisted'), false); + assert.deepEqual(parseQueueTask(payload), { + source: 'provider-import', + userId: 'user-1', + bookId: 'book-1', + title: 'A book', + extension: 'epub' + }); + }); + + test('invalid or unknown task payloads are rejected', () => { + assert.equal(parseQueueTask('{"version":999,"task":{}}'), null); + assert.equal(parseQueueTask('{"version":1,"task":null}'), null); + assert.equal(parseQueueTask('not-json'), null); + }); });