From 4bdbaa006249fe3877606618c86ed42d6076f180 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Fri, 26 Jun 2026 22:22:00 -0300 Subject: [PATCH] fix(usage): dedupe request-usage logging and debounce stats (#4940) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.38 (rebased on tip; DB-handle hang was stale-base artifact β€” resetDbInstance already closes the handle, test green 5/5; file-size drift consolidated at release; CHANGELOG re-injected) --- CHANGELOG.md | 1 + src/lib/usage/usageEvents.ts | 51 ++++++++ src/lib/usage/usageHistory.ts | 116 ++++++++++++----- tests/unit/usage/usageHistoryDedup.test.ts | 137 +++++++++++++++++++++ 4 files changed, 273 insertions(+), 32 deletions(-) create mode 100644 tests/unit/usage/usageHistoryDedup.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index ff56346aad..c1c3648c8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,7 @@ _In development β€” bullets added per PR; finalized at release._ ### πŸ”§ Bug Fixes +- **fix(usage): dedupe request-usage logging and debounce stats events** β€” `saveRequestUsage` now guards against duplicate inserts (natural key: timestamp + provider + model + connection + api-key + token counts), back-fills a missing `endpoint`, and only emits `usageRecorded` when a row was actually inserted; stats `update`/`pending` event bursts are collapsed into a single debounced notification to reduce churn. ([#4940](https://github.com/diegosouzapw/OmniRoute/pull/4940) β€” thanks @nguyenxvotanminh3) - **fix(sse): convert the native Gemini request body to OpenAI format in the Antigravity MITM handler** β€” `contents` / `systemInstruction` / `generationConfig` / `thinkingConfig` are now translated to OpenAI chat-completions format before forwarding to `/v1/chat/completions`, so thinking-capable models (e.g. `ag/claude-opus-4-6-thinking`) no longer fail with provider-side 400 "invalid argument" errors. ([#4845](https://github.com/diegosouzapw/OmniRoute/pull/4845) β€” thanks @anuragg-saxenaa) - **fix(db): translate the two pt-BR SQLite driver-fallback log lines to English** β€” `[DB] PrΓ©-inicializando sql.js WASM…` and `[DB] Drivers sΓ­ncronos indisponΓ­veis…` were the only non-English server log strings, mixing languages in the logs. Now `[DB] Pre-initializing sql.js WASM (synchronous drivers unavailable)…` / `[DB] Synchronous drivers unavailable β€” falling back to sql.js (WASM)`, guarded by a test that scans the driver path for accented log strings. ([#5103](https://github.com/diegosouzapw/OmniRoute/issues/5103)) - **fix(diagnostics): non-streaming Claude responses no longer false-502 as `empty_choices`** β€” the v3.8.37 malformed-200 detector (#4942) only understood OpenAI `choices` and Responses-API `output` shapes, so a `/v1/messages` response that stays in Claude shape (`{type:"message", content:[…]}`) fell through to `empty_choices` β†’ 502 (cascading to "All models failed" in a combo). Most visibly, an extended-thinking turn whose buffered body is a single **empty thinking block with a valid `signature`** (Claude Code's non-streaming Bash classifier) 502'd on every call. `detectMalformedNonStream` now understands the Claude shape: text/tool_use blocks and thinking blocks carrying a signature count as valid output, while a genuinely empty `content:[]` is still flagged. ([#5108](https://github.com/diegosouzapw/OmniRoute/issues/5108), thanks @insoln) diff --git a/src/lib/usage/usageEvents.ts b/src/lib/usage/usageEvents.ts index dbd2a507eb..cdef3438d2 100644 --- a/src/lib/usage/usageEvents.ts +++ b/src/lib/usage/usageEvents.ts @@ -38,3 +38,54 @@ export function emitUsageRecorded( } } } + +// ── Stats-event debounce ───────────────────────────────────────────────────── +// +// Rapid back-to-back inserts (e.g. combo routing that fans out to multiple +// models simultaneously) can fire dozens of "update" events per second. +// scheduleStatsEvent collapses bursts: the first call within a window sets a +// timer; subsequent calls within the same window are no-ops. The timer fires +// once at the end of the window. + +type StatsEventKey = "update" | "pending"; + +const statsEmitTimers: Record | null> = { + update: null, + pending: null, +}; + +const statsListeners: Record void>> = { + update: new Set(), + pending: new Set(), +}; + +/** Register a debounced stats listener. Returns an unsubscribe fn. */ +export function onStatsEvent(event: StatsEventKey, listener: () => void): () => void { + statsListeners[event].add(listener); + return () => { + statsListeners[event].delete(listener); + }; +} + +/** + * Schedule a stats event emission after `delayMs`, collapsing rapid bursts + * into a single notification. Safe to call from hot paths β€” subsequent calls + * within the same window are no-ops. + */ +export function scheduleStatsEvent(event: StatsEventKey, delayMs = 200): void { + if (statsEmitTimers[event] != null) return; + statsEmitTimers[event] = setTimeout(() => { + statsEmitTimers[event] = null; + for (const listener of statsListeners[event]) { + try { + listener(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + console.warn(`[usageEvents] stats listener (${event}) failed: ${message}`); + } + } + }, delayMs); + // Allow Node.js to exit naturally even if the timer is still pending + // (avoids keeping the event loop alive for a stray stats notification). + (statsEmitTimers[event] as any)?.unref?.(); +} diff --git a/src/lib/usage/usageHistory.ts b/src/lib/usage/usageHistory.ts index 0e5ed1fe82..12edcc92be 100644 --- a/src/lib/usage/usageHistory.ts +++ b/src/lib/usage/usageHistory.ts @@ -637,42 +637,94 @@ export async function saveRequestUsage(entry: any) { const timestamp = entry.timestamp || new Date().toISOString(); const serviceTier = normalizeServiceTier(entry.serviceTier ?? entry.service_tier); - db.prepare( + const tokensInput = getLoggedInputTokens(entry.tokens); + const tokensOutput = getLoggedOutputTokens(entry.tokens); + + // Dedup guard: skip INSERT when an identical row already exists in the same + // second. This prevents double-counting when onRequestSuccess fires more + // than once (e.g. combo routing calling the callback from both the + // streaming and non-streaming paths for the same underlying request). + // Keyed on the natural identity of a request: timestamp + provider + model + // + connectionId + apiKeyId + token counts. If only the endpoint is missing + // on the existing row, fill it in rather than inserting a duplicate. + let inserted = false; + + db.transaction(() => { + const existing = db + .prepare( + `SELECT id, endpoint FROM usage_history + WHERE timestamp = ? + AND COALESCE(provider, '') = COALESCE(?, '') + AND COALESCE(model, '') = COALESCE(?, '') + AND COALESCE(connection_id, '') = COALESCE(?, '') + AND COALESCE(api_key_id, '') = COALESCE(?, '') + AND tokens_input = ? + AND tokens_output = ? + ORDER BY id DESC LIMIT 1` + ) + .get( + timestamp, + entry.provider || null, + entry.model || null, + entry.connectionId || null, + entry.apiKeyId || null, + tokensInput, + tokensOutput + ) as { id: number; endpoint: string | null } | undefined; + + if (existing) { + // Back-fill endpoint if the original row missed it. + if (!existing.endpoint && entry.endpoint) { + db.prepare(`UPDATE usage_history SET endpoint = ? WHERE id = ?`).run( + entry.endpoint, + existing.id + ); + } + return; // duplicate β€” do not insert + } + + db.prepare( + ` + INSERT INTO usage_history (provider, model, connection_id, api_key_id, api_key_name, + tokens_input, tokens_output, tokens_cache_read, tokens_cache_creation, tokens_reasoning, + service_tier, status, success, latency_ms, ttft_ms, error_code, combo_strategy, endpoint, timestamp) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ` - INSERT INTO usage_history (provider, model, connection_id, api_key_id, api_key_name, - tokens_input, tokens_output, tokens_cache_read, tokens_cache_creation, tokens_reasoning, - service_tier, status, success, latency_ms, ttft_ms, error_code, combo_strategy, endpoint, timestamp) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ` - ).run( - entry.provider || null, - entry.model || null, - entry.connectionId || null, - entry.apiKeyId || null, - entry.apiKeyName || null, - getLoggedInputTokens(entry.tokens), - getLoggedOutputTokens(entry.tokens), - getPromptCacheReadTokens(entry.tokens), - getPromptCacheCreationTokens(entry.tokens), - getReasoningTokens(entry.tokens), - serviceTier, - entry.status || null, - entry.success === false ? 0 : 1, - Number.isFinite(Number(entry.latencyMs)) ? Number(entry.latencyMs) : 0, - Number.isFinite(Number(entry.timeToFirstTokenMs)) - ? Number(entry.timeToFirstTokenMs) - : Number.isFinite(Number(entry.latencyMs)) - ? Number(entry.latencyMs) - : 0, - entry.errorCode || null, - entry.comboStrategy || entry.combo_strategy || null, - entry.endpoint || null, - timestamp - ); + ).run( + entry.provider || null, + entry.model || null, + entry.connectionId || null, + entry.apiKeyId || null, + entry.apiKeyName || null, + tokensInput, + tokensOutput, + getPromptCacheReadTokens(entry.tokens), + getPromptCacheCreationTokens(entry.tokens), + getReasoningTokens(entry.tokens), + serviceTier, + entry.status || null, + entry.success === false ? 0 : 1, + Number.isFinite(Number(entry.latencyMs)) ? Number(entry.latencyMs) : 0, + Number.isFinite(Number(entry.timeToFirstTokenMs)) + ? Number(entry.timeToFirstTokenMs) + : Number.isFinite(Number(entry.latencyMs)) + ? Number(entry.latencyMs) + : 0, + entry.errorCode || null, + entry.comboStrategy || entry.combo_strategy || null, + entry.endpoint || null, + timestamp + ); + + inserted = true; + })(); // Decoupled via the event bus so usageHistory never imports providerLimits // (which would pull the executors/translator graph into the type-check surface). - emitUsageRecorded(entry.provider, entry.connectionId); + // Only emit when a row was actually inserted β€” not on dedup no-ops. + if (inserted) { + emitUsageRecorded(entry.provider, entry.connectionId); + } } catch (error) { console.error("Failed to save usage stats:", error); } diff --git a/tests/unit/usage/usageHistoryDedup.test.ts b/tests/unit/usage/usageHistoryDedup.test.ts new file mode 100644 index 0000000000..7cbc43af45 --- /dev/null +++ b/tests/unit/usage/usageHistoryDedup.test.ts @@ -0,0 +1,137 @@ +/** + * TDD regression for port of https://github.com/decolua/9router/pull/2044: + * "Fix usage logging dedupe and reduce stats churn" + * + * Asserts: + * 1. Inserting the same request usage entry twice results in exactly ONE row + * in usage_history (dedup guard). + * 2. emitUsageRecorded fires only when a row is actually inserted β€” not on + * a duplicate. + */ + +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"; + +// Isolate the DB from other tests and from the real data dir. +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-usage-dedup-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +// Dynamic imports so DATA_DIR is set before any module initialises the DB. +const { resetDbInstance, getDbInstance } = await import("../../../src/lib/db/core.ts"); +const { onUsageRecorded } = await import("../../../src/lib/usage/usageEvents.ts"); +const { saveRequestUsage } = await import("../../../src/lib/usage/usageHistory.ts"); + +// Cleanup: close DB handle and temp directory so the test runner doesn't hang. +test.after(() => { + resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +// ── helpers ────────────────────────────────────────────────────────────────── + +function makeEntry(overrides: Record = {}) { + return { + provider: "test-provider", + model: "test-model", + connectionId: "conn-abc123", + apiKeyId: null, + apiKeyName: null, + tokens: { input_tokens: 10, output_tokens: 20 }, + status: "success", + success: true, + latencyMs: 100, + timeToFirstTokenMs: 50, + errorCode: null, + comboStrategy: null, + endpoint: "/v1/chat/completions", + timestamp: new Date().toISOString(), + ...overrides, + }; +} + +function countRows(db: ReturnType): number { + const row = db.prepare("SELECT COUNT(*) AS cnt FROM usage_history").get() as { cnt: number }; + return row.cnt; +} + +// ── tests ───────────────────────────────────────────────────────────────────── + +test("saveRequestUsage: first insert creates exactly one row", async () => { + const db = getDbInstance(); + const before = countRows(db); + const entry = makeEntry({ timestamp: new Date().toISOString() }); + + await saveRequestUsage(entry); + + assert.equal(countRows(db), before + 1, "Expected exactly one new row after first insert"); +}); + +test("saveRequestUsage: duplicate entry (same key fields) inserts only ONE row", async () => { + const db = getDbInstance(); + const ts = new Date().toISOString(); + const entry = makeEntry({ timestamp: ts }); + + await saveRequestUsage(entry); + const afterFirst = countRows(db); + + // Insert identical entry a second time β€” should be a no-op. + await saveRequestUsage(entry); + const afterSecond = countRows(db); + + assert.equal( + afterSecond, + afterFirst, + "Duplicate insert must not create a second row (dedup guard)" + ); +}); + +test("saveRequestUsage: emitUsageRecorded fires on real insert but NOT on duplicate", async () => { + const ts = new Date().toISOString(); + const entry = makeEntry({ timestamp: ts }); + + let fireCount = 0; + const unsub = onUsageRecorded(() => { + fireCount++; + }); + + try { + await saveRequestUsage(entry); // real insert β†’ should fire + await saveRequestUsage(entry); // duplicate β†’ must NOT fire + + assert.equal(fireCount, 1, "emitUsageRecorded should fire exactly once (not on duplicate)"); + } finally { + unsub(); + } +}); + +test("saveRequestUsage: two entries with different timestamps are both inserted", async () => { + const db = getDbInstance(); + const before = countRows(db); + + await saveRequestUsage(makeEntry({ timestamp: new Date(Date.now() - 5000).toISOString() })); + await saveRequestUsage(makeEntry({ timestamp: new Date(Date.now() - 4000).toISOString() })); + + assert.equal( + countRows(db), + before + 2, + "Two distinct entries (different timestamps) should both be inserted" + ); +}); + +test("saveRequestUsage: two entries with different providers are both inserted", async () => { + const db = getDbInstance(); + const before = countRows(db); + const ts = new Date().toISOString(); + + await saveRequestUsage(makeEntry({ timestamp: ts, provider: "provider-A" })); + await saveRequestUsage(makeEntry({ timestamp: ts, provider: "provider-B" })); + + assert.equal( + countRows(db), + before + 2, + "Two entries with different providers should both be inserted" + ); +});