#233: prune conversion job payloads for every job kind

The raw input/result bytes of import/export conversion jobs were kept
forever; a deleted classified page could live on inside its last export.
A new daily conversion-payload-prune job nulls both once a finished
(succeeded or failed) job passes conversion.payloadRetentionDays
(instance setting, default 30) — the row survives for status/audit.
PENDING and RUNNING rows keep their payload, so the worker's stale-lock
recovery path is untouched; a hand-requeued pruned job fails finally
via conversionInputOf instead of crashing the worker.

The input column becomes nullable; the migration backfills by clearing
payloads of jobs already finished longer ago than the default period
(recent results stay downloadable until they age out).

Job-count fence in system.spec: 7 -> 8 (new scheduler registration).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0168Ph5uBmHm8X28CSVpbpnJ

apps/api/prisma/migrations/20260730170000_prune_conversion_job_payloads/migration.sql

@@ -0,0 +1,16 @@
+-- #233: conversion job payloads become prunable. The raw input/result bytes
+-- are transient; a daily job nulls them once a finished job passes
+-- `conversion.payloadRetentionDays` (default 30). The row survives for
+-- status/audit purposes.
+ALTER TABLE "conversion_jobs" ALTER COLUMN "input" DROP NOT NULL;
+
+-- Backfill: clear the payloads of jobs that already finished longer ago than
+-- the default period. Recently finished jobs keep their bytes so a pending
+-- download still works; the scheduled job picks them up when they age out.
+-- PENDING/RUNNING rows are untouched (the worker's stale-lock recovery may
+-- still re-run them).
+UPDATE "conversion_jobs"
+   SET "input" = NULL, "result" = NULL, "result_mime_type" = NULL
+ WHERE "status" IN ('SUCCEEDED', 'FAILED')
+   AND "updated_at" < now() - interval '30 days'
+   AND ("input" IS NOT NULL OR "result" IS NOT NULL);

apps/api/prisma/schema.prisma

@@ -742,9 +742,11 @@ enum ConversionJobStatus {
 /// enqueued PENDING, a worker claims it (`FOR UPDATE SKIP LOCKED`, `lockedAt`
 /// recovers a crashed run), calls the pandoc sidecar with a timeout, and
 /// stores the output bytes or an `errorCode`. `input`/`result` are the raw
-/// document bytes — kept small by the request size limit and pruned by a
+/// document bytes — kept small by the request size limit and transient, not
-/// later maintenance job (they are transient, not the durable copy an
+/// the durable copy an Attachment is: the daily `conversion-payload-prune`
-/// Attachment is). The polling endpoint `GET /jobs/:id` is owner-scoped.
+/// job (#233) nulls both once a finished job passes
+/// `conversion.payloadRetentionDays`; the row survives for status/audit.
+/// The polling endpoint `GET /jobs/:id` is owner-scoped.
 model ConversionJob {
   id             String              @id @default(uuid())
   ownerId        String              @map("owner_id")
@@ -754,7 +756,9 @@ model ConversionJob {
   sourceFormat   String              @map("source_format")
   targetFormat   String              @map("target_format")
   standalone     Boolean             @default(true)
-  input          Bytes
+  /// Null once the retention job (#233) pruned a finished job's payload —
+  /// never while the job is PENDING/RUNNING (incl. stale-lock recovery).
+  input          Bytes?
   status         ConversionJobStatus @default(PENDING)
   attempts       Int                 @default(0)
   result         Bytes?
@@ -763,7 +767,7 @@ model ConversionJob {
   lockedAt       DateTime?           @map("locked_at")
   /// For a data-export job (#68): when its stored result stops being
   /// downloadable and is purged (GDPR data minimization). Null for every
-  /// other job kind, whose result never expires.
+  /// other job kind, whose payload the general retention (#233) prunes.
   expiresAt      DateTime?           @map("expires_at")
   /// Kind-specific job options (issue #117): a vault import carries
   /// `{parentPageId, labelIds, frontmatterMode}`. Null for other kinds.

apps/api/src/import-export/conversion-job.service.ts

@@ -3,11 +3,15 @@ import { ConversionJob, ConversionJobStatus as PrismaStatus } from '@prisma/clie
 import { ConversionJobStatus, ConversionJobView } from '@dorfteich/shared';
 import { PinoLogger } from 'nestjs-pino';
 
+import { ClockService } from '../common/clock.service';
 import { PrismaService } from '../prisma/prisma.service';
+import { InstanceSettingsService } from '../settings/instance-settings.service';
 
 import { ConversionWorker } from './conversion-worker.service';
 import { MAX_CONVERSION_INPUT_BYTES } from './pandoc.converter';
 
+const MS_PER_DAY = 24 * 60 * 60 * 1000;
+
 export interface EnqueueConversion {
   ownerId: string;
   kind: string;
@@ -50,6 +54,8 @@ export class ConversionJobService {
   constructor(
     private readonly prisma: PrismaService,
     private readonly worker: ConversionWorker,
+    private readonly settings: InstanceSettingsService,
+    private readonly clock: ClockService,
     private readonly logger: PinoLogger,
   ) {
     this.logger.setContext(ConversionJobService.name);
@@ -106,6 +112,35 @@ export class ConversionJobService {
     };
   }
 
+  /**
+   * Null the raw payload bytes of jobs that finished longer ago than
+   * `conversion.payloadRetentionDays` (issue #233) — every kind, input AND
+   * result. Only terminal jobs are touched: a PENDING row and a crashed
+   * RUNNING row awaiting stale-lock recovery keep their input so the worker
+   * can still (re)process them. The row itself survives for status/audit;
+   * `updatedAt` marks completion because a terminal row is never written
+   * again (the payload guard below keeps this run from re-matching rows).
+   */
+  async pruneExpiredPayloads(): Promise<number> {
+    const retentionDays = await this.settings.get('conversion.payloadRetentionDays');
+    const cutoff = new Date(this.clock.now().getTime() - retentionDays * MS_PER_DAY);
+    const result = await this.prisma.conversionJob.updateMany({
+      where: {
+        status: { in: ['SUCCEEDED', 'FAILED'] },
+        updatedAt: { lt: cutoff },
+        OR: [{ input: { not: null } }, { result: { not: null } }],
+      },
+      data: { input: null, result: null, resultMimeType: null },
+    });
+    if (result.count > 0) {
+      this.logger.info(
+        { pruned: result.count, cutoff: cutoff.toISOString(), retentionDays },
+        'audit: conversion job payloads pruned',
+      );
+    }
+    return result.count;
+  }
+
   private async ownedJob(id: string, userId: string): Promise<ConversionJob> {
     const job = await this.prisma.conversionJob.findFirst({ where: { id, ownerId: userId } });
     if (!job) throw new NotFoundException();

apps/api/src/import-export/conversion-payload-prune.e2e.db.test.ts

@@ -0,0 +1,120 @@
+import { INestApplication } from '@nestjs/common';
+import { PrismaClient } from '@prisma/client';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+
+import { createTestApp } from '../testing/test-app';
+import { createTestPrisma, hasTestDb, uniqueSuffix } from '../testing/test-db';
+import { UsersService } from '../users/users.service';
+
+import { ConversionJobService } from './conversion-job.service';
+
+const DAY = 24 * 60 * 60 * 1000;
+
+/**
+ * Conversion payload retention (issue #233): finished jobs past
+ * `conversion.payloadRetentionDays` lose their raw input/result bytes while
+ * the row survives for status; pending and stale-RUNNING rows (the worker's
+ * lock-recovery path) keep their payload untouched.
+ */
+describe.skipIf(!hasTestDb)('conversion payload prune (e2e, issue #233)', () => {
+  let app: INestApplication;
+  let prisma: PrismaClient;
+  let ownerId: string;
+  const suffix = uniqueSuffix();
+
+  const bytes = () => new Uint8Array(Buffer.from(`payload-${suffix}`));
+
+  async function jobRow(
+    status: 'PENDING' | 'RUNNING' | 'SUCCEEDED' | 'FAILED',
+    ageDays: number,
+    lockedAt: Date | null = null,
+  ) {
+    return prisma.conversionJob.create({
+      data: {
+        ownerId,
+        kind: `test_prune_${suffix}`,
+        sourceFormat: 'markdown',
+        targetFormat: 'html',
+        input: bytes(),
+        status,
+        result: status === 'SUCCEEDED' ? bytes() : null,
+        resultMimeType: status === 'SUCCEEDED' ? 'text/html' : null,
+        errorCode: status === 'FAILED' ? 'conversion_failed' : null,
+        lockedAt,
+        updatedAt: new Date(Date.now() - ageDays * DAY),
+      },
+    });
+  }
+
+  beforeAll(async () => {
+    prisma = createTestPrisma();
+    // A short period so ages are unambiguous; written straight to the row
+    // BEFORE the app boots (the settings cache is in-process and fills on
+    // first read). The key is cleaned afterAll.
+    await prisma.instanceSetting.upsert({
+      where: { key: 'conversion.payloadRetentionDays' },
+      create: { key: 'conversion.payloadRetentionDays', value: 10 },
+      update: { value: 10 },
+    });
+    app = await createTestApp();
+    const user = await app.get(UsersService).createUser({
+      username: `pia-prune-${suffix}`,
+      email: `pia-prune-${suffix}@example.org`,
+      displayName: `Pia Prune ${suffix}`,
+      password: 'bytes verschwinden fristgerecht 1',
+      locale: 'en',
+    });
+    ownerId = user.id;
+  });
+
+  afterAll(async () => {
+    await prisma.instanceSetting.deleteMany({
+      where: { key: 'conversion.payloadRetentionDays' },
+    });
+    await prisma.conversionJob.deleteMany({ where: { kind: `test_prune_${suffix}` } });
+    await prisma.user.deleteMany({ where: { username: { contains: suffix } } });
+    await prisma.$disconnect();
+    await app.close();
+  });
+
+  it('prunes finished jobs past the period, keeps everything else', async () => {
+    const oldSucceeded = await jobRow('SUCCEEDED', 15);
+    const oldFailed = await jobRow('FAILED', 15);
+    const freshSucceeded = await jobRow('SUCCEEDED', 5);
+    const oldPending = await jobRow('PENDING', 15);
+    // A crashed run the worker's stale-lock recovery will pick up again —
+    // its input must survive or the retry would fail (#233 acceptance).
+    const oldStaleRunning = await jobRow('RUNNING', 15, new Date(Date.now() - 15 * DAY));
+
+    // >=: the shared suite database may hold other files' aged rows.
+    const pruned = await app.get(ConversionJobService).pruneExpiredPayloads();
+    expect(pruned).toBeGreaterThanOrEqual(2);
+
+    const byId = new Map(
+      (await prisma.conversionJob.findMany({ where: { kind: `test_prune_${suffix}` } })).map(
+        (job) => [job.id, job],
+      ),
+    );
+    // The finished rows survive with status and error code, only bytes-free.
+    expect(byId.get(oldSucceeded.id)).toMatchObject({
+      status: 'SUCCEEDED',
+      input: null,
+      result: null,
+      resultMimeType: null,
+    });
+    expect(byId.get(oldFailed.id)).toMatchObject({
+      status: 'FAILED',
+      errorCode: 'conversion_failed',
+      input: null,
+      result: null,
+    });
+    expect(byId.get(freshSucceeded.id)!.input).not.toBeNull();
+    expect(byId.get(freshSucceeded.id)!.result).not.toBeNull();
+    expect(byId.get(oldPending.id)!.input).not.toBeNull();
+    expect(byId.get(oldStaleRunning.id)!.input).not.toBeNull();
+  });
+
+  it('is a no-op when nothing is due', async () => {
+    expect(await app.get(ConversionJobService).pruneExpiredPayloads()).toBe(0);
+  });
+});

apps/api/src/import-export/conversion-worker.service.ts

@@ -14,7 +14,12 @@ import {
 } from './data-export.constants';
 import { GotenbergRenderer, RenderError } from './gotenberg.renderer';
 import { IMPORT_PROCESSOR, ImportProcessor, isImportKind } from './import.constants';
-import { ConversionError, ConversionResult, PandocConverter } from './pandoc.converter';
+import {
+  ConversionError,
+  ConversionResult,
+  conversionInputOf,
+  PandocConverter,
+} from './pandoc.converter';
 
 /** How often the worker sweeps for pending jobs on its own — the safety net
  * that makes a queued conversion survive an API restart even if no new
@@ -59,7 +64,7 @@ export class ConversionWorker implements OnModuleInit, OnModuleDestroy {
     const result: ConversionResult = await this.converter.convert({
       from: job.sourceFormat,
       to: job.targetFormat,
-      input: Buffer.from(job.input),
+      input: Buffer.from(conversionInputOf(job)),
       standalone: job.standalone,
     });
     return { bytes: result.output, mimeType: result.mimeType };
@@ -157,7 +162,9 @@ export class ConversionWorker implements OnModuleInit, OnModuleDestroy {
         expiresAt = new Date(Date.now() + DATA_EXPORT_TTL_MS);
       } else if (job.targetFormat === 'pdf') {
         output = {
-          bytes: await this.renderer.renderHtmlToPdf(Buffer.from(job.input).toString('utf8')),
+          bytes: await this.renderer.renderHtmlToPdf(
+            Buffer.from(conversionInputOf(job)).toString('utf8'),
+          ),
           mimeType: 'application/pdf',
         };
       } else {

apps/api/src/import-export/import-export.module.ts

@@ -1,11 +1,13 @@
 import { Module, OnModuleInit } from '@nestjs/common';
 
+import { CommonModule } from '../common/common.module';
 import { FilesModule } from '../files/files.module';
 import { LabelsModule } from '../labels/labels.module';
 import { PagesModule } from '../pages/pages.module';
 import { PluginsModule } from '../plugins/plugins.module';
 import { SchedulerModule } from '../scheduler/scheduler.module';
 import { SchedulerService } from '../scheduler/scheduler.service';
+import { SettingsModule } from '../settings/settings.module';
 
 import { ConversionJobService } from './conversion-job.service';
 import { ConversionWorker } from './conversion-worker.service';
@@ -25,13 +27,25 @@ import { PandocConverter, PandocServerConverter } from './pandoc.converter';
  * the link's own expiry check already stops downloads the moment it lapses. */
 const EXPORT_PURGE_CADENCE_SECONDS = 60 * 60;
 
+/** Daily, per operations.md's maintenance-jobs table (issue #233): the
+ * payload retention works in days, so a tighter cadence buys nothing. */
+const PAYLOAD_PRUNE_CADENCE_SECONDS = 24 * 60 * 60;
+
 /**
  * Import/export orchestration (ADR 0009): the conversion job queue, its worker,
  * the pandoc-server client (#62), the document import pipeline (#63), the
  * feature exports (#65/#67), and the GDPR account data export (#68).
  */
 @Module({
-  imports: [FilesModule, LabelsModule, PagesModule, PluginsModule, SchedulerModule],
+  imports: [
+    CommonModule,
+    FilesModule,
+    LabelsModule,
+    PagesModule,
+    PluginsModule,
+    SchedulerModule,
+    SettingsModule,
+  ],
   controllers: [JobsController, ImportController, ExportController, DataExportController],
   providers: [
     ConversionJobService,
@@ -56,6 +70,7 @@ export class ImportExportModule implements OnModuleInit {
   constructor(
     private readonly scheduler: SchedulerService,
     private readonly dataExport: DataExportService,
+    private readonly jobs: ConversionJobService,
   ) {}
 
   onModuleInit(): void {
@@ -64,5 +79,10 @@ export class ImportExportModule implements OnModuleInit {
       cadenceSeconds: EXPORT_PURGE_CADENCE_SECONDS,
       run: () => this.dataExport.purgeExpired().then(() => undefined),
     });
+    this.scheduler.register({
+      name: 'conversion-payload-prune',
+      cadenceSeconds: PAYLOAD_PRUNE_CADENCE_SECONDS,
+      run: () => this.jobs.pruneExpiredPayloads().then(() => undefined),
+    });
   }
 }

apps/api/src/import-export/import.service.ts

@@ -32,7 +32,7 @@ import {
   parseVaultZip,
   planVaultImport,
 } from './obsidian-vault';
-import { ConversionError, PandocConverter } from './pandoc.converter';
+import { ConversionError, conversionInputOf, PandocConverter } from './pandoc.converter';
 import { ConversionJobService } from './conversion-job.service';
 
 /** Source format per accepted upload extension (ADR 0009). `md` is our own
@@ -270,7 +270,7 @@ export class ImportService implements ImportProcessor {
     const rawMarkdown = await convertImportedDocument(
       this.converter,
       job.sourceFormat,
-      Buffer.from(job.input),
+      Buffer.from(conversionInputOf(job)),
     );
     const page = await this.createPageFromMarkdown(user, job.pondId, rawMarkdown, job.sourceName);
     await this.prisma.conversionJob.update({
@@ -321,7 +321,7 @@ export class ImportService implements ImportProcessor {
     let plan: VaultImportPlan;
     let assetBytes: Map<string, { data: Uint8Array; name: string }>;
     try {
-      const zip = new Uint8Array(job.input);
+      const zip = new Uint8Array(conversionInputOf(job));
       plan = planVaultImport(zip, {
         frontmatterMode: options.frontmatterMode,
         existingSlugs: new Set(existing.map((row) => row.slug)),

apps/api/src/import-export/pandoc.converter.ts

@@ -50,6 +50,17 @@ export class ConversionError extends Error {
   }
 }
 
+/** The job's input bytes. Since #233 the column is nullable — the retention
+ * job prunes finished jobs' payloads. It never touches PENDING/RUNNING rows
+ * (incl. stale-lock recovery), so a claimed job without input was re-queued
+ * by hand; fail it finally instead of crashing the worker. */
+export function conversionInputOf(job: { input: Uint8Array | null }): Uint8Array {
+  if (!job.input) {
+    throw new ConversionError('conversion_failed', false, 'input payload was pruned');
+  }
+  return job.input;
+}
+
 /** Server-side conversion limits (ADR 0009). Input is checked before the
  * sidecar call; output is capped while reading the response so a runaway
  * conversion can't exhaust memory. */

apps/api/src/settings/instance-settings.service.ts

@@ -40,6 +40,11 @@ export const INSTANCE_SETTINGS = {
   // recorded (`audit.pruned`) so the gap is explainable. The read-access
   // trail (#224) is deliberately NOT covered — it gets its own period.
   'audit.retentionDays': z.number().int().min(1).default(365),
+  // Conversion-job payload retention (issue #233): days a finished
+  // (succeeded/failed) import/export job keeps its raw input/result bytes
+  // before the daily prune job nulls them. The row itself survives for
+  // status/audit purposes; PENDING/RUNNING jobs are never touched.
+  'conversion.payloadRetentionDays': z.number().int().min(1).default(30),
   // Non-image upload allowlist (ADR 0011, issue #61): lowercase extensions
   // without the dot. Images are always allowed regardless; SVG is governed
   // by `upload.svgPolicy`. Normalized (lowercased, dot-stripped, deduped) so

apps/web/e2e/system.spec.ts

@@ -19,8 +19,9 @@ test('lists maintenance jobs and triggers one manually', async ({ browser }) =>
   // All registered jobs appear (language-neutral: row count + button).
   // Keep in sync with the scheduler registrations: trash-purge,
   // version-thinning, page-compaction, data-export-purge,
-  // notification-digest, orphan-file-sweep (#194), audit-retention (#196).
+  // notification-digest, orphan-file-sweep (#194), audit-retention (#196),
-  await expect(jobsTable.locator('tbody tr')).toHaveCount(7);
+  // conversion-payload-prune (#233).
+  await expect(jobsTable.locator('tbody tr')).toHaveCount(8);
 
   const firstRow = jobsTable.locator('tbody tr').first();
   await firstRow.getByRole('button').click();

docs/architecture/operations.md

@@ -99,13 +99,14 @@ monitoring, structured logs, backup alerting — no dedicated metrics stack.
 ## Maintenance jobs (in-app scheduler, `jobs` table)
 
 | Job                      | Cadence                 | Purpose                                             |
-| --------------------- | ----------------------- | --------------------------------------------------- |
+| ------------------------ | ----------------------- | --------------------------------------------------- |
 | trash purge              | daily                   | delete pages/ponds past trash retention (ADR 0013)  |
 | version thinning         | daily                   | auto-version retention policy (ADR 0013)            |
 | update-log compaction    | hourly, idle pages only | bound Yjs log growth                                |
 | quota reconciliation     | nightly                 | recompute `pond_usage`, report drift                |
 | orphan file sweep        | nightly                 | volume ↔ DB consistency (ADR 0011)                  |
 | audit retention          | daily                   | prune `audit_log` past `audit.retentionDays` (#196) |
+| conversion payload prune | daily                   | null finished conversion jobs' raw bytes (#233)     |
 | mail outbox retry        | every minute            | e-mail delivery with backoff                        |
 
 Job outcomes are visible in the Site Admin UI (last run, status) — that
@@ -123,6 +124,16 @@ optional there), so cleanup of those is a human decision in the panel or
 the pond file manager. Attachment deletion is hard everywhere — the
 unused `deleted_at` column was removed with #194.
 
+Conversion payload prune (issue #233): daily. Nulls the raw `input` and
+`result` bytes of import/export conversion jobs that finished (succeeded
+or failed) more than `conversion.payloadRetentionDays` (default 30) ago —
+the bytes are transient, not the durable copy an Attachment is, so a
+deleted page cannot live on inside its last export. The row itself
+survives for status/audit purposes. Pending or running jobs — including
+a crashed RUNNING row the worker's stale-lock recovery will re-run —
+keep their payload. Data-export downloads additionally expire much
+earlier through their own link TTL (#68).
+
 Pond purge (issue #193): a trashed pond past the trash retention is
 removed with everything it holds — pages (cascading versions, comments,
 content cache incl. the search vector, update log, mentions, label

docs/vs-nfd/20-massnahmenplan.md

@@ -249,7 +249,7 @@ fertiges Produkt ohne Gesprächspartner.
 Befunde der nachgezogenen `10-ist-aufnahme.md`, die in diesem Plan
 fehlten — als Issues angelegt:
 
-- [ ] Conversion-Job-Payloads prunen — Rohbytes jedes Im-/Exports liegen
+- [x] Conversion-Job-Payloads prunen — Rohbytes jedes Im-/Exports liegen
       unbefristet in `conversion_jobs` · 2 AT · #233 (M24, I-22)
 - [ ] Retention für `mail_outbox` — Digest-Mails tragen Seitentitel
       · 1 AT · #234 (M24, I-23)