From 1be2c27b2738c4a28a53c2da6c1723b52dbbb309 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 21 Oct 2026 00:06:21 +0200 Subject: [PATCH 1/2] fix(ai): serialize Codex refreshes so a lost race never parks a live sign-in Two callers refreshing one Codex token raced: the loser heard refresh_token_reused, read it as a dead grant and parked the connection before the winner wrote its rotated tokens, leaving fresh tokens beside an expired status. Refreshes now share one flight per process and queue on the per-credential advisory lock across processes, re-read the row inside it, use a token a peer already rotated, write the rotated tokens compare-and-set on the spent one together with connected, and re-read after a refusal before parking. refresh_token_reused no longer counts as a dead grant, and neither does a bare 403 without an OAuth error code: an edge challenge answers a valid grant so, and it keeps the ledger's timed cooldown instead of parking until the next sign-in. --- src/lib/ai/__tests__/codex-oauth.test.ts | 17 + src/lib/ai/codex-oauth.ts | 17 +- src/lib/ai/provider.ts | 453 +++++++++++------- src/lib/integrations/oauth-refresh.ts | 3 +- .../integration/codex-credential-park.test.ts | 58 +++ 5 files changed, 362 insertions(+), 186 deletions(-) diff --git a/src/lib/ai/__tests__/codex-oauth.test.ts b/src/lib/ai/__tests__/codex-oauth.test.ts index 1092e26af..dbbc695cf 100644 --- a/src/lib/ai/__tests__/codex-oauth.test.ts +++ b/src/lib/ai/__tests__/codex-oauth.test.ts @@ -225,6 +225,23 @@ describe("codex-oauth", () => { expect(err.oauthError).toBeNull(); }); + it("does not read a bare 403 as a dead sign-in", async () => { + // An edge challenge in front of the issuer answers a valid grant so. + failWith(403, "challenge"); + const err = await refreshError(); + expect(err.credentialExpired).toBe(false); + expect(err.httpStatus).toBe(403); + }); + + it("does not read refresh_token_reused as a dead sign-in", async () => { + // The loser of two refreshes of one token hears this; the winner's + // rotated token is alive. + failWith(400, JSON.stringify({ error: "refresh_token_reused" })); + const err = await refreshError(); + expect(err.credentialExpired).toBe(false); + expect(err.oauthError).toBe("refresh_token_reused"); + }); + it("never quotes the response body in the message", async () => { failWith( 401, diff --git a/src/lib/ai/codex-oauth.ts b/src/lib/ai/codex-oauth.ts index 6d389160c..bcfd2c630 100644 --- a/src/lib/ai/codex-oauth.ts +++ b/src/lib/ai/codex-oauth.ts @@ -296,14 +296,18 @@ export async function refreshDeviceTokens( /** * OAuth error codes on the refresh grant that mean the stored refresh token - * is dead for good: revoked, expired, already rotated, or issued to another - * client. Retrying cannot help; only signing in again can. + * is dead for good: revoked, expired, or issued to another client. Retrying + * cannot help; only signing in again can. + * + * `refresh_token_reused` is deliberately not among them. It is what the loser + * of two refreshes of one token hears while the winner's rotated token is + * perfectly alive, so it says nothing about the sign-in: the refresh re-reads + * the row and carries on with the newer token (`refreshCodexCredential`). */ const DEAD_REFRESH_TOKEN_CODES: ReadonlySet = new Set([ "invalid_grant", "invalid_refresh_token", "refresh_token_expired", - "refresh_token_reused", "refresh_token_invalidated", "invalid_token", ]); @@ -333,9 +337,12 @@ export class CodexRefreshError extends Error { readonly oauthError: string | null; constructor(status: number, oauthError: string | null) { + // A bare 403 is not proof: an edge challenge in front of the issuer + // answers 403 to a perfectly valid grant. It keeps its own status and + // with it the ledger's timed auth cooldown; only a 401 or the issuer's + // own dead-grant code parks the sign-in until the person reconnects. const credentialExpired = status === 401 || - status === 403 || (oauthError !== null && DEAD_REFRESH_TOKEN_CODES.has(oauthError)); super( credentialExpired @@ -345,7 +352,7 @@ export class CodexRefreshError extends Error { this.name = "CodexRefreshError"; this.credentialExpired = credentialExpired; this.oauthError = oauthError; - this.httpStatus = credentialExpired && status !== 403 ? 401 : status; + this.httpStatus = credentialExpired ? 401 : status; } } diff --git a/src/lib/ai/provider.ts b/src/lib/ai/provider.ts index e05758bed..2a5e24b1e 100644 --- a/src/lib/ai/provider.ts +++ b/src/lib/ai/provider.ts @@ -14,6 +14,7 @@ import { decryptCodexCreds, encryptAdminCodexCreds, decryptAdminCodexCreds, + type CodexCreds, } from "./codex-oauth"; import { isPublicUrl } from "@/lib/validations/notifications"; import { isLocalAiHostAllowed } from "./local-host-allowlist"; @@ -31,6 +32,11 @@ import { } from "@/lib/sharing/provider-work-authority"; import { memoizePerRequest } from "@/lib/request-cache"; import { emitSignal } from "@/lib/logging/signal"; +import { + acquireProviderTokenRefreshLock, + PROVIDER_REFRESH_TRANSACTION_OPTIONS, +} from "@/lib/integrations/oauth-refresh"; +import type { Prisma } from "@/generated/prisma/client"; import type { AiModality } from "@/lib/ai/capabilities/types"; import type { ProviderEntryPresence, @@ -62,72 +68,253 @@ function oauthErrorOf(error: unknown): string | null { return typeof code === "string" ? code : null; } +type CodexCredentialOwner = "codex" | "admin-codex"; + /** - * Park a Codex connection whose sign-in the issuer has refused for good. - * - * Retrying a dead refresh token cannot succeed; until this existed every - * generation of every card of every run asked the issuer again, failed with - * the same 401, and logged it. Parking flips the connection to the existing - * `expired` state — the one Settings already renders as "reconnect ChatGPT" — - * so the chain, the presence probe and with it every AI capability stop - * counting it, and nothing is sent until the person signs in again (the - * device-code flow writes `connected`). - * - * Compare-and-set on the refresh token that was refused: a reconnect that - * landed in between wrote a different one and is left alone, and of several - * workers failing at once exactly one flips the row, so the transition is - * signalled once and never again per run. + * One Codex credential row as a refresh sees it: the user's own connection or + * the operator's central one. `writeRotated` and `park` are compare-and-set on + * the refresh token this refresh spent, so a sign-in that landed meanwhile (the + * device-code flow writes without the refresh lock) is never overwritten. */ -async function parkUserCodex( - userId: string, - refusedRefreshEncrypted: string, - error: unknown, -): Promise { - try { - const parked = await prisma.user.updateMany({ - where: { - id: userId, - codexConnectionStatus: "connected", - codexRefreshTokenEncrypted: refusedRefreshEncrypted, - }, - data: { codexConnectionStatus: "expired" }, - }); - if (parked.count === 1) { - emitSignal({ - action: "ai.credential.expired", - level: "warn", - meta: { provider: "codex", oauth_error: oauthErrorOf(error) }, - }); - } - } catch { - // Best effort: an unparked connection is retried exactly as before. - } +interface CodexCredentialStore { + owner: CodexCredentialOwner; + lockId: string; + read(tx: Prisma.TransactionClient): Promise<{ + status: string | null; + refreshEncrypted: string | null; + creds: CodexCreds | null; + } | null>; + writeRotated( + tx: Prisma.TransactionClient, + spentRefreshEncrypted: string, + fresh: CodexCreds, + ): Promise; + park( + tx: Prisma.TransactionClient, + spentRefreshEncrypted: string, + ): Promise; } -/** The operator's central Codex, parked the same way. */ -async function parkCentralCodex( - refusedRefreshEncrypted: string, - error: unknown, -): Promise { - try { - const parked = await prisma.appSettings.updateMany({ +function userCodexStore(userId: string): CodexCredentialStore { + return { + owner: "codex", + lockId: userId, + async read(tx) { + const row = await tx.user.findUnique({ + where: { id: userId }, + select: { + codexConnectionStatus: true, + codexAccessTokenEncrypted: true, + codexRefreshTokenEncrypted: true, + }, + }); + if (!row) return null; + return { + status: row.codexConnectionStatus, + refreshEncrypted: row.codexRefreshTokenEncrypted, + creds: + row.codexAccessTokenEncrypted && row.codexRefreshTokenEncrypted + ? decryptCodexCreds({ + accessEncrypted: row.codexAccessTokenEncrypted, + refreshEncrypted: row.codexRefreshTokenEncrypted, + }) + : null, + }; + }, + async writeRotated(tx, spent, fresh) { + const enc = encryptCodexCreds(fresh); + const written = await tx.user.updateMany({ + where: { id: userId, codexRefreshTokenEncrypted: spent }, + data: { + codexAccessTokenEncrypted: enc.accessEncrypted, + codexRefreshTokenEncrypted: enc.refreshEncrypted, + codexTokenExpiresAt: fresh.expiresAt, + codexConnectionStatus: "connected", + }, + }); + return written.count; + }, + async park(tx, spent) { + const parked = await tx.user.updateMany({ + where: { + id: userId, + codexConnectionStatus: "connected", + codexRefreshTokenEncrypted: spent, + }, + data: { codexConnectionStatus: "expired" }, + }); + return parked.count; + }, + }; +} + +const centralCodexStore: CodexCredentialStore = { + owner: "admin-codex", + lockId: "singleton", + async read(tx) { + const row = await tx.appSettings.findUnique({ + where: { id: "singleton" }, + select: { + adminCodexConnectionStatus: true, + adminCodexAccessTokenEncrypted: true, + adminCodexRefreshTokenEncrypted: true, + adminCodexAccountIdEncrypted: true, + adminCodexTokenExpiresAt: true, + }, + }); + if (!row) return null; + return { + status: row.adminCodexConnectionStatus, + refreshEncrypted: row.adminCodexRefreshTokenEncrypted, + creds: + row.adminCodexAccessTokenEncrypted && + row.adminCodexRefreshTokenEncrypted && + row.adminCodexAccountIdEncrypted + ? decryptAdminCodexCreds({ + accessEncrypted: row.adminCodexAccessTokenEncrypted, + refreshEncrypted: row.adminCodexRefreshTokenEncrypted, + accountIdEncrypted: row.adminCodexAccountIdEncrypted, + expiresAt: row.adminCodexTokenExpiresAt, + }) + : null, + }; + }, + async writeRotated(tx, spent, fresh) { + const enc = encryptAdminCodexCreds(fresh); + const written = await tx.appSettings.updateMany({ + where: { id: "singleton", adminCodexRefreshTokenEncrypted: spent }, + data: { + adminCodexAccessTokenEncrypted: enc.accessEncrypted, + adminCodexRefreshTokenEncrypted: enc.refreshEncrypted, + adminCodexAccountIdEncrypted: enc.accountIdEncrypted, + adminCodexTokenExpiresAt: enc.expiresAt, + adminCodexConnectionStatus: "connected", + }, + }); + return written.count; + }, + async park(tx, spent) { + const parked = await tx.appSettings.updateMany({ where: { id: "singleton", adminCodexConnectionStatus: "connected", - adminCodexRefreshTokenEncrypted: refusedRefreshEncrypted, + adminCodexRefreshTokenEncrypted: spent, }, data: { adminCodexConnectionStatus: "expired" }, }); - if (parked.count === 1) { - emitSignal({ - action: "ai.credential.expired", - level: "warn", - meta: { provider: "admin-codex", oauth_error: oauthErrorOf(error) }, - }); - } - } catch { - // Best effort, as above. + return parked.count; + }, +}; + +/** Refreshes in flight in this process, keyed on credential and spent token. */ +const codexRefreshesInFlight = new Map>(); + +/** + * Refresh one Codex credential, at most once per spent token. + * + * The issuer rotates the refresh token on every refresh and honours each one + * once. Two callers that refreshed the same token used to race: the loser got + * `refresh_token_reused`, read it as a dead sign-in and parked the connection + * while the winner's fresh tokens landed beside an `expired` status. So: + * + * - callers in one process share one refresh (single flight), and callers + * in different processes queue on the per-credential advisory lock the + * provider syncs use (`acquireProviderTokenRefreshLock`); + * - inside the lock the row is read again, and a token a peer already + * rotated is used as it is, without asking the issuer; + * - the rotated tokens are written compare-and-set on the spent token, and + * the write sets `connected`; + * - after a refusal the row is read again too, and only a refusal of the + * token still stored, with the issuer's dead-grant answer, parks it + * (`expired`, the state Settings renders as "reconnect ChatGPT"). The + * transition is signalled once: only the lock holder can make it. + * + * Parking stops the chain, the presence probe and every AI capability from + * counting the connection until the person signs in again, which is what + * stopped every run from asking the issuer about the same dead token. + */ +function refreshCodexCredential( + store: CodexCredentialStore, + spentRefreshEncrypted: string, +): Promise { + const key = `${store.owner}:${store.lockId}:${spentRefreshEncrypted}`; + const inFlight = codexRefreshesInFlight.get(key); + if (inFlight) return inFlight; + const run = runCodexRefresh(store, spentRefreshEncrypted).finally(() => { + codexRefreshesInFlight.delete(key); + }); + codexRefreshesInFlight.set(key, run); + return run; +} + +async function runCodexRefresh( + store: CodexCredentialStore, + spent: string, +): Promise { + const outcome = await prisma.$transaction( + async ( + tx, + ): Promise< + | { fresh: CodexCreds } + | { refused: unknown; parked: boolean } + | { failed: unknown } + > => { + await acquireProviderTokenRefreshLock(tx, store.owner, store.lockId); + const row = await store.read(tx); + if (!row || row.status !== "connected" || !row.refreshEncrypted) { + return { + refused: codexReconnectRequired("no refresh token"), + parked: false, + }; + } + if (row.refreshEncrypted !== spent && row.creds) { + // A peer rotated while this caller waited: its token is the live one. + return { fresh: row.creds }; + } + if (!row.creds) { + const refused = codexReconnectRequired("token storage corrupt"); + return { + refused, + parked: (await store.park(tx, row.refreshEncrypted)) === 1, + }; + } + let fresh: CodexCreds; + try { + fresh = await refreshDeviceTokens(row.creds.refreshToken); + } catch (error) { + const now = await store.read(tx); + if (now?.refreshEncrypted !== row.refreshEncrypted && now?.creds) { + // A sign-in replaced the token while the issuer refused the old one. + return { fresh: now.creds }; + } + if (!isDeadCodexSignIn(error)) return { failed: error }; + return { + refused: error, + parked: (await store.park(tx, row.refreshEncrypted)) === 1, + }; + } + if ((await store.writeRotated(tx, row.refreshEncrypted, fresh)) === 0) { + // A sign-in replaced the row meanwhile; it wins over this refresh. + const now = await store.read(tx); + if (now?.creds) return { fresh: now.creds }; + } + return { fresh }; + }, + PROVIDER_REFRESH_TRANSACTION_OPTIONS, + ); + if ("fresh" in outcome) return outcome.fresh; + if ("failed" in outcome) throw outcome.failed; + if (outcome.parked) { + emitSignal({ + action: "ai.credential.expired", + level: "warn", + meta: { + provider: store.owner, + oauth_error: oauthErrorOf(outcome.refused), + }, + }); } + throw outcome.refused; } class NoProvider implements AIProvider { @@ -384,27 +571,16 @@ async function resolveCodexProvider( // Proactive refresh if the access token is within 5 min of expiry. if (stored.expiresAt.getTime() < Date.now() + TOKEN_REFRESH_BUFFER_MS) { try { - const fresh = await refreshDeviceTokens(stored.refreshToken); - active = fresh; - - const enc = encryptCodexCreds(fresh); - await prisma.user.update({ - where: { id: userId }, - data: { - codexAccessTokenEncrypted: enc.accessEncrypted, - codexRefreshTokenEncrypted: enc.refreshEncrypted, - codexTokenExpiresAt: fresh.expiresAt, - }, - }); + active = await refreshCodexCredential( + userCodexStore(userId), + user.codexRefreshTokenEncrypted, + ); } catch (error) { - // A refused grant parks the connection and drops it from this chain: - // calling Codex with the expired access token could only 401 into a - // second refusal. Anything else falls through — CodexClient will - // trigger an on-401 refresh and persist there. - if (isDeadCodexSignIn(error)) { - await parkUserCodex(userId, user.codexRefreshTokenEncrypted, error); - return null; - } + // A refused grant drops the connection from this chain (the refresh + // parked it): calling Codex with the expired access token could only + // 401 into a second refusal. Anything else falls through — CodexClient + // will trigger an on-401 refresh. + if (isDeadCodexSignIn(error)) return null; } } @@ -412,54 +588,17 @@ async function resolveCodexProvider( accessToken: active.accessToken, accountId: active.accountId, onTokenRefresh: async () => { - const freshUser = await prisma.user.findUnique({ + const current = await prisma.user.findUnique({ where: { id: userId }, - select: { - codexAccessTokenEncrypted: true, - codexRefreshTokenEncrypted: true, - }, + select: { codexRefreshTokenEncrypted: true }, }); - if ( - !freshUser?.codexAccessTokenEncrypted || - !freshUser.codexRefreshTokenEncrypted - ) { + if (!current?.codexRefreshTokenEncrypted) { throw codexReconnectRequired("no refresh token"); } - const decoded = decryptCodexCreds({ - accessEncrypted: freshUser.codexAccessTokenEncrypted, - refreshEncrypted: freshUser.codexRefreshTokenEncrypted, - }); - if (!decoded) { - const error = codexReconnectRequired("token storage corrupt"); - await parkUserCodex( - userId, - freshUser.codexRefreshTokenEncrypted, - error, - ); - throw error; - } - let fresh; - try { - fresh = await refreshDeviceTokens(decoded.refreshToken); - } catch (error) { - if (isDeadCodexSignIn(error)) { - await parkUserCodex( - userId, - freshUser.codexRefreshTokenEncrypted, - error, - ); - } - throw error; - } - const enc = encryptCodexCreds(fresh); - await prisma.user.update({ - where: { id: userId }, - data: { - codexAccessTokenEncrypted: enc.accessEncrypted, - codexRefreshTokenEncrypted: enc.refreshEncrypted, - codexTokenExpiresAt: fresh.expiresAt, - }, - }); + const fresh = await refreshCodexCredential( + userCodexStore(userId), + current.codexRefreshTokenEncrypted, + ); return { accessToken: fresh.accessToken, accountId: fresh.accountId }; }, }); @@ -521,26 +660,15 @@ async function resolveAdminCodexProvider(): Promise { if (stored.expiresAt.getTime() < Date.now() + TOKEN_REFRESH_BUFFER_MS) { try { - const fresh = await refreshDeviceTokens(stored.refreshToken); - active = fresh; - const enc = encryptAdminCodexCreds(fresh); - await prisma.appSettings.update({ - where: { id: "singleton" }, - data: { - adminCodexAccessTokenEncrypted: enc.accessEncrypted, - adminCodexRefreshTokenEncrypted: enc.refreshEncrypted, - adminCodexAccountIdEncrypted: enc.accountIdEncrypted, - adminCodexTokenExpiresAt: enc.expiresAt, - }, - }); + active = await refreshCodexCredential( + centralCodexStore, + settings.adminCodexRefreshTokenEncrypted, + ); } catch (error) { - // A refused grant parks the central connection (see `parkUserCodex`); - // anything else falls through — CodexClient triggers an on-401 refresh - // and persists via the callback below. - if (isDeadCodexSignIn(error)) { - await parkCentralCodex(settings.adminCodexRefreshTokenEncrypted, error); - return null; - } + // A refused grant parked the central connection (see + // `refreshCodexCredential`); anything else falls through — CodexClient + // triggers an on-401 refresh via the callback below. + if (isDeadCodexSignIn(error)) return null; } } @@ -548,52 +676,17 @@ async function resolveAdminCodexProvider(): Promise { accessToken: active.accessToken, accountId: active.accountId, onTokenRefresh: async () => { - const fresh = await prisma.appSettings.findUnique({ + const current = await prisma.appSettings.findUnique({ where: { id: "singleton" }, - select: { - adminCodexAccessTokenEncrypted: true, - adminCodexRefreshTokenEncrypted: true, - adminCodexAccountIdEncrypted: true, - adminCodexTokenExpiresAt: true, - }, + select: { adminCodexRefreshTokenEncrypted: true }, }); - if ( - !fresh?.adminCodexAccessTokenEncrypted || - !fresh.adminCodexRefreshTokenEncrypted || - !fresh.adminCodexAccountIdEncrypted - ) { + if (!current?.adminCodexRefreshTokenEncrypted) { throw codexReconnectRequired("no central refresh token"); } - const decoded = decryptAdminCodexCreds({ - accessEncrypted: fresh.adminCodexAccessTokenEncrypted, - refreshEncrypted: fresh.adminCodexRefreshTokenEncrypted, - accountIdEncrypted: fresh.adminCodexAccountIdEncrypted, - expiresAt: fresh.adminCodexTokenExpiresAt, - }); - if (!decoded) { - const error = codexReconnectRequired("central token storage corrupt"); - await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error); - throw error; - } - let refreshed; - try { - refreshed = await refreshDeviceTokens(decoded.refreshToken); - } catch (error) { - if (isDeadCodexSignIn(error)) { - await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error); - } - throw error; - } - const enc = encryptAdminCodexCreds(refreshed); - await prisma.appSettings.update({ - where: { id: "singleton" }, - data: { - adminCodexAccessTokenEncrypted: enc.accessEncrypted, - adminCodexRefreshTokenEncrypted: enc.refreshEncrypted, - adminCodexAccountIdEncrypted: enc.accountIdEncrypted, - adminCodexTokenExpiresAt: enc.expiresAt, - }, - }); + const refreshed = await refreshCodexCredential( + centralCodexStore, + current.adminCodexRefreshTokenEncrypted, + ); return { accessToken: refreshed.accessToken, accountId: refreshed.accountId, diff --git a/src/lib/integrations/oauth-refresh.ts b/src/lib/integrations/oauth-refresh.ts index 89c424051..b2ecaec70 100644 --- a/src/lib/integrations/oauth-refresh.ts +++ b/src/lib/integrations/oauth-refresh.ts @@ -61,7 +61,8 @@ export async function persistRotatedToken( return opts.readPeerAccessToken(); } -export type SerializedRefreshProvider = "whoop" | "withings"; +export type SerializedRefreshProvider = + "whoop" | "withings" | "codex" | "admin-codex"; /** * Provider refreshes hold a database transaction open around one bounded diff --git a/tests/integration/codex-credential-park.test.ts b/tests/integration/codex-credential-park.test.ts index 95ea57391..811fefd0c 100644 --- a/tests/integration/codex-credential-park.test.ts +++ b/tests/integration/codex-credential-park.test.ts @@ -261,3 +261,61 @@ describe("a refused Codex sign-in", () => { expect(vi.mocked(refreshDeviceTokens)).toHaveBeenCalledTimes(1); }); }); + +describe("two refreshes of one Codex token", () => { + async function codexStatus(userId: string) { + return ( + await prisma.user.findUniqueOrThrow({ + where: { id: userId }, + select: { codexConnectionStatus: true }, + }) + ).codexConnectionStatus; + } + + it("share one refresh: the slower caller never parks the winner's sign-in", async () => { + const user = await seedCodexUser(); + // The issuer honours each refresh token once. The first caller's refresh + // takes a moment; a second refresh of the same token is refused as reused. + let asked = 0; + vi.mocked(refreshDeviceTokens).mockImplementation(async () => { + asked += 1; + if (asked > 1) throw new CodexRefreshError(400, "refresh_token_reused"); + await new Promise((resolve) => setTimeout(resolve, 200)); + return creds("rotated", 10 * HOUR_MS); + }); + + const [first, second] = await Promise.all([ + resolveProviderChain(user.id), + resolveProviderChain(user.id), + ]); + + expect(first.map((p) => p.providerType)).toContain("codex"); + expect(second.map((p) => p.providerType)).toContain("codex"); + expect(asked).toBe(1); + expect(await codexStatus(user.id)).toBe("connected"); + expect(parkSignals("codex")).toHaveLength(0); + }); + + it("carries on with the token a peer rotated while the issuer refused the old one", async () => { + const user = await seedCodexUser(); + // Another process won the race: by the time the issuer answers this + // caller "reused", the peer's rotated tokens are in the row. + vi.mocked(refreshDeviceTokens).mockImplementation(async () => { + const peer = encryptCodexCreds(creds("peer", 10 * HOUR_MS)); + await prisma.user.update({ + where: { id: user.id }, + data: { + codexAccessTokenEncrypted: peer.accessEncrypted, + codexRefreshTokenEncrypted: peer.refreshEncrypted, + }, + }); + throw new CodexRefreshError(400, "refresh_token_reused"); + }); + + const chain = await resolveProviderChain(user.id); + + expect(chain.map((p) => p.providerType)).toContain("codex"); + expect(await codexStatus(user.id)).toBe("connected"); + expect(parkSignals("codex")).toHaveLength(0); + }); +}); From 769d0334a7d502540b6afda14d0f49250f9373fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 21 Oct 2026 00:06:21 +0200 Subject: [PATCH 2/2] fix(rollups): match the keyed refold's days in UTC whatever the session zone The keyed refold's day filter and delete compared the naive UTC measured_at and bucket_start with a timestamptz[] parameter, which goes through the session time zone; in any zone but UTC neither matched and the refold left its buckets stale. Both now bind timestamp[] literals. Keys are refolded in (type, day) order in chunks of 500, and the rollup-recompute queue describes the runs of buckets a job now covers. --- src/lib/jobs/queue-runtime.ts | 4 +- src/lib/rollups/measurement-rollups.ts | 37 +++++++-- .../rollup-keyed-refold-session-tz.test.ts | 76 +++++++++++++++++++ 3 files changed, 108 insertions(+), 9 deletions(-) create mode 100644 tests/integration/rollup-keyed-refold-session-tz.test.ts diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 3c3d007ff..3bc085fd0 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -190,7 +190,9 @@ export const QUEUE_RUNTIME: Readonly> = { "daily-briefing": short("A light cohort scan every fifteen minutes."), // ── Rollups ──────────────────────────────────────────────────────── - "rollup-recompute": short("One bucket."), + "rollup-recompute": short( + "One type at one granularity over one bucket or a run of adjacent buckets, inside the five-year fold window.", + ), "rollup-full-backfill": { runtime: "long", expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 0a9eae21b..a5277687f 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -362,8 +362,19 @@ export async function recomputeBucketsForMeasurement( ); } +/** + * UTC wall-clock literals for a `timestamp[]` parameter. `measured_at` and + * `bucket_start` are naive UTC timestamps, so the instants they are compared + * with must be too: a `timestamptz[]` parameter is converted through the + * session time zone, and in any zone but UTC neither the day filter nor the + * keyed delete matched a single day, so a refold left its buckets stale. + */ +function utcStamps(dates: readonly Date[]): string[] { + return dates.map((d) => d.toISOString().replace("Z", "")); +} + /** Keys one set-based DAY refold carries; bounds the statement's arrays. */ -const KEYED_REFOLD_CHUNK = 2_000; +const KEYED_REFOLD_CHUNK = 500; /** * The set-based sibling of {@link recomputeBucketsForMeasurement}, for a write @@ -381,7 +392,17 @@ export async function recomputeBucketsForKeys( userId: string, rows: Array<{ type: MeasurementType; measuredAt: Date }>, ): Promise { - const keys = collapseToTypeDayKeys(rows); + // Sorted on (type, day), the order every upsert of the tier takes its row + // locks in (`persistRollupRows` sorts the same way), and chunked small, so + // a keyed refold racing the full fold or another refold holds a short, + // deterministic slice and waits rather than interleaving lock orders. + const keys = collapseToTypeDayKeys(rows).sort((a, b) => + a.type !== b.type + ? a.type < b.type + ? -1 + : 1 + : a.measuredAt.getTime() - b.measuredAt.getTime(), + ); for (let i = 0; i < keys.length; i += KEYED_REFOLD_CHUNK) { await refoldDayBuckets(userId, keys.slice(i, i + KEYED_REFOLD_CHUNK)); } @@ -441,11 +462,11 @@ async function refoldDayBuckets( AND granularity = 'DAY'::"measurement_rollup_granularity" AND (type, bucket_start) IN ( SELECT k.t::"measurement_type", k.b - FROM unnest($2::text[], $3::timestamptz[]) AS k(t, b) + FROM unnest($2::text[], $3::timestamp[]) AS k(t, b) )`, userId, keys.map((k) => k.type), - keys.map((k) => k.measuredAt.toISOString()), + utcStamps(keys.map((k) => k.measuredAt)), ); for (let i = 0; i < ordered.length; i += 500) { const { sql, params } = buildRollupUpsert( @@ -642,9 +663,9 @@ async function runRollupAggregate( if (input.types && input.types.length === 0) return []; // Bound as the fourth parameter, never spliced. const dayFilter = input.days - ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])` : ""; - const dayParams = input.days ? [input.days.map((d) => d.toISOString())] : []; + const dayParams = input.days ? [utcStamps(input.days)] : []; // Postgres rejects `date_trunc($1::text, m."measured_at")` in a // GROUP BY because the planner sees the parameterised expression @@ -857,7 +878,7 @@ async function applyHourlyMeanDay( AND m."deleted_at" IS NULL ${ input.days - ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamp[])` : "" } GROUP BY m."type", m."source", date_trunc('hour', m."measured_at") @@ -869,7 +890,7 @@ async function applyHourlyMeanDay( input.userId, input.from, input.to, - ...(input.days ? [input.days.map((d) => d.toISOString())] : []), + ...(input.days ? [utcStamps(input.days)] : []), ); const byKey = new Map( means.map((m) => [ diff --git a/tests/integration/rollup-keyed-refold-session-tz.test.ts b/tests/integration/rollup-keyed-refold-session-tz.test.ts new file mode 100644 index 000000000..30e88bb3c --- /dev/null +++ b/tests/integration/rollup-keyed-refold-session-tz.test.ts @@ -0,0 +1,76 @@ +/** + * The keyed DAY refold does not depend on the session time zone. + * + * `measured_at` and `bucket_start` are naive UTC timestamps. The refold's day + * filter and its delete compared them with a `timestamptz[]` parameter, which + * Postgres converts through the session time zone: outside UTC neither matched + * a single day, so the refold silently left every touched bucket stale. The + * app pins its sessions to UTC, so this runs the pool in another zone on + * purpose. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db-session", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, TIME_ZONE_OPTION: "-c TimeZone=Pacific/Auckland" }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { prisma } from "@/lib/db"; +import { + recomputeBucketsForKeys, + recomputeBucketsForMeasurement, +} from "@/lib/rollups/measurement-rollups"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +describe("keyed DAY refold in a non-UTC session", () => { + it("folds a new reading into its day", async () => { + const [{ zone }] = await prisma.$queryRaw>` + SELECT current_setting('TimeZone') AS zone + `; + expect(zone).toBe("Pacific/Auckland"); + + const user = await getPrismaClient().user.create({ + data: { username: "tz-refold", email: "tz-refold@example.test" }, + }); + const day = new Date("2026-05-10T00:00:00.000Z"); + await getPrismaClient().measurement.createMany({ + data: [7, 19].map((hour, i) => ({ + userId: user.id, + type: "WEIGHT" as const, + value: 80 + i, + unit: "kg", + measuredAt: new Date(day.getTime() + hour * 3_600_000), + })), + }); + await recomputeBucketsForMeasurement(user.id, "WEIGHT", day); + // A third reading lands; the refold has to fold it in. + await getPrismaClient().measurement.create({ + data: { + userId: user.id, + type: "WEIGHT", + value: 82, + unit: "kg", + measuredAt: new Date(day.getTime() + 21 * 3_600_000), + }, + }); + + await recomputeBucketsForKeys(user.id, [ + { type: "WEIGHT", measuredAt: new Date(day.getTime() + 7 * 3_600_000) }, + ]); + + const rows = await getPrismaClient().measurementRollup.findMany({ + where: { userId: user.id, granularity: "DAY" }, + }); + expect(rows).toHaveLength(1); + expect(rows[0]!.bucketStart.toISOString()).toBe(day.toISOString()); + expect(rows[0]!.count).toBe(3); + }); +});