fix(sse): keep one proxy journal row per request actually sent

changelog.d/fixes/14637-proxy-attempt-log-rows.md

@@ -0,0 +1 @@
+- **fix(sse):** the proxy journal keeps one row per request actually sent, so refused outlets stop looking like first-try successes — count requests with `COUNT(DISTINCT correlation_id)` ([#14637](https://github.com/diegosouzapw/OmniRoute/pull/14637)) — thanks @maxmad64bis

open-sse/utils/upstreamStatusCapture.ts

@@ -1,4 +1,44 @@
 import { isProviderRequestCaptureActive, onDispatchStart } from "./providerRequestLogging.ts";
+import { sanitizeErrorMessage } from "./errorSanitization.ts";
+
+/** One request actually sent: the outlet snapshot plus what came back. */
+export type AttemptRecord = {
+  proxy: unknown;
+  rotationAccount?: string | null;
+  upstreamStatus?: number;
+  error?: string | null;
+  startedAt?: number;
+  durationMs?: number | null;
+};
+
+export type AttemptSink = {
+  upstreamStatus?: number;
+  attempts?: AttemptRecord[];
+};
+
+/** Marker for the in-flight record, so two stacked wrappers complete one row, not two. */
+const IN_FLIGHT = "omniroute.attempt.in-flight";
+
+type FlightSink = AttemptSink & Record<string, AttemptRecord | undefined>;
+
+function readFlight(sink: AttemptSink): AttemptRecord | undefined {
+  return (sink as FlightSink)[IN_FLIGHT];
+}
+
+function writeFlight(sink: AttemptSink, record: AttemptRecord | undefined): void {
+  if (record === undefined) delete (sink as FlightSink)[IN_FLIGHT];
+  else (sink as FlightSink)[IN_FLIGHT] = record;
+}
+
+function closeAttemptRecord(
+  record: AttemptRecord,
+  patch: { upstreamStatus?: number; error?: string | null }
+): void {
+  if (patch.upstreamStatus === undefined) record.upstreamStatus = undefined;
+  else record.upstreamStatus = patch.upstreamStatus;
+  record.error = patch.error ?? null;
+  record.durationMs = Date.now() - (record.startedAt ?? Date.now());
+}
 
 /**
  * Wrap the process-wide fetch so the HTTP status the provider actually returned lands on
@@ -18,7 +58,7 @@ import { isProviderRequestCaptureActive, onDispatchStart } from "./providerReque
  */
 export function withUpstreamStatusCapture<A extends unknown[]>(
   inner: (...args: A) => Promise<Response>,
-  getSink: () => { upstreamStatus?: number } | undefined,
+  getSink: () => AttemptSink | undefined,
   isDispatching: () => boolean = isProviderRequestCaptureActive
 ): (...args: A) => Promise<Response> {
   if (isDispatching === isProviderRequestCaptureActive) {
@@ -30,13 +70,39 @@ export function withUpstreamStatusCapture<A extends unknown[]>(
   return async (...args: A) => {
     const sink = isDispatching() ? getSink() : undefined;
     if (!sink) return inner(...args);
+    // One journal row per request actually sent: snapshot the outlet at entry,
+    // complete the row when the send settles. A stacked wrapper reuses the
+    // in-flight row instead of opening a second one. Sends outside a dispatch
+    // (side calls, background work) leave no trace, as before.
+    let record: AttemptRecord | undefined;
+    const inFlight = readFlight(sink);
+    if (inFlight) {
+      record = inFlight;
+    } else {
+      record = {
+        proxy: (sink as { proxy?: unknown }).proxy ?? null,
+        rotationAccount: (sink as { rotationAccount?: string | null }).rotationAccount ?? null,
+        startedAt: Date.now(),
+      };
+      (sink.attempts ??= []).push(record);
+      writeFlight(sink, record);
+    }
+    const active = record === readFlight(sink);
     let response: Response;
     try {
       response = await inner(...args);
     } catch (error) {
+      if (isDispatching() && active) {
+        closeAttemptRecord(record, { error: sanitizeErrorMessage(error) || "Request failed" });
+        writeFlight(sink, undefined);
+      }
       if (isDispatching()) sink.upstreamStatus = undefined;
       throw error;
     }
+    if (isDispatching() && active) {
+      closeAttemptRecord(record, { upstreamStatus: response.status, error: null });
+      writeFlight(sink, undefined);
+    }
     if (isDispatching()) sink.upstreamStatus = response.status;
     return response;
   };

src/lib/db/migrationRunner.ts

@@ -613,6 +613,11 @@ function isSchemaAlreadyApplied(
       // correlation_id at boot. Keyed by version only — a stale number here
       // would answer for another migration's schema and skip it.
       return hasColumn(db, "proxy_logs", "correlation_id");
+    case "187":
+      // Same shape as 179/181/183/184: ensureProxyLogsColumns may have added
+      // attempt_number at boot. Keyed by version only — a stale number here
+      // would answer for another migration's schema and skip it.
+      return hasColumn(db, "proxy_logs", "attempt_number");
     default:
       return false;
   }

src/lib/db/migrations/187_proxy_logs_attempts.sql

@@ -0,0 +1,7 @@
+-- attempt_number: 1-based position of the row within its request journal
+-- (1..N for N sends sharing one correlation_id). NULL for legacy rows.
+-- attempt_issue: outcome of the send within its journal — served for the send
+-- the request settled on, abandoned for earlier sends. NULL for legacy rows.
+-- No index: not a query dimension (count requests via correlation_id).
+ALTER TABLE proxy_logs ADD COLUMN attempt_number INTEGER;
+ALTER TABLE proxy_logs ADD COLUMN attempt_issue TEXT;

src/lib/db/proxyLatency.ts

@@ -13,7 +13,10 @@ type LatencyLogRow = {
 };
 
 // Builds a `"host:port" -> avg_latency_ms` map from proxy_logs rows recorded
-// within the trailing PROXY_LATENCY_WINDOW_HOURS window.
+// within the trailing PROXY_LATENCY_WINDOW_HOURS window. The latency strategy
+// measures the latency of sends served: a fast refusal does not make an
+// outlet fast, so abandoned sends are excluded (legacy rows predate the
+// journal and stay included via the NULL guard).
 function buildLatencyMap(db: ReturnType<typeof getDbInstance>): Map<string, number> {
   const sinceIso = new Date(Date.now() - PROXY_LATENCY_WINDOW_HOURS * 60 * 60 * 1000).toISOString();
 
@@ -22,6 +25,7 @@ function buildLatencyMap(db: ReturnType<typeof getDbInstance>): Map<string, numb
       `SELECT proxy_host, proxy_port, AVG(latency_ms) as avg_latency
        FROM proxy_logs
        WHERE timestamp >= ?
+         AND (attempt_issue IS NULL OR attempt_issue <> 'abandoned')
        GROUP BY proxy_host, proxy_port`
     )
     .all(sinceIso) as LatencyLogRow[];

src/lib/db/schemaColumns.ts

@@ -308,6 +308,14 @@ export function ensureProxyLogsColumns(db: SqliteDatabase) {
       db.exec("ALTER TABLE proxy_logs ADD COLUMN correlation_id TEXT");
       console.log("[DB] Added proxy_logs.correlation_id column");
     }
+    if (!columnNames.has("attempt_number")) {
+      db.exec("ALTER TABLE proxy_logs ADD COLUMN attempt_number INTEGER");
+      console.log("[DB] Added proxy_logs.attempt_number column");
+    }
+    if (!columnNames.has("attempt_issue")) {
+      db.exec("ALTER TABLE proxy_logs ADD COLUMN attempt_issue TEXT");
+      console.log("[DB] Added proxy_logs.attempt_issue column");
+    }
   } catch (error: unknown) {
     const message = error instanceof Error ? error.message : String(error);
     console.warn("[DB] Failed to verify proxy_logs schema:", message);

src/lib/proxyLogger.ts

@@ -72,6 +72,10 @@ interface ProxyLogEntry {
   tlsFingerprint: boolean;
   /** HTTP status the provider actually returned; null when no response was received. */
   upstreamStatus: number | null;
+  /** 1-based position of this row within its request journal; null for unjournaled rows. */
+  attemptNumber: number | null;
+  /** Outcome of this send within its request journal; null for unjournaled rows. */
+  attemptIssue: "served" | "abandoned" | null;
 }
 
 type ProxyLogInput = Partial<ProxyLogEntry> & {
@@ -132,6 +136,11 @@ function loadFromDb() {
         correlationId: row.correlation_id || null,
         tlsFingerprint: row.tls_fingerprint === 1,
         upstreamStatus: typeof row.upstream_status === "number" ? row.upstream_status : null,
+        attemptNumber: typeof row.attempt_number === "number" ? row.attempt_number : null,
+        attemptIssue:
+          row.attempt_issue === "served" || row.attempt_issue === "abandoned"
+            ? row.attempt_issue
+            : null,
       });
     }
 
@@ -223,6 +232,14 @@ export function logProxyEvent(entry: ProxyLogInput) {
     correlationId: entry.correlationId || null,
     tlsFingerprint: entry.tlsFingerprint || false,
     upstreamStatus: entry.upstreamStatus ?? null,
+    attemptNumber:
+      typeof entry.attemptNumber === "number" && Number.isInteger(entry.attemptNumber)
+        ? entry.attemptNumber
+        : null,
+    attemptIssue:
+      entry.attemptIssue === "served" || entry.attemptIssue === "abandoned"
+        ? entry.attemptIssue
+        : null,
   };
 
   // Structured egress line so the operator can confirm, in the proxy logs, which
@@ -313,10 +330,12 @@ export function flushProxyLogsSync() {
     const insertStmt = db.prepare(
       `INSERT INTO proxy_logs (id, timestamp, status, proxy_type, proxy_host, proxy_port, proxy_name,
         level, level_id, provider, target_url, public_ip, egress_ip, latency_ms, error,
-        connection_id, combo_id, account, rotation_account, correlation_id, tls_fingerprint, upstream_status)
+        connection_id, combo_id, account, rotation_account, correlation_id, tls_fingerprint, upstream_status,
+        attempt_number, attempt_issue)
       VALUES (@id, @timestamp, @status, @proxyType, @proxyHost, @proxyPort, @proxyName,
         @level, @levelId, @provider, @targetUrl, @clientIp, @egressIp, @latencyMs, @error,
-        @connectionId, @comboId, @account, @rotationAccount, @correlationId, @tlsFingerprint, @upstreamStatus)`
+        @connectionId, @comboId, @account, @rotationAccount, @correlationId, @tlsFingerprint, @upstreamStatus,
+        @attemptNumber, @attemptIssue)`
     );
 
     const transaction = db.transaction((entries: ProxyLogEntry[]) => {
@@ -344,6 +363,8 @@ export function flushProxyLogsSync() {
           correlationId: item.correlationId,
           tlsFingerprint: item.tlsFingerprint ? 1 : 0,
           upstreamStatus: item.upstreamStatus,
+          attemptNumber: item.attemptNumber,
+          attemptIssue: item.attemptIssue,
         });
       }
     });

src/sse/handlers/chatHelpers.ts

@@ -49,8 +49,9 @@ import { classify429FromError, type FailureKind } from "../../shared/utils/class
 import { resolveUseUpstream429BreakerHints } from "../../shared/utils/providerHints";
 import { isFeatureFlagEnabled } from "../../shared/utils/featureFlags";
 
-import { logProxyEvent } from "../../lib/proxyLogger";
 import { noteProxyOutcome } from "./proxyOutcomeMemory";
+import { logProxyJournal } from "./proxyJournal";
+import type { AttemptJournalEntry } from "./proxyJournal";
 import { logTranslationEvent } from "../../lib/translatorEvents";
 import { getRuntimeProviderProfile } from "@omniroute/open-sse/services/accountFallback.ts";
 
@@ -1061,11 +1062,14 @@ export function withUpstreamStatus<T extends object>(
 }
 
 /** Merge both things the applied-proxy sink captured: the executor proxy, then the status. */
+export type { AttemptJournalEntry } from "./proxyJournal";
 export function mergeAppliedProxySink(
   proxyInfo: { proxy?: unknown; level?: string; levelId?: string | null } | null | undefined,
-  sink: { proxy: unknown; upstreamStatus?: number }
+  sink: { proxy: unknown; upstreamStatus?: number; attempts?: AttemptJournalEntry[] }
 ) {
-  return withUpstreamStatus(applyExecutorProxyToInfo(proxyInfo, sink.proxy), sink);
+  const merged = withUpstreamStatus(applyExecutorProxyToInfo(proxyInfo, sink.proxy), sink);
+  if (sink.attempts?.length) return { ...(merged || {}), attempts: sink.attempts };
+  return merged;
 }
 
 // Async because the egress-IP lookup lazy-imports proxyEgress; callers treat
@@ -1097,50 +1101,18 @@ export async function safeLogEvents({
   }
 
   try {
-    const rawIp =
-      clientRawRequest?.headers?.["x-forwarded-for"] ||
-      clientRawRequest?.headers?.["x-real-ip"] ||
-      clientRawRequest?.headers?.["cf-connecting-ip"] ||
-      null;
-    const rawIpValue = Array.isArray(rawIp) ? rawIp[0] : rawIp;
-    const clientIp = typeof rawIpValue === "string" ? rawIpValue.split(",")[0].trim() : null;
-
-    // Resolve the egress IP (the IP the upstream actually saw) from cache — never
-    // blocking the request. Warm it in the background for next time. null until
-    // the first warm completes; direct (no proxy) is also tracked.
-    let egressIp: string | null = null;
-    try {
-      const { getCachedEgressIp, warmEgressIp } = await import("../../lib/proxyEgress");
-      const { proxyConfigToUrl } = await import("@omniroute/open-sse/utils/proxyDispatcher.ts");
-      const proxyUrl = proxyInfo?.proxy ? proxyConfigToUrl(proxyInfo.proxy) : null;
-      egressIp = getCachedEgressIp(proxyUrl);
-      warmEgressIp(proxyUrl);
-    } catch {
-      // egress visibility is best-effort; never break the request path
-    }
-
-    logProxyEvent({
-      status: result.success
-        ? "success"
-        : result.status === 408 || result.status === 504
-          ? "timeout"
-          : "error",
-      proxy: proxyInfo?.proxy || null,
-      level: proxyInfo?.level || "direct",
-      levelId: proxyInfo?.levelId || null,
+    await logProxyJournal({
+      result,
+      proxyInfo,
+      proxyLatency,
       provider,
-      targetUrl: `${provider}/${model}`,
-      clientIp,
-      egressIp,
-      latencyMs: proxyLatency,
-      error: result.success ? null : result.error || null,
-      connectionId: credentials.connectionId,
-      comboId: comboName || null,
-      account: credentials.connectionId?.slice(0, 8) || null,
-      rotationAccount: rotationAccount || null,
-      correlationId: correlationId || null,
-      tlsFingerprint: tlsFingerprintUsed,
-      upstreamStatus: proxyInfo?.upstreamStatus ?? null,
+      model,
+      credentials,
+      comboName,
+      clientRawRequest,
+      tlsFingerprintUsed,
+      rotationAccount,
+      correlationId,
     });
   } catch {}