Feature/outbox structured observability

docs/OBSERVABILITY.md

@@ -440,6 +440,9 @@ It also exports the following Prometheus metrics to track throughput, lag, and f
 - **`outbox_pending_gauge`** (Gauge): Current number of pending outbox events (lag/backlog).
 - **`outbox_lease_renew_total`** (Counter): Total number of outbox events whose lease was renewed, indicating processing duration or stalls.
 - **`outbox_dead_letter_total`** (Counter): Total number of outbox events moved to dead-letter, labeled by `error_code`.
+- **`outbox_quarantine_total`** (Counter): Total number of outbox events moved to quarantine, labeled by `reason`.
+- **`outbox_leader_acquired_total`** (Counter): Total number of times this instance acquired outbox leadership.
+- **`outbox_leader_lost_total`** (Counter): Total number of times this instance lost outbox leadership.
 
 ## Timeouts & Retry Policies
 

docs/observability.md

@@ -440,6 +440,9 @@ It also exports the following Prometheus metrics to track throughput, lag, and f
 - **`outbox_pending_gauge`** (Gauge): Current number of pending outbox events (lag/backlog).
 - **`outbox_lease_renew_total`** (Counter): Total number of outbox events whose lease was renewed, indicating processing duration or stalls.
 - **`outbox_dead_letter_total`** (Counter): Total number of outbox events moved to dead-letter, labeled by `error_code`.
+- **`outbox_quarantine_total`** (Counter): Total number of outbox events moved to quarantine, labeled by `reason`.
+- **`outbox_leader_acquired_total`** (Counter): Total number of times this instance acquired outbox leadership.
+- **`outbox_leader_lost_total`** (Counter): Total number of times this instance lost outbox leadership.
 
 ## Timeouts & Retry Policies
 

monitoring/grafana/dashboard.json

@@ -1152,6 +1152,93 @@
         "y": 32
       }
     },
+    {
+      "id": 17,
+      "title": "Outbox Dead Letter",
+      "type": "timeseries",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "rate(outbox_dead_letter_total[5m])",
+          "legendFormat": "{{error_code}}",
+          "refId": "A"
+        }
+      ],
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 0,
+        "y": 40
+      }
+    },
+    {
+      "id": 18,
+      "title": "Outbox Quarantine",
+      "type": "timeseries",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "rate(outbox_quarantine_total[5m])",
+          "legendFormat": "{{reason}}",
+          "refId": "A"
+        }
+      ],
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 8,
+        "y": 40
+      }
+    },
+    {
+      "id": 19,
+      "title": "Outbox Leader Elections",
+      "type": "timeseries",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "rate(outbox_leader_acquired_total[5m])",
+          "legendFormat": "Acquired",
+          "refId": "A"
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "rate(outbox_leader_lost_total[5m])",
+          "legendFormat": "Lost",
+          "refId": "B"
+        }
+      ],
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 16,
+        "y": 40
+      }
+    },
     {
       "id": 15,
       "title": "Horizon Listener Lag (seconds)",
@@ -1206,7 +1293,7 @@
         "h": 8,
         "w": 12,
         "x": 0,
-        "y": 40
+        "y": 48
       },
       "alert": {
         "name": "Horizon listener lag exceeds TTL",
@@ -1250,7 +1337,7 @@
         "h": 8,
         "w": 12,
         "x": 12,
-        "y": 40
+        "y": 48
       }
     }
   ],

src/db/outbox/publisher.test.ts

@@ -4,6 +4,21 @@ import { OutboxPublisher } from './publisher'
 import { OutboxRepository } from './repository'
 import type { OutboxEvent } from './types'
 import crypto from 'crypto'
+import { vi, beforeEach, describe, it, expect } from 'vitest'
+
+vi.mock('../pool.js', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../pool.js')>()
+  const mockPool = {
+    query: vi.fn().mockResolvedValue({ rows: [] }),
+    connect: vi.fn(),
+    end: vi.fn(),
+    on: vi.fn(),
+  }
+  return {
+    ...actual,
+    pool: mockPool as any,
+  }
+})
 
 async function buildTestPool(): Promise<Pool> {
   const db = newDb()
@@ -601,4 +616,50 @@ describe('OutboxRepository correlation id persistence', () => {
     const [claimed] = await repo.claimEvents(pool, 'consumer-a', 10, 60)
     expect(claimed.correlationId).toBeFalsy()
   })
+})
+
+describe('OutboxPublisher edge cases (#1003)', () => {
+  beforeEach(() => {
+    vi.restoreAllMocks()
+  })
+
+  it('resets pending gauge to 0 on stop', async () => {
+    const obs = await import('../../observability/index.js')
+    const spy = vi.spyOn(obs, 'setOutboxPendingGauge')
+
+    const publisher = new OutboxPublisher({ publish: async () => undefined })
+    ;(publisher as any).running = true
+    ;(publisher as any).metricsTimer = setTimeout(() => {}, 1_000_000)
+
+    await publisher.stop()
+
+    expect(spy).toHaveBeenCalledWith(0)
+    clearTimeout((publisher as any).metricsTimer)
+  })
+
+  it('does not log event payload content', async () => {
+    const { logger } = await import('../../utils/logger.js')
+    const infoSpy = vi.spyOn(logger, 'info')
+    const errorSpy = vi.spyOn(logger, 'error')
+    const warnSpy = vi.spyOn(logger, 'warn')
+
+    vi.spyOn(OutboxRepository.prototype, 'markPublished').mockResolvedValue(undefined)
+
+    const publisher = new OutboxPublisher({ publish: async () => undefined })
+    const event = baseEvent({
+      publishIdempotencyKey: 'already-published',
+      payload: { secret: 'should-not-appear-in-logs' },
+    })
+
+    await (publisher as any).processEvent(event)
+
+    const allLogArgs = [
+      ...infoSpy.mock.calls,
+      ...errorSpy.mock.calls,
+      ...warnSpy.mock.calls,
+    ].flat().map(String)
+
+    const leaked = allLogArgs.filter(arg => arg.includes('should-not-appear-in-logs'))
+    expect(leaked).toEqual([])
+  })
 })