fix(server): let upload sessions recover and refuse bad input early

Five defects in resumable uploads, found while planning the Android
client's transfers:

- A create under a key whose session had failed, been cancelled or
  expired handed that dead session back, and the unique index refused a
  new one, so a client that keeps its key across Try again could never
  succeed. A dead session now releases its key, keeping its row and
  outcome, and a new session is made under it.
- A failed import was stored as UPLOAD_IMPORT_FAILED whatever the error,
  with the raw exception message, which can carry server paths. A coded
  upload error now keeps its own code and message; anything else is
  stored with a fixed message and the raw error goes to the log.
- Upload capabilities listed every reachable library whatever its type,
  and the session routes had no type check. Only book libraries are
  offered or accepted, as on the older library upload route.
- Contents that do not match the format were refused only at complete,
  after every byte was sent, and the session stayed receiving. The first
  chunk is now checked against the format's signature, and refused
  contents end the session and free its staged bytes.
- A chunk's body was read while the session's lock was held, with no
  timeout, so a connection gone half-open blocked every later chunk,
  complete and cancel for that session until the server restarted. The
  body is now read before the lock is taken, and a chunk that sends
  nothing for two minutes is given up.

The response shapes and request fields are unchanged.

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;
   }
 
-  /**
+  /** The session holding this key, whatever its state. The service decides whether it may be resumed. */
-   * 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.
-   */
   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<Omit<UploadSessionInsert, 'id' | 'userId' | 'idempotencyKey'>>) {
     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,

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) =>
+        (userId: number, key: string) => [...rows.values()].find((row) => row.userId === userId && row.idempotencyKey === key) ?? null,
-          [...rows.values()].find((row) => row.userId === userId && row.idempotencyKey === key && row.expiresAt > new Date()) ?? 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({

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<string, Promise<void>>();
 
@@ -91,10 +96,14 @@ export class UploadSessionService implements OnApplicationBootstrap {
 
   async create(dto: CreateUploadSessionDto, user: RequestUser): Promise<UploadSessionResponse> {
     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<UploadSessionResponse> {
-    return this.withLock(id, async () => {
+    // The body is read before the session's lock is taken. Read under the lock, a chunk whose
-      let row = await this.owned(id, user);
+    // connection went half-open held it for as long as the connection lasted, and every later
-      row = await this.reconcileSize(row);
+    // chunk, complete and cancel for the session queued behind it.
-      this.assertReceiving(row);
+    const before = await this.owned(id, user);
-      if (offset !== row.receivedBytes) {
+    this.assertReceiving(before);
-        throw new ConflictException({
+    if (offset !== before.receivedBytes) throw offsetMismatch(before.receivedBytes, offset);
-          errorCode: UploadErrorCode.OffsetMismatch,
+    const allowedChunk = Math.min(UPLOAD_CHUNK_SIZE_BYTES, before.sizeBytes - offset);
-          message: `Expected upload offset ${row.receivedBytes}, received ${offset}`,
+    const stagedChunk = await this.storageService.streamToTemp(chunkStream as Readable, allowedChunk, { idleTimeoutMs: CHUNK_IDLE_TIMEOUT_MS });
-          expectedOffset: row.receivedBytes,
+    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;
+      return await this.withLock(id, async () => {
-      const allowedChunk = Math.min(UPLOAD_CHUNK_SIZE_BYTES, remaining);
+        let row = await this.owned(id, user);
-      const stagedChunk = await this.storageService.streamToTemp(chunkStream as Readable, allowedChunk);
+        row = await this.reconcileSize(row);
-      try {
+        this.assertReceiving(row);
-        if (stagedChunk.sizeBytes === 0) throw uploadError.empty();
+        if (offset !== row.receivedBytes) throw offsetMismatch(row.receivedBytes, offset);
-        if (stagedChunk.sizeBytes > remaining) throw uploadError.tooLarge('Chunk exceeds the remaining upload size');
+        if (stagedChunk.sizeBytes > row.sizeBytes - row.receivedBytes) throw uploadError.tooLarge('Chunk exceeds the remaining upload size');
-        if (checksum) {
+        if (row.receivedBytes === 0) await this.refuseUnreadableStart(row, stagedChunk);
-          const normalized = normalizeChecksum(checksum);
-          const actual = await sha256File(stagedChunk.tempPath);
-          if (normalized !== actual) throw uploadError.checksumMismatch('Chunk checksum does not match');
-        }
 
         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<UploadSessionResponse> {
@@ -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.failImport(row.id, error);
-        await this.fail(row.id, UploadErrorCode.ImportFailed, message);
         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<void> {
+    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<void>): Promise<void> {
+    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<void> {
+    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<void> {
     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()

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);

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) {

server/src/modules/upload/upload-validator.service.ts

@@ -7,6 +7,9 @@ import { uploadError } from './upload-errors';
 
 export const SUPPORTED_BOOK_FORMATS = new Set<string>(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<void> {
     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)) {