#233: prune conversion job payloads for every job kind
Some checks failed
CI / Lint, typecheck, test (pull_request) Successful in 5m12s
CI / Build container images (pull_request) Successful in 3m28s
CI / Auth e2e pack (pull_request) Successful in 8m33s
CI / Import/export fidelity gate (pull_request) Successful in 1m2s
CD / Build and push images (push) Successful in 29s
CD / Deploy to Test (push) Successful in 12s
CD / Smoke tests against Test (push) Successful in 1m22s
CD / Promote to Int (push) Successful in 11s
CI / Lint, typecheck, test (push) Failing after 5m9s
CI / Auth e2e pack (push) Has been skipped
CI / Import/export fidelity gate (push) Has been skipped
CI / Build container images (push) Has been skipped

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
This commit is contained in:
Claude Fable 5 2026-07-30 22:12:32 +02:00
parent ff842f97e2
commit ff505bc752
12 changed files with 254 additions and 24 deletions

View File

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

View File

@ -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
/// later maintenance job (they are transient, not the durable copy an
/// Attachment is). The polling endpoint `GET /jobs/:id` is owner-scoped.
/// document bytes — kept small by the request size limit and transient, not
/// the durable copy an Attachment is: the daily `conversion-payload-prune`
/// 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.

View File

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

View File

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

View File

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

View File

@ -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),
});
}
}

View File

@ -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)),

View File

@ -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. */

View File

@ -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

View File

@ -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).
await expect(jobsTable.locator('tbody tr')).toHaveCount(7);
// notification-digest, orphan-file-sweep (#194), audit-retention (#196),
// conversion-payload-prune (#233).
await expect(jobsTable.locator('tbody tr')).toHaveCount(8);
const firstRow = jobsTable.locator('tbody tr').first();
await firstRow.getByRole('button').click();

View File

@ -98,15 +98,16 @@ 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) |
| mail outbox retry | every minute | e-mail delivery with backoff |
| 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
panel is the operator's single glance for instance health.
@ -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

View File

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