From 7ac6c5c8ccfd4fdb3ccc614759244b2e48f18fb3 Mon Sep 17 00:00:00 2001 From: 404oops Date: Sun, 1 Nov 2026 19:04:20 +0100 Subject: [PATCH] fix(realtime): reuse settled credit for post-record continuation --- .../OpenAIRealtimeRelayService.test.ts | 128 ++++++++++++++++++ .../realtime/OpenAIRealtimeRelayService.ts | 12 +- src/backend/services/realtime/README.md | 10 +- src/docs/src/AI/realtime.md | 2 + src/docs/src/rate-limits-and-quotas.md | 2 +- 5 files changed, 148 insertions(+), 6 deletions(-) diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts index 50d3fcb21..880848a69 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts @@ -1783,6 +1783,134 @@ describe('OpenAIRealtimeRelayService relay', () => { }, ); + it('uses cached continuation without re-reading headroom or subtracting its own hold', async () => { + const holder = await createTestUser(server, { + username: 'continuation-fraction', + password: 'continuation-fraction-password', + }); + paidPolicy.monthUsageAllowance = MINI_HOLD + 0.5; + const paid = vi.mocked(server.services.metering.getActorSubscription); + const plan = paid.getMockImplementation()!; + paid.mockImplementation(async (actor) => ({ + ...(await plan(actor)), + monthUsageAllowance: MINI_HOLD + 0.5, + })); + const c = open(`${relayUrl}?auth_token=${holder.token}`); + await c.waitFor('session.created'); + c.ws.send(JSON.stringify({ type: 'hold.response' })); + await c.waitFor('response.created'); + const row = await server.stores.user.getByUsername(holder.username); + const actor = { user: row } as never; + await vi.waitFor(async () => + expect( + await server.services.metering.getOutstandingHolds(actor), + ).toBe(MINI_HOLD), + ); + const headroom = vi.spyOn(server.services.metering, 'hasEnoughCredits'); + const redis = vi.spyOn( + server.clients.redis as typeof server.clients.redis & { + creditHoldSum( + key: string, + now: string, + ): Promise<[string, number]>; + }, + 'creditHoldSum', + ); + const kvRead = vi.spyOn(server.stores.kv, 'get'); + const dbRead = vi.spyOn(server.clients.db, 'read'); + const dbPrimaryRead = vi.spyOn(server.clients.db, 'pread'); + const readCached = server.services.metering.hasAnyUsageCached.bind( + server.services.metering, + ); + const cached = vi + .spyOn(server.services.metering, 'hasAnyUsageCached') + .mockImplementation(async (actor) => { + const before = [ + redis.mock.calls.length, + kvRead.mock.calls.length, + dbRead.mock.calls.length + dbPrimaryRead.mock.calls.length, + ]; + const result = await readCached(actor); + expect([ + redis.mock.calls.length - before[0], + kvRead.mock.calls.length - before[1], + dbRead.mock.calls.length + + dbPrimaryRead.mock.calls.length - + before[2], + ]).toEqual([0, 0, 0]); + return result; + }); + providerEvent(c, { + type: 'response.done', + response: { + id: 'resp_hold', + usage: { + input_token_details: { audio_tokens: 8000 }, + output_token_details: { audio_tokens: 4096 }, + }, + }, + }); + await c.waitFor('response.done'); + await vi.waitFor(() => expect(cached).toHaveBeenCalled()); + expect(headroom).not.toHaveBeenCalled(); + expect(c.ws.readyState).toBe(WebSocket.OPEN); + const beforeHeadroom = [ + redis.mock.calls.length, + kvRead.mock.calls.length, + dbRead.mock.calls.length + dbPrimaryRead.mock.calls.length, + ]; + await server.services.metering.hasEnoughCredits(actor, 1); + const comparison = [ + redis.mock.calls.length - beforeHeadroom[0], + kvRead.mock.calls.length - beforeHeadroom[1], + dbRead.mock.calls.length + + dbPrimaryRead.mock.calls.length - + beforeHeadroom[2], + ]; + process.stdout.write( + `[realtime continuation I/O] Redis hold sum / KV get / SQL read: cached [0,0,0], headroom ${JSON.stringify(comparison)}\n`, + ); + c.ws.close(); + }); + + it('closes an exhausted account after recording and releases its response hold', async () => { + paidPolicy.monthUsageAllowance = MINI_HOLD; + const holder = await createTestUser(server, { + username: 'continuation-empty', + password: 'continuation-empty-password', + }); + const row = await server.stores.user.getByUsername(holder.username); + const actor = { user: row } as never; + const c = open(`${relayUrl}?auth_token=${holder.token}`); + await c.waitFor('session.created'); + c.ws.send(JSON.stringify({ type: 'hold.response' })); + await c.waitFor('response.created'); + await vi.waitFor(async () => + expect( + await server.services.metering.getOutstandingHolds(actor), + ).toBe(MINI_HOLD), + ); + providerEvent(c, { + type: 'response.done', + response: { + id: 'resp_hold', + usage: { + input_token_details: { audio_tokens: 8000 }, + output_token_details: { audio_tokens: 4096 }, + }, + }, + }); + expect(await c.waitFor('error')).toMatchObject({ + error: { code: 'insufficient_funds' }, + }); + await c.closed; + await vi.waitFor(async () => + expect( + await server.services.metering.getOutstandingHolds(actor), + ).toBe(0), + ); + }); + it('preserves the oldest billing identity when pending capacity is exceeded', async () => { const holder = await createTestUser(server, { username: 'transcription-cap', diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts index adb620982..bcd567066 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts @@ -1946,8 +1946,16 @@ export class OpenAIRealtimeRelayService extends PuterService { onExhausted: () => void, ): Promise { try { - await this.#metering.batchIncrementUsages(actor, rows); - if (!(await this.services.metering.hasEnoughCredits(actor, 1))) { + const recorded = await this.#metering.batchIncrementUsages( + actor, + rows, + ); + // Continuation asks whether settled credit remains; new work still + // requires atomic admission including every outstanding hold. + if ( + recorded.total > 0 && + !(await this.services.metering.hasAnyUsageCached(actor)) + ) { onExhausted(); } } catch (err) { diff --git a/src/backend/services/realtime/README.md b/src/backend/services/realtime/README.md index d8bd0d804..2538bdd4f 100644 --- a/src/backend/services/realtime/README.md +++ b/src/backend/services/realtime/README.md @@ -84,6 +84,11 @@ Each committed item retains its model until completion, failure, or bounded settlement. Missing usage is billed the item's fixed budget. Overflow closes the connection without evicting existing pending identities. +The post-record continuation check asks whether settled credit remains. It +excludes outstanding holds and accepts fractional positive credit. Admission of +new billable work continues to use the atomic hold path, which includes every +outstanding reservation. + Text, transcripts and raw audio are forwarded without new content logging. The relay does not invoke `ai.prompt.validate`. A realtime adapter needs the host listeners' chat payload and denial contracts before that hook can safely apply @@ -93,9 +98,8 @@ separate policy decisions. Credential checks run in batches of at most 16 with no overlapping sweeps. Session and access-token revocation events close affected sockets and invalidate their auth cache entries. Subscription eligibility is checked independently of -authentication and after subscription invalidations. A sustained verification -outage gets a 60-second retry grace from the first failed lookup, then closes -with a retryable internal error. Periodic checks remain the missed-event fallback. +authentication and after subscription invalidations. A failed verification check after a 60-second +retry grace from the first failed lookup closes with a retryable internal error. Periodic checks remain the missed-event fallback. When the drain starts, open sessions get a `puter.reconnect` event and new sessions receive `server_restarting` and close with 1012. Open sessions can finish an in-flight response for up to five seconds at shutdown, then close diff --git a/src/docs/src/AI/realtime.md b/src/docs/src/AI/realtime.md index cea7a760e..d7ca356a1 100644 --- a/src/docs/src/AI/realtime.md +++ b/src/docs/src/AI/realtime.md @@ -120,3 +120,5 @@ The endpoint requires a paid Puter plan. Usage is charged after completed respon Input transcription reserves a whole window before its configuration or audio is forwarded. The reservation remains held while that window has unused capacity or unsettled items. When the window is consumed, audio waits until its items settle and another reservation succeeds. Failed transcriptions and items without final usage are charged a conservative fixed amount. The reservation amounts, timeout and audio budget are listed in [rate limits and quotas](/rate-limits-and-quotas/). Provider-reported realtime models must have a known price. A missing model uses the last known model. An explicitly unpriced model closes the connection with `upstream_error`. + +After recording usage, the connection can continue while settled credit is positive, including a fractional balance. Each new billable operation still needs its full reservation after accounting for other outstanding holds. diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index abea96335..0ad7ceb0f 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -110,7 +110,7 @@ Realtime sessions ([`puter.ai.convo()`](/AI/convo/), `/puterai/openai/v1/realtim | Session setup timeout | 15 seconds | Error `connection_timeout`, close 1011. | | Upstream handshake timeout | 10 seconds | Close 1011. | | Heartbeat interval | 30 seconds | A missing pong terminates the connection on the next heartbeat. | -| Credential cache and recheck | Cache 60 seconds, sweep every 15 seconds, at most 16 checks in parallel | Invalid credentials close with 1008. A check that fails at connection gets `internal_error` and closes with 1011. A failed recheck retries on the next sweep for up to 60 seconds from the first failed lookup, then closes with `internal_error` and 1011. Subscription eligibility is checked independently on each sweep and after a plan invalidation. Revocation events close the affected credentials. Overlapping sweeps are skipped. | +| Credential cache and recheck | Cache 60 seconds, sweep every 15 seconds, at most 16 checks in parallel | Invalid credentials close with 1008. A check that fails at connection gets `internal_error` and closes with 1011. Once failures have continued for 60 seconds from the first failed lookup, the next failed check closes with `internal_error` and 1011. Subscription eligibility is checked independently on each sweep and after a plan invalidation. Revocation events close the affected credentials. Overlapping sweeps are skipped. | | Session attempts per signature-verified credential | 60 per minute | Error `rate_limit_exceeded`, close 1008. | | Pre-auth attempts | 120 per client IP and 600 per node per minute | Local memory admission runs before the WebSocket handshake and signature verification. Exceeding either rate returns HTTP 429. Client IP follows the configured proxy trust policy. The shared memory limiter caps its state at 10,000 keys with a 1-hour retention floor and a 60-second sweep. | | Unauthenticated upgrade sockets, including closing sockets | 64 per node | Further upgrades receive HTTP 503. The slot is retained until successful admission or transport closure. |