fix(api): save call logs and add endpoint fallback for local rerank providers (#11081)

Validated on a combined board over tip aa128736: focused tests local-rerank-logging + call-logs-row-filter green (7/7 across runs), typecheck:core clean. One pre-merge fix: widened waitForCallLogSaves 5s→15s — the call-log artifact writer's fsync latency exceeds 5s under a loaded host (pre-existing call-log-save-drain flakes identically on the pure tip), so the budget was load-flaky, not the code. Local rerank now logs call entries on success and error paths like embeddings/cloud rerank, captures full request/response payloads, attaches x-omniroute-* headers, and falls back /v1/rerank→/rerank on 404. Thank you @AndrianBalanescu!
2026-08-22 23:22:09 +03:00 · 2026-08-22 14:54:17 -05:00
parent aa12873686
commit 56540f24c5
3 changed files with 345 additions and 14 deletions
--- a/open-sse/handlers/rerank.ts
+++ b/open-sse/handlers/rerank.ts
@@ -199,6 +199,8 @@ export async function handleRerank({
  return_documents,
  credentials,
  connectionId = null,
+  apiKeyId = null,
+  apiKeyName = null,
 }) {
  const startTime = Date.now();
  if (!model) return errorResponse(400, "model is required");
@@ -267,10 +269,23 @@ export async function handleRerank({

    if (!res.ok) {
      const errData = await res.json().catch(() => ({}));
-      return errorResponse(
-        res.status,
-        errData.message || errData.error?.message || `Provider returned HTTP ${res.status}`
-      );
+      const errorMessage =
+        errData.message || errData.error?.message || `Provider returned HTTP ${res.status}`;
+      saveCallLog({
+        method: "POST",
+        path: "/v1/rerank",
+        status: res.status,
+        model: `${providerId}/${modelId}`,
+        provider: providerId,
+        connectionId: connectionId || undefined,
+        duration: Date.now() - startTime,
+        requestBody,
+        responseBody: errData,
+        error: errorMessage,
+        apiKeyId: apiKeyId || undefined,
+        apiKeyName: apiKeyName || undefined,
+      }).catch(() => {});
+      return errorResponse(res.status, errorMessage);
    }

    const data = await res.json();
@@ -289,10 +304,13 @@ export async function handleRerank({
      status: 200,
      model: `${providerId}/${modelId}`,
      provider: providerId,
+      connectionId: connectionId || undefined,
      duration: Date.now() - startTime,
      tokens: { prompt_tokens: 0, completion_tokens: 0 },
-      responseBody: { results_count: Array.isArray(result?.results) ? result.results.length : 0 },
-      connectionId,
+      requestBody,
+      responseBody: result,
+      apiKeyId: apiKeyId || undefined,
+      apiKeyName: apiKeyName || undefined,
    }).catch(() => {});

    const headers = new Headers({ ...CORS_HEADERS, "Content-Type": "application/json" });
--- a/src/app/api/v1/rerank/route.ts
+++ b/src/app/api/v1/rerank/route.ts
@@ -10,11 +10,15 @@ import { HTTP_STATUS } from "@omniroute/open-sse/config/constants.ts";
 import { enforceApiKeyPolicy } from "@/shared/utils/apiKeyPolicy";
 import { v1RerankSchema } from "@/shared/validation/schemas";
 import { isValidationFailure, validateBody } from "@/shared/validation/helpers";
-import { getCachedProviderNodes } from "@/lib/localDb";
+import { getCachedProviderNodes } from "@/lib/db/readCache";
 import {
  isAllRateLimitedCredentials,
  rateLimitedProviderResponse,
 } from "@/app/api/v1/_shared/rateLimit";
+import { saveCallLog } from "@/lib/usageDb";
+import { attachOmniRouteMetaHeaders } from "@/domain/omnirouteResponseMeta";
+import { generateRequestId } from "@/shared/utils/requestId";
+import { CORS_HEADERS } from "@omniroute/open-sse/utils/cors.ts";

 /**
 * Handle CORS preflight
@@ -121,6 +125,8 @@ async function postHandler(request, context) {
      return_documents: body.return_documents,
      credentials,
      connectionId: (credentials as { connectionId?: string } | null)?.connectionId || null,
+      apiKeyId: policy.apiKeyInfo?.id || null,
+      apiKeyName: policy.apiKeyInfo?.name || null,
    });
    if (response?.ok) {
      await clearRecoveredProviderState(credentials);
@@ -148,8 +154,9 @@ async function postHandler(request, context) {
      }

      const token = credentials?.apiKey || credentials?.accessToken;
+      const startTime = Date.now();
      try {
-        const res = await fetch(localProvider.baseUrl, {
+        let res = await fetch(localProvider.baseUrl, {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
@@ -164,19 +171,110 @@ async function postHandler(request, context) {
          }),
        });

+        // Some local providers (e.g. Infinity, TEI) mount at /rerank rather than /v1/rerank
+        if (res.status === 404 && localProvider.baseUrl.endsWith("/v1/rerank")) {
+          const fallbackUrl = localProvider.baseUrl.replace(/\/v1\/rerank$/, "/rerank");
+          try {
+            const fallbackRes = await fetch(fallbackUrl, {
+              method: "POST",
+              headers: {
+                "Content-Type": "application/json",
+                Authorization: `Bearer ${token}`,
+              },
+              body: JSON.stringify({
+                model: localModel,
+                query: body.query,
+                documents: body.documents,
+                top_n: body.top_n || body.documents.length,
+                return_documents: body.return_documents !== false,
+              }),
+            });
+            if (fallbackRes.ok || fallbackRes.status !== 404) {
+              res = fallbackRes;
+            }
+          } catch {
+            // retain original 404 response if fallback fetch fails
+          }
+        }
+
        if (!res.ok) {
          const errData = await res.json().catch(() => ({}));
-          return errorResponse(
-            res.status,
-            errData.message || errData.detail || `Provider returned HTTP ${res.status}`
-          );
+          const errorMessage =
+            errData.message || errData.detail || `Provider returned HTTP ${res.status}`;
+          saveCallLog({
+            method: "POST",
+            path: "/v1/rerank",
+            status: res.status,
+            model: body.model,
+            provider: prefix,
+            connectionId:
+              (credentials as { connectionId?: string } | null)?.connectionId || undefined,
+            duration: Date.now() - startTime,
+            requestBody: {
+              model: body.model,
+              query: body.query,
+              documents: body.documents,
+              top_n: body.top_n,
+              return_documents: body.return_documents,
+            },
+            responseBody: errData,
+            error: errorMessage,
+            apiKeyId: policy.apiKeyInfo?.id || undefined,
+            apiKeyName: policy.apiKeyInfo?.name || undefined,
+          }).catch(() => {});
+          return errorResponse(res.status, errorMessage);
        }

        const data = await res.json();
-        return Response.json(data, {
-          headers: {},
+        const latencyMs = Date.now() - startTime;
+        saveCallLog({
+          method: "POST",
+          path: "/v1/rerank",
+          status: 200,
+          model: body.model,
+          provider: prefix,
+          connectionId:
+            (credentials as { connectionId?: string } | null)?.connectionId || undefined,
+          duration: latencyMs,
+          tokens: { prompt_tokens: 0, completion_tokens: 0 },
+          requestBody: {
+            model: body.model,
+            query: body.query,
+            documents: body.documents,
+            top_n: body.top_n,
+            return_documents: body.return_documents,
+          },
+          responseBody: data,
+          apiKeyId: policy.apiKeyInfo?.id || undefined,
+          apiKeyName: policy.apiKeyInfo?.name || undefined,
+        }).catch(() => {});
+
+        const headers = new Headers({ ...CORS_HEADERS, "Content-Type": "application/json" });
+        attachOmniRouteMetaHeaders(headers, {
+          provider: prefix,
+          model: localModel,
+          costUsd: 0,
+          latencyMs,
+          requestId: generateRequestId(),
+        });
+        return new Response(JSON.stringify(data), {
+          status: 200,
+          headers,
        });
      } catch (err: any) {
+        saveCallLog({
+          method: "POST",
+          path: "/v1/rerank",
+          status: 500,
+          model: body.model,
+          provider: prefix,
+          connectionId:
+            (credentials as { connectionId?: string } | null)?.connectionId || undefined,
+          duration: Date.now() - startTime,
+          error: err.message,
+          apiKeyId: policy.apiKeyInfo?.id || undefined,
+          apiKeyName: policy.apiKeyInfo?.name || undefined,
+        }).catch(() => {});
        return errorResponse(500, `Rerank request failed: ${err.message}`);
      }
    }
--- a/tests/unit/local-rerank-logging.test.ts
+++ b/tests/unit/local-rerank-logging.test.ts
@@ -0,0 +1,215 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rerank-test-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+
+const core = await import("../../src/lib/db/core.ts");
+const { invalidateDbCache } = await import("../../src/lib/db/readCache.ts");
+const { createProviderNode, createProviderConnection } =
+  await import("../../src/lib/db/providers.ts");
+const { getCallLogs, getCallLogById, waitForCallLogSaves } =
+  await import("../../src/lib/usage/callLogs.ts");
+const { POST } = await import("../../src/app/api/v1/rerank/route.ts");
+
+interface RerankSuccessResponse {
+  results: Array<{ index: number; relevance_score: number }>;
+}
+
+interface CallLogRow {
+  id: string;
+  model: string;
+  provider: string;
+  status: number;
+  error?: string;
+  connectionId?: string;
+}
+
+test.describe("Local rerank provider logging and fallback", () => {
+  const originalFetch = globalThis.fetch;
+
+  test.after(() => {
+    globalThis.fetch = originalFetch;
+    core.resetDbInstance();
+    try {
+      fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+    } catch {
+      // ignore
+    }
+  });
+
+  test("successfully logs local rerank calls and attaches metadata headers", async () => {
+    const now = new Date().toISOString();
+    await createProviderNode({
+      id: "vram",
+      name: "vram",
+      type: "openai",
+      prefix: "vram",
+      baseUrl: "http://127.0.0.1:8000/v1",
+      createdAt: now,
+      updatedAt: now,
+    });
+
+    await createProviderConnection({
+      id: "conn-vram-1",
+      provider: "vram",
+      authType: "apikey",
+      name: "vram-local",
+      apiKey: "test-token",
+      createdAt: now,
+      updatedAt: now,
+    });
+
+    invalidateDbCache("nodes");
+    invalidateDbCache("connections");
+
+    globalThis.fetch = async (url: string | URL | Request, init?: RequestInit) => {
+      assert.equal(String(url), "http://127.0.0.1:8000/v1/rerank");
+      const parsedBody = JSON.parse(String(init?.body || "{}"));
+      assert.equal(parsedBody.model, "BAAI/bge-reranker-v2-m3");
+      assert.equal(parsedBody.query, "test query");
+      assert.deepEqual(parsedBody.documents, ["doc1", "doc2"]);
+
+      return new Response(
+        JSON.stringify({
+          results: [
+            { index: 0, relevance_score: 0.95 },
+            { index: 1, relevance_score: 0.2 },
+          ],
+        }),
+        { status: 200, headers: { "Content-Type": "application/json" } }
+      );
+    };
+
+    const req = new Request("http://localhost:20128/api/v1/rerank", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        model: "vram/BAAI/bge-reranker-v2-m3",
+        query: "test query",
+        documents: ["doc1", "doc2"],
+      }),
+    });
+
+    const res = await POST(req, {} as Record<string, unknown>);
+    assert.equal(res.status, 200);
+    assert.equal(res.headers.get("x-omniroute-provider"), "vram");
+    assert.equal(res.headers.get("x-omniroute-model"), "BAAI/bge-reranker-v2-m3");
+
+    const json = (await res.json()) as RerankSuccessResponse;
+    assert.equal(json.results.length, 2);
+
+    await waitForCallLogSaves(15000);
+
+    const logs = (await getCallLogs({ limit: 10 })) as unknown as CallLogRow[];
+    const logEntry = logs.find((l) => l.model === "vram/BAAI/bge-reranker-v2-m3");
+    assert.ok(logEntry, "Expected call log entry for local rerank");
+    assert.equal(logEntry.provider, "vram");
+    assert.equal(logEntry.status, 200);
+
+    const detail = await getCallLogById(logEntry.id);
+    assert.deepEqual(detail?.requestBody, {
+      model: "vram/BAAI/bge-reranker-v2-m3",
+      query: "test query",
+      documents: ["doc1", "doc2"],
+    });
+    assert.deepEqual(detail?.responseBody, {
+      results: [
+        { index: 0, relevance_score: 0.95 },
+        { index: 1, relevance_score: 0.2 },
+      ],
+    });
+  });
+
+  test("falls back from /v1/rerank to /rerank when local provider returns 404", async () => {
+    const now = new Date().toISOString();
+    await createProviderNode({
+      id: "infinity",
+      name: "infinity",
+      type: "openai",
+      prefix: "infinity",
+      baseUrl: "http://127.0.0.1:7997",
+      createdAt: now,
+      updatedAt: now,
+    });
+
+    await createProviderConnection({
+      id: "conn-infinity-1",
+      provider: "infinity",
+      authType: "apikey",
+      name: "infinity-local",
+      apiKey: "test-token",
+      createdAt: now,
+      updatedAt: now,
+    });
+
+    invalidateDbCache("nodes");
+    invalidateDbCache("connections");
+
+    const urlsAttempted: string[] = [];
+    globalThis.fetch = async (url: string | URL | Request) => {
+      urlsAttempted.push(String(url));
+      if (String(url).endsWith("/v1/rerank")) {
+        return new Response("Not Found", { status: 404 });
+      }
+      return new Response(
+        JSON.stringify({
+          results: [{ index: 0, relevance_score: 0.99 }],
+        }),
+        { status: 200, headers: { "Content-Type": "application/json" } }
+      );
+    };
+
+    const req = new Request("http://localhost:20128/api/v1/rerank", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        model: "infinity/bge-reranker-large",
+        query: "search",
+        documents: ["doc1"],
+      }),
+    });
+
+    const res = await POST(req, {} as Record<string, unknown>);
+    assert.equal(res.status, 200);
+    assert.deepEqual(urlsAttempted, [
+      "http://127.0.0.1:7997/v1/rerank",
+      "http://127.0.0.1:7997/rerank",
+    ]);
+  });
+
+  test("records error call log when local provider returns 500", async () => {
+    globalThis.fetch = async () => {
+      return new Response(JSON.stringify({ detail: "Local backend failure" }), {
+        status: 500,
+        headers: { "Content-Type": "application/json" },
+      });
+    };
+
+    const req = new Request("http://localhost:20128/api/v1/rerank", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        model: "vram/BAAI/bge-reranker-v2-m3",
+        query: "test query",
+        documents: ["doc1"],
+      }),
+    });
+
+    const res = await POST(req, {} as Record<string, unknown>);
+    assert.equal(res.status, 500);
+
+    await waitForCallLogSaves(15000);
+
+    const logs = (await getCallLogs({ limit: 10 })) as unknown as CallLogRow[];
+    const logEntry = logs.find(
+      (l) => l.model === "vram/BAAI/bge-reranker-v2-m3" && l.status === 500
+    );
+    assert.ok(logEntry, "Expected 500 call log entry for local rerank failure");
+    assert.equal(logEntry.provider, "vram");
+    assert.equal(logEntry.error, "Local backend failure");
+  });
+});