feat(backend): telemetry, retry, distributed locking and stress tests for Log Retention & Archival Service

backend/.env.example

@@ -97,3 +97,15 @@ RESEND_API_KEY=your_resend_api_key
 # Sender address for receipt emails (must be a verified domain in Resend)
 # Defaults to: receipts@notifications.dripsnetwork.com
 RECEIPT_FROM_EMAIL=receipts@yourdomain.com
+
+# Log Retention & Archival Service (docs/LOG_RETENTION.md)
+# LOG_RETENTION_DAYS=30
+# LOG_PURGE_BATCH_SIZE=1000
+# LOG_PURGE_MAX_DURATION_MS=300000
+# LOG_PURGE_RETRY_MAX_ATTEMPTS=4
+# LOG_PURGE_RETRY_BASE_DELAY_MS=250
+# LOG_PURGE_RETRY_MAX_DELAY_MS=5000
+# Run the purge inside the API on a cron schedule (safe on every instance):
+# LOG_RETENTION_PURGE_CRON=30 3 * * *
+# Push the cron script's metrics here (http/https only):
+# LOG_RETENTION_PUSHGATEWAY_URL=http://pushgateway:9091

backend/docs/alerts/log-retention.rules.yml

@@ -0,0 +1,83 @@
+# Prometheus alerting rules for the Log Retention & Archival Service (issue #1473).
+#
+# Load with:   rule_files: ["docs/alerts/log-retention.rules.yml"]
+# Validate:    promtool check rules docs/alerts/log-retention.rules.yml
+#
+# Metric reference: docs/LOG_RETENTION.md
+#
+# The purge usually runs as a cron script that pushes to a Pushgateway
+# (LOG_RETENTION_PUSHGATEWAY_URL), or inside the API when
+# LOG_RETENTION_PURGE_CRON is set. Either way the series below are emitted.
+
+groups:
+  - name: log-retention
+    rules:
+      - alert: LogRetentionPurgeUnhealthy
+        expr: max(log_retention_health_state) >= 2
+        for: 10m
+        labels:
+          severity: critical
+        annotations:
+          summary: Webhook log purge reports unhealthy
+          description: >
+            The purge failed several times in a row or has not succeeded within
+            the staleness limit (LOG_RETENTION_HEALTH_MAX_STALENESS_MS). Old
+            webhook delivery logs are accumulating. See GET /health/log-retention
+            for the reasons and the last error in the logs.
+
+      - alert: LogRetentionPurgeFailing
+        expr: max(log_retention_purge_consecutive_failures) >= 1
+        for: 30m
+        labels:
+          severity: warning
+        annotations:
+          summary: Last webhook log purge run failed
+          description: >
+            At least one purge run failed and no run has succeeded since.
+            Check the "Log retention purge failed" log line for the error code.
+
+      - alert: LogRetentionPurgeStale
+        expr: |
+          max(log_retention_purge_last_success_timestamp_seconds) > 0
+          and
+          time() - max(log_retention_purge_last_success_timestamp_seconds) > 2 * 86400
+        for: 15m
+        labels:
+          severity: warning
+        annotations:
+          summary: No successful webhook log purge in 48 hours
+          description: >
+            The daily purge has not succeeded for two days. Check that the cron
+            (npm run purge:webhook-logs) or LOG_RETENTION_PURGE_CRON still runs.
+
+      - alert: LogRetentionPurgeDurationExceeded
+        expr: increase(log_retention_purge_duration_exceeded_total[1d]) > 0
+        labels:
+          severity: warning
+        annotations:
+          summary: Webhook log purge overran its time budget
+          description: >
+            A purge run took longer than LOG_PURGE_MAX_DURATION_MS. The backlog
+            may be large or the database slow; consider a larger
+            LOG_PURGE_BATCH_SIZE or running the purge more often.
+
+      - alert: LogRetentionRetriesExhausted
+        expr: increase(log_retention_purge_retries_total{result="exhausted"}[1h]) > 0
+        labels:
+          severity: warning
+        annotations:
+          summary: Webhook log purge gave up after retries
+          description: >
+            A purge query kept failing with transient database errors until the
+            retry budget (LOG_PURGE_RETRY_MAX_ATTEMPTS) ran out.
+
+      - alert: LogRetentionLockErrors
+        expr: increase(log_retention_purge_lock_total{result="error"}[1h]) > 0
+        labels:
+          severity: warning
+        annotations:
+          summary: Webhook log purge could not take its advisory lock
+          description: >
+            Acquiring the Postgres advisory lock failed (not merely contended).
+            The purge did not run. Check database connectivity from the host
+            that runs the purge.

backend/load-tests/log-retention-stress.test.js

@@ -0,0 +1,190 @@
+/**
+ * Log Retention & Archival Service — stress suite (issue #1476).
+ *
+ * Many simulated instances (one pool each) purge one shared in-memory
+ * Postgres double concurrently, with injected latency and transient
+ * failures. No network and no real database.
+ *
+ * Invariants checked under load:
+ *   - every expired row is deleted exactly once; rows inside the retention
+ *     window are never touched
+ *   - at most one instance holds the purge lock at any moment
+ *   - no advisory lock, connection or in-process guard leaks, including
+ *     after failures
+ *   - run counters add up: success + failure + skipped == attempts
+ *   - retries absorb a high rate of transient errors
+ *
+ * Run with:  npm run test:load -- log-retention-stress
+ */
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+
+vi.mock("../src/lib/logger.js", () => ({
+  logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+}));
+
+import { purgeWebhookLogs, runLogRetentionPurge } from "../src/lib/log-retention.js";
+import {
+  logRetentionRunsTotal,
+  resetLogRetentionHealth,
+} from "../src/lib/log-retention-telemetry.js";
+import { resetLogRetentionLocks } from "../src/lib/log-retention-lock.js";
+import { createFakeLogRetentionDb, makeRows, pgError } from "../tests/helpers/fake-log-retention-db.js";
+
+const fastRetry = { sleep: async () => {}, baseDelayMs: 0, maxAttempts: 8 };
+
+/** Deterministic PRNG so a failing seed can be replayed. */
+function prng(seed) {
+  let s = seed >>> 0;
+  return () => {
+    s = (s * 1664525 + 1013904223) >>> 0;
+    return s / 2 ** 32;
+  };
+}
+
+async function runs(outcome) {
+  const { values } = await logRetentionRunsTotal.get();
+  return values.find((v) => v.labels.outcome === outcome)?.value ?? 0;
+}
+
+describe("log retention stress", () => {
+  beforeEach(() => {
+    resetLogRetentionHealth();
+    resetLogRetentionLocks();
+    vi.spyOn(console, "log").mockImplementation(() => {});
+    vi.spyOn(console, "error").mockImplementation(() => {});
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it("50 instances racing: one purges, the rest skip, every expired row deleted once", async () => {
+    const expired = makeRows(5_000, 90, "old");
+    const kept = makeRows(500, 3, "new");
+    const db = createFakeLogRetentionDb({ rows: [...expired, ...kept], latencyMs: 1 });
+
+    const before = { success: await runs("success"), skipped: await runs("skipped") };
+    const results = await Promise.all(
+      Array.from({ length: 50 }, () =>
+        runLogRetentionPurge({
+          pool: db.createPool(),
+          retentionDays: 30,
+          batchSize: 250,
+          retry: fastRetry,
+        }),
+      ),
+    );
+
+    const ran = results.filter((r) => !r.skipped);
+    expect(ran).toHaveLength(1);
+    expect(ran[0].totalDeleted).toBe(5_000);
+    expect(results.filter((r) => r.skipped)).toHaveLength(49);
+    expect(db.stats.maxLockHolders).toBe(1);
+    expect(db.doubleDeletes).toEqual([]);
+    expect(new Set(db.deleted).size).toBe(db.deleted.length);
+    expect(db.remainingIds().sort()).toEqual(kept.map((r) => r.id).sort());
+    expect(db.heldLocks()).toEqual([]);
+    expect(db.stats.openConnections).toBe(0);
+    expect(await runs("success")).toBe(before.success + 1);
+    expect(await runs("skipped")).toBe(before.skipped + 49);
+  });
+
+  it("repeated waves with 20% transient failures converge with no leaks", async () => {
+    const random = prng(1476);
+    const transient = ["40001", "40P01", "55P03", "57014", "08006", "ECONNRESET"];
+    const db = createFakeLogRetentionDb({
+      rows: [...makeRows(3_000, 45, "old"), ...makeRows(300, 10, "new")],
+      latencyMs: () => Math.floor(random() * 3),
+      fail: (kind) =>
+        kind !== "unlock" && random() < 0.2
+          ? pgError(transient[Math.floor(random() * transient.length)])
+          : null,
+    });
+
+    let attempts = 0;
+    let outcomes = { ok: 0, skipped: 0, failed: 0 };
+    for (let wave = 0; wave < 10 && db.remainingIds().length > 300; wave += 1) {
+      const results = await Promise.allSettled(
+        Array.from({ length: 8 }, () => {
+          attempts += 1;
+          return runLogRetentionPurge({
+            pool: db.createPool(),
+            retentionDays: 30,
+            batchSize: 100,
+            retry: fastRetry,
+          });
+        }),
+      );
+      for (const r of results) {
+        if (r.status === "rejected") outcomes.failed += 1;
+        else if (r.value.skipped) outcomes.skipped += 1;
+        else outcomes.ok += 1;
+      }
+      expect(db.heldLocks()).toEqual([]);
+      expect(db.stats.openConnections).toBe(0);
+    }
+
+    expect(outcomes.ok + outcomes.skipped + outcomes.failed).toBe(attempts);
+    expect(outcomes.ok).toBeGreaterThanOrEqual(1);
+    expect(db.remainingIds()).toHaveLength(300);
+    expect(db.remainingIds().every((id) => id.startsWith("new-"))).toBe(true);
+    expect(db.doubleDeletes).toEqual([]);
+    expect(db.stats.maxLockHolders).toBe(1);
+  });
+
+  it("unlocked direct callers under contention still delete each row once (SKIP LOCKED)", async () => {
+    const db = createFakeLogRetentionDb({
+      rows: makeRows(4_000, 120),
+      latencyMs: () => 1,
+    });
+
+    const results = await Promise.all(
+      Array.from({ length: 16 }, () =>
+        purgeWebhookLogs({
+          pool: db.createPool(),
+          retentionDays: 30,
+          batchSize: 64,
+          retry: fastRetry,
+        }),
+      ),
+    );
+
+    expect(results.reduce((sum, r) => sum + r.totalDeleted, 0)).toBe(4_000);
+    expect(db.doubleDeletes).toEqual([]);
+    expect(db.remainingIds()).toEqual([]);
+  });
+
+  it("a burst of triggers on one instance runs at most one purge at a time", async () => {
+    const db = createFakeLogRetentionDb({ rows: makeRows(1_000, 40), latencyMs: 1 });
+    const pool = db.createPool();
+
+    const results = await Promise.all(
+      Array.from({ length: 100 }, () =>
+        runLogRetentionPurge({ pool, retentionDays: 30, batchSize: 100, retry: fastRetry }),
+      ),
+    );
+
+    expect(results.filter((r) => !r.skipped)).toHaveLength(1);
+    // Skipped in-process: only the winner ever reached the database lock.
+    expect(db.callCount("lock")).toBe(1);
+    expect(db.remainingIds()).toEqual([]);
+  });
+
+  it("keeps purge throughput reasonable for a large backlog", async () => {
+    const db = createFakeLogRetentionDb({ rows: makeRows(50_000, 365) });
+    const started = Date.now();
+
+    const result = await runLogRetentionPurge({
+      pool: db.createPool(),
+      retentionDays: 30,
+      batchSize: 1_000,
+      retry: fastRetry,
+    });
+
+    expect(result.totalDeleted).toBe(50_000);
+    expect(db.callCount("delete")).toBe(51);
+    // Generous ceiling: this guards against accidental O(n^2) per-row work
+    // in the service, not against a slow CI machine.
+    expect(Date.now() - started).toBeLessThan(10_000);
+  });
+});

backend/scripts/purge-webhook-logs.js

@@ -5,11 +5,19 @@ import { purgeWebhookLogs, sanitizeLogMessage } from "../src/lib/log-retention.j
 async function run() {
   const startedAt = Date.now();
   try {
-    const result = await purgeWebhookLogs({ pool });
+    // Takes the cross-instance purge lock (issue #1475): when the cron runs
+    // on several hosts, exactly one purges and the others exit cleanly.
+    const result = await runLogRetentionPurge({ pool });
     const elapsedMs = Date.now() - startedAt;
-    console.log(
-      `[log-retention] Run succeeded: table=${result.tableName} deleted_rows=${result.totalDeleted} elapsed_ms=${elapsedMs}`,
-    );
+    if (result.skipped) {
+      console.log(
+        `[log-retention] Run skipped: table=${result.tableName} another purge holds the lock elapsed_ms=${elapsedMs}`,
+      );
+    } else {
+      console.log(
+        `[log-retention] Run succeeded: table=${result.tableName} deleted_rows=${result.totalDeleted} elapsed_ms=${elapsedMs}`,
+      );
+    }
   } catch (error) {
     const elapsedMs = Date.now() - startedAt;
     // Driver errors can echo the connection string; log a scrubbed message
@@ -20,6 +28,11 @@ async function run() {
     );
     process.exitCode = 1;
   } finally {
+    // This process exits before any scrape could reach it, so push the run's
+    // metrics (issue #1473) when a Pushgateway is configured.
+    if (process.env.LOG_RETENTION_PUSHGATEWAY_URL) {
+      await pushLogRetentionMetrics(process.env.LOG_RETENTION_PUSHGATEWAY_URL);
+    }
     await closePool();
   }
 }