fix(retention): persist daily cleanup outcomes

backend/__tests__/service/pgRetentionRun.test.js

@@ -0,0 +1,80 @@
+/**
+ * Retention-run ledger — executed against real PostgreSQL.
+ *
+ * The claim is persistence across a backend restart, which pg-mem cannot
+ * establish. A separate Pool is the relevant control: the model writes with
+ * the application's pool; this reader observes the committed outcome through
+ * a fresh connection.
+ */
+const fs = require('fs');
+const path = require('path');
+const { Pool } = require('pg');
+const PgRetentionRun = require('../../models/pg/PgRetentionRun');
+
+const RUN = process.env.INTEGRATION_TEST === 'true';
+const d = RUN ? describe : describe.skip;
+const TEST_DETAIL = 'pg-retention-run-tier1-test';
+
+let pool;
+
+const connect = () => new Pool({
+  host: process.env.PG_HOST,
+  port: Number(process.env.PG_PORT || 5432),
+  database: process.env.PG_DATABASE,
+  user: process.env.PG_USER,
+  password: process.env.PG_PASSWORD,
+  ssl: false,
+});
+
+d('pg_retention_runs, against a real database', () => {
+  beforeAll(async () => {
+    pool = connect();
+    await pool.query(fs.readFileSync(path.join(__dirname, '../../config/schema.sql'), 'utf8'));
+  });
+
+  beforeEach(async () => {
+    await pool.query('DELETE FROM pg_retention_runs WHERE detail = $1', [TEST_DETAIL]);
+  });
+
+  afterAll(async () => {
+    if (pool) {
+      await pool.query('DELETE FROM pg_retention_runs WHERE detail = $1', [TEST_DETAIL]);
+      await pool.end();
+    }
+    // The model owns a second pool; close it so Jest does not retain a handle.
+    // eslint-disable-next-line global-require
+    const { pool: modelPool } = require('../../config/db-pg');
+    if (modelPool?.end) await modelPool.end();
+  });
+
+  test('a completed deletion count survives a new database connection', async () => {
+    const runId = await PgRetentionRun.start({ configuredRetentionDays: 30, targetBytes: 6 });
+    await PgRetentionRun.finish(runId, {
+      status: 'completed',
+      finalRetentionDays: 30,
+      protectedPodCount: 71,
+      deletedMessageCount: 9,
+      reRootedCount: 2,
+      initialSizeBytes: 1024,
+      finalSizeBytes: 1000,
+      detail: TEST_DETAIL,
+    });
+
+    const observer = connect();
+    try {
+      const { rows } = await observer.query(
+        `SELECT status, deleted_message_count, re_rooted_count, protected_pod_count, finished_at
+           FROM pg_retention_runs WHERE id = $1`,
+        [runId],
+      );
+      expect(rows).toHaveLength(1);
+      expect(rows[0].status).toBe('completed');
+      expect(Number(rows[0].deleted_message_count)).toBe(9);
+      expect(Number(rows[0].re_rooted_count)).toBe(2);
+      expect(Number(rows[0].protected_pod_count)).toBe(71);
+      expect(rows[0].finished_at).toBeTruthy();
+    } finally {
+      await observer.end();
+    }
+  });
+});

backend/__tests__/service/threading.retention.test.js

@@ -97,9 +97,10 @@ d('deleting a thread root, against a real database', () => {
     const { R, C, G } = await chain();
     await pool.query("UPDATE messages SET created_at = NOW() - INTERVAL '400 days' WHERE id = $1", [R]);
 
-    const { deleted } = await PGMessage.deleteOlderThan(30);
+    const { deleted, reRooted } = await PGMessage.deleteOlderThan(30);
 
     expect(deleted).toBeGreaterThanOrEqual(1);
+    expect(reRooted).toBe(1);
     expect(await rootOf(G)).toBe(C);
   });
 

backend/__tests__/services/pgRetentionService.test.js

@@ -4,6 +4,10 @@ jest.mock('../../config/db-pg', () => ({
 jest.mock('../../models/pg/Message', () => ({
   deleteOlderThan: jest.fn(),
 }));
+jest.mock('../../models/pg/PgRetentionRun', () => ({
+  start: jest.fn(),
+  finish: jest.fn(),
+}));
 jest.mock('node-cron', () => ({ schedule: jest.fn(() => ({ start: jest.fn(), stop: jest.fn() })) }));
 
 // `find().select().lean()` — chainable, so the service's real call shape works.
@@ -25,6 +29,7 @@ jest.mock('../../models/Pod', () => ({
 
 const { pool } = require('../../config/db-pg');
 const Message = require('../../models/pg/Message');
+const PgRetentionRun = require('../../models/pg/PgRetentionRun');
 const cron = require('node-cron');
 const { runMessageRetention, initPgRetention } = require('../../services/pgRetentionService');
 
@@ -51,6 +56,10 @@ describe('pgRetentionService.runMessageRetention', () => {
   beforeEach(() => {
     pool.query.mockReset();
     Message.deleteOlderThan.mockReset();
+    PgRetentionRun.start.mockReset();
+    PgRetentionRun.start.mockResolvedValue(42);
+    PgRetentionRun.finish.mockReset();
+    PgRetentionRun.finish.mockResolvedValue(undefined);
     process.env = { ...ORIGINAL_ENV };
     delete process.env.PG_MESSAGE_RETENTION_DAYS;
     delete process.env.PG_CAPACITY_BYTES;
@@ -81,12 +90,21 @@ describe('pgRetentionService.runMessageRetention', () => {
       expect.stringContaining('invalid PG_MESSAGE_RETENTION_DAYS'),
       '0',
     );
+    expect(PgRetentionRun.start).toHaveBeenCalledWith({
+      configuredRetentionDays: null,
+      targetBytes: null,
+    });
+    expect(PgRetentionRun.finish).toHaveBeenCalledWith(42, expect.objectContaining({
+      status: 'skipped',
+      deletedMessageCount: 0,
+      detail: 'invalid PG_MESSAGE_RETENTION_DAYS',
+    }));
   });
 
   it('runs single pass and skips tiering when size already under target', async () => {
     // 8 GiB cap, 75% target = 6 GiB. 4 GiB everywhere = under target.
     mockSizeQueries([4, 4, 4]);
-    Message.deleteOlderThan.mockResolvedValue({ deleted: 10 });
+    Message.deleteOlderThan.mockResolvedValue({ deleted: 10, reRooted: 2 });
 
     await runMessageRetention();
 
@@ -96,6 +114,13 @@ describe('pgRetentionService.runMessageRetention', () => {
     expect(vacuumCalls).toHaveLength(1);
     expect(vacuumCalls[0][0]).toMatch(/VACUUM ANALYZE messages/i);
     expect(warnSpy).not.toHaveBeenCalled();
+    expect(PgRetentionRun.finish).toHaveBeenCalledWith(42, expect.objectContaining({
+      status: 'completed',
+      deletedMessageCount: 10,
+      reRootedCount: 2,
+      protectedPodCount: 0,
+      finalRetentionDays: 30,
+    }));
   });
 
   it('steps down retention when still above target, stops when under', async () => {
@@ -183,6 +208,27 @@ describe('pgRetentionService.runMessageRetention', () => {
 
     await expect(runMessageRetention()).resolves.toBeUndefined();
     expect(errSpy).toHaveBeenCalledWith('[pg-retention] failed:', 'db down');
+    expect(PgRetentionRun.finish).toHaveBeenCalledWith(42, expect.objectContaining({
+      status: 'failed',
+      deletedMessageCount: 0,
+      detail: 'db down',
+    }));
+  });
+
+  it('leaves the run as running when recording a completed outcome fails', async () => {
+    mockSizeQueries([4, 4]);
+    Message.deleteOlderThan.mockResolvedValue({ deleted: 5 });
+    PgRetentionRun.finish.mockRejectedValue(new Error('ledger update lost'));
+
+    await expect(runMessageRetention()).resolves.toBeUndefined();
+
+    // Do not retry with `failed`: the start record remains the honest durable
+    // signal that the run completed but its terminal outcome was not written.
+    expect(PgRetentionRun.finish.mock.calls.map(([, outcome]) => outcome.status)).toEqual(['completed']);
+    expect(errSpy).toHaveBeenCalledWith(
+      '[pg-retention] could not persist run outcome:',
+      'ledger update lost',
+    );
   });
   /*
    * The Pro tier's headline promise is "Unlimited message history — nothing
@@ -305,6 +351,11 @@ describe('pgRetentionService.runMessageRetention', () => {
         expect.stringContaining('ABORT'),
         expect.stringContaining('mongo unreachable'),
       );
+      expect(PgRetentionRun.finish).toHaveBeenCalledWith(42, expect.objectContaining({
+        status: 'aborted',
+        deletedMessageCount: 0,
+        detail: expect.stringContaining('mongo unreachable'),
+      }));
     });
 
     it('with no Pro users the delete is unchanged', async () => {
@@ -316,6 +367,17 @@ describe('pgRetentionService.runMessageRetention', () => {
       expect(Message.deleteOlderThan).toHaveBeenCalledWith(30, []);
     });
   });
+
+  it('refuses to delete if the durable run record cannot start', async () => {
+    PgRetentionRun.start.mockRejectedValue(new Error('ledger unavailable'));
+    mockSizeQueries([1, 1]);
+    Message.deleteOlderThan.mockResolvedValue({ deleted: 9 });
+
+    await runMessageRetention();
+
+    expect(Message.deleteOlderThan).not.toHaveBeenCalled();
+    expect(errSpy).toHaveBeenCalledWith('[pg-retention] failed:', 'ledger unavailable');
+  });
 });
 
 
@@ -338,4 +400,4 @@ describe('pgRetentionService.initPgRetention', () => {
       expect(cron.schedule).not.toHaveBeenCalled();
     }
   });
-});
+});

backend/__tests__/unit/models/pg/message.retentionExempt.test.js

@@ -38,7 +38,7 @@ describe('Message.deleteOlderThan retention exemption', () => {
 
   it('unset env keeps the original unconditional delete', async () => {
     const res = await Message.deleteOlderThan(30);
-    expect(res).toEqual({ deleted: 3 });
+    expect(res).toEqual({ deleted: 3, reRooted: 0 });
     const [sql, params] = pool.query.mock.calls[0];
     expect(sql).not.toMatch(/pod_id/);
     expect(params).toEqual(['30 days']);
@@ -73,11 +73,21 @@ describe('Message.deleteOlderThan retention exemption', () => {
 
   it('invalid day counts still refuse to run at all', async () => {
     process.env.PG_RETENTION_EXEMPT_POD_IDS = SHOWCASE;
-    expect(await Message.deleteOlderThan(0)).toEqual({ deleted: 0 });
-    expect(await Message.deleteOlderThan(NaN)).toEqual({ deleted: 0 });
+    expect(await Message.deleteOlderThan(0)).toEqual({ deleted: 0, reRooted: 0 });
+    expect(await Message.deleteOlderThan(NaN)).toEqual({ deleted: 0, reRooted: 0 });
     expect(pool.query).not.toHaveBeenCalled();
   });
 
+  it('reports a repair failure as unknown, never as zero re-rooted rows', async () => {
+    const repair = jest.spyOn(Message, 'reRootOrphanedChains')
+      .mockRejectedValue(new Error('repair database unavailable'));
+    try {
+      await expect(Message.deleteOlderThan(30)).resolves.toEqual({ deleted: 3, reRooted: null });
+    } finally {
+      repair.mockRestore();
+    }
+  });
+
   /*
    * The Pro tier sells "Unlimited message history — nothing expires at 30
    * days". These pin the mechanism that makes that sentence true.

backend/__tests__/unit/models/pg/pgRetentionRun.test.js

@@ -0,0 +1,54 @@
+jest.mock('../../../../config/db-pg', () => ({
+  pool: { query: jest.fn() },
+}));
+
+const { pool } = require('../../../../config/db-pg');
+const PgRetentionRun = require('../../../../models/pg/PgRetentionRun');
+
+describe('PgRetentionRun', () => {
+  beforeEach(() => pool.query.mockReset());
+
+  it('starts a durable row before retention work begins', async () => {
+    pool.query.mockResolvedValue({ rows: [{ id: '7' }], rowCount: 1 });
+
+    await expect(PgRetentionRun.start({ configuredRetentionDays: 30, targetBytes: 6 })).resolves.toBe(7);
+
+    const [sql, params] = pool.query.mock.calls[0];
+    expect(sql).toMatch(/INSERT INTO pg_retention_runs/);
+    expect(params).toEqual([30, 6]);
+  });
+
+  it('persists the outcome fields that a restart would otherwise discard', async () => {
+    pool.query.mockResolvedValue({ rows: [], rowCount: 1 });
+
+    await PgRetentionRun.finish(7, {
+      status: 'completed',
+      finalRetentionDays: 30,
+      protectedPodCount: 71,
+      deletedMessageCount: 9,
+      reRootedCount: 2,
+      initialSizeBytes: 1024,
+      finalSizeBytes: 1000,
+    });
+
+    const [sql, params] = pool.query.mock.calls[0];
+    expect(sql).toMatch(/UPDATE pg_retention_runs/);
+    expect(sql).toMatch(/finished_at = CURRENT_TIMESTAMP/);
+    expect(params).toEqual([7, 'completed', 30, 71, 9, 2, 1024, 1000, null]);
+  });
+
+  it('refuses to report an outcome for a run that was not recorded', async () => {
+    pool.query.mockResolvedValue({ rows: [], rowCount: 0 });
+
+    await expect(PgRetentionRun.finish(7, {
+      status: 'failed',
+      finalRetentionDays: null,
+      protectedPodCount: null,
+      deletedMessageCount: 0,
+      reRootedCount: null,
+      initialSizeBytes: null,
+      finalSizeBytes: null,
+      detail: 'database down',
+    })).rejects.toThrow('run ledger row 7 was not found');
+  });
+});

backend/config/schema.sql

@@ -120,6 +120,32 @@ CREATE TABLE IF NOT EXISTS migration_records (
   details JSONB
 );
 
+-- Retention execution ledger. This is deliberately NOT `migration_records`:
+-- migration records describe one-off changes, while this records every run of
+-- a destructive recurring job. A backend restart discards pod logs, so the
+-- deletion count and the reasons no deletion occurred must live with the data.
+-- A `running` row left behind is evidence of an interrupted run, not success.
+CREATE TABLE IF NOT EXISTS pg_retention_runs (
+  id BIGSERIAL PRIMARY KEY,
+  status VARCHAR(16) NOT NULL DEFAULT 'running'
+    CHECK (status IN ('running', 'completed', 'aborted', 'failed', 'skipped')),
+  configured_retention_days INTEGER,
+  final_retention_days INTEGER,
+  protected_pod_count INTEGER,
+  deleted_message_count BIGINT NOT NULL DEFAULT 0,
+  re_rooted_count BIGINT,
+  target_bytes BIGINT,
+  initial_size_bytes BIGINT,
+  final_size_bytes BIGINT,
+  detail TEXT,
+  started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  finished_at TIMESTAMP WITH TIME ZONE
+);
+-- `CREATE TABLE IF NOT EXISTS` does not add fields to an already-created
+-- ledger, so keep this additive form for instances that booted an earlier DDL.
+ALTER TABLE pg_retention_runs ADD COLUMN IF NOT EXISTS re_rooted_count BIGINT;
+CREATE INDEX IF NOT EXISTS idx_pg_retention_runs_started_at ON pg_retention_runs(started_at DESC);
+
 -- Sprint B5: message reactions. One row per (message, user, emoji) — a user
 -- can stack different emojis on the same message but each emoji is binary
 -- (toggle on/off). PG-only; Mongo fallback path doesn't get reactions in v1.
@@ -184,4 +210,4 @@ CREATE INDEX IF NOT EXISTS idx_pod_members_pod_id ON pod_members(pod_id);
 CREATE INDEX IF NOT EXISTS idx_pod_members_user_id ON pod_members(user_id);
 CREATE INDEX IF NOT EXISTS idx_pods_created_by ON pods(created_by);
 CREATE INDEX IF NOT EXISTS idx_pods_type ON pods(type);
-CREATE INDEX IF NOT EXISTS idx_users_username ON users(username); 
+CREATE INDEX IF NOT EXISTS idx_users_username ON users(username);

backend/models/pg/Message.ts

@@ -393,9 +393,9 @@ class Message {
   static async deleteOlderThan(
     days: number,
     protectedPodIds: string[] = [],
-  ): Promise<{ deleted: number }> {
+  ): Promise<{ deleted: number; reRooted: number | null }> {
     if (!Number.isFinite(days) || days <= 0) {
-      return { deleted: 0 };
+      return { deleted: 0, reRooted: 0 };
     }
     const fromEnv = String(process.env.PG_RETENTION_EXEMPT_POD_IDS || '')
       .split(',')
@@ -419,9 +419,11 @@ class Message {
     // here rather than in the retention cron means every caller of this method
     // gets it, including a future one written by someone who never read
     // TASK-043.
+    let reRooted: number | null = 0;
     if (deleted > 0) {
       try {
         const repair = await Message.reRootOrphanedChains();
+        reRooted = repair.reRooted;
         if (repair.reRooted > 0) {
           console.log(
             `[pg-retention] re-rooted ${repair.reRooted} orphaned reply row(s) `
@@ -434,9 +436,12 @@ class Message {
         // unknown and renders expanded — noisy and non-destructive. Throwing
         // here would make retention look broken for a cosmetic consequence.
         console.warn('[pg-retention] re-root after delete failed:', (err as Error).message);
+        // The delete happened; retaining a zero here would make the ledger
+        // claim the repair executed and found nothing. Preserve that unknown.
+        reRooted = null;
       }
     }
-    return { deleted };
+    return { deleted, reRooted };
   }
 
   static async findActivityHint(podId: unknown, since: unknown): Promise<ActivityHintResult> {