-
Notifications
You must be signed in to change notification settings - Fork 192
fix(cache): hotpath db queries caching #2662
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,204 @@ | ||
| import { serve } from "@hono/node-server"; | ||
| import "dotenv/config"; | ||
| import { Hono } from "hono"; | ||
| import { | ||
| afterAll, | ||
| beforeAll, | ||
| beforeEach, | ||
| describe, | ||
| expect, | ||
| test, | ||
| vi, | ||
| } from "vitest"; | ||
|
|
||
| import { app } from "@/app.js"; | ||
| import { clearCache } from "@/test-utils/test-helpers.js"; | ||
|
|
||
| import { db, pool, tables } from "@llmgateway/db"; | ||
|
|
||
| // A trivial upstream that always returns a valid completion, so the gateway | ||
| // path runs end-to-end without touching a real provider. | ||
| const mockServer = new Hono(); | ||
| let server: ReturnType<typeof serve> | null = null; | ||
| const MOCK_PORT = 3099; | ||
|
|
||
| mockServer.post("/v1/chat/completions", async (c) => { | ||
| return c.json({ | ||
| id: "chatcmpl-mock", | ||
| object: "chat.completion", | ||
| created: Math.floor(Date.now() / 1000), | ||
| model: "mock-model", | ||
| choices: [ | ||
| { | ||
| index: 0, | ||
| message: { role: "assistant", content: "Hello" }, | ||
| finish_reason: "stop", | ||
| }, | ||
| ], | ||
| usage: { prompt_tokens: 10, completion_tokens: 20, total_tokens: 30 }, | ||
| }); | ||
| }); | ||
|
|
||
| /** | ||
| * Tables whose reads MUST be served from the Redis/Drizzle cache once warm. | ||
| * A SELECT against any of these on a repeated chat request means the per-request | ||
| * metadata lookup regressed to hitting Postgres (unstable cache key or use of | ||
| * the uncached client) — the exact class of bug this test guards against. | ||
| */ | ||
| const CACHED_READ_TABLES = [ | ||
| "api_key", | ||
| "api_key_iam_rule", | ||
| "project", | ||
| "organization", | ||
| "provider_key", | ||
| "rate_limit", | ||
| "discount", | ||
| "model_provider_mapping_history", | ||
| "user_organization", | ||
| "wallet", | ||
| "end_user_session", | ||
| "end_customer", | ||
| "routing_config", | ||
| ]; | ||
|
|
||
| function statementText(args: unknown[]): string { | ||
| const first = args[0]; | ||
| if (typeof first === "string") { | ||
| return first; | ||
| } | ||
| if (first && typeof first === "object") { | ||
| const o = first as { text?: unknown; sql?: unknown }; | ||
| if (typeof o.text === "string") { | ||
| return o.text; | ||
| } | ||
| if (typeof o.sql === "string") { | ||
| return o.sql; | ||
| } | ||
| } | ||
| return ""; | ||
| } | ||
|
|
||
| function cachedTableReads(statements: string[]): string[] { | ||
| return statements.filter((sql) => { | ||
| const lower = sql.toLowerCase(); | ||
| if (!lower.includes("select")) { | ||
| return false; | ||
| } | ||
| return CACHED_READ_TABLES.some((t) => lower.includes(`from "${t}"`)); | ||
| }); | ||
| } | ||
|
|
||
| async function sendChat(prompt: string) { | ||
| return await app.request("/v1/chat/completions", { | ||
| method: "POST", | ||
| headers: { | ||
| "Content-Type": "application/json", | ||
| Authorization: "Bearer cache-token", | ||
| }, | ||
| body: JSON.stringify({ | ||
| model: "llmgateway/custom", | ||
| messages: [{ role: "user", content: prompt }], | ||
| }), | ||
| }); | ||
| } | ||
|
|
||
| describe("Chat completions caching: repeated requests do not re-read Postgres", () => { | ||
| beforeAll(async () => { | ||
| server = serve({ fetch: mockServer.fetch, port: MOCK_PORT }); | ||
| }); | ||
|
|
||
| afterAll(() => { | ||
| if (server) { | ||
| server.close(); | ||
| } | ||
| }); | ||
|
|
||
| beforeEach(async () => { | ||
| await clearCache(); | ||
|
|
||
| await Promise.all([ | ||
| db.delete(tables.log), | ||
| db.delete(tables.apiKey), | ||
| db.delete(tables.providerKey), | ||
| ]); | ||
| await Promise.all([ | ||
| db.delete(tables.userOrganization), | ||
| db.delete(tables.project), | ||
| ]); | ||
| await Promise.all([db.delete(tables.organization), db.delete(tables.user)]); | ||
|
|
||
| await db.insert(tables.user).values({ | ||
| id: "cache-user", | ||
| name: "user", | ||
| email: "cache@test.com", | ||
| }); | ||
| // Keep credits positive: findOrganizationById refetches UNCACHED when an | ||
| // org is out of credits, which would legitimately hit Postgres every time. | ||
| await db.insert(tables.organization).values({ | ||
| id: "cache-org", | ||
| name: "Cache Org", | ||
| billingEmail: "cache@test.com", | ||
| plan: "pro", | ||
| credits: "100.00", | ||
| }); | ||
| await db.insert(tables.userOrganization).values({ | ||
| id: "cache-user-org", | ||
| userId: "cache-user", | ||
| organizationId: "cache-org", | ||
| }); | ||
| await db.insert(tables.project).values({ | ||
| id: "cache-project", | ||
| name: "Cache Project", | ||
| organizationId: "cache-org", | ||
| mode: "api-keys", | ||
| }); | ||
| await db.insert(tables.apiKey).values({ | ||
| id: "cache-key", | ||
| token: "cache-token", | ||
| projectId: "cache-project", | ||
| description: "Cache Key", | ||
| createdBy: "cache-user", | ||
| }); | ||
| await db.insert(tables.providerKey).values({ | ||
| id: "cache-provider-key", | ||
| token: "sk-mock", | ||
| provider: "llmgateway", | ||
| organizationId: "cache-org", | ||
| baseUrl: `http://localhost:${MOCK_PORT}`, | ||
| }); | ||
| }); | ||
|
|
||
| test("a warm chat request issues zero Postgres reads on cached metadata tables", async () => { | ||
| // Warm the cache: api key, project, org, provider key, rate limits, | ||
| // discounts and routing metrics are all looked up and cached here. | ||
| expect((await sendChat("warm up one")).status).toBe(200); | ||
| expect((await sendChat("warm up two")).status).toBe(200); | ||
|
|
||
| // Record every statement the shared pool executes during a fresh request. | ||
| // Both the cached (cdb) and uncached (db) clients use this same pool, so | ||
| // this captures any query that actually reaches Postgres. | ||
| const statements: string[] = []; | ||
| const original = pool.query.bind(pool); | ||
| const spy = vi.spyOn(pool, "query").mockImplementation((( | ||
| ...args: Parameters<typeof original> | ||
| ) => { | ||
| statements.push(statementText(args)); | ||
| return original(...args); | ||
| }) as typeof pool.query); | ||
|
|
||
| try { | ||
| // A unique prompt guarantees the response cache cannot short-circuit | ||
| // the request, so the full auth/routing/pricing path runs. | ||
| const res = await sendChat("measured request with a unique prompt"); | ||
| expect(res.status).toBe(200); | ||
| } finally { | ||
| spy.mockRestore(); | ||
| } | ||
|
|
||
| const leaked = cachedTableReads(statements); | ||
| expect( | ||
| leaked, | ||
| `Expected zero cached-table SELECTs on a warm request, but these hit Postgres:\n${leaked.join("\n")}`, | ||
| ).toEqual([]); | ||
| }); | ||
| }); | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,4 @@ | ||
| import { and, eq, gte, isNull, or } from "drizzle-orm"; | ||
| import { and, eq, isNull, or } from "drizzle-orm"; | ||
|
|
||
| import { logger } from "@llmgateway/logger"; | ||
|
|
||
|
|
@@ -51,25 +51,25 @@ export async function getEffectiveDiscount( | |
| model: string, | ||
| ): Promise<EffectiveDiscount> { | ||
| try { | ||
| const now = new Date(); | ||
|
|
||
| const notExpiredCondition = or( | ||
| isNull(discountTable.expiresAt), | ||
| gte(discountTable.expiresAt, now), | ||
| ); | ||
|
|
||
| const discounts = await cdb | ||
| // The expiry filter is applied in JS below, NOT in SQL: a `now` Date in the | ||
| // WHERE clause becomes a query parameter, and the cached client keys its | ||
| // cache on hashQuery(sql, params). A per-request millisecond `now` would | ||
| // make that key unique every call, so the cache would never hit and this | ||
| // (hot, per-provider-candidate) lookup would query Postgres on every | ||
| // request. Keeping the SQL time-independent lets the cache key stay stable | ||
| // while expiry is still evaluated fresh on each call. | ||
| const rows = await cdb | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
With this lookup now using a stable Useful? React with 👍 / 👎. |
||
| .select({ | ||
| id: discountTable.id, | ||
| organizationId: discountTable.organizationId, | ||
| provider: discountTable.provider, | ||
| model: discountTable.model, | ||
| discountPercent: discountTable.discountPercent, | ||
| expiresAt: discountTable.expiresAt, | ||
| }) | ||
| .from(discountTable) | ||
| .where( | ||
| and( | ||
| notExpiredCondition, | ||
| or( | ||
| isNull(discountTable.organizationId), | ||
| organizationId | ||
|
|
@@ -84,6 +84,15 @@ export async function getEffectiveDiscount( | |
| ), | ||
| ); | ||
|
|
||
| const now = Date.now(); | ||
| const discounts = rows.filter( | ||
| // expiresAt is a Date on both a fresh query and a Drizzle cache hit (the | ||
| // cache stores the raw pg result and re-applies the timestamp parser on | ||
| // restore). Wrap in new Date() defensively so the compare is robust even | ||
| // if a serialized value ever reaches here. | ||
| (d) => d.expiresAt === null || new Date(d.expiresAt).getTime() >= now, | ||
| ); | ||
|
|
||
| const modelMatches = (discountModel: string | null): boolean => | ||
| discountModel !== null && discountModel === model; | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,6 @@ | ||
| import { and, eq, isNull, or } from "drizzle-orm"; | ||
|
|
||
| import { db } from "./db.js"; | ||
| import { cdb } from "./cdb.js"; | ||
| import { rateLimit as rateLimitTable } from "./schema.js"; | ||
|
|
||
| export type RateLimitSource = | ||
|
|
@@ -149,7 +149,10 @@ function pickRateLimitByPrecedence( | |
|
|
||
| /** | ||
| * Get the effective rate limits for a given organization, provider, and model. | ||
| * Uses the uncached database client so admin changes take effect immediately. | ||
| * Uses the cached database client (cdb) so this hot, per-request lookup is served | ||
| * from the Drizzle cache instead of hitting Postgres on every gateway request. | ||
| * The WHERE clause is time-independent, so the cache key stays stable. Admin | ||
| * changes propagate within the cache TTL (default 60s). | ||
| * | ||
| * Rate limits are always keyed by the root model ID — provider-specific model | ||
| * names are reserved for upstream requests and are never persisted as a | ||
|
|
@@ -168,7 +171,7 @@ export async function getEffectiveRateLimit( | |
| provider: string, | ||
| model: string, | ||
| ): Promise<EffectiveRateLimit> { | ||
| const rateLimits = await db | ||
| const rateLimits = await cdb | ||
| .select({ | ||
|
Comment on lines
+174
to
175
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
When an admin creates or deletes a rate limit to stop traffic, this cached lookup can keep returning the previous result because the admin handlers mutate Useful? React with 👍 / 👎. |
||
| id: rateLimitTable.id, | ||
| organizationId: rateLimitTable.organizationId, | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.