From 499306379b62911406abc2f1223a50115e188f19 Mon Sep 17 00:00:00 2001 From: hmjn Date: Fri, 14 Aug 2026 13:32:18 +0900 Subject: [PATCH 1/4] fix(transfer): stream exports and clean orphaned artifacts --- .../application/services/backup-service.ts | 326 ++++++++++++------ .../services/job-transfer-storage.ts | 221 +++++++++++- .../services/source-transfer-job-service.ts | 35 +- .../src/infrastructure/jobs/job-worker.ts | 47 ++- .../src/infrastructure/storage/local.ts | 13 + .../src/infrastructure/storage/schema.ts | 8 + .../services/job-transfer-storage.test.ts | 116 +++++++ .../infrastructure/jobs/job-worker.test.ts | 52 +++ 8 files changed, 697 insertions(+), 121 deletions(-) create mode 100644 apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts diff --git a/apps/server/src/application/services/backup-service.ts b/apps/server/src/application/services/backup-service.ts index b314875a2..89636bb8d 100644 --- a/apps/server/src/application/services/backup-service.ts +++ b/apps/server/src/application/services/backup-service.ts @@ -1,5 +1,8 @@ +import { createReadStream, createWriteStream } from "node:fs"; import fs from "node:fs/promises"; import path from "node:path"; +import type { Writable } from "node:stream"; +import { finished } from "node:stream/promises"; import { type AuthorPlatform, type MediaDumpItem, @@ -10,6 +13,7 @@ import { getErrorMessage } from "@solid-imager/core/utils/get-error-message"; import type { Table } from "drizzle-orm"; import { and, eq, gt, inArray, sql } from "drizzle-orm"; import type { PgColumn } from "drizzle-orm/pg-core"; +import { getJobTransferRoot } from "~/application/services/job-transfer-storage"; import { db } from "~/infrastructure/db"; import { authorAccounts, @@ -127,6 +131,15 @@ type BackupDbClient = Pick< "delete" | "insert" | "query" | "select" | "update" >; +type TarArchive = import("archiver").Archiver; +type TarEntryData = import("archiver").TarEntryData; + +const TarStagingDirectory = path.resolve( + getJobTransferRoot(), + "..", + "tar-staging", +); + // const _IMAGES_PREFIX = /^images\//; /** @@ -150,6 +163,160 @@ function validateArchiveEntries(entries: string[]): void { } } +async function writeWithBackpressure( + stream: Writable, + chunk: string, +): Promise { + if (stream.write(chunk)) { + return; + } + + await new Promise((resolve, reject) => { + const cleanup = () => { + stream.removeListener("drain", onDrain); + stream.removeListener("error", onError); + stream.removeListener("close", onClose); + }; + const onDrain = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const onClose = () => { + cleanup(); + reject(new Error("Dump stream closed before it drained")); + }; + + stream.once("drain", onDrain); + stream.once("error", onError); + stream.once("close", onClose); + }); +} + +async function* iterateMediaDumpItems( + mediaSourceId: string, + transform: (mediaList: Partial[]) => MediaDumpItem[], +): AsyncGenerator { + const limit = 1000; + let lastId: string | null = null; + let hasMore = true; + + while (hasMore) { + const mediaList: MediaListQueryItem[] = await db.query.medias.findMany({ + where: lastId + ? and(eq(medias.mediaSourceId, mediaSourceId), gt(medias.id, lastId)) + : eq(medias.mediaSourceId, mediaSourceId), + limit, + with: { + generationInfo: true, + urls: true, + tags: { with: { tag: true } }, + authors: { + with: { author: { with: { accounts: true } } }, + }, + characters: { + with: { + character: { with: { ips: { with: { ip: true } } } }, + }, + }, + ips: { with: { ip: true } }, + projects: { with: { project: true } }, + }, + orderBy: medias.id, + }); + + if (mediaList.length < limit) { + hasMore = false; + } + lastId = mediaList.at(-1)?.id ?? lastId; + + if (mediaList.length > 0) { + yield* transform(mediaList); + } + } +} + +async function writeNdjsonDump( + mediaSourceId: string, + outputPath: string, + transform: (mediaList: Partial[]) => MediaDumpItem[], +): Promise { + const output = createWriteStream(outputPath); + try { + for await (const item of iterateMediaDumpItems(mediaSourceId, transform)) { + await writeWithBackpressure(output, `${JSON.stringify(item)}\n`); + } + output.end(); + await finished(output); + } catch (error) { + output.destroy(); + throw error; + } +} + +async function* iterateNdjsonDumpItems( + inputPath: string, +): AsyncGenerator { + const readline = await import("node:readline"); + const input = createReadStream(inputPath); + const lines = readline.createInterface({ + input, + crlfDelay: Infinity, + }); + + try { + for await (const line of lines) { + if (line.trim()) { + yield mediaDumpItemSchema.parse(JSON.parse(line)); + } + } + } finally { + lines.close(); + input.destroy(); + } +} + +function appendTarEntry( + archive: TarArchive, + source: import("node:stream").Readable, + data: TarEntryData, +): Promise { + return new Promise((resolve, reject) => { + const cleanup = () => { + archive.removeListener("entry", onEntry); + archive.removeListener("error", onArchiveError); + source.removeListener("error", onSourceError); + }; + const onEntry = (entry: TarEntryData) => { + if (entry.name !== data.name) { + return; + } + cleanup(); + resolve(); + }; + const onArchiveError = (error: Error) => { + cleanup(); + reject(error); + }; + const onSourceError = (error: Error) => { + cleanup(); + reject(error); + }; + + archive.on("entry", onEntry); + archive.once("error", onArchiveError); + source.once("error", onSourceError); + try { + archive.append(source, data); + } catch (error) { + onArchiveError(error instanceof Error ? error : new Error(String(error))); + } + }); +} + /** * Service for handling media source backups, restoration, and imports. */ @@ -931,51 +1098,16 @@ export const BackupService = { const { PassThrough } = await import("node:stream"); const passThrough = new PassThrough(); - (async () => { + void (async () => { try { - const limit = 1000; - let lastId: string | null = null; - let hasMore = true; - - while (hasMore) { - const mediaList: MediaListQueryItem[] = - await db.query.medias.findMany({ - where: lastId - ? and( - eq(medias.mediaSourceId, mediaSourceId), - gt(medias.id, lastId), - ) - : eq(medias.mediaSourceId, mediaSourceId), - limit, - with: { - generationInfo: true, - urls: true, - tags: { with: { tag: true } }, - authors: { - with: { author: { with: { accounts: true } } }, - }, - characters: { - with: { - character: { with: { ips: { with: { ip: true } } } }, - }, - }, - ips: { with: { ip: true } }, - projects: { with: { project: true } }, - }, - orderBy: medias.id, - }); - - if (mediaList.length < limit) { - hasMore = false; - } - lastId = mediaList.at(-1)?.id ?? lastId; - - if (mediaList.length > 0) { - const transformedItems = this._transformMediaList(mediaList); - for (const item of transformedItems) { - passThrough.write(`${JSON.stringify(item)}\n`); - } - } + for await (const item of iterateMediaDumpItems( + mediaSourceId, + this._transformMediaList, + )) { + await writeWithBackpressure( + passThrough, + `${JSON.stringify(item)}\n`, + ); } passThrough.end(); } catch (err) { @@ -995,76 +1127,68 @@ export const BackupService = { const { PassThrough } = await import("node:stream"); const passThrough = new PassThrough(); - const ndjsonStream = new PassThrough(); const archive = new archiverMod.TarArchive(); archive.pipe(passThrough); - archive.append(ndjsonStream, { name: "dump.ndjson" }); + archive.on("error", (error) => { + if (!passThrough.destroyed) { + passThrough.destroy(error); + } + }); - (async () => { + void (async () => { + let stagingDirectory: string | undefined; try { - const limit = 1000; - let lastId: string | null = null; - let hasMore = true; - - while (hasMore) { - const mediaList: MediaListQueryItem[] = - await db.query.medias.findMany({ - where: lastId - ? and( - eq(medias.mediaSourceId, mediaSourceId), - gt(medias.id, lastId), - ) - : eq(medias.mediaSourceId, mediaSourceId), - limit, - with: { - generationInfo: true, - urls: true, - tags: { with: { tag: true } }, - authors: { - with: { author: { with: { accounts: true } } }, - }, - characters: { - with: { - character: { with: { ips: { with: { ip: true } } } }, - }, - }, - ips: { with: { ip: true } }, - projects: { with: { project: true } }, - }, - orderBy: medias.id, - }); + await fs.mkdir(TarStagingDirectory, { recursive: true }); + stagingDirectory = await fs.mkdtemp( + path.join(TarStagingDirectory, "export-"), + ); + const ndjsonPath = path.join(stagingDirectory, "dump.ndjson"); + await writeNdjsonDump( + mediaSourceId, + ndjsonPath, + this._transformMediaList, + ); - if (mediaList.length < limit) { - hasMore = false; - } - lastId = mediaList.at(-1)?.id ?? lastId; - - if (mediaList.length > 0) { - const transformedItems = this._transformMediaList(mediaList); - const includeImages = options?.includeImages ?? true; - for (const item of transformedItems) { - ndjsonStream.write(`${JSON.stringify(item)}\n`); - - if (includeImages && item.filePath) { - try { - const buffer = await driver.get(item.filePath); - archive.append(buffer, { name: `images/${item.filePath}` }); - } catch { - // ignore missing files - } - } + const ndjsonStats = await fs.stat(ndjsonPath); + await appendTarEntry(archive, createReadStream(ndjsonPath), { + name: "dump.ndjson", + stats: ndjsonStats, + }); + + const includeImages = options?.includeImages ?? true; + if (includeImages) { + for await (const item of iterateNdjsonDumpItems(ndjsonPath)) { + if (!item.filePath) { + continue; } + + let file: Awaited>; + try { + file = await driver.getStream(item.filePath); + } catch { + // Ignore missing files, matching the previous export behavior. + continue; + } + + await appendTarEntry(archive, file.stream, { + name: `images/${item.filePath}`, + stats: file.stats, + }); } } - ndjsonStream.end(); await archive.finalize(); } catch (err) { logger.error({ err }, "Error generating TAR dump"); - ndjsonStream.destroy( - err instanceof Error ? err : new Error(String(err)), - ); + const error = err instanceof Error ? err : new Error(String(err)); archive.abort(); + passThrough.destroy(error); + } finally { + if (stagingDirectory) { + await fs + .rm(stagingDirectory, { recursive: true, force: true }) + .catch(() => {}); + } } })(); diff --git a/apps/server/src/application/services/job-transfer-storage.ts b/apps/server/src/application/services/job-transfer-storage.ts index cdf558c35..614065f6f 100644 --- a/apps/server/src/application/services/job-transfer-storage.ts +++ b/apps/server/src/application/services/job-transfer-storage.ts @@ -3,14 +3,20 @@ import { createWriteStream } from "node:fs"; import fs from "node:fs/promises"; import path from "node:path"; import { pipeline } from "node:stream/promises"; +import type { Job } from "@solid-imager/core/domain/repositories/job-repository"; import { webReadableToNodeStream } from "~/infrastructure/utils/stream-utils"; +const configuredTransferDirectory = process.env.SOLID_IMAGER_JOB_TRANSFER_DIR; +const isolatedRuntimeDirectory = + process.env.E2E_RUNTIME_DIR ?? process.env.DEV_STARTUP_RUNTIME_DIR; const JobTransferDirectory = path.resolve( - process.cwd(), - ".cache", - "job-transfers", + configuredTransferDirectory ?? + (isolatedRuntimeDirectory + ? path.join(isolatedRuntimeDirectory, ".cache", "job-transfers") + : path.join(process.cwd(), ".cache", "job-transfers")), ); const JobArtifactTtlMs = 24 * 60 * 60 * 1000; +const JobTransferStaleFileTtlMs = 60 * 60 * 1000; export type JobTransferMode = "json" | "zip"; @@ -38,6 +44,13 @@ export function getInputPath(jobId: string, mode: JobTransferMode): string { ); } +export function getInputPartialPath( + jobId: string, + mode: JobTransferMode, +): string { + return `${getInputPath(jobId, mode)}.partial`; +} + export function getArtifactPath(jobId: string, mode: JobTransferMode): string { return path.join( JobTransferDirectory, @@ -46,6 +59,13 @@ export function getArtifactPath(jobId: string, mode: JobTransferMode): string { ); } +export function getArtifactPartialPath( + jobId: string, + mode: JobTransferMode, +): string { + return `${getArtifactPath(jobId, mode)}.partial`; +} + export function isJobTransferPath(jobId: string, targetPath: string): boolean { const resolvedTarget = path.resolve(targetPath); const resolvedRoot = path.resolve(JobTransferDirectory); @@ -83,12 +103,20 @@ export async function persistJobInput( file: File, ): Promise { const inputPath = getInputPath(jobId, mode); + const partialPath = getInputPartialPath(jobId, mode); await fs.mkdir(path.dirname(inputPath), { recursive: true }); - await pipeline( - webReadableToNodeStream(file.stream()), - createWriteStream(inputPath), - ); - return inputPath; + await removeJobTransferFile(inputPath); + await removeJobTransferFile(partialPath); + try { + await pipeline( + webReadableToNodeStream(file.stream()), + createWriteStream(partialPath), + ); + await fs.rename(partialPath, inputPath); + return inputPath; + } finally { + await removeJobTransferFile(partialPath); + } } export async function removeJobTransferFile(targetPath: string): Promise { @@ -135,3 +163,180 @@ export async function cleanupExpiredJobTransferFiles( } } } + +type TransferFileKind = "inputs" | "artifacts"; + +type JobLookup = (jobId: string) => Promise; + +type JobTransferCleanupResult = { + removedFiles: number; + removedBytes: number; +}; + +const TransferFileNamePattern = + /^([0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})\.(?:ndjson|tar)$/i; + +function readTransferJobId(fileName: string): string | null { + return TransferFileNamePattern.exec(fileName)?.[1] ?? null; +} + +function readRestoreInputPath(payload: unknown): string | null { + if ( + typeof payload !== "object" || + payload === null || + Array.isArray(payload) + ) { + return null; + } + const inputPath = (payload as { inputPath?: unknown }).inputPath; + return typeof inputPath === "string" ? inputPath : null; +} + +function isExpectedTransferFile( + kind: TransferFileKind, + targetPath: string, + job: Job | null, +): boolean { + if (!job) { + return false; + } + + if (kind === "artifacts") { + return job.status === "completed" && job.artifactPath === targetPath; + } + + return ( + (job.status === "pending" || job.status === "in_progress") && + readRestoreInputPath(job.payload) === targetPath + ); +} + +async function latestModificationTime(targetPath: string): Promise { + let stat: Awaited>; + try { + stat = await fs.stat(targetPath); + } catch { + return 0; + } + + if (!stat.isDirectory()) { + return stat.mtimeMs; + } + + let latest = stat.mtimeMs; + let entries: Dirent[]; + try { + entries = await fs.readdir(targetPath, { withFileTypes: true }); + } catch { + return latest; + } + + for (const entry of entries) { + latest = Math.max( + latest, + await latestModificationTime(path.join(targetPath, entry.name)), + ); + } + return latest; +} + +async function cleanupOrphanedTarStaging( + expirationTime: number, +): Promise { + const stagingDirectory = path.join(JobTransferDirectory, "..", "tar-staging"); + let entries: Dirent[]; + try { + entries = await fs.readdir(stagingDirectory, { withFileTypes: true }); + } catch (error) { + if (isNodeErrorCode(error, "ENOENT")) { + return { removedFiles: 0, removedBytes: 0 }; + } + throw error; + } + + let removedFiles = 0; + let removedBytes = 0; + for (const entry of entries) { + if (!entry.isDirectory()) { + continue; + } + const targetPath = path.join(stagingDirectory, entry.name); + if ((await latestModificationTime(targetPath)) > expirationTime) { + continue; + } + const stat = await fs.stat(targetPath).catch(() => null); + await fs.rm(targetPath, { recursive: true, force: true }); + removedFiles++; + removedBytes += stat?.isDirectory() ? 0 : (stat?.size ?? 0); + } + + return { removedFiles, removedBytes }; +} + +/** + * Removes transfer files left by failed, cancelled, stale, or deleted jobs. + * Completed artifacts and restore inputs still referenced by pending jobs are + * retained; age-based expiry remains handled by cleanupExpiredJobTransferFiles. + */ +export async function cleanupOrphanedJobTransferFiles( + findJob: JobLookup, + now = Date.now(), +): Promise { + const expirationTime = now - JobTransferStaleFileTtlMs; + const jobCache = new Map(); + let removedFiles = 0; + let removedBytes = 0; + + for (const kind of ["inputs", "artifacts"] as const) { + const directoryPath = path.join(JobTransferDirectory, kind); + let entries: Dirent[]; + try { + entries = await fs.readdir(directoryPath, { withFileTypes: true }); + } catch (error) { + if (isNodeErrorCode(error, "ENOENT")) { + continue; + } + throw error; + } + + for (const entry of entries) { + if (!entry.isFile()) { + continue; + } + const targetPath = path.join(directoryPath, entry.name); + const stat = await fs.stat(targetPath).catch(() => null); + if (!stat || stat.mtimeMs > expirationTime) { + continue; + } + + const isPartial = entry.name.endsWith(".partial"); + const jobId = readTransferJobId( + isPartial ? entry.name.slice(0, -".partial".length) : entry.name, + ); + let shouldRemove = isPartial; + if (!isPartial && jobId) { + if (!jobCache.has(jobId)) { + jobCache.set(jobId, await findJob(jobId)); + } + const job = jobCache.get(jobId) ?? null; + // An unknown job may belong to another isolated database/runtime. + // Keep it for age-based expiry instead of deleting user data. + shouldRemove = + job !== null && !isExpectedTransferFile(kind, targetPath, job); + } + + if (!shouldRemove) { + continue; + } + await removeJobTransferFile(targetPath); + removedFiles++; + removedBytes += stat.size; + } + } + + const stagingCleanup = await cleanupOrphanedTarStaging(expirationTime); + return { + removedFiles: removedFiles + stagingCleanup.removedFiles, + removedBytes: removedBytes + stagingCleanup.removedBytes, + }; +} diff --git a/apps/server/src/application/services/source-transfer-job-service.ts b/apps/server/src/application/services/source-transfer-job-service.ts index 8f63df663..4ae77abda 100644 --- a/apps/server/src/application/services/source-transfer-job-service.ts +++ b/apps/server/src/application/services/source-transfer-job-service.ts @@ -11,6 +11,7 @@ import { services } from "~/application/registry"; import { BackupService } from "~/application/services/backup-service"; import { getArtifactMetadata, + getArtifactPartialPath, isJobTransferPath, removeJobTransferFile, } from "~/application/services/job-transfer-storage"; @@ -29,17 +30,33 @@ export async function processSourceExportJob(job: Job): Promise { includeImages: payload.includeImages, }); const artifact = getArtifactMetadata(job.id, job.mediaSourceId, payload.mode); + const partialPath = getArtifactPartialPath(job.id, payload.mode); + let artifactCommitted = false; await fs.mkdir(path.dirname(artifact.path), { recursive: true }); - await pipeline( - webReadableToNodeStream(asDumpStream(dump)), - createWriteStream(artifact.path), - ); - const stat = await fs.stat(artifact.path); - await services.getJobRepository().setArtifact(job.id, { - ...artifact, - size: stat.size, - }); + await removeJobTransferFile(artifact.path); + await removeJobTransferFile(partialPath); + try { + await pipeline( + webReadableToNodeStream(asDumpStream(dump)), + createWriteStream(partialPath), + ); + await fs.rename(partialPath, artifact.path); + artifactCommitted = true; + + const stat = await fs.stat(artifact.path); + await services.getJobRepository().setArtifact(job.id, { + ...artifact, + size: stat.size, + }); + } catch (error) { + if (artifactCommitted) { + await removeJobTransferFile(artifact.path); + } + throw error; + } finally { + await removeJobTransferFile(partialPath); + } } export async function processSourceRestoreJob(job: Job): Promise { diff --git a/apps/server/src/infrastructure/jobs/job-worker.ts b/apps/server/src/infrastructure/jobs/job-worker.ts index d5b872574..bebed45b3 100644 --- a/apps/server/src/infrastructure/jobs/job-worker.ts +++ b/apps/server/src/infrastructure/jobs/job-worker.ts @@ -1,6 +1,7 @@ import type { AppConfig } from "@solid-imager/core/domain/config/config-schema"; import { cleanupExpiredJobTransferFiles, + cleanupOrphanedJobTransferFiles, removeJobTransferFile, } from "~/application/services/job-transfer-storage"; import type { IJobRepository } from "~/domain/repositories/job-repository"; @@ -56,6 +57,7 @@ export class JobWorker { private activeJobs = 0; private activeAiJobs = 0; private activeThumbnailJobs = 0; + private activeExportJobs = 0; private readonly jobRepo: IJobRepository; private readonly processor: (job: Job) => Promise; @@ -65,6 +67,7 @@ export class JobWorker { "extract_ccip_vector", ]); private readonly thumbnailJobTypes = new Set(["generate_thumbnail"]); + private readonly exportJobTypes = new Set(["source_export"]); constructor( jobRepo: IJobRepository, @@ -157,15 +160,34 @@ export class JobWorker { } } - // 3. Poll Other Jobs + // 3. Keep source exports in a single-worker pool. A TAR export is + // disk-backed, but concurrent archives still multiply stream buffers and + // database page memory. + if (this.activeExportJobs < 1) { + const jobs = await this.jobRepo.claimPending(1, { + includeTypes: Array.from(this.exportJobTypes), + }); + for (const job of jobs) { + void this.tryProcessJob(job); + } + } + + // 4. Poll Other Jobs // "concurrency" is treated as the limit for NON-AI jobs in this independent pool model const activeOtherJobs = - this.activeJobs - this.activeAiJobs - this.activeThumbnailJobs; + this.activeJobs - + this.activeAiJobs - + this.activeThumbnailJobs - + this.activeExportJobs; if (activeOtherJobs < this.concurrency) { const slots = this.concurrency - activeOtherJobs; if (slots > 0) { const jobs = await this.jobRepo.claimPending(slots, { - excludeTypes: [...this.aiJobTypes, ...this.thumbnailJobTypes], + excludeTypes: [ + ...this.aiJobTypes, + ...this.thumbnailJobTypes, + ...this.exportJobTypes, + ], }); for (const job of jobs) { void this.tryProcessJob(job); @@ -190,6 +212,7 @@ export class JobWorker { this.activeJobs++; const isAiJob = this.aiJobTypes.has(job.type); const isThumbnailJob = this.thumbnailJobTypes.has(job.type); + const isExportJob = this.exportJobTypes.has(job.type); const startedAt = Date.now(); if (isAiJob) { this.activeAiJobs++; @@ -197,6 +220,9 @@ export class JobWorker { if (isThumbnailJob) { this.activeThumbnailJobs++; } + if (isExportJob) { + this.activeExportJobs++; + } const heartbeatId = setInterval(() => { void Promise.resolve( this.jobRepo.update(job.id, { updatedAt: new Date() }), @@ -302,6 +328,9 @@ export class JobWorker { if (isThumbnailJob) { this.activeThumbnailJobs--; } + if (isExportJob) { + this.activeExportJobs--; + } } } @@ -342,6 +371,18 @@ export class JobWorker { if (count > 0) { logger.warn({ count, olderThan }, "Requeued stale in-progress jobs"); } + const orphaned = await cleanupOrphanedJobTransferFiles( + this.jobRepo.findById.bind(this.jobRepo), + ); + if (orphaned.removedFiles > 0) { + logger.info( + { + removedFiles: orphaned.removedFiles, + removedBytes: orphaned.removedBytes, + }, + "Removed orphaned job transfer files", + ); + } await cleanupExpiredJobTransferFiles(); } catch (error) { logger.error( diff --git a/apps/server/src/infrastructure/storage/local.ts b/apps/server/src/infrastructure/storage/local.ts index e87e13ef4..022ec1c01 100644 --- a/apps/server/src/infrastructure/storage/local.ts +++ b/apps/server/src/infrastructure/storage/local.ts @@ -3,6 +3,7 @@ * Extracted from src/lib/drivers/local.ts */ +import { createReadStream } from "node:fs"; import fs from "node:fs/promises"; import path from "node:path"; import type { LocalConnection } from "@solid-imager/core/domain/sources/schemas"; @@ -92,6 +93,18 @@ export class LocalDriver implements MediaSourceDriver { return Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength); } + async getStream(p: string) { + const absolutePath = this.getAbsolutePath(p); + const stats = await fs.stat(absolutePath); + if (!stats.isFile()) { + throw new Error(`Not a regular file: ${p}`); + } + return { + stream: createReadStream(absolutePath), + stats, + }; + } + /** * Writes content to a file within the local media source. * Creates parent directories if they don't exist. diff --git a/apps/server/src/infrastructure/storage/schema.ts b/apps/server/src/infrastructure/storage/schema.ts index d534d65cf..38c6fb896 100644 --- a/apps/server/src/infrastructure/storage/schema.ts +++ b/apps/server/src/infrastructure/storage/schema.ts @@ -1,3 +1,5 @@ +import type { Stats } from "node:fs"; +import type { Readable } from "node:stream"; import type { LocalConnection, S3Connection, @@ -45,6 +47,12 @@ export type MediaSourceDriver = { */ get(path: string): Promise; + /** + * Opens a file as a stream and returns its metadata without buffering its + * contents in memory. + */ + getStream(path: string): Promise<{ stream: Readable; stats: Stats }>; + /** * Writes content to a file within the media source. */ diff --git a/apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts b/apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts new file mode 100644 index 000000000..c394e90a8 --- /dev/null +++ b/apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts @@ -0,0 +1,116 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { Job } from "@solid-imager/core/domain/repositories/job-repository"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +type JobTransferStorage = + typeof import("~/application/services/job-transfer-storage"); + +let runtimeDirectory: string; +let storage: JobTransferStorage; + +function createJob(overrides: Partial): Job { + const now = new Date(); + return { + id: randomUUID(), + type: "source_export", + mediaSourceId: randomUUID(), + status: "in_progress", + payload: { mode: "zip", includeImages: false }, + result: null, + error: null, + createdAt: now, + updatedAt: now, + parentId: null, + attemptCount: 1, + startedAt: now, + finishedAt: null, + ...overrides, + }; +} + +async function writeOldFile(targetPath: string): Promise { + await fs.mkdir(path.dirname(targetPath), { recursive: true }); + await fs.writeFile(targetPath, "stale transfer"); + const oldTime = (Date.now() - 2 * 60 * 60 * 1000) / 1000; + await fs.utimes(targetPath, oldTime, oldTime); +} + +describe("job transfer storage cleanup", () => { + beforeEach(async () => { + runtimeDirectory = await fs.mkdtemp( + path.join(os.tmpdir(), "solid-imager-transfer-test-"), + ); + vi.stubEnv( + "SOLID_IMAGER_JOB_TRANSFER_DIR", + path.join(runtimeDirectory, "job-transfers"), + ); + vi.resetModules(); + storage = await import("~/application/services/job-transfer-storage"); + }); + + afterEach(async () => { + vi.unstubAllEnvs(); + await fs.rm(runtimeDirectory, { recursive: true, force: true }); + }); + + it("removes stale files for a known non-completed job", async () => { + const jobId = randomUUID(); + const targetPath = storage.getArtifactPath(jobId, "zip"); + await writeOldFile(targetPath); + + const result = await storage.cleanupOrphanedJobTransferFiles(async (id) => + id === jobId ? createJob({ id, status: "in_progress" }) : null, + ); + + expect(result.removedFiles).toBe(1); + expect(await fs.stat(targetPath).catch(() => null)).toBeNull(); + }); + + it("keeps a valid completed artifact", async () => { + const jobId = randomUUID(); + const targetPath = storage.getArtifactPath(jobId, "zip"); + await writeOldFile(targetPath); + const job = createJob({ + id: jobId, + status: "completed", + artifactPath: targetPath, + finishedAt: new Date(), + }); + + const result = await storage.cleanupOrphanedJobTransferFiles( + async () => job, + ); + + expect(result.removedFiles).toBe(0); + expect((await fs.stat(targetPath)).isFile()).toBe(true); + }); + + it("keeps a file whose job is unknown until TTL expiry", async () => { + const jobId = randomUUID(); + const targetPath = storage.getArtifactPath(jobId, "zip"); + await writeOldFile(targetPath); + + const result = await storage.cleanupOrphanedJobTransferFiles( + async () => null, + ); + + expect(result.removedFiles).toBe(0); + expect((await fs.stat(targetPath)).isFile()).toBe(true); + }); + + it("removes stale partial output without a database lookup", async () => { + const jobId = randomUUID(); + const targetPath = storage.getArtifactPartialPath(jobId, "zip"); + await writeOldFile(targetPath); + const findJob = vi.fn(async () => null); + + const result = await storage.cleanupOrphanedJobTransferFiles(findJob); + + expect(result.removedFiles).toBe(1); + expect(findJob).not.toHaveBeenCalled(); + expect(await fs.stat(targetPath).catch(() => null)).toBeNull(); + }); +}); diff --git a/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts b/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts index 960f4d3ad..577727486 100644 --- a/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts +++ b/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts @@ -17,6 +17,14 @@ vi.mock("~/infrastructure/logger", () => ({ updateLogLevel: vi.fn(), })); +vi.mock("~/application/services/job-transfer-storage", () => ({ + cleanupExpiredJobTransferFiles: vi.fn().mockResolvedValue(undefined), + cleanupOrphanedJobTransferFiles: vi + .fn() + .mockResolvedValue({ removedFiles: 0, removedBytes: 0 }), + removeJobTransferFile: vi.fn(), +})); + describe("JobWorker", () => { let jobRepo: IJobRepository; let processor: (job: Job) => Promise; @@ -100,6 +108,7 @@ describe("JobWorker", () => { "auto_tagging", "extract_ccip_vector", "generate_thumbnail", + "source_export", ], }), ); @@ -129,6 +138,9 @@ describe("JobWorker", () => { if (options?.includeTypes?.includes("generate_thumbnail")) { return Promise.resolve([]); } + if (options?.includeTypes?.includes("source_export")) { + return Promise.resolve([]); + } if (options?.includeTypes) { return Promise.resolve(aiJobs.slice(0, limit)); } @@ -180,6 +192,9 @@ describe("JobWorker", () => { if (options?.includeTypes?.includes("generate_thumbnail")) { return Promise.resolve([]); } + if (options?.includeTypes?.includes("source_export")) { + return Promise.resolve([]); + } if (options?.includeTypes) { // AI request return Promise.resolve([aiJob].slice(0, limit)); @@ -209,6 +224,7 @@ describe("JobWorker", () => { "auto_tagging", "extract_ccip_vector", "generate_thumbnail", + "source_export", ], }), ); @@ -295,4 +311,40 @@ describe("JobWorker", () => { resolveThumbnail(); await vi.runOnlyPendingTimersAsync(); }); + + it("processes at most one source export job at a time", async () => { + let resolveExport: () => void = () => {}; + processor = vi.fn( + () => + new Promise((resolve) => { + resolveExport = resolve; + }), + ); + worker = new JobWorker(jobRepo, processor); + const exportJob = { + id: "export-1", + type: "source_export", + status: "pending", + } as Job; + (jobRepo.claimPending as any).mockImplementation( + (limit: number, options: any) => + Promise.resolve( + options?.includeTypes?.includes("source_export") + ? [exportJob].slice(0, limit) + : [], + ), + ); + + worker.start(); + await vi.advanceTimersByTimeAsync(TimerDelay); + await vi.advanceTimersByTimeAsync(1000); + + expect(processor).toHaveBeenCalledTimes(1); + expect(jobRepo.claimPending).toHaveBeenCalledWith(1, { + includeTypes: ["source_export"], + }); + + resolveExport(); + await vi.runOnlyPendingTimersAsync(); + }); }); From 4d225d252989a6be2b7368fd08bba59c4798090a Mon Sep 17 00:00:00 2001 From: hmjn Date: Fri, 14 Aug 2026 13:34:23 +0900 Subject: [PATCH 2/4] fix(download): stream completed transfer artifacts --- .../src/infrastructure/api/job-artifact.ts | 74 ++++++++++++++++++ .../infrastructure/api/routers/jobs-router.ts | 38 +--------- .../api/rpc-response-headers.ts | 16 ++++ .../src/routes/api/jobs.$jobId.artifact.ts | 25 ++++++ apps/server/src/routes/api/rpc.$.ts | 8 +- apps/server/src/routes/v2/jobs.tsx | 30 ++------ .../api/rpc-response-headers.test.ts | 57 ++++++++++++++ apps/server/vite.config.ts | 15 +++- packages/client/src/create-client.test.ts | 39 +++++++++- packages/client/src/create-client.ts | 76 ++++++++++++++++++- 10 files changed, 314 insertions(+), 64 deletions(-) create mode 100644 apps/server/src/infrastructure/api/job-artifact.ts create mode 100644 apps/server/src/infrastructure/api/rpc-response-headers.ts create mode 100644 apps/server/src/routes/api/jobs.$jobId.artifact.ts create mode 100644 apps/server/src/tests/unit/infrastructure/api/rpc-response-headers.test.ts diff --git a/apps/server/src/infrastructure/api/job-artifact.ts b/apps/server/src/infrastructure/api/job-artifact.ts new file mode 100644 index 000000000..cc6d9b208 --- /dev/null +++ b/apps/server/src/infrastructure/api/job-artifact.ts @@ -0,0 +1,74 @@ +import { createReadStream } from "node:fs"; +import fs from "node:fs/promises"; +import type { Job } from "@solid-imager/core/domain/repositories/job-repository"; +import { isJobTransferPath } from "~/application/services/job-transfer-storage"; +import { JobRepository } from "~/infrastructure/repositories/job-repository"; +import { nodeStreamToWebReadable } from "~/infrastructure/utils/stream-utils"; + +export type ResolvedJobArtifact = { + job: Job; + path: string; + fileName: string; + contentType: string; + size: number; + stream: ReadableStream; +}; + +export async function resolveJobArtifact( + jobId: string, +): Promise { + const job = await JobRepository.findById(jobId); + if ( + job?.status !== "completed" || + !job.artifactPath || + !job.artifactFileName || + !job.artifactContentType || + !isJobTransferPath(job.id, job.artifactPath) + ) { + return null; + } + + if (job.artifactExpiresAt && job.artifactExpiresAt <= new Date()) { + return null; + } + + let stat: Awaited>; + try { + stat = await fs.stat(job.artifactPath); + } catch { + return null; + } + if (!stat.isFile()) { + return null; + } + + return { + job, + path: job.artifactPath, + fileName: job.artifactFileName, + contentType: job.artifactContentType, + size: stat.size, + stream: nodeStreamToWebReadable(createReadStream(job.artifactPath)), + }; +} + +function encodeContentDispositionFilename(fileName: string): string { + return encodeURIComponent(fileName).replace( + /[!'()*]/g, + (character) => `%${character.charCodeAt(0).toString(16).toUpperCase()}`, + ); +} + +export function createJobArtifactHeaders( + artifact: Pick, +): HeadersInit { + const fallbackFileName = + artifact.fileName.replace(/[^\x20-\x7e]/g, "_").replace(/["\\;]/g, "_") || + "job-artifact"; + return { + "Cache-Control": "no-store", + "Content-Disposition": `attachment; filename="${fallbackFileName}"; filename*=UTF-8''${encodeContentDispositionFilename(artifact.fileName)}`, + "Content-Length": String(artifact.size), + "Content-Type": artifact.contentType, + }; +} diff --git a/apps/server/src/infrastructure/api/routers/jobs-router.ts b/apps/server/src/infrastructure/api/routers/jobs-router.ts index 53c785c9a..f925f147a 100644 --- a/apps/server/src/infrastructure/api/routers/jobs-router.ts +++ b/apps/server/src/infrastructure/api/routers/jobs-router.ts @@ -1,5 +1,3 @@ -import { createReadStream } from "node:fs"; -import fs from "node:fs/promises"; import { eventIterator, ORPCError, os } from "@orpc/server"; import { isBatchParentJobType, @@ -16,12 +14,11 @@ import { } from "@solid-imager/core/domain/sources/events"; import { and, count, desc, eq } from "drizzle-orm"; import { z } from "zod"; -import { isJobTransferPath } from "~/application/services/job-transfer-storage"; +import { resolveJobArtifact } from "~/infrastructure/api/job-artifact"; import { db } from "~/infrastructure/db"; import { jobs } from "~/infrastructure/db/schema"; import { RealtimeEventBus } from "~/infrastructure/events/realtime-event-bus"; import { JobRepository } from "~/infrastructure/repositories/job-repository"; -import { nodeStreamToWebReadable } from "~/infrastructure/utils/stream-utils"; const PublicJobFailureMessage = "Job failed"; @@ -139,41 +136,14 @@ export const jobsRouter = { .input(jobIdRequestSchema) .output(z.instanceof(ReadableStream)) .handler(async ({ input }) => { - const job = await JobRepository.findById(input.id); - if ( - job?.status !== "completed" || - !job.artifactPath || - !job.artifactFileName || - !job.artifactContentType || - !isJobTransferPath(job.id, job.artifactPath) - ) { - throw new ORPCError("NOT_FOUND", { - message: "Artifact not found", - }); - } - - if (job.artifactExpiresAt && job.artifactExpiresAt <= new Date()) { - throw new ORPCError("NOT_FOUND", { - message: "Artifact not found", - }); - } - - let stat: Awaited>; - try { - stat = await fs.stat(job.artifactPath); - } catch { - throw new ORPCError("NOT_FOUND", { - message: "Artifact not found", - }); - } - - if (!stat.isFile()) { + const artifact = await resolveJobArtifact(input.id); + if (!artifact) { throw new ORPCError("NOT_FOUND", { message: "Artifact not found", }); } - return nodeStreamToWebReadable(createReadStream(job.artifactPath)); + return artifact.stream; }), retry: os diff --git a/apps/server/src/infrastructure/api/rpc-response-headers.ts b/apps/server/src/infrastructure/api/rpc-response-headers.ts new file mode 100644 index 000000000..1b134dbc0 --- /dev/null +++ b/apps/server/src/infrastructure/api/rpc-response-headers.ts @@ -0,0 +1,16 @@ +const UrlBase = "http://localhost"; + +export function isJobArtifactDownloadPath(pathname: string): boolean { + return pathname.replace(/\/+$/, "").endsWith("/jobs/downloadArtifact"); +} + +export function createRpcResponseHeaders(url: URL | string): Headers { + const parsedUrl = typeof url === "string" ? new URL(url, UrlBase) : url; + const headers = new Headers(); + + if (isJobArtifactDownloadPath(parsedUrl.pathname)) { + headers.set("content-type", "application/octet-stream"); + } + + return headers; +} diff --git a/apps/server/src/routes/api/jobs.$jobId.artifact.ts b/apps/server/src/routes/api/jobs.$jobId.artifact.ts new file mode 100644 index 000000000..3b9e1c39f --- /dev/null +++ b/apps/server/src/routes/api/jobs.$jobId.artifact.ts @@ -0,0 +1,25 @@ +import { createFileRoute } from "@tanstack/solid-router"; +import { + createJobArtifactHeaders, + resolveJobArtifact, +} from "~/infrastructure/api/job-artifact"; +import type { ServerRouteContext } from "~/infrastructure/router/route-types"; +import { bootstrapServerRoute } from "~/infrastructure/server-route-bootstrap"; + +export const Route = createFileRoute("/api/jobs/$jobId/artifact")({ + server: { + handlers: { + GET: async ({ params }: ServerRouteContext<{ jobId: string }>) => { + bootstrapServerRoute(); + const artifact = await resolveJobArtifact(params.jobId); + if (!artifact) { + return new Response("Artifact not found", { status: 404 }); + } + + return new Response(artifact.stream, { + headers: createJobArtifactHeaders(artifact), + }); + }, + }, + }, +}); diff --git a/apps/server/src/routes/api/rpc.$.ts b/apps/server/src/routes/api/rpc.$.ts index 37edbe5c3..dacce0468 100644 --- a/apps/server/src/routes/api/rpc.$.ts +++ b/apps/server/src/routes/api/rpc.$.ts @@ -1,11 +1,15 @@ import { RPCHandler } from "@orpc/server/fetch"; +import { ResponseHeadersPlugin } from "@orpc/server/plugins"; import { createFileRoute } from "@tanstack/solid-router"; import { appRouter } from "~/domain/shared/api-contract"; +import { createRpcResponseHeaders } from "~/infrastructure/api/rpc-response-headers"; import { logger } from "~/infrastructure/logger"; import type { ServerRouteContext } from "~/infrastructure/router/route-types"; import { bootstrapServerRoute } from "~/infrastructure/server-route-bootstrap"; -const handler = new RPCHandler(appRouter); +const handler = new RPCHandler(appRouter, { + plugins: [new ResponseHeadersPlugin()], +}); export const Route = createFileRoute("/api/rpc/$")({ server: { @@ -14,7 +18,7 @@ export const Route = createFileRoute("/api/rpc/$")({ bootstrapServerRoute(); const { response } = await handler.handle(request, { prefix: "/api/rpc", - context: {}, + context: { resHeaders: createRpcResponseHeaders(request.url) }, }); if (response) { return response; diff --git a/apps/server/src/routes/v2/jobs.tsx b/apps/server/src/routes/v2/jobs.tsx index eae613495..0c9290615 100644 --- a/apps/server/src/routes/v2/jobs.tsx +++ b/apps/server/src/routes/v2/jobs.tsx @@ -62,29 +62,15 @@ function V2JobsRoute() { throw error; } }} - onDownload={async (job) => { + onDownload={(job) => { if (!job.artifact) return; - try { - const stream = await orpc.jobs.downloadArtifact({ id: job.id }); - const blob = await new Response(stream, { - headers: { "content-type": job.artifact.contentType }, - }).blob(); - const url = URL.createObjectURL(blob); - const anchor = document.createElement("a"); - anchor.href = url; - anchor.download = job.artifact.fileName; - document.body.appendChild(anchor); - anchor.click(); - anchor.remove(); - setTimeout(() => URL.revokeObjectURL(url), 0); - } catch (error) { - toast.error( - error instanceof Error - ? error.message - : "Failed to download artifact", - ); - throw error; - } + const anchor = document.createElement("a"); + anchor.href = `/api/jobs/${encodeURIComponent(job.id)}/artifact`; + anchor.download = job.artifact.fileName; + anchor.rel = "noopener"; + document.body.appendChild(anchor); + anchor.click(); + anchor.remove(); }} state={() => toQueryUiState(jobsQuery)} /> diff --git a/apps/server/src/tests/unit/infrastructure/api/rpc-response-headers.test.ts b/apps/server/src/tests/unit/infrastructure/api/rpc-response-headers.test.ts new file mode 100644 index 000000000..16d2f6814 --- /dev/null +++ b/apps/server/src/tests/unit/infrastructure/api/rpc-response-headers.test.ts @@ -0,0 +1,57 @@ +import { os } from "@orpc/server"; +import { RPCHandler } from "@orpc/server/fetch"; +import { ResponseHeadersPlugin } from "@orpc/server/plugins"; +import { describe, expect, it } from "vitest"; +import { z } from "zod"; +import { createRpcResponseHeaders } from "~/infrastructure/api/rpc-response-headers"; + +describe("RPC response headers", () => { + it("marks job artifact streams as binary responses", async () => { + const router = { + jobs: { + downloadArtifact: os.output(z.instanceof(ReadableStream)).handler( + async () => + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("tar")); + controller.close(); + }, + }), + ), + }, + } as const; + const handler = new RPCHandler(router, { + plugins: [new ResponseHeadersPlugin()], + }); + const request = new Request( + "http://localhost/api/rpc/jobs/downloadArtifact", + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ json: null }), + }, + ); + + const result = await handler.handle(request, { + prefix: "/api/rpc", + context: { resHeaders: createRpcResponseHeaders(request.url) }, + }); + + expect(result.matched).toBe(true); + expect(result.response?.headers.get("content-type")).toBe( + "application/octet-stream", + ); + const response = result.response as Response; + const artifact = await response.blob(); + expect(artifact.type).toBe("application/octet-stream"); + expect(await artifact.text()).toBe("tar"); + }); + + it("does not mark regular RPC responses as binary", () => { + expect( + createRpcResponseHeaders( + "http://localhost/api/rpc/jobs/get?data=%7B%22json%22%3Anull%7D", + ).has("content-type"), + ).toBe(false); + }); +}); diff --git a/apps/server/vite.config.ts b/apps/server/vite.config.ts index 1bc45dce9..b4ec37f21 100644 --- a/apps/server/vite.config.ts +++ b/apps/server/vite.config.ts @@ -1,6 +1,7 @@ import path from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; import { RPCHandler } from "@orpc/server/node"; +import { ResponseHeadersPlugin } from "@orpc/server/plugins"; import { defineConfig } from "vite"; import type { Plugin } from "vite"; import tailwindcss from "@tailwindcss/vite"; @@ -103,7 +104,7 @@ const bypassSecFetchDestPlugin = (): Plugin => ({ }); const loadDevRpcHandler = async () => { - const [{ appRouter }, { initServices, startBackgroundWorker }, { logger }] = await Promise.all([ + const [{ appRouter }, { initServices, startBackgroundWorker }, { logger }, { createRpcResponseHeaders }] = await Promise.all([ runtimeImport( serverModuleUrl("src/domain/shared/api-contract.ts"), ), @@ -113,10 +114,16 @@ const loadDevRpcHandler = async () => { runtimeImport( serverModuleUrl("src/infrastructure/logger.ts"), ), + runtimeImport( + serverModuleUrl("src/infrastructure/api/rpc-response-headers.ts"), + ), ]); return { - handler: new RPCHandler(appRouter), + handler: new RPCHandler(appRouter, { + plugins: [new ResponseHeadersPlugin()], + }), + createRpcResponseHeaders, initServices, startBackgroundWorker, logger, @@ -165,7 +172,7 @@ const devOrpcNodeMiddlewarePlugin = (): Plugin => ({ try { const devRpcHandler = await getDevRpcHandler(); logger = devRpcHandler.logger; - const { handler, initServices } = devRpcHandler; + const { handler, initServices, createRpcResponseHeaders } = devRpcHandler; startBackgroundWorker = devRpcHandler.startBackgroundWorker; initServices(); @@ -181,7 +188,7 @@ const devOrpcNodeMiddlewarePlugin = (): Plugin => ({ const result = await handler.handle(req, res, { prefix: "/api/rpc", - context: {}, + context: { resHeaders: createRpcResponseHeaders(req.url ?? "/") }, }); matched = result.matched; if (matched && responseFinished && res.statusCode >= 200 && res.statusCode < 300) { diff --git a/packages/client/src/create-client.test.ts b/packages/client/src/create-client.test.ts index 939e5b64a..906e17026 100644 --- a/packages/client/src/create-client.test.ts +++ b/packages/client/src/create-client.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; import { APIError } from "./api-error"; -import { createTimedFetch } from "./create-client"; +import { createClient, createTimedFetch } from "./create-client"; describe("createTimedFetch", () => { it("propagates an upstream abort signal to fetch", async () => { @@ -40,4 +40,41 @@ describe("createTimedFetch", () => { expect.any(APIError), ); }); + + it("downloads job artifacts as a raw stream", async () => { + const requests: Request[] = []; + const client = createClient({ + url: "http://localhost", + fetch: async (request) => { + requests.push(request.clone()); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("tar")); + controller.close(); + }, + }), + { headers: { "content-type": "application/octet-stream" } }, + ); + }, + }) as unknown as { + jobs: { + downloadArtifact: ( + input: { id: string }, + options?: { signal?: AbortSignal }, + ) => Promise>; + }; + }; + + const stream = await client.jobs.downloadArtifact({ id: "job-id" }); + const body = await new Response(stream).text(); + + expect(body).toBe("tar"); + expect(requests).toHaveLength(1); + expect(requests[0]?.url).toBe( + "http://localhost/api/rpc/jobs/downloadArtifact", + ); + expect(requests[0]?.method).toBe("POST"); + expect(await requests[0]?.json()).toEqual({ json: { id: "job-id" } }); + }); }); diff --git a/packages/client/src/create-client.ts b/packages/client/src/create-client.ts index 42f8d4c5c..61ab7a761 100644 --- a/packages/client/src/create-client.ts +++ b/packages/client/src/create-client.ts @@ -16,6 +16,78 @@ export type ClientOptions = { timeoutMs?: number; }; +type JobArtifactInput = { id: string }; +type JobArtifactOptions = { signal?: AbortSignal }; + +// The server intentionally returns this root-level artifact as a raw stream. +// The regular oRPC JSON decoder cannot deserialize a stream body, so this one +// procedure must be fetched directly while the other procedures use RPCLink. +async function downloadJobArtifact( + url: URL, + fetchImpl: (request: Request, init?: FetchInit) => Promise, + input: JobArtifactInput, + options?: JobArtifactOptions, +): Promise> { + const response = await fetchImpl( + new Request(url, { + method: "POST", + headers: { + accept: "application/octet-stream", + "content-type": "application/json", + }, + body: JSON.stringify({ json: input }), + signal: options?.signal, + }), + { signal: options?.signal }, + ); + + if (!response.ok) { + const body = await response.text().catch(() => ""); + throw new APIError( + body || `Artifact download failed with status ${response.status}`, + "SERVER_ERROR", + ); + } + if (!response.body) { + throw new APIError( + "Artifact download returned an empty body", + "SERVER_ERROR", + ); + } + + return response.body; +} + +function withStreamingJobArtifactDownload( + client: ContractRouterClient, + artifactUrl: URL, + fetchImpl: (request: Request, init?: FetchInit) => Promise, +): ContractRouterClient { + return new Proxy(client as object, { + get(target, property, receiver) { + if (property !== "jobs") { + return Reflect.get(target, property, receiver); + } + + const jobs = Reflect.get(target, property, receiver); + if (typeof jobs !== "function") { + return jobs; + } + + return new Proxy(jobs, { + get(jobsTarget, jobsProperty, jobsReceiver) { + if (jobsProperty !== "downloadArtifact") { + return Reflect.get(jobsTarget, jobsProperty, jobsReceiver); + } + + return (input: JobArtifactInput, options?: JobArtifactOptions) => + downloadJobArtifact(artifactUrl, fetchImpl, input, options); + }, + }); + }, + }) as ContractRouterClient; +} + function isAbortError(error: unknown): boolean { return error instanceof Error && error.name === "AbortError"; } @@ -84,5 +156,7 @@ export function createClient( fetch: fetchImpl, }); - return createORPCClient(link) as ContractRouterClient; + const client = createORPCClient(link) as ContractRouterClient; + const artifactUrl = new URL("jobs/downloadArtifact", rpcUrl); + return withStreamingJobArtifactDownload(client, artifactUrl, fetchImpl); } From 1c7fd98c461f820deb5bc03058e448cab65c9de2 Mon Sep 17 00:00:00 2001 From: hmjn Date: Fri, 14 Aug 2026 13:36:56 +0900 Subject: [PATCH 3/4] fix(v2): expose TAR transfer formats --- .../infrastructure/api-clients/sources-api.ts | 2 +- .../tests/e2e/v2-routes.responsive.spec.ts | 73 +++++++++++++++++++ .../api-clients/sources-api-ext.test.ts | 4 +- .../infrastructure/api-clients/sources-api.ts | 2 +- .../src/screens/v2-manager/data-transfer.tsx | 14 ++++ 5 files changed, 91 insertions(+), 4 deletions(-) diff --git a/apps/server/src/infrastructure/api-clients/sources-api.ts b/apps/server/src/infrastructure/api-clients/sources-api.ts index 2cc341b35..1179110b6 100644 --- a/apps/server/src/infrastructure/api-clients/sources-api.ts +++ b/apps/server/src/infrastructure/api-clients/sources-api.ts @@ -96,7 +96,7 @@ export async function fetchSourceDump( mode: "json" | "zip" = "json", opts?: { includeImages?: boolean }, ): Promise { - const includeImages = opts?.includeImages ?? false; + const includeImages = opts?.includeImages ?? mode === "zip"; const job = await orpc.sources.enqueueExport({ id, mode, diff --git a/apps/server/src/tests/e2e/v2-routes.responsive.spec.ts b/apps/server/src/tests/e2e/v2-routes.responsive.spec.ts index b8bf58f68..28da134d7 100644 --- a/apps/server/src/tests/e2e/v2-routes.responsive.spec.ts +++ b/apps/server/src/tests/e2e/v2-routes.responsive.spec.ts @@ -95,6 +95,79 @@ test("V2 wide collection uses selection preview before detail navigation", async await expect(page).toHaveURL(v2MediaPath(E2E_SIMILAR_MEDIA_ID)); }); +test("V2 restore exposes and selects the TAR format", async ({ page }) => { + await page.goto("/v2/manager"); + await waitForAppHydration(page); + + const categoryNavigation = page.locator( + 'nav[aria-label="Manager categories"]:visible', + ); + const transferButton = categoryNavigation + .getByRole("button", { name: /Data transfer/ }) + .first(); + await transferButton.scrollIntoViewIfNeeded(); + await transferButton.click(); + await expect( + page.getByRole("heading", { name: "Data transfer", exact: true }), + ).toBeVisible(); + + const selectTriggers = page.locator('button[aria-haspopup="listbox"]'); + await selectTriggers.nth(0).click(); + await page + .getByRole("option", { name: E2E_SOURCE_NAME, exact: true }) + .click(); + + await selectTriggers.nth(2).click(); + const tarOption = page.getByRole("option", { + name: "TAR archive", + exact: true, + }); + await expect(tarOption).toBeVisible(); + await tarOption.click(); + await expect(selectTriggers.nth(2)).toContainText("TAR archive"); + await expect(page.locator('input[type="file"]')).toHaveAttribute( + "accept", + ".tar,.zip,application/x-tar,application/zip", + ); +}); + +test("V2 completed export starts a native streaming download", async ({ + page, +}) => { + await page.goto("/v2/manager"); + await waitForAppHydration(page); + + const categoryNavigation = page.locator( + 'nav[aria-label="Manager categories"]:visible', + ); + await categoryNavigation + .getByRole("button", { name: /Data transfer/ }) + .first() + .click(); + + const selectTriggers = page.locator('button[aria-haspopup="listbox"]'); + await selectTriggers.nth(0).click(); + await page + .getByRole("option", { name: E2E_SOURCE_NAME, exact: true }) + .click(); + await selectTriggers.nth(1).click(); + await page.getByRole("option", { name: "TAR archive", exact: true }).click(); + await page.getByRole("button", { name: "Queue export", exact: true }).click(); + await expect(page.getByText(/Export queued/)).toBeVisible(); + + await page.goto("/v2/jobs"); + await waitForAppHydration(page); + const exportJob = page.getByRole("button", { name: /Source Export/ }).first(); + await expect(exportJob).toBeVisible({ timeout: 30_000 }); + await exportJob.click(); + const inspector = page.getByRole("complementary", { name: "Job details" }); + await expect(inspector).toContainText("Completed", { timeout: 30_000 }); + + const download = page.waitForEvent("download"); + await inspector.getByRole("button", { name: /Download source-/ }).click(); + expect((await download).suggestedFilename()).toMatch(/\.tar$/); +}); + test("V2 search filter opens without remounting media results", async ({ page, }) => { diff --git a/apps/server/src/tests/unit/infrastructure/api-clients/sources-api-ext.test.ts b/apps/server/src/tests/unit/infrastructure/api-clients/sources-api-ext.test.ts index 8b05f9eab..c44a416d5 100644 --- a/apps/server/src/tests/unit/infrastructure/api-clients/sources-api-ext.test.ts +++ b/apps/server/src/tests/unit/infrastructure/api-clients/sources-api-ext.test.ts @@ -65,7 +65,7 @@ describe("Sources API Client Extensions", () => { expect(await result.text()).toBe("dump"); }); - it("should pass the selected mode to export jobs", async () => { + it("should include images by default for TAR exports", async () => { const id = "test-source-id"; const mockBlob = new Blob(["zip content"], { type: "application/zip" }); ((orpc.sources as any).enqueueExport as any).mockResolvedValue({ @@ -84,7 +84,7 @@ describe("Sources API Client Extensions", () => { expect((orpc.sources as any).enqueueExport).toHaveBeenCalledWith({ id, mode: "zip", - includeImages: false, + includeImages: true, }); expect(await result.text()).toBe("zip content"); }); diff --git a/apps/tauri/src/infrastructure/api-clients/sources-api.ts b/apps/tauri/src/infrastructure/api-clients/sources-api.ts index ab60200d5..6932fafa7 100644 --- a/apps/tauri/src/infrastructure/api-clients/sources-api.ts +++ b/apps/tauri/src/infrastructure/api-clients/sources-api.ts @@ -15,7 +15,7 @@ export async function fetchSourceDump( mode: "json" | "zip" = "json", opts?: { includeImages?: boolean }, ): Promise { - const includeImages = opts?.includeImages ?? false; + const includeImages = opts?.includeImages ?? mode === "zip"; const job = await client.sources.enqueueExport({ id, mode, includeImages }); return downloadCompletedJobArtifact(client.jobs, job.id); } diff --git a/packages/ui/src/screens/v2-manager/data-transfer.tsx b/packages/ui/src/screens/v2-manager/data-transfer.tsx index 915574769..28c194dc2 100644 --- a/packages/ui/src/screens/v2-manager/data-transfer.tsx +++ b/packages/ui/src/screens/v2-manager/data-transfer.tsx @@ -18,6 +18,10 @@ import type { V2ManagerTransferFormat, } from "./types"; +function formatOptionLabel(format: V2ManagerTransferFormat): string { + return format === "ndjson" ? "NDJSON metadata" : "TAR archive"; +} + export function DataTransferPanel(props: { actions: V2ManagerTransferActions; manager: UseManagerPageResult; @@ -127,6 +131,11 @@ export function DataTransferPanel(props: {
( + + {formatOptionLabel(selectProps.item.rawValue)} + + )} onChange={(value) => value && setImportFormat(value)} options={["ndjson", "tar"] as const} value={importFormat()} From ad156567b7dc9037fe6a3fa0d66e59cc3b8a157d Mon Sep 17 00:00:00 2001 From: hmjn Date: Fri, 14 Aug 2026 14:08:13 +0900 Subject: [PATCH 4/4] fix(transfer): address review findings --- .../application/services/backup-service.ts | 30 ++++++++++++++++--- .../services/job-transfer-storage.ts | 19 +++++++++++- .../services/source-transfer-job-service.ts | 1 + .../src/infrastructure/jobs/job-worker.ts | 27 ++++++++++------- .../services/job-transfer-storage.test.ts | 26 ++++++++++++++++ .../infrastructure/jobs/job-worker.test.ts | 27 ++++++++++++++--- 6 files changed, 111 insertions(+), 19 deletions(-) diff --git a/apps/server/src/application/services/backup-service.ts b/apps/server/src/application/services/backup-service.ts index 89636bb8d..cbcdaaa37 100644 --- a/apps/server/src/application/services/backup-service.ts +++ b/apps/server/src/application/services/backup-service.ts @@ -245,15 +245,28 @@ async function writeNdjsonDump( transform: (mediaList: Partial[]) => MediaDumpItem[], ): Promise { const output = createWriteStream(outputPath); + let streamError: Error | undefined; + const onOutputError = (error: Error) => { + streamError ??= error; + }; + output.on("error", onOutputError); try { for await (const item of iterateMediaDumpItems(mediaSourceId, transform)) { + if (streamError) { + throw streamError; + } await writeWithBackpressure(output, `${JSON.stringify(item)}\n`); } + if (streamError) { + throw streamError; + } output.end(); await finished(output); } catch (error) { output.destroy(); throw error; + } finally { + output.removeListener("error", onOutputError); } } @@ -279,11 +292,17 @@ async function* iterateNdjsonDumpItems( } } +function normalizeArchiveEntryName(name: string): string { + return name.replaceAll("\\", "/").replace(/\/+/g, "/"); +} + function appendTarEntry( archive: TarArchive, source: import("node:stream").Readable, data: TarEntryData, ): Promise { + const normalizedName = normalizeArchiveEntryName(data.name); + const normalizedData = { ...data, name: normalizedName }; return new Promise((resolve, reject) => { const cleanup = () => { archive.removeListener("entry", onEntry); @@ -291,7 +310,7 @@ function appendTarEntry( source.removeListener("error", onSourceError); }; const onEntry = (entry: TarEntryData) => { - if (entry.name !== data.name) { + if (entry.name !== normalizedName) { return; } cleanup(); @@ -310,7 +329,7 @@ function appendTarEntry( archive.once("error", onArchiveError); source.once("error", onSourceError); try { - archive.append(source, data); + archive.append(source, normalizedData); } catch (error) { onArchiveError(error instanceof Error ? error : new Error(String(error))); } @@ -1079,7 +1098,7 @@ export const BackupService = { async createDump( mediaSourceId: string, mode: "json" | "zip" | "ndjson" | "tar" = "ndjson", - options?: { includeImages: boolean }, + options?: { includeImages: boolean; jobId?: string }, ) { // 1. Fetch Media Source Info (needed for Driver) const mediaSource = await db.query.mediaSources.findFirst({ @@ -1139,8 +1158,11 @@ export const BackupService = { let stagingDirectory: string | undefined; try { await fs.mkdir(TarStagingDirectory, { recursive: true }); + const stagingPrefix = options?.jobId + ? `${options.jobId}-export-` + : "export-"; stagingDirectory = await fs.mkdtemp( - path.join(TarStagingDirectory, "export-"), + path.join(TarStagingDirectory, stagingPrefix), ); const ndjsonPath = path.join(stagingDirectory, "dump.ndjson"); await writeNdjsonDump( diff --git a/apps/server/src/application/services/job-transfer-storage.ts b/apps/server/src/application/services/job-transfer-storage.ts index 614065f6f..f61e19d52 100644 --- a/apps/server/src/application/services/job-transfer-storage.ts +++ b/apps/server/src/application/services/job-transfer-storage.ts @@ -175,11 +175,17 @@ type JobTransferCleanupResult = { const TransferFileNamePattern = /^([0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})\.(?:ndjson|tar)$/i; +const TarStagingDirectoryNamePattern = + /^([0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})-export-/i; function readTransferJobId(fileName: string): string | null { return TransferFileNamePattern.exec(fileName)?.[1] ?? null; } +function readTarStagingJobId(directoryName: string): string | null { + return TarStagingDirectoryNamePattern.exec(directoryName)?.[1] ?? null; +} + function readRestoreInputPath(payload: unknown): string | null { if ( typeof payload !== "object" || @@ -242,6 +248,7 @@ async function latestModificationTime(targetPath: string): Promise { async function cleanupOrphanedTarStaging( expirationTime: number, + findJob: JobLookup, ): Promise { const stagingDirectory = path.join(JobTransferDirectory, "..", "tar-staging"); let entries: Dirent[]; @@ -261,6 +268,13 @@ async function cleanupOrphanedTarStaging( continue; } const targetPath = path.join(stagingDirectory, entry.name); + const jobId = readTarStagingJobId(entry.name); + if (jobId) { + const job = await findJob(jobId); + if (job?.status === "pending" || job?.status === "in_progress") { + continue; + } + } if ((await latestModificationTime(targetPath)) > expirationTime) { continue; } @@ -334,7 +348,10 @@ export async function cleanupOrphanedJobTransferFiles( } } - const stagingCleanup = await cleanupOrphanedTarStaging(expirationTime); + const stagingCleanup = await cleanupOrphanedTarStaging( + expirationTime, + findJob, + ); return { removedFiles: removedFiles + stagingCleanup.removedFiles, removedBytes: removedBytes + stagingCleanup.removedBytes, diff --git a/apps/server/src/application/services/source-transfer-job-service.ts b/apps/server/src/application/services/source-transfer-job-service.ts index 4ae77abda..3092a4491 100644 --- a/apps/server/src/application/services/source-transfer-job-service.ts +++ b/apps/server/src/application/services/source-transfer-job-service.ts @@ -28,6 +28,7 @@ export async function processSourceExportJob(job: Job): Promise { const payload = sourceExportJobPayloadSchema.parse(job.payload); const dump = await BackupService.createDump(job.mediaSourceId, payload.mode, { includeImages: payload.includeImages, + jobId: job.id, }); const artifact = getArtifactMetadata(job.id, job.mediaSourceId, payload.mode); const partialPath = getArtifactPartialPath(job.id, payload.mode); diff --git a/apps/server/src/infrastructure/jobs/job-worker.ts b/apps/server/src/infrastructure/jobs/job-worker.ts index bebed45b3..aab04ef1a 100644 --- a/apps/server/src/infrastructure/jobs/job-worker.ts +++ b/apps/server/src/infrastructure/jobs/job-worker.ts @@ -371,16 +371,23 @@ export class JobWorker { if (count > 0) { logger.warn({ count, olderThan }, "Requeued stale in-progress jobs"); } - const orphaned = await cleanupOrphanedJobTransferFiles( - this.jobRepo.findById.bind(this.jobRepo), - ); - if (orphaned.removedFiles > 0) { - logger.info( - { - removedFiles: orphaned.removedFiles, - removedBytes: orphaned.removedBytes, - }, - "Removed orphaned job transfer files", + try { + const orphaned = await cleanupOrphanedJobTransferFiles( + this.jobRepo.findById.bind(this.jobRepo), + ); + if (orphaned.removedFiles > 0) { + logger.info( + { + removedFiles: orphaned.removedFiles, + removedBytes: orphaned.removedBytes, + }, + "Removed orphaned job transfer files", + ); + } + } catch (error) { + logger.error( + { err: error }, + "Failed to clean up orphaned transfer files", ); } await cleanupExpiredJobTransferFiles(); diff --git a/apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts b/apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts index c394e90a8..f24d01260 100644 --- a/apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts +++ b/apps/server/src/tests/unit/application/services/job-transfer-storage.test.ts @@ -38,6 +38,12 @@ async function writeOldFile(targetPath: string): Promise { await fs.utimes(targetPath, oldTime, oldTime); } +async function writeOldDirectory(targetPath: string): Promise { + await writeOldFile(path.join(targetPath, "dump.ndjson")); + const oldTime = (Date.now() - 2 * 60 * 60 * 1000) / 1000; + await fs.utimes(targetPath, oldTime, oldTime); +} + describe("job transfer storage cleanup", () => { beforeEach(async () => { runtimeDirectory = await fs.mkdtemp( @@ -113,4 +119,24 @@ describe("job transfer storage cleanup", () => { expect(findJob).not.toHaveBeenCalled(); expect(await fs.stat(targetPath).catch(() => null)).toBeNull(); }); + + it("keeps staging for an active export job", async () => { + const jobId = randomUUID(); + const stagingDirectory = path.join( + storage.getJobTransferRoot(), + "..", + "tar-staging", + `${jobId}-export-stale`, + ); + await writeOldDirectory(stagingDirectory); + + const result = await storage.cleanupOrphanedJobTransferFiles(async (id) => + id === jobId ? createJob({ id, status: "in_progress" }) : null, + ); + + expect(result.removedFiles).toBe(0); + expect( + (await fs.stat(path.join(stagingDirectory, "dump.ndjson"))).isFile(), + ).toBe(true); + }); }); diff --git a/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts b/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts index 577727486..d58ebbb95 100644 --- a/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts +++ b/apps/server/src/tests/unit/infrastructure/jobs/job-worker.test.ts @@ -4,6 +4,13 @@ import type { IJobRepository } from "~/domain/repositories/job-repository"; import type { Job } from "~/infrastructure/db/schema"; import { JobWorker } from "~/infrastructure/jobs/job-worker"; +const { mockCleanupExpired, mockCleanupOrphaned } = vi.hoisted(() => ({ + mockCleanupExpired: vi.fn().mockResolvedValue(undefined), + mockCleanupOrphaned: vi + .fn() + .mockResolvedValue({ removedFiles: 0, removedBytes: 0 }), +})); + // Mock logger to avoid noise vi.mock("~/infrastructure/logger", () => ({ logger: { @@ -18,10 +25,8 @@ vi.mock("~/infrastructure/logger", () => ({ })); vi.mock("~/application/services/job-transfer-storage", () => ({ - cleanupExpiredJobTransferFiles: vi.fn().mockResolvedValue(undefined), - cleanupOrphanedJobTransferFiles: vi - .fn() - .mockResolvedValue({ removedFiles: 0, removedBytes: 0 }), + cleanupExpiredJobTransferFiles: mockCleanupExpired, + cleanupOrphanedJobTransferFiles: mockCleanupOrphaned, removeJobTransferFile: vi.fn(), })); @@ -35,6 +40,10 @@ describe("JobWorker", () => { beforeEach(() => { vi.useFakeTimers(); + mockCleanupExpired.mockReset().mockResolvedValue(undefined); + mockCleanupOrphaned + .mockReset() + .mockResolvedValue({ removedFiles: 0, removedBytes: 0 }); // Mock Repository jobRepo = { @@ -347,4 +356,14 @@ describe("JobWorker", () => { resolveExport(); await vi.runOnlyPendingTimersAsync(); }); + + it("still expires transfer files when orphan cleanup fails", async () => { + mockCleanupOrphaned.mockRejectedValueOnce(new Error("cleanup failed")); + + await ( + worker as unknown as { recoverStaleJobs: () => Promise } + ).recoverStaleJobs(); + + expect(mockCleanupExpired).toHaveBeenCalledTimes(1); + }); });