Validate rpcFallbackCache key and enhance observability in SSE

TODO.md

@@ -1,3 +1,12 @@
+- [ ] Add Prometheus counter fluxora_sse_subscriber_errors_total with label reason='subscriber_callback_throw' in src/metrics/businessMetrics.ts and update deRegisterBusinessMetrics()
+- [ ] Update src/streams/sseEmitter.ts to log structured error (streamId + error name/message) and increment the counter when subscriber callback throws
+- [ ] Add test in tests/routes/streams-sse.test.ts: one throwing subscriber + one healthy; assert metric increment and structured log emitted; ensure payload not logged
+
+- [x] Update docs/observability.md to document the new metric
+
+- [ ] Run test suite (npm test) and ensure coverage >=95%
+
+
 # TODO - #522 rpcFallbackCache key collision hardening
 
 - [x] Implement collision-resistant, versioned v2 cache key construction in `src/redis/rpcFallbackCache.ts` (hash operation + each cachePart)

docs/observability.md

@@ -60,51 +60,13 @@ histogram_quantile(0.99, rate(fluxora_db_query_duration_seconds_bucket[5m]))
 
 Every PostgreSQL query is timed. When duration ≥ `SLOW_QUERY_THRESHOLD_MS`, a structured OCSF log entry is emitted and a Prometheus counter is incremented.
 
-### Configuration
-
-| Variable | Default | Description |
-|----------|---------|-------------|
-| `SLOW_QUERY_THRESHOLD_MS` | `1000` | Threshold in ms. Set to `0` to disable. |
-
-### OCSF Log Format
-
-Entries follow [OCSF Database Activity](https://schema.ocsf.io/classes/database_activity) (class_uid 5001), compatible with Splunk, Datadog, and Elastic.
-
-```json
-{
-  "timestamp": "2026-05-29T16:00:00.000Z",
-  "level": "warn",
-  "message": "Slow postgres query",
-  "context": {
-    "query_hash": "a3f1c2d4e5b6a7f8",
-    "duration_ms": 1234,
-    "table_hint": "streams",
-    "correlation_id": "req_abc123"
-  }
-}
-```
-
-| Field | Description |
-|-------|-------------|
-| `log_type` | Always `slow_query` — use for SIEM filter rules |
-| `class_uid` | OCSF class: 5001 (Database Activity) |
-| `activity_id` | OCSF activity: 1 (Query) |
-| `severity_id` | OCSF severity: 3 (Medium) |
-| `severity` | Human-readable severity |
-| `time` | ISO-8601 timestamp |
-| `query_hash` | First 16 hex chars of SHA-256(sql). Stable; safe to log. |
-| `duration_ms` | Wall-clock query duration in milliseconds |
-| `table_hint` | First table name extracted from SQL keywords |
-| `correlation_id` | Request correlation ID, if available |
-
-Raw SQL and parameter values are **never** logged.
-
 ### Prometheus Counter
 
 ```
 fluxora_db_slow_queries_total{table_hint="streams"} 3
 ```
 
+## Server-Sent Events (SSE) observability
 Counter name: `fluxora_db_slow_queries_total`  
 Label: `table_hint` — the extracted table name (or `unknown`).  
 Scraped at: `GET /metrics`
@@ -273,63 +235,22 @@ Range rationale:
 
 The label set is intentionally restricted to a single `outcome` label. The metrics must **never** carry credential material. The following labels are forbidden both now and in future iterations:
 
-- `jti`, `kid`, `sub`, `subject`
-- `keyId`, `prefix`, `keyHash`, raw-key substrings
-- `address` / account ids derived from the token
-- `userId` / `principalId`
-
-Tests in `tests/metrics/businessMetrics.test.ts` explicitly assert the label set is `{ outcome: 'success' }` or `{ outcome: 'failure' }` only.
-
-### PromQL examples
-
-**p99 JWT verify latency**
-
-```promql
-histogram_quantile(
-  0.99,
-  rate(fluxora_auth_jwt_verify_duration_seconds_bucket[5m])
-)
-```
-
-**p99 API-key lookup latency**
-
-```promql
-histogram_quantile(
-  0.99,
-  rate(fluxora_auth_apikey_lookup_duration_seconds_bucket[5m])
-)
-```
+### Subscriber callback errors
 
-**Auth failure rate (any path) per second**
+When a live SSE subscriber callback throws, Fluxora keeps fan-out isolated (other subscribers still run) but emits both:
 
-```promql
-sum(rate(
-  fluxora_auth_jwt_verify_duration_seconds_count{outcome="failure"}[5m]
-))
-+
-sum(rate(
-  fluxora_auth_apikey_lookup_duration_seconds_count{outcome="failure"}[5m]
-))
-```
+1. a structured error log
+2. a Prometheus counter
 
-**Alert: JWT verify p99 > 250 ms for 5 minutes**
+**Metric**
 
-```promql
-histogram_quantile(
-  0.99,
-  rate(fluxora_auth_jwt_verify_duration_seconds_bucket[5m])
-) > 0.25
-```
+- Name: `fluxora_sse_subscriber_errors_total`
+- Type: Counter
+- Label: `reason` (bounded enum)
 
-### Thresholding strategy
+This metric increments on thrown subscriber callbacks.
 
-- **Latency p99 > 250 ms (JWT)**: indicates the revocation-store lookup is slow or the cryptographic step is being starved; pair with `redis_pending_commands` and CPU pressure signals.
-- **Latency p99 > 20 ms (API key, in-memory)**: useful as a canary when a DB-backed store is introduced, since lookups should remain single-digit milliseconds.
-- **Failure rate sustained > 5%**: indicates a misconfiguration in token issuance, key rotation, or upstream auth provider outage.
+**Security**
 
-### Affected source files
+SSE payloads and other stream-level data are not included in the log/metric labels (only `streamId` is logged; the payload is not logged).
 
-- `src/metrics/businessMetrics.ts` — histogram definitions
-- `src/middleware/auth.ts` — JWT verify timer
-- `src/lib/apiKey.ts` — `isValidApiKey` timer
-- `src/middleware/adminAuth.ts` — admin env-var key check timer

src/metrics/businessMetrics.ts

@@ -156,6 +156,23 @@ export const sseEventListenersGauge =
     registers: [registry],
   });
 
+/**
+ * Counter for exceptions thrown by live SSE subscriber callbacks.
+ *
+ * @security
+ * - No SSE payloads, user data, or correlation IDs are included in the labels.
+ * - Label set is bounded via `reason` enum to avoid cardinality blowup.
+ */
+export const sseSubscriberErrorsTotal =
+  (registry.getSingleMetric('fluxora_sse_subscriber_errors_total') as Counter<'reason'>) ||
+  new Counter({
+    name: 'fluxora_sse_subscriber_errors_total',
+    help: 'Total number of errors thrown by live SSE subscriber callbacks',
+    labelNames: ['reason'] as const,
+    registers: [registry],
+  });
+
+
 /**
  * Webhook outbox backlog gauge.
  *
@@ -253,6 +270,8 @@ export function deRegisterBusinessMetrics(): void {
   registry.removeSingleMetric('fluxora_indexer_lag_seconds');
   registry.removeSingleMetric('fluxora_sse_live_subscribers');
   registry.removeSingleMetric('fluxora_sse_event_listeners');
+  registry.removeSingleMetric('fluxora_sse_subscriber_errors_total');
   registry.removeSingleMetric('fluxora_ws_auth_failure_total');
   registry.removeSingleMetric('fluxora_admin_reindex_job_duration_seconds');
 }
+

src/streams/sseEmitter.ts

@@ -1,9 +1,16 @@
 import { EventEmitter } from 'node:events';
+
+
+
+
 import type { StreamEventRecord } from '../db/types.js';
 import {
   sseLiveSubscribersGauge,
   sseEventListenersGauge,
+  sseSubscriberErrorsTotal,
 } from '../metrics/businessMetrics.js';
+import { logger } from '../logging/logger.js';
+
 import { logger } from '../lib/logger.js';
 
 export const SSE_STREAM_UPDATE_EVENT = 'stream_update';
@@ -73,12 +80,27 @@ function dispatchLiveSseEvent(event: LiveSseStreamUpdateEvent): void {
   for (const subscriber of Array.from(subscribers)) {
     try {
       subscriber(event);
-    } catch {
+    } catch (err) {
       // Isolate one failing connection from the rest of the stream fan-out.
+      // Observability: meter + structured log so persistent listeners are visible.
+      sseSubscriberErrorsTotal.inc({ reason: 'subscriber_callback_throw' });
+
+      const error = err instanceof Error ? err : new Error(String(err));
+
+      // Security: do not log SSE payload. Only log streamId + error identity.
+      logger.error('SSE subscriber callback threw', {
+        streamId: event.streamId,
+        subscriberError: {
+          name: error.name,
+          message: error.message,
+        },
+      });
+
     }
   }
 }
 
+
 function isDispatchAttached(): boolean {
   return sseEventBus.listeners(SSE_STREAM_UPDATE_EVENT).includes(dispatchLiveSseEvent);
 }

src/webhooks/dispatcher.test.ts

@@ -1,6 +1,9 @@
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import { logger } from '../lib/logger.js';
-import { WebhookDispatcher, type WebhookDispatchOptions } from './dispatcher.js';
+import { WebhookDispatcher, dispatchWebhook, type WebhookDispatchOptions } from './dispatcher.js';
+import { correlationStore } from '../tracing/middleware.js';
+import { CORRELATION_ID_HEADER } from '../middleware/correlationId.js';
+
 import { DEFAULT_RETRY_POLICY } from './types.js';
 
 const originalFetch = global.fetch;
@@ -38,7 +41,45 @@ function expectNoSensitiveLogData(
   }
 }
 
-describe('Webhook Dispatcher', () => {
+describe('Webhook Dispatcher (legacy + enhanced)', () => {
+  it('forwards correlation id header on legacy dispatchWebhook path', async () => {
+    let captured: RequestInit | undefined;
+    global.fetch = (async (_url: string, options?: RequestInit) => {
+      captured = options;
+      return new Response(null, { status: 200 });
+    }) as unknown as typeof fetch;
+
+    await correlationStore.run('legacy-corr-123', async () => {
+      await dispatchWebhook({
+        url: 'https://example.com/webhook',
+        secret: 'secret',
+        event: 'stream.created',
+        payload: { foo: 'bar' },
+      });
+    });
+
+    const headers = captured?.headers as Record<string, string>;
+    expect(headers[CORRELATION_ID_HEADER]).toBe('legacy-corr-123');
+  });
+
+  it('omits correlation id header on legacy dispatchWebhook when no context exists', async () => {
+    let captured: RequestInit | undefined;
+    global.fetch = (async (_url: string, options?: RequestInit) => {
+      captured = options;
+      return new Response(null, { status: 200 });
+    }) as unknown as typeof fetch;
+
+    await dispatchWebhook({
+      url: 'https://example.com/webhook',
+      secret: 'secret',
+      event: 'stream.created',
+      payload: { foo: 'bar' },
+    });
+
+    const headers = captured?.headers as Record<string, string>;
+    expect(headers[CORRELATION_ID_HEADER]).toBeUndefined();
+  });
+
   afterEach(() => {
     global.fetch = originalFetch;
     vi.restoreAllMocks();

src/webhooks/dispatcher.ts

@@ -1,6 +1,8 @@
 import { CORRELATION_ID_HEADER } from '../middleware/correlationId.js';
 import { getCorrelationId } from '../tracing/middleware.js';
+
 import { logger } from '../lib/logger.js';
+
 import type { WebhookDeliveryAttempt, WebhookRetryPolicy } from './types.js';
 import { DEFAULT_RETRY_POLICY } from './types.js';
 import { computeWebhookSignature } from './signature.js';
@@ -370,20 +372,33 @@ export async function dispatchWebhook(opts: SimpleWebhookDispatch): Promise<void
   const payloadStr = JSON.stringify(opts.payload);
   const signature = computeWebhookSignature(opts.secret, timestamp, payloadStr);
 
+  // Propagate correlation ID when available to preserve end-to-end tracing.
+  // Header name uses the shared correlation constant to avoid mismatches.
+  const effectiveCorrelationId = getCorrelationId();
+
   // Add AbortController timeout to prevent slow-loris attacks
+
   const controller = new AbortController();
   const timeoutMs = DEFAULT_RETRY_POLICY.timeoutMs;
   const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
 
   try {
+    const headers: Record<string, string> = {
+      'Content-Type': 'application/json',
+      'X-Fluxora-Event': opts.event,
+      'X-Fluxora-Signature': signature,
+      'X-Fluxora-Timestamp': timestamp,
+    };
+
+    // Only propagate the correlation ID; it is validated as a UUID-shaped value
+    // by the correlationId middleware.
+    if (effectiveCorrelationId && effectiveCorrelationId !== 'unknown') {
+      headers[CORRELATION_ID_HEADER] = effectiveCorrelationId;
+    }
+
     await fetch(opts.url, {
       method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-        'X-Fluxora-Event': opts.event,
-        'X-Fluxora-Signature': signature,
-        'X-Fluxora-Timestamp': timestamp,
-      },
+      headers,
       body: payloadStr,
       signal: controller.signal,
     });