diff --git a/apps/api/.env.example b/apps/api/.env.example index e59aebf87..90c38cbbd 100644 --- a/apps/api/.env.example +++ b/apps/api/.env.example @@ -232,6 +232,7 @@ PASSKEY_ORIGIN=http://127.0.0.1:3000,openmapx:// # apps/api/src/server.ts. Set lower for tight-disk deployments, higher to # keep more history. In-flight jobs are never pruned. # AUDIT_LOG_RETENTION_DAYS=90 +# Finished admin jobs and data-manager pipeline runs older than this are pruned. # ADMIN_JOB_RETENTION_DAYS=30 # Isochrone provider override (default: valhalla). Use "otp" to route diff --git a/apps/api/src/privacy/preservation.test.ts b/apps/api/src/privacy/preservation.test.ts index f6c52bfe8..50fa026ab 100644 --- a/apps/api/src/privacy/preservation.test.ts +++ b/apps/api/src/privacy/preservation.test.ts @@ -71,6 +71,31 @@ describe("privacy preservation", () => { expect(query.params).toEqual(["admin-audit-attribution"]); }); + it("also keeps rows through the statuses a caller names", () => { + const query = dialect.sqlToQuery( + preservationAllowsDeletion({ + registrationId: "data-manager-trigger-attribution", + candidateTimestamp: sql.identifier("started_at"), + subjectIds: [sql.identifier("subject")], + blockingStatuses: ["held", "action_required"], + }), + ); + + expect(query.sql).toContain("preservation.status in ('held', 'action_required')"); + expect(query.params).toEqual(["data-manager-trigger-attribution"]); + }); + + it("refuses a guard that no hold status could block", () => { + expect(() => + preservationAllowsDeletion({ + registrationId: "data-manager-trigger-attribution", + candidateTimestamp: sql.identifier("started_at"), + subjectIds: [sql.identifier("subject")], + blockingStatuses: [], + }), + ).toThrow("requires a blocking hold status"); + }); + it("takes the transaction lock in a separate statement before retention work", async () => { const order: string[] = []; const transaction = { diff --git a/apps/api/src/privacy/preservation.ts b/apps/api/src/privacy/preservation.ts index ca151ad0a..1f63dbb51 100644 --- a/apps/api/src/privacy/preservation.ts +++ b/apps/api/src/privacy/preservation.ts @@ -41,6 +41,19 @@ export function preservationReleaseTransition( : { status: "released", released: true }; } +export type PreservationBlockingStatus = "held" | "action_required"; + +const BLOCKING_STATUS_LITERALS: Record = { + held: "'held'", + action_required: "'action_required'", +}; + +/** The closed set of statuses renders as fixed literals, never caller text. */ +function blockingStatusCondition(statuses: readonly PreservationBlockingStatus[]): SQL { + const literals = [...new Set(statuses)].map((status) => BLOCKING_STATUS_LITERALS[status]); + return sql.raw(literals.length === 1 ? `= ${literals[0]}` : `in (${literals.join(", ")})`); +} + /** * SQL guard for a source retention DELETE. Keeping the hold check in the same * statement as the delete prevents a capture/release worker from creating a @@ -54,9 +67,18 @@ export function preservationAllowsDeletion(input: { registrationId: string; candidateTimestamp: SQLWrapper; subjectIds: readonly SQLWrapper[]; + /** + * Hold statuses that keep the rows. An `operator_task` registration's hold + * becomes `action_required` while the operator still has to preserve the + * rows by hand, so its source must also stay through that status. + */ + blockingStatuses?: readonly PreservationBlockingStatus[]; }): SQL { if (input.subjectIds.length === 0) throw new Error("A preservation deletion guard requires an exact subject locator"); + const blockingStatuses = input.blockingStatuses ?? ["held"]; + if (blockingStatuses.length === 0) + throw new Error("A preservation deletion guard requires a blocking hold status"); const subjectMatch = sql.join( input.subjectIds.map((subjectId) => sql`preservation_request.user_id = ${subjectId}`), sql` or `, @@ -67,7 +89,7 @@ export function preservationAllowsDeletion(input: { inner join data_subject_request preservation_request on preservation_request.id = preservation.request_id where preservation.registration_id = ${input.registrationId} - and preservation.status = 'held' + and preservation.status ${blockingStatusCondition(blockingStatuses)} and preservation.released_at is null and preservation.source_cutoff_at >= ${input.candidateTimestamp} and preservation_request.user_id is not null diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 1f736999f..c80cef51d 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -79,6 +79,7 @@ import { pruneAppLogs, pruneAuditLog, pruneCompletedJobs, + pruneDataManagerJobs, pruneExpiredVerifications, } from "./services/activity-retention"; import { @@ -644,6 +645,9 @@ const runRetention = () => { void pruneCompletedJobs(JOB_RETENTION_DAYS).catch((err) => server.log.warn(err, "Admin job prune failed"), ); + void pruneDataManagerJobs(JOB_RETENTION_DAYS).catch((err) => + server.log.warn(err, "Data-manager job prune failed"), + ); void pruneAppLogs(APP_LOG_RETENTION_DAYS).catch((err) => server.log.warn(err, "Application log prune failed"), ); diff --git a/apps/api/src/services/__tests__/activity-retention.test.ts b/apps/api/src/services/__tests__/activity-retention.test.ts index 6fbfcfc1c..02a107a93 100644 --- a/apps/api/src/services/__tests__/activity-retention.test.ts +++ b/apps/api/src/services/__tests__/activity-retention.test.ts @@ -12,6 +12,13 @@ vi.mock("../../db/schema", () => ({ }, adminJob: { id: "jobId", createdBy: "jobCreatedBy", finishedAt: "jobFinishedAt" }, appLog: { id: "appLogId", metadata: "appLogMetadata", createdAt: "appLogCreatedAt" }, + dataManagerJobs: { + id: "dataManagerJobId", + kind: "dataManagerJobKind", + startedAt: "dataManagerJobStartedAt", + finishedAt: "dataManagerJobFinishedAt", + triggeredBy: "dataManagerJobTriggeredBy", + }, verification: { id: "verificationId", expiresAt: "verificationExpiresAt" }, })); vi.mock("drizzle-orm", async (importOriginal) => { @@ -25,6 +32,7 @@ const { pruneAppLogs, pruneAuditLog, pruneCompletedJobs, + pruneDataManagerJobs, pruneExpiredVerifications, } = await import("../activity-retention.js"); @@ -100,6 +108,24 @@ describe("pruneAuditLog / pruneCompletedJobs retention guard", () => { }); }); +// Its deletion against PostgreSQL is data-manager-job-retention-postgres.test.ts's. +describe("pruneDataManagerJobs retention guard", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.spyOn(console, "warn").mockImplementation(() => {}); + }); + afterEach(() => vi.restoreAllMocks()); + + it.each([0, -5, Number.NaN, Number.POSITIVE_INFINITY])( + "days = %s is a no-op guard", + async (days) => { + expect(await pruneDataManagerJobs(days)).toBe(0); + expect(dbMock.db.delete).not.toHaveBeenCalled(); + expect(console.warn).toHaveBeenCalled(); + }, + ); +}); + describe("parseRequiredRetentionDays", () => { it("uses the documented fallback when the setting is absent", () => { expect(parseRequiredRetentionDays("BACKUP_RETENTION_DAYS", 30, {})).toBe(30); diff --git a/apps/api/src/services/__tests__/data-manager-job-retention-postgres.test.ts b/apps/api/src/services/__tests__/data-manager-job-retention-postgres.test.ts new file mode 100644 index 000000000..e7f502c97 --- /dev/null +++ b/apps/api/src/services/__tests__/data-manager-job-retention-postgres.test.ts @@ -0,0 +1,157 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, inArray, like } from "drizzle-orm"; +import { afterEach, describe, expect, it } from "vitest"; +import { db } from "../../db/index.js"; +import { + dataManagerJobStages, + dataManagerJobs, + dataSubjectRequest, + dataSubjectRequestPreservation, + user, +} from "../../db/schema.js"; +import { loadMasterKeyRing } from "../../privacy/crypto.js"; +import { releasePreservation } from "../../privacy/preservation.js"; +import { PrivacyRequestService } from "../../privacy/request-service.js"; +import { pruneDataManagerJobs } from "../activity-retention.js"; + +const DAY = 86_400_000; +const users: string[] = []; +const requests: string[] = []; + +afterEach(async () => { + await db.delete(dataManagerJobs).where(like(dataManagerJobs.kind, "retention-fixture-%")); + for (const id of requests.splice(0)) + await db.delete(dataSubjectRequest).where(eq(dataSubjectRequest.id, id)); + for (const id of users.splice(0)) await db.delete(user).where(eq(user.id, id)); +}); + +interface Fixture { + status?: string; + daysAgo: number; + triggeredBy?: string; + finished?: boolean; +} + +async function insertJob(kind: string, fixture: Fixture): Promise { + const startedAt = new Date(Date.now() - fixture.daysAgo * DAY); + const [row] = await db + .insert(dataManagerJobs) + .values({ + kind, + status: fixture.status ?? "ok", + startedAt, + finishedAt: fixture.finished === false ? null : new Date(startedAt.getTime() + 60_000), + triggeredBy: fixture.triggeredBy ?? "cron:data-manager-cron", + }) + .returning({ id: dataManagerJobs.id }); + await db.insert(dataManagerJobStages).values({ + jobId: row.id, + stage: "fetch", + status: "ok", + startedAt, + finishedAt: startedAt, + durationMs: 1, + }); + return row.id; +} + +async function remaining(ids: string[]): Promise { + const rows = await db + .select({ id: dataManagerJobs.id }) + .from(dataManagerJobs) + .where(inArray(dataManagerJobs.id, ids)); + return rows.map((row) => row.id).sort(); +} + +describe.skipIf(process.env.OPENMAPX_RUN_DATABASE_TESTS !== "1")( + "data-manager job retention on PostgreSQL", + () => { + it("deletes finished jobs past retention with their stages and keeps the rest", async () => { + const kind = `retention-fixture-${randomUUID()}`; + const otherKind = `retention-fixture-${randomUUID()}`; + const expired = await insertJob(kind, { daysAgo: 40 }); + const interrupted = await insertJob(kind, { daysAgo: 35, status: "interrupted" }); + const running = await insertJob(kind, { daysAgo: 45, status: "running", finished: false }); + const newest = await insertJob(kind, { daysAgo: 2 }); + // A rarely run kind keeps its only run, however old. + const onlyRunOfOtherKind = await insertJob(otherKind, { daysAgo: 90 }); + + expect(await pruneDataManagerJobs(30)).toBeGreaterThanOrEqual(2); + + expect(await remaining([expired, interrupted, running, newest, onlyRunOfOtherKind])).toEqual( + [running, newest, onlyRunOfOtherKind].sort(), + ); + const stages = await db + .select({ jobId: dataManagerJobStages.jobId }) + .from(dataManagerJobStages) + .where(inArray(dataManagerJobStages.jobId, [expired, interrupted])); + expect(stages).toEqual([]); + }); + + it("keeps the jobs an account triggered while its data-subject request needs them", async () => { + const userId = randomUUID(); + users.push(userId); + await db.insert(user).values({ + id: userId, + name: "Retention fixture", + email: `${userId}@example.test`, + updatedAt: new Date(), + }); + const ring = loadMasterKeyRing({ + env: { + NODE_ENV: "development", + OPENMAPX_EXPORTS_KEY: Buffer.alloc(32, 9).toString("base64url"), + }, + }); + const service = new PrivacyRequestService({ database: db, keyRing: ring }); + const request = await service.create({ + userId, + kind: "access", + channel: "self_service", + locale: "en", + timeZone: "UTC", + }); + requests.push(request.id); + + const kind = `retention-fixture-${randomUUID()}`; + const heldApi = await insertJob(kind, { daysAgo: 40, triggeredBy: `api:user:${userId}` }); + const heldManual = await insertJob(kind, { + daysAgo: 39, + triggeredBy: `manual:admin:${userId}`, + }); + const otherUser = await insertJob(kind, { + daysAgo: 38, + triggeredBy: `api:user:${randomUUID()}`, + }); + const scheduled = await insertJob(kind, { daysAgo: 37 }); + const newest = await insertJob(kind, { daysAgo: 1 }); + + await pruneDataManagerJobs(30); + expect(await remaining([heldApi, heldManual, otherUser, scheduled, newest])).toEqual( + [heldApi, heldManual, newest].sort(), + ); + + const holds = await db + .select({ + id: dataSubjectRequestPreservation.id, + status: dataSubjectRequestPreservation.status, + }) + .from(dataSubjectRequestPreservation) + .where( + and( + eq(dataSubjectRequestPreservation.requestId, request.id), + eq(dataSubjectRequestPreservation.registrationId, "data-manager-trigger-attribution"), + ), + ); + // The operator still has to preserve this registration's rows by hand. + expect(holds).toEqual([{ id: holds[0]?.id, status: "action_required" }]); + await pruneDataManagerJobs(30); + expect(await remaining([heldApi, heldManual, newest])).toHaveLength(3); + + await releasePreservation({ id: holds[0].id, database: db }); + + await pruneDataManagerJobs(30); + expect(await remaining([heldApi, heldManual, newest])).toEqual([newest]); + }); + }, +); diff --git a/apps/api/src/services/activity-jobs.test.ts b/apps/api/src/services/activity-jobs.test.ts index 831b10729..8fb1e65c9 100644 --- a/apps/api/src/services/activity-jobs.test.ts +++ b/apps/api/src/services/activity-jobs.test.ts @@ -17,6 +17,7 @@ describe("activity job normalization", () => { ["manual:admin:user-2", "user-2", "Administrator"], ["cron:data-manager-cron", null, "Scheduled"], ["bootstrap", null, "Bootstrap"], + ["cron:data-manager-auto-bump", null, "Automatic version bump"], ["data-manager-auto-bump", null, "Automatic version bump"], [null, null, "System"], ])("parses data-manager actor %s", (triggeredBy, userId, label) => { diff --git a/apps/api/src/services/activity-jobs.ts b/apps/api/src/services/activity-jobs.ts index d78885a0d..965e87928 100644 --- a/apps/api/src/services/activity-jobs.ts +++ b/apps/api/src/services/activity-jobs.ts @@ -74,7 +74,8 @@ export function parseDataManagerActor(triggeredBy: string | null): { return { userId: null, label: "Scheduled" }; } if (triggeredBy === "bootstrap") return { userId: null, label: "Bootstrap" }; - if (triggeredBy === "data-manager-auto-bump") { + // The data-manager stores its scheduled runs as `cron:`. + if (triggeredBy === "data-manager-auto-bump" || triggeredBy === "cron:data-manager-auto-bump") { return { userId: null, label: "Automatic version bump" }; } diff --git a/apps/api/src/services/activity-retention.ts b/apps/api/src/services/activity-retention.ts index 9a113d537..b37291192 100644 --- a/apps/api/src/services/activity-retention.ts +++ b/apps/api/src/services/activity-retention.ts @@ -1,6 +1,6 @@ -import { and, lt, sql } from "drizzle-orm"; +import { and, desc, isNotNull, lt, notInArray, sql } from "drizzle-orm"; import { db } from "../db"; -import { adminAuditLog, adminJob, appLog, verification } from "../db/schema"; +import { adminAuditLog, adminJob, appLog, dataManagerJobs, verification } from "../db/schema"; import { preservationAllowsDeletion, withPreservationRetentionLock, @@ -84,6 +84,52 @@ export async function pruneCompletedJobs(days: number): Promise { return result.length; } +/** + * The account a data-manager job's `triggered_by` names, in the formats + * `parseDataManagerActor` reads; NULL for scheduled and other system runs. + */ +const dataManagerJobSubject = sql`substring(${dataManagerJobs.triggeredBy} from '^(?:api:user:|manual:admin:|manual:user:)(.+)$')`; + +/** + * Delete finished data-manager pipeline jobs, and by cascade their stages, + * that finished more than `days` ago. A running job has no finish time and is + * never deleted. The newest finished job of each kind is kept, so the transit + * state still reports the last sync when the schedule runs rarely or is off. + */ +export async function pruneDataManagerJobs(days: number): Promise { + if (!Number.isFinite(days) || days <= 0) { + console.warn( + `Skipping data-manager job prune: retention days must be a positive finite number, got ${days}`, + ); + return 0; + } + const cutoff = new Date(Date.now() - days * MS_PER_DAY); + const result = await withPreservationRetentionLock(db, (tx) => { + const newestFinishedOfEachKind = tx + .selectDistinctOn([dataManagerJobs.kind], { id: dataManagerJobs.id }) + .from(dataManagerJobs) + .where(isNotNull(dataManagerJobs.finishedAt)) + .orderBy(dataManagerJobs.kind, desc(dataManagerJobs.startedAt)); + return tx + .delete(dataManagerJobs) + .where( + and( + lt(dataManagerJobs.finishedAt, cutoff), + notInArray(dataManagerJobs.id, newestFinishedOfEachKind), + preservationAllowsDeletion({ + registrationId: "data-manager-trigger-attribution", + candidateTimestamp: dataManagerJobs.startedAt, + subjectIds: [dataManagerJobSubject], + // An operator preserves this registration's rows by hand. + blockingStatuses: ["held", "action_required"], + }), + ), + ) + .returning({ id: dataManagerJobs.id }); + }); + return result.length; +} + /** Delete persisted operational log records after the disclosed retention period. */ export async function pruneAppLogs(days: number): Promise { if (!Number.isFinite(days) || days <= 0) { diff --git a/docs/docs/administration/monitoring.md b/docs/docs/administration/monitoring.md index 10721a1f1..4764279c4 100644 --- a/docs/docs/administration/monitoring.md +++ b/docs/docs/administration/monitoring.md @@ -480,9 +480,12 @@ The server writes the trail independently of the client. Its operational limits `AUDIT_LOG_RETENTION_DAYS` (default 90), so the table doesn't grow without limit on a long-lived instance. Raise or lower it via the env var. -The **Jobs** tab lists running and recently finished background operations with their logs. Job rows are pruned after `ADMIN_JOB_RETENTION_DAYS` (default -30). Between them, the audit log tells you the _intent_ of every admin action and -the jobs view shows the _execution_. +The **Jobs** tab lists running and recently finished background operations with their logs. A daily prune deletes finished jobs after `ADMIN_JOB_RETENTION_DAYS` (default +30). That covers the data-manager's pipeline runs too, with their stages, except +the newest finished run of each kind, so the transit page keeps showing the last +sync. Jobs that are still running are never pruned, and neither are jobs that a +pending data-subject request still needs. Between them, the audit log tells you +the _intent_ of every admin action and the jobs view shows the _execution_. Application job details are pushed to the browser over Server-Sent Events (`GET /api/admin/jobs//events`). Every event carries a per-job cursor; when diff --git a/docs/docs/install/configuration.md b/docs/docs/install/configuration.md index 9ce31d792..d119861ab 100644 --- a/docs/docs/install/configuration.md +++ b/docs/docs/install/configuration.md @@ -512,7 +512,7 @@ Lower-level toggles, retention, and legal-page metadata. All optional. | `OPENMAPX_DISABLE_LOCALHOST_AUTH` | Set to `1` to disable the loopback admin short-circuit entirely, regardless of `OPENMAPX_LOCAL_ADMIN_TOKEN`. Use on multi-tenant hosts where another local process could reach the API's loopback. | Optional. Commented | | `ISOCHRONE_PROVIDER` | Ground-mode isochrone provider. Only `valhalla` is supported; unset or empty also selects it. Other values fail. Transit reachability uses the separate MOTIS workflow. | Optional. Commented. Default `valhalla` | | `AUDIT_LOG_RETENTION_DAYS` | Days to keep admin audit-log entries before the daily prune. | Optional. Commented `90` | -| `ADMIN_JOB_RETENTION_DAYS` | Days to keep finished admin jobs (in-flight jobs are never pruned). | Optional. Commented `30` | +| `ADMIN_JOB_RETENTION_DAYS` | Days to keep finished admin jobs and data-manager pipeline runs. In-flight jobs and the newest finished run of each pipeline are never pruned. | Optional. Commented `30` | | `BACKUP_RETENTION_DAYS` | Days to keep local backups. Expired backups are pruned and refused by restore; apply the same lifecycle to off-host copies. | Optional. Default `30` | | `GITHUB_TOKEN` | GitHub API token for catalog requests. Access and rate limits depend on the token and GitHub policy. | Optional. Commented | | `EXTENSION_CATALOG_URL` | Default catalog URL for the **Extensions** store; the curated (verified-tier) list shown under `/admin/extensions`. | Optional. `https://raw.githubusercontent.com/openmapx/community-extensions/main/catalog.json` | diff --git a/infra/docker/.env.example b/infra/docker/.env.example index 85505ab4d..11b0542e7 100644 --- a/infra/docker/.env.example +++ b/infra/docker/.env.example @@ -534,6 +534,7 @@ THUNDERFOREST_API_KEY= # admin jobs once a day. In-flight jobs (finishedAt IS NULL) are never # pruned. Defaults track apps/api/src/server.ts. # AUDIT_LOG_RETENTION_DAYS=90 +# Finished admin jobs and data-manager pipeline runs older than this are pruned. # ADMIN_JOB_RETENTION_DAYS=30 # Backups older than this are pruned and refused by restore. BACKUP_RETENTION_DAYS=30 diff --git a/packages/db-schema/src/data-manager.ts b/packages/db-schema/src/data-manager.ts index a3bd2484a..e30109bf5 100644 --- a/packages/db-schema/src/data-manager.ts +++ b/packages/db-schema/src/data-manager.ts @@ -28,7 +28,10 @@ export const jobs = dataManager.table("jobs", { finishedAt: timestamp("finished_at", { withTimezone: true }), /** "running" | "ok" | "partial" | "error" */ status: text("status").notNull(), - /** "cron" | "manual:" */ + /** + * `:`, e.g. "cron:data-manager-cron" or "api:user:". + * A user id here is personal data: retention prunes the row once it ages out. + */ triggeredBy: text("triggered_by"), /** * Client-supplied idempotency token. A POST /sync that replays the same diff --git a/scripts/database-test-groups.json b/scripts/database-test-groups.json index 1d76f6dd7..92c674cd8 100644 --- a/scripts/database-test-groups.json +++ b/scripts/database-test-groups.json @@ -18,6 +18,7 @@ "apps/api/src/routes/__tests__/reviews-keypair-postgres.test.ts", "apps/api/src/routes/__tests__/saved-import-postgres.test.ts", "apps/api/src/routes/__tests__/shares-postgres.test.ts", + "apps/api/src/services/__tests__/data-manager-job-retention-postgres.test.ts", "apps/api/src/services/__tests__/extension-store-valkey.test.ts", "apps/api/src/services/__tests__/health-history.integration.test.ts", "apps/api/src/services/capability-bindings.test.ts",