diff --git a/server/src/modules/upload/upload-session.repository.ts b/server/src/modules/upload/upload-session.repository.ts index 1da52fc69..bef9eb053 100644 --- a/server/src/modules/upload/upload-session.repository.ts +++ b/server/src/modules/upload/upload-session.repository.ts @@ -1,5 +1,5 @@ import { Inject, Injectable } from '@nestjs/common'; -import { and, eq, gt, inArray, lt } from 'drizzle-orm'; +import { and, eq, inArray, lt } from 'drizzle-orm'; import type { NodePgDatabase } from 'drizzle-orm/node-postgres'; import { DB } from '../../db'; @@ -23,25 +23,27 @@ export class UploadSessionRepository { return row ?? null; } - /** - * Finds a session that can still be resumed under this key. - * - * Expired sessions are excluded deliberately. Idempotency exists to deduplicate a repeated - * `create` for work still in progress; it must not hand back a session that can no longer receive - * bytes. The client's key is the queue record's own id and does not change when it retries, so - * without this predicate a client that deliberately abandons an expired session gets the same - * dead one back, and its next chunk fails `assertReceiving` with the error it was retrying to - * escape. Retry then cannot ever succeed. - */ + /** The session holding this key, whatever its state. The service decides whether it may be resumed. */ async findByIdempotencyKey(userId: number, idempotencyKey: string) { const [row] = await this.db .select() .from(uploadSessions) - .where(and(eq(uploadSessions.userId, userId), eq(uploadSessions.idempotencyKey, idempotencyKey), gt(uploadSessions.expiresAt, new Date()))) + .where(and(eq(uploadSessions.userId, userId), eq(uploadSessions.idempotencyKey, idempotencyKey))) .limit(1); return row ?? null; } + /** + * Frees a dead session's key for a new session, keeping the row and its outcome. A client key + * never contains `#`, so the replacement cannot collide with one. + */ + async releaseIdempotencyKey(id: string) { + await this.db + .update(uploadSessions) + .set({ idempotencyKey: `#released-${id}`, updatedAt: new Date() }) + .where(eq(uploadSessions.id, id)); + } + async update(id: string, values: Partial>) { const [row] = await this.db .update(uploadSessions) @@ -73,7 +75,7 @@ export class UploadSessionRepository { return this.db .select({ id: libraries.id, name: libraries.name, allowedFormats: libraries.allowedFormats, organizationMode: libraries.organizationMode }) .from(libraries) - .where(inArray(libraries.id, libraryIds)); + .where(and(inArray(libraries.id, libraryIds), eq(libraries.type, 'books'))); } findCapabilityFolders(libraryIds: number[]) { @@ -102,6 +104,7 @@ export class UploadSessionRepository { const [row] = await this.db .select({ libraryId: books.libraryId, + libraryType: libraries.type, allowedFormats: libraries.allowedFormats, organizationMode: libraries.organizationMode, // Needed by validateTarget: a book whose folder path is a file cannot receive another file, diff --git a/server/src/modules/upload/upload-session.service.test.ts b/server/src/modules/upload/upload-session.service.test.ts index 0498013ea..36a46744a 100644 --- a/server/src/modules/upload/upload-session.service.test.ts +++ b/server/src/modules/upload/upload-session.service.test.ts @@ -7,6 +7,7 @@ import { join } from 'path'; import { Readable } from 'stream'; import { beforeEach, afterEach, describe, expect, it, vi } from 'vitest'; +import { uploadError } from './upload-errors'; import { UploadSessionService } from './upload-session.service'; import { UploadValidatorService } from './upload-validator.service'; @@ -26,18 +27,22 @@ describe('UploadSessionService', () => { await mkdir(join(root, 'upload-sessions'), { recursive: true }); rows = new Map(); repo = { - // Mirrors the repository's predicate, expiry included. Without the expiry check this fake is - // more permissive than the real query, and no test here could fail when a dead session is - // handed back under a live key. findByIdempotencyKey: vi.fn( - (userId: number, key: string) => - [...rows.values()].find((row) => row.userId === userId && row.idempotencyKey === key && row.expiresAt > new Date()) ?? null, + (userId: number, key: string) => [...rows.values()].find((row) => row.userId === userId && row.idempotencyKey === key) ?? null, ), + // Enforces upload_sessions_user_idempotency_uidx, as Postgres does. Without it a test could + // pass by making a second session under a key the database would refuse. create: vi.fn((values: any) => { + if ([...rows.values()].some((row) => row.userId === values.userId && row.idempotencyKey === values.idempotencyKey)) { + throw new Error('duplicate key value violates unique constraint "upload_sessions_user_idempotency_uidx"'); + } const row = makeRow(values); rows.set(row.id, row); return row; }), + releaseIdempotencyKey: vi.fn((id: string) => { + rows.set(id, { ...rows.get(id), idempotencyKey: `#released-${id}` }); + }), findById: vi.fn((id: string) => rows.get(id) ?? null), update: vi.fn((id: string, values: any) => { const current = rows.get(id); @@ -51,18 +56,30 @@ describe('UploadSessionService', () => { deleteTerminalBefore: vi.fn(() => undefined), findCapabilityLibraries: vi.fn(() => [{ id: 1, name: 'Main', allowedFormats: ['epub', 'pdf'], organizationMode: 'book_per_folder' }]), findCapabilityFolders: vi.fn(() => [{ id: 2, libraryId: 1, path: '/books/Main Shelf' }]), - findLibrary: vi.fn((id: number) => (id === 1 ? { id: 1, allowedFormats: ['epub', 'pdf'], organizationMode: 'book_per_folder' } : null)), + findLibrary: vi.fn((id: number) => { + if (id === 1) return { id: 1, type: 'books', allowedFormats: ['epub', 'pdf'], organizationMode: 'book_per_folder' }; + if (id === 3) return { id: 3, type: 'podcasts', allowedFormats: [], organizationMode: 'book_per_folder' }; + return null; + }), folderBelongsToLibrary: vi.fn((folderId: number, libraryId: number) => folderId === 2 && libraryId === 1), // Book 9's folder is a real directory; book 8's "folder" is a file, which is the shape that // reached `complete` and threw ENOTDIR before validateTarget checked it. See F-006. findBookTarget: vi.fn((id: number) => { if (id === 9) { - return { id: 9, libraryId: 1, allowedFormats: ['epub', 'pdf'], organizationMode: 'book_per_folder', folderPath: root }; + return { + id: 9, + libraryId: 1, + libraryType: 'books', + allowedFormats: ['epub', 'pdf'], + organizationMode: 'book_per_folder', + folderPath: root, + }; } if (id === 8) { return { id: 8, libraryId: 1, + libraryType: 'books', allowedFormats: ['epub', 'pdf'], organizationMode: 'book_per_folder', folderPath: join(root, 'a-book-that-is-a-file.cbz'), @@ -140,6 +157,30 @@ describe('UploadSessionService', () => { expect(repo.create).toHaveBeenCalledTimes(2); }); + it.each(['failed', 'cancelled', 'expired'])('creates a new session when the key belongs to a %s one, keeping its outcome', async (status) => { + const dto = createDto(); + const first = await service.create(dto, user); + rows.set(first.id, { ...rows.get(first.id), status, errorCode: UploadErrorCode.ImportFailed }); + + // Even for different details: the dead session no longer claims the key. + const second = await service.create({ ...dto, sizeBytes: 11 }, user); + + expect(second.id).not.toBe(first.id); + expect(second.status).toBe('receiving'); + expect(rows.get(first.id)).toMatchObject({ status, errorCode: UploadErrorCode.ImportFailed, idempotencyKey: `#released-${first.id}` }); + }); + + it.each(['processing', 'completed'])('hands back a %s session under its key rather than uploading twice', async (status) => { + const dto = createDto(); + const first = await service.create(dto, user); + rows.set(first.id, { ...rows.get(first.id), status }); + + const second = await service.create(dto, user); + + expect(second.id).toBe(first.id); + expect(repo.releaseIdempotencyKey).not.toHaveBeenCalled(); + }); + it('rejects an existing_book target whose folder path is a file', async () => { await writeFile(join(root, 'a-book-that-is-a-file.cbz'), 'not a directory'); @@ -165,6 +206,13 @@ describe('UploadSessionService', () => { await expect(service.create({ ...dto, sizeBytes: 11 }, user)).rejects.toBeInstanceOf(ConflictException); }); + it('refuses a library that does not hold books', async () => { + await expect(service.create({ ...createDto(), target: { kind: 'library', libraryId: 3 } } as any, user)).rejects.toMatchObject({ + response: { errorCode: UploadErrorCode.InvalidTarget }, + }); + expect(repo.create).not.toHaveBeenCalled(); + }); + it('enforces upload permission and library-folder ownership before creating files', async () => { await expect(service.create(createDto(), { ...user, permissions: [] })).rejects.toBeInstanceOf(ForbiddenException); await expect(service.create({ ...createDto(), target: { kind: 'library', libraryId: 1, folderId: 99 } } as any, user)).rejects.toMatchObject({ @@ -201,6 +249,76 @@ describe('UploadSessionService', () => { expect(await readFile(rows.get(response.id).stagingPath)).toHaveLength(0); }); + it('reads a chunk before taking the session lock, so a stalled chunk does not hold up the next', async () => { + const response = await service.create(createDto(), user); + const chunkPath = join(root, 'chunk'); + await writeFile(chunkPath, '%PDF-1234'); + storage.streamToTemp.mockImplementationOnce(() => new Promise(() => undefined)).mockResolvedValueOnce({ tempPath: chunkPath, sizeBytes: 9 }); + + void service.appendChunk(response.id, 0, undefined, Readable.from('stalled'), user); + const result = await service.appendChunk(response.id, 0, undefined, Readable.from('%PDF-1234'), user); + + expect(result.receivedBytes).toBe(9); + }); + + it('refuses contents that do not match the format at the first chunk and ends the session', async () => { + const response = await service.create({ ...createDto(), sizeBytes: 1_024 }, user); + const chunkPath = join(root, 'not-a-pdf'); + await writeFile(chunkPath, Buffer.alloc(600, 0x41)); + storage.streamToTemp.mockResolvedValue({ tempPath: chunkPath, sizeBytes: 600 }); + + await expect(service.appendChunk(response.id, 0, undefined, Readable.from('ignored'), user)).rejects.toMatchObject({ + response: { errorCode: UploadErrorCode.InvalidContent }, + }); + expect(rows.get(response.id)).toMatchObject({ status: 'failed', errorCode: UploadErrorCode.InvalidContent, receivedBytes: 0 }); + expect(storage.cleanup).toHaveBeenCalledWith(rows.get(response.id).stagingPath); + }); + + it('leaves a first chunk shorter than the signature for complete to judge', async () => { + const response = await service.create({ ...createDto(), sizeBytes: 1_024 }, user); + const chunkPath = join(root, 'short'); + await writeFile(chunkPath, 'AAAA'); + storage.streamToTemp.mockResolvedValue({ tempPath: chunkPath, sizeBytes: 4 }); + + await expect(service.appendChunk(response.id, 0, undefined, Readable.from('AAAA'), user)).resolves.toMatchObject({ receivedBytes: 4 }); + }); + + it('ends the session when complete refuses its contents', async () => { + const response = await service.create({ ...createDto(), sizeBytes: 9 }, user); + const row = rows.get(response.id); + await writeFile(row.stagingPath, 'AAAAAAAAA'); + rows.set(response.id, { ...row, receivedBytes: 9 }); + + await expect(service.complete(response.id, user)).rejects.toMatchObject({ response: { errorCode: UploadErrorCode.InvalidContent } }); + expect(rows.get(response.id)).toMatchObject({ status: 'failed', errorCode: UploadErrorCode.InvalidContent }); + }); + + it('stores the code an import failed with, and never a raw exception message', async () => { + const conflict = await service.create({ ...createDto(), sizeBytes: 9, idempotencyKey: 'upload-key-conflict' }, user); + const raw = await service.create({ ...createDto(), sizeBytes: 9, idempotencyKey: 'upload-key-raw' }, user); + for (const id of [conflict.id, raw.id]) { + await writeFile(rows.get(id).stagingPath, '%PDF-1234'); + rows.set(id, { ...rows.get(id), receivedBytes: 9 }); + } + upload.uploadForSession + .mockRejectedValueOnce(uploadError.destinationConflict('A file named "book.pdf" already exists at the target location')) + .mockRejectedValueOnce(new Error("ENOTDIR: not a directory, open '/library/Some Book/book.pdf'")); + + await expect(service.complete(conflict.id, user)).rejects.toMatchObject({ response: { errorCode: UploadErrorCode.DestinationConflict } }); + await expect(service.complete(raw.id, user)).rejects.toThrow('ENOTDIR'); + + expect(rows.get(conflict.id)).toMatchObject({ + status: 'failed', + errorCode: UploadErrorCode.DestinationConflict, + errorMessage: 'A file named "book.pdf" already exists at the target location', + }); + expect(rows.get(raw.id)).toMatchObject({ + status: 'failed', + errorCode: UploadErrorCode.ImportFailed, + errorMessage: 'The file could not be added', + }); + }); + it('will not complete a partial upload', async () => { const response = await service.create(createDto(), user); await expect(service.complete(response.id, user)).rejects.toMatchObject({ diff --git a/server/src/modules/upload/upload-session.service.ts b/server/src/modules/upload/upload-session.service.ts index d6f31f10c..fa5090d04 100644 --- a/server/src/modules/upload/upload-session.service.ts +++ b/server/src/modules/upload/upload-session.service.ts @@ -1,4 +1,4 @@ -import { ConflictException, ForbiddenException, HttpException, Injectable, NotFoundException, OnApplicationBootstrap } from '@nestjs/common'; +import { ConflictException, ForbiddenException, HttpException, Injectable, Logger, NotFoundException, OnApplicationBootstrap } from '@nestjs/common'; import { ModuleRef } from '@nestjs/core'; import { Interval } from '@nestjs/schedule'; import { Permission, UploadErrorCode, UPLOAD_SUPPORTED_FORMATS } from '@bookorbit/types'; @@ -15,6 +15,7 @@ import { Inject } from '@nestjs/common'; import type { UploadSessionRow } from '../../db/schema'; import type { RequestUser } from '../../common/types/request-user'; import { storageConfig } from '../../config/config'; +import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils'; import { HARD_MAX_UPLOAD_BYTES, UPLOAD_CHUNK_SIZE_BYTES, @@ -29,11 +30,15 @@ import { UploadProcessorService } from './upload-processor.service'; import { UploadService } from './upload.service'; import { UploadSessionRepository } from './upload-session.repository'; import { UploadStorageService } from './upload-storage.service'; -import { UploadValidatorService } from './upload-validator.service'; +import { CONTENT_SIGNATURE_BYTES, UploadValidatorService } from './upload-validator.service'; import { uploadError } from './upload-errors'; +const IMPORT_FAILED_MESSAGE = 'The file could not be added'; +const CHUNK_IDLE_TIMEOUT_MS = 120_000; + @Injectable() export class UploadSessionService implements OnApplicationBootstrap { + private readonly logger = new Logger(UploadSessionService.name); private readonly stagingDirectory: string; private readonly locks = new Map>(); @@ -91,10 +96,14 @@ export class UploadSessionService implements OnApplicationBootstrap { async create(dto: CreateUploadSessionDto, user: RequestUser): Promise { const existing = await this.repo.findByIdempotencyKey(user.id, dto.idempotencyKey); - if (existing) { + if (existing && isResumable(existing)) { this.assertIdempotentMatch(existing, dto); return this.toResponse(existing); } + // A key belongs to one live upload, not to every attempt made under it. Clients that keep their + // key across a retry were handed the failed, cancelled or expired session back, which can never + // receive or complete, and the unique index refused them a new one. + if (existing) await this.repo.releaseIdempotencyKey(existing.id); const maxMb = await this.appSettings.getMaxUploadSizeMb(); const maxBytes = Math.min(HARD_MAX_UPLOAD_BYTES, maxMb * 1_024 * 1_024); @@ -143,29 +152,29 @@ export class UploadSessionService implements OnApplicationBootstrap { chunkStream: NodeJS.ReadableStream, user: RequestUser, ): Promise { - return this.withLock(id, async () => { - let row = await this.owned(id, user); - row = await this.reconcileSize(row); - this.assertReceiving(row); - if (offset !== row.receivedBytes) { - throw new ConflictException({ - errorCode: UploadErrorCode.OffsetMismatch, - message: `Expected upload offset ${row.receivedBytes}, received ${offset}`, - expectedOffset: row.receivedBytes, - }); + // The body is read before the session's lock is taken. Read under the lock, a chunk whose + // connection went half-open held it for as long as the connection lasted, and every later + // chunk, complete and cancel for the session queued behind it. + const before = await this.owned(id, user); + this.assertReceiving(before); + if (offset !== before.receivedBytes) throw offsetMismatch(before.receivedBytes, offset); + const allowedChunk = Math.min(UPLOAD_CHUNK_SIZE_BYTES, before.sizeBytes - offset); + const stagedChunk = await this.storageService.streamToTemp(chunkStream as Readable, allowedChunk, { idleTimeoutMs: CHUNK_IDLE_TIMEOUT_MS }); + try { + if (stagedChunk.sizeBytes === 0) throw uploadError.empty(); + if (checksum) { + const normalized = normalizeChecksum(checksum); + const actual = await sha256File(stagedChunk.tempPath); + if (normalized !== actual) throw uploadError.checksumMismatch('Chunk checksum does not match'); } - const remaining = row.sizeBytes - row.receivedBytes; - const allowedChunk = Math.min(UPLOAD_CHUNK_SIZE_BYTES, remaining); - const stagedChunk = await this.storageService.streamToTemp(chunkStream as Readable, allowedChunk); - try { - if (stagedChunk.sizeBytes === 0) throw uploadError.empty(); - if (stagedChunk.sizeBytes > remaining) throw uploadError.tooLarge('Chunk exceeds the remaining upload size'); - if (checksum) { - const normalized = normalizeChecksum(checksum); - const actual = await sha256File(stagedChunk.tempPath); - if (normalized !== actual) throw uploadError.checksumMismatch('Chunk checksum does not match'); - } + return await this.withLock(id, async () => { + let row = await this.owned(id, user); + row = await this.reconcileSize(row); + this.assertReceiving(row); + if (offset !== row.receivedBytes) throw offsetMismatch(row.receivedBytes, offset); + if (stagedChunk.sizeBytes > row.sizeBytes - row.receivedBytes) throw uploadError.tooLarge('Chunk exceeds the remaining upload size'); + if (row.receivedBytes === 0) await this.refuseUnreadableStart(row, stagedChunk); try { await pipeline(createReadStream(stagedChunk.tempPath), createWriteStream(row.stagingPath, { flags: 'a' })); @@ -180,10 +189,10 @@ export class UploadSessionService implements OnApplicationBootstrap { const receivedBytes = row.receivedBytes + stagedChunk.sizeBytes; row = (await this.repo.update(row.id, { receivedBytes, expiresAt: new Date(Date.now() + UPLOAD_SESSION_TTL_MS) }))!; return this.toResponse(row); - } finally { - await this.storageService.cleanup(stagedChunk.tempPath); - } - }); + }); + } finally { + await this.storageService.cleanup(stagedChunk.tempPath); + } } async complete(id: string, user: RequestUser): Promise { @@ -199,7 +208,8 @@ export class UploadSessionService implements OnApplicationBootstrap { if (actual !== row.expectedSha256) throw uploadError.checksumMismatch('File checksum does not match'); } - await this.validator.validateContent(row.stagingPath, this.validator.validateFormat(row.filename, await this.allowedFormats(row))); + const format = this.validator.validateFormat(row.filename, await this.allowedFormats(row)); + await this.refuseUnreadable(row, () => this.validator.validateContent(row.stagingPath, format)); row = (await this.repo.update(row.id, { status: 'processing', errorCode: null, errorMessage: null }))!; try { @@ -237,8 +247,7 @@ export class UploadSessionService implements OnApplicationBootstrap { } return this.toResponse(row); } catch (error) { - const message = error instanceof Error ? error.message : String(error); - await this.fail(row.id, UploadErrorCode.ImportFailed, message); + await this.failImport(row.id, error); throw error; } }); @@ -271,7 +280,7 @@ export class UploadSessionService implements OnApplicationBootstrap { await this.repo.update(row.id, { status: 'completed', completedAt: new Date() }); await this.storageService.cleanup(row.stagingPath); } catch (error) { - await this.fail(row.id, UploadErrorCode.ImportFailed, error instanceof Error ? error.message : String(error)); + await this.failImport(row.id, error); } } @@ -286,7 +295,7 @@ export class UploadSessionService implements OnApplicationBootstrap { await this.repo.update(row.id, { status: 'completed', completedAt: new Date() }); await this.storageService.cleanup(row.stagingPath); } catch (error) { - await this.fail(row.id, UploadErrorCode.ImportFailed, error instanceof Error ? error.message : String(error)); + await this.failImport(row.id, error); } } @@ -301,6 +310,7 @@ export class UploadSessionService implements OnApplicationBootstrap { const library = await this.repo.findLibrary(target.libraryId); if (!library) throw uploadError.invalidTarget('Library not found'); await this.libraryService.verifyUserAccess(user.id, library.id, user.isSuperuser); + if (library.type !== 'books') throw uploadError.invalidTarget('Files can only be uploaded to a book library'); if (target.folderId) { if (!(await this.repo.folderBelongsToLibrary(target.folderId, target.libraryId))) { throw uploadError.invalidTarget('Folder does not belong to this library'); @@ -312,6 +322,7 @@ export class UploadSessionService implements OnApplicationBootstrap { const book = await this.repo.findBookTarget(target.bookId); if (!book) throw uploadError.invalidTarget('Book not found'); await this.libraryService.verifyUserAccess(user.id, book.libraryId, user.isSuperuser); + if (book.libraryType !== 'books') throw uploadError.invalidTarget('Files can only be uploaded to a book library'); if (book.organizationMode === 'book_per_file') throw uploadError.invalidTarget('This library does not allow multiple files per book'); await this.assertBookFolderIsADirectory(book.folderPath); this.validator.validateFormat(filename, book.allowedFormats); @@ -401,6 +412,46 @@ export class UploadSessionService implements OnApplicationBootstrap { return user.isSuperuser || user.permissions.includes(permission); } + /** + * A format's signature is in its first bytes, so a file that is not what its name says is refused + * at its first chunk rather than after every byte has been sent. + */ + private async refuseUnreadableStart(row: UploadSessionRow, chunk: { tempPath: string; sizeBytes: number }): Promise { + if (chunk.sizeBytes < CONTENT_SIGNATURE_BYTES && chunk.sizeBytes < row.sizeBytes) return; + const format = this.validator.validateFormat(row.filename, []); + await this.refuseUnreadable(row, () => this.validator.validateContent(chunk.tempPath, format)); + } + + /** Contents that do not match the format will not get better, so the session ends with them. */ + private async refuseUnreadable(row: UploadSessionRow, check: () => Promise): Promise { + try { + await check(); + } catch (error) { + if (uploadErrorCodeOf(error) === UploadErrorCode.InvalidContent) { + await this.fail(row.id, UploadErrorCode.InvalidContent, (error as HttpException).message); + await this.storageService.cleanup(row.stagingPath); + } + throw error; + } + } + + /** + * Stores why an import failed. A coded upload error keeps its code and its message, which are + * written for the client. Anything else is stored as an import failure with a fixed message, + * because a raw exception message can carry server paths; the raw error goes to the log. + */ + private async failImport(id: string, error: unknown): Promise { + const code = uploadErrorCodeOf(error); + if (code) { + await this.fail(id, code, (error as HttpException).message); + return; + } + const errorClass = error instanceof Error ? error.name : typeof error; + const message = sanitizeLogValue(error instanceof Error ? error.message : String(error)); + this.logger.warn(`[upload_session.import] [fail] sessionId=${id} errorClass=${errorClass} error="${message}" - upload import failed`); + await this.fail(id, UploadErrorCode.ImportFailed, IMPORT_FAILED_MESSAGE); + } + private async fail(id: string, errorCode: string, errorMessage: string): Promise { await this.repo.update(id, { status: 'failed', errorCode, errorMessage: errorMessage.slice(0, 2_000), completedAt: new Date() }); } @@ -448,6 +499,27 @@ export class UploadSessionService implements OnApplicationBootstrap { } } +/** A session a repeated create may hand back: still able to take bytes, importing, or added. */ +function isResumable(row: UploadSessionRow): boolean { + if (row.status === 'processing' || row.status === 'completed') return true; + return row.status === 'receiving' && row.expiresAt.getTime() > Date.now(); +} + +function offsetMismatch(expectedOffset: number, offset: number): ConflictException { + return new ConflictException({ + errorCode: UploadErrorCode.OffsetMismatch, + message: `Expected upload offset ${expectedOffset}, received ${offset}`, + expectedOffset, + }); +} + +function uploadErrorCodeOf(error: unknown): UploadErrorCode | null { + if (!(error instanceof HttpException)) return null; + const response = error.getResponse(); + const code = typeof response === 'object' && response !== null ? (response as { errorCode?: unknown }).errorCode : undefined; + return (Object.values(UploadErrorCode) as unknown[]).includes(code) ? (code as UploadErrorCode) : null; +} + function normalizeChecksum(value: string): string { const normalized = value .trim() diff --git a/server/src/modules/upload/upload-storage.service.test.ts b/server/src/modules/upload/upload-storage.service.test.ts index 876a94d89..9bc718524 100644 --- a/server/src/modules/upload/upload-storage.service.test.ts +++ b/server/src/modules/upload/upload-storage.service.test.ts @@ -1,5 +1,5 @@ import { Test, type TestingModule } from '@nestjs/testing'; -import { PayloadTooLargeException } from '@nestjs/common'; +import { PayloadTooLargeException, RequestTimeoutException } from '@nestjs/common'; import { createReadStream, type WriteStream } from 'fs'; import { copyFile, mkdir, mkdtemp, open, readFile, readdir, rename, rm, stat, symlink, unlink, writeFile } from 'fs/promises'; import { join } from 'path'; @@ -109,6 +109,29 @@ describe('UploadStorageService', () => { expect(await readdir(control.tempDir)).toEqual(['source.epub']); }); + it('gives up a source that sends nothing for the idle timeout and removes its temporary file', async () => { + const input = new PassThrough(); + input.write(payload); + + await expect(service.streamToTemp(input, undefined, { idleTimeoutMs: 20 })).rejects.toBeInstanceOf(RequestTimeoutException); + expect(await readdir(control.tempDir)).toEqual(['source.epub']); + }); + + it('keeps reading a source that keeps sending, however long it takes in all', async () => { + const input = new PassThrough(); + const sent = (async () => { + for (let index = 0; index < 4; index += 1) { + input.write(payload); + await new Promise((resolve) => setTimeout(resolve, 15)); + } + input.end(); + })(); + + const result = await service.streamToTemp(input, undefined, { idleTimeoutMs: 40 }); + await sent; + expect(result.sizeBytes).toBe(payload.length * 4); + }); + it('uses rename directly on the same filesystem, without opening a staging file', async () => { await service.moveToPath(source, destination); expect(await readFile(destination)).toEqual(payload); diff --git a/server/src/modules/upload/upload-storage.service.ts b/server/src/modules/upload/upload-storage.service.ts index 7a3b8de3c..265e42727 100644 --- a/server/src/modules/upload/upload-storage.service.ts +++ b/server/src/modules/upload/upload-storage.service.ts @@ -1,4 +1,4 @@ -import { Injectable, Logger } from '@nestjs/common'; +import { Injectable, Logger, RequestTimeoutException } from '@nestjs/common'; import { mkdir, open, rename, stat, unlink } from 'fs/promises'; import { createReadStream, createWriteStream } from 'fs'; import { tmpdir } from 'os'; @@ -24,17 +24,30 @@ export class UploadStorageService { constructor(private readonly appSettings: AppSettingsService) {} /** - * Streams the multipart file to a temp path on disk. + * Streams the multipart file to a temp path on disk. With `idleTimeoutMs`, a source that sends + * nothing for that long is given up, so a connection gone half-open does not hold a request open + * until the server restarts. */ - async streamToTemp(source: Readable, requestedLimitBytes?: number): Promise<{ tempPath: string; sizeBytes: number }> { + async streamToTemp( + source: Readable, + requestedLimitBytes?: number, + options: { idleTimeoutMs?: number } = {}, + ): Promise<{ tempPath: string; sizeBytes: number }> { const tempPath = join(tmpdir(), `bookorbit-upload-${randomUUID()}`); const writeStream = createWriteStream(tempPath); const configuredMb = await this.appSettings.getMaxUploadSizeMb(); const configuredBytes = Math.min(HARD_MAX_UPLOAD_BYTES, configuredMb * 1_024 * 1_024); const limitBytes = Math.min(configuredBytes, requestedLimitBytes ?? configuredBytes); let totalBytes = 0; + let idleTimer: NodeJS.Timeout | undefined; + const armIdleTimer = () => { + if (!options.idleTimeoutMs) return; + clearTimeout(idleTimer); + idleTimer = setTimeout(() => source.destroy(new RequestTimeoutException('No upload data arrived in time')), options.idleTimeoutMs); + }; const limiter = new Transform({ transform(chunk: Buffer, _encoding, callback) { + armIdleTimer(); totalBytes += chunk.length; if (totalBytes > limitBytes) { callback(uploadError.tooLarge(`File exceeds the ${Math.floor(limitBytes / 1_024 / 1_024)} MB upload limit`)); @@ -45,6 +58,7 @@ export class UploadStorageService { }); try { + armIdleTimer(); await pipeline(source, limiter, writeStream); } catch (err) { await this.cleanup(tempPath); @@ -52,6 +66,8 @@ export class UploadStorageService { throw uploadError.storageFull('The server does not have enough storage for this upload'); } throw err; + } finally { + clearTimeout(idleTimer); } if ((source as Readable & { truncated?: boolean }).truncated) { diff --git a/server/src/modules/upload/upload-validator.service.ts b/server/src/modules/upload/upload-validator.service.ts index 43f9e079e..f751d7e6f 100644 --- a/server/src/modules/upload/upload-validator.service.ts +++ b/server/src/modules/upload/upload-validator.service.ts @@ -7,6 +7,9 @@ import { uploadError } from './upload-errors'; export const SUPPORTED_BOOK_FORMATS = new Set(UPLOAD_SUPPORTED_FORMATS); +/** How much of a file's start `validateContent` reads to recognise its format. */ +export const CONTENT_SIGNATURE_BYTES = 512; + @Injectable() export class UploadValidatorService { /** @@ -63,7 +66,7 @@ export class UploadValidatorService { async validateContent(absolutePath: string, format: string): Promise { const handle = await open(absolutePath, 'r'); try { - const header = Buffer.alloc(512); + const header = Buffer.alloc(CONTENT_SIGNATURE_BYTES); const { bytesRead } = await handle.read(header, 0, header.length, 0); const bytes = header.subarray(0, bytesRead); if (!matchesFormatSignature(bytes, format)) {