From 38e3bbd5690d070e34f5e8c51e892d9e6bae2d6d Mon Sep 17 00:00:00 2001 From: Parv Ahuja <17094219+parvahuja@users.noreply.github.com> Date: Wed, 26 Aug 2026 16:12:12 -0400 Subject: [PATCH] fix(tempo): preserve ambiguous session vouchers --- .changeset/quiet-sessions-recover.md | 5 ++ src/client/internal/Fetch.test.ts | 47 +++++++++++++--- src/client/internal/Fetch.ts | 29 ++++++---- src/client/internal/MethodResponse.ts | 19 +++---- src/tempo/session/client/CredentialState.ts | 3 +- src/tempo/session/client/Session.test.ts | 4 +- src/tempo/session/client/Session.ts | 54 ++++++++++--------- .../session/client/SessionManager.test.ts | 52 ++++++++++++++++++ src/tempo/session/client/SessionManager.ts | 10 ++-- src/tempo/session/client/Transports.test.ts | 27 +++++++++- src/tempo/session/client/Transports.ts | 13 +++-- 11 files changed, 196 insertions(+), 67 deletions(-) create mode 100644 .changeset/quiet-sessions-recover.md diff --git a/.changeset/quiet-sessions-recover.md b/.changeset/quiet-sessions-recover.md new file mode 100644 index 000000000..b9367cfff --- /dev/null +++ b/.changeset/quiet-sessions-recover.md @@ -0,0 +1,5 @@ +--- +'mppx': patch +--- + +Preserved dispatched session voucher authorization after ambiguous HTTP response failures. diff --git a/src/client/internal/Fetch.test.ts b/src/client/internal/Fetch.test.ts index 4de3acf17..6d3df7ecd 100644 --- a/src/client/internal/Fetch.test.ts +++ b/src/client/internal/Fetch.test.ts @@ -448,26 +448,57 @@ describe('Fetch.from: method responses', () => { }) test.each([ - ['a rejected paid response', () => new Response('rejected', { status: 500 }), 500, undefined], - ['the final payment retry', () => make402(), 402, 1], + [ + 'a rejected paid response', + () => new Response('rejected', { status: 500 }), + 500, + undefined, + 'rejected', + ], + ['the final payment retry', () => make402(), 402, 1, 'rejected'], [ 'the paid request throwing', () => Promise.reject(new Error('paid request failed')), undefined, undefined, + 'unknown', ], - ])('settles credential state after %s', async (_name, paidRequest, status, maxPaymentRetries) => { + ])( + 'settles credential state after %s', + async (_name, paidRequest, status, maxPaymentRetries, expected) => { + const settlements: MethodResponse.AttemptOutcome['status'][] = [] + let calls = 0 + const fetch = Fetch.from({ + fetch: async () => (++calls === 1 ? make402() : paidRequest()), + maxPaymentRetries, + methods: [trackedMethod(settlements)], + }) + + const result = fetch('https://example.com/paid') + if (status) expect((await result).status).toBe(status) + else await expect(result).rejects.toThrow('paid request failed') + expect(settlements).toEqual([expected]) + }, + ) + + test('rejects credential state when an already-aborted paid request throws', async () => { + const controller = new AbortController() const settlements: MethodResponse.AttemptOutcome['status'][] = [] let calls = 0 const fetch = Fetch.from({ - fetch: async () => (++calls === 1 ? make402() : paidRequest()), - maxPaymentRetries, + fetch: async () => { + if (++calls === 1) { + controller.abort() + return make402() + } + throw new Error('paid request aborted') + }, methods: [trackedMethod(settlements)], }) - const result = fetch('https://example.com/paid') - if (status) expect((await result).status).toBe(status) - else await expect(result).rejects.toThrow('paid request failed') + await expect(fetch('https://example.com/paid', { signal: controller.signal })).rejects.toThrow( + 'paid request aborted', + ) expect(settlements).toEqual(['rejected']) }) }) diff --git a/src/client/internal/Fetch.ts b/src/client/internal/Fetch.ts index a53d0b381..c30ae4668 100644 --- a/src/client/internal/Fetch.ts +++ b/src/client/internal/Fetch.ts @@ -363,17 +363,24 @@ export function from( }), ) - response = await baseFetch( - paymentInput, - transport.setCredential( - { - ...fetchInit, - headers: initialRequest.headers, - }, - credential, - { challenge: selectedChallenge }, - ), + const credentialInit = transport.setCredential( + { + ...fetchInit, + headers: initialRequest.headers, + }, + credential, + { challenge: selectedChallenge }, ) + const signal = resolveRequestSignal(paymentInput, credentialInit) + const dispatched = !signal?.aborted + try { + response = await baseFetch(paymentInput, credentialInit) + } catch (error) { + await settleAttempt(prepared, { + status: dispatched ? 'unknown' : 'rejected', + }) + throw error + } const paymentRequired = await transport.isPaymentRequired( response, transportRequest as never, @@ -985,7 +992,7 @@ async function resolveCredential( const parsedContext = mi.context && context !== undefined ? mi.context.parse(context) : undefined const parameters = parsedContext !== undefined ? { challenge, context: parsedContext } : { challenge } - if (attempt) MethodResponse.attachAttempt(mi, parameters, attempt) + if (attempt) MethodResponse.attachAttempt(parameters, attempt) return mi.createCredential(parameters as never) } diff --git a/src/client/internal/MethodResponse.ts b/src/client/internal/MethodResponse.ts index 85dd8ab1e..916a782cb 100644 --- a/src/client/internal/MethodResponse.ts +++ b/src/client/internal/MethodResponse.ts @@ -8,7 +8,7 @@ const attempts = new WeakMap() export type AttemptOutcome = { challenges?: readonly Challenge.Challenge[] | undefined response?: Response | undefined - status: 'accepted' | 'pending' | 'rejected' + status: 'accepted' | 'pending' | 'rejected' | 'unknown' } export type Attempt = { @@ -47,16 +47,17 @@ export function unregister(method: Method.AnyClient): void { handlers.delete(method) } -/** Adds response lifecycle state to internal credential parameters. */ -export function attachAttempt( - method: Method.AnyClient, - parameters: object, - attempt: Attempt, -): void { - if (handlers.has(method)) attempts.set(parameters, attempt) +/** Returns whether Fetch owns successful-response handling for this method. */ +export function hasHandler(method: Method.AnyClient): boolean { + return handlers.has(method) } -/** Reads response lifecycle state when Fetch owns this credential. */ +/** Adds transport lifecycle state to internal credential parameters. */ +export function attachAttempt(parameters: object, attempt: Attempt): void { + attempts.set(parameters, attempt) +} + +/** Reads transport lifecycle state for this credential attempt. */ export function getAttempt(parameters: object): Attempt | undefined { return attempts.get(parameters) } diff --git a/src/tempo/session/client/CredentialState.ts b/src/tempo/session/client/CredentialState.ts index 394e01194..2c5983b33 100644 --- a/src/tempo/session/client/CredentialState.ts +++ b/src/tempo/session/client/CredentialState.ts @@ -808,8 +808,7 @@ async function voucher( resolved.chainId, resolved.escrow, ) - entry.cumulativeAmount = cumulativeAmount - await storeChannelEntry(sink, entry) + await storeChannelEntry(sink, { ...entry, cumulativeAmount }) return payload } diff --git a/src/tempo/session/client/Session.test.ts b/src/tempo/session/client/Session.test.ts index 42d7041f1..68b97b4fb 100644 --- a/src/tempo/session/client/Session.test.ts +++ b/src/tempo/session/client/Session.test.ts @@ -586,8 +586,8 @@ describe('precompile client session', () => { expect((await first).status).toBe(200) expect((await second).status).toBe(500) expect((await third).status).toBe(500) - expect(actions).toEqual(['open', 'topUp', 'voucher', 'topUp', 'voucher']) - expect(vouchers).toEqual([200n, 300n]) + expect(actions).toEqual(['open', 'topUp', 'voucher', 'voucher']) + expect(vouchers).toEqual([200n, 200n]) expect(await channelStore.get(defaultChannelKey)).toMatchObject({ opened: true }) }) diff --git a/src/tempo/session/client/Session.ts b/src/tempo/session/client/Session.ts index 855b2978b..7cee9e178 100644 --- a/src/tempo/session/client/Session.ts +++ b/src/tempo/session/client/Session.ts @@ -212,22 +212,22 @@ export function session(parameters: session.Parameters = {}) { plan = await resolveCredentialPlan(resolved, context, entry) } } - let pendingOpen: ChannelEntry | undefined - // Defer opens only when low-level Fetch owns the response lifecycle. - // SessionManager unregisters this hook and keeps its existing transaction boundary. - const credentialSink = - attempt && plan.type === 'open' - ? { - store: { - get: (key: string) => store.get(key), - async set(next: ChannelEntry) { - pendingOpen = next - }, - delete: (key: string) => store.delete(key), + let pendingEntry: ChannelEntry | undefined + const deferEntry = + attempt && + (plan.type === 'voucher' || (plan.type === 'open' && MethodResponse.hasHandler(method))) + const credentialSink = deferEntry + ? { + store: { + get: (key: string) => store.get(key), + async set(next: ChannelEntry) { + pendingEntry = next }, - notifyUpdate() {}, - } - : sink + delete: (key: string) => store.delete(key), + }, + notifyUpdate() {}, + } + : sink const payload = await executeCredentialPlan( plan, credentialSink, @@ -239,21 +239,27 @@ export function session(parameters: session.Parameters = {}) { resolved.chainId, plan.account, ) - if (attempt && pendingOpen) { - const opened = pendingOpen + if (attempt && pendingEntry) { + const staged = pendingEntry + const voucher = plan.type === 'voucher' attempt.settle = async (outcome) => { const accepted = - outcome.status === 'accepted' || acknowledgesOpen(outcome, challenge, opened) + outcome.status === 'accepted' || + (voucher + ? outcome.status === 'unknown' + : acknowledgesOpen(outcome, challenge, staged)) if (!accepted && outcome.status === 'pending') return false try { if (accepted) { const current = await store.get(resolved.key) - if ( - !current?.opened || - current.channelId.toLowerCase() !== opened.channelId.toLowerCase() - ) { - await store.set(opened) - sink.notifyUpdate(opened) + const sameChannel = + current?.channelId.toLowerCase() === staged.channelId.toLowerCase() + const shouldStore = voucher + ? !current || (sameChannel && current.cumulativeAmount < staged.cumulativeAmount) + : !current?.opened || !sameChannel + if (shouldStore) { + await store.set(staged) + sink.notifyUpdate(staged) } } return true diff --git a/src/tempo/session/client/SessionManager.test.ts b/src/tempo/session/client/SessionManager.test.ts index f9c1298a0..ab4e61255 100644 --- a/src/tempo/session/client/SessionManager.test.ts +++ b/src/tempo/session/client/SessionManager.test.ts @@ -794,6 +794,58 @@ describe('Session', () => { expect(s.state.status).toBe('active') }) + test('preserves a dispatched voucher when the response is lost', async () => { + const { map, store } = makeChannelStore([channelEntry()]) + let paidRequests = 0 + const mockFetch = vi.fn().mockImplementation((_input, init?: RequestInit) => { + const authorization = new Headers(init?.headers).get(Constants.Headers.authorization) + if (!authorization) return Promise.resolve(make402Response()) + paidRequests += 1 + if (paidRequests === 1) return Promise.resolve(makeOkResponse()) + throw new Error('connection reset after request dispatch') + }) + const s = sessionManager({ + account, + client, + fetch: mockFetch as typeof globalThis.fetch, + channelStore: store, + }) + + expect((await s.fetch('https://api.example.com/data')).status).toBe(200) + await expect(s.fetch('https://api.example.com/data')).rejects.toThrow( + 'connection reset after request dispatch', + ) + + expect(s.cumulative).toBe(3_000_000n) + expect(map.get(entryKey(channelEntry()))?.cumulativeAmount).toBe(3_000_000n) + }) + + test('preserves a first dispatched voucher when resuming a stored channel', async () => { + const { delete: remove, map, store } = makeChannelStore([channelEntry()]) + let responseLost = false + const mockFetch = vi.fn().mockImplementation((_input, init?: RequestInit) => { + const authorization = new Headers(init?.headers).get(Constants.Headers.authorization) + if (!authorization) + return Promise.resolve(responseLost ? makeOkResponse() : make402Response()) + responseLost = true + throw new Error('connection reset after resumed voucher dispatch') + }) + const s = sessionManager({ + account, + client, + fetch: mockFetch as typeof globalThis.fetch, + channelStore: store, + }) + + await expect(s.fetch('https://api.example.com/data')).rejects.toThrow( + 'connection reset after resumed voucher dispatch', + ) + + expect(remove).not.toHaveBeenCalled() + expect(map.get(entryKey(channelEntry()))?.cumulativeAmount).toBe(2_000_000n) + expect((await s.fetch('https://api.example.com/data')).status).toBe(200) + }) + test('does not bootstrap when disabled', async () => { const mockFetch = vi.fn().mockResolvedValue(makeOkResponse()) const s = sessionManager({ diff --git a/src/tempo/session/client/SessionManager.ts b/src/tempo/session/client/SessionManager.ts index 8d2fe7fd7..052a4f81e 100644 --- a/src/tempo/session/client/SessionManager.ts +++ b/src/tempo/session/client/SessionManager.ts @@ -301,13 +301,13 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa return state?.status === 'active' ? state.units : 0 } - function commitDurableTopUp(entry: ChannelEntry) { + function commitDurableEntry(entry: ChannelEntry) { const use = channelUse - const baseline = use?.committed?.channel?.entry ?? use?.resumed + const baseline = use?.committed?.channel?.entry ?? use?.resumed ?? use?.previous.channel?.entry if ( !use || baseline?.channelId.toLowerCase() !== entry.channelId.toLowerCase() || - entry.deposit <= baseline.deposit + (entry.deposit <= baseline.deposit && entry.cumulativeAmount <= baseline.cumulativeAmount) ) return use.resumed = undefined @@ -341,7 +341,7 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa units: activeUnits(), }) } - commitDurableTopUp(entry) + commitDurableEntry(entry) }, }) MethodResponse.unregister(method) @@ -594,7 +594,7 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa spent: runtime.spent.toString(), units: activeUnits(), }) - commitDurableTopUp(applied.channel) + commitDurableEntry(applied.channel) } return receipt } diff --git a/src/tempo/session/client/Transports.test.ts b/src/tempo/session/client/Transports.test.ts index d01ad6786..01256ba2a 100644 --- a/src/tempo/session/client/Transports.test.ts +++ b/src/tempo/session/client/Transports.test.ts @@ -310,7 +310,7 @@ describe('HttpManagement', () => { expect(restoreCumulative).toHaveBeenCalledWith(channelId, 5n) }) - test('retryHttpPaymentRequired restores cumulative authorization when retry throws', async () => { + test('retryHttpPaymentRequired preserves cumulative authorization when dispatch throws', async () => { const entry = channel({ cumulativeAmount: 5n }) const restoreCumulative = vi.fn() @@ -329,6 +329,31 @@ describe('HttpManagement', () => { }), ).rejects.toThrow('network failed') + expect(restoreCumulative).not.toHaveBeenCalled() + }) + + test('retryHttpPaymentRequired restores cumulative authorization when already aborted', async () => { + const controller = new AbortController() + controller.abort() + const entry = channel({ cumulativeAmount: 5n }) + const restoreCumulative = vi.fn() + + await expect( + retryHttpPaymentRequired({ + createSessionCredential: async () => 'voucher-credential', + fetch: async () => { + throw new Error('request aborted before dispatch') + }, + getChannel: () => entry, + init: { signal: controller.signal }, + input: 'https://example.test/resource', + response: response402(challenge(snapshot())), + restoreCumulative, + setChallenge() {}, + topUpIfNeeded: async () => {}, + }), + ).rejects.toThrow('request aborted before dispatch') + expect(restoreCumulative).toHaveBeenCalledWith(channelId, 5n) }) diff --git a/src/tempo/session/client/Transports.ts b/src/tempo/session/client/Transports.ts index da8fccb79..3b00c481f 100644 --- a/src/tempo/session/client/Transports.ts +++ b/src/tempo/session/client/Transports.ts @@ -698,6 +698,7 @@ export async function retryHttpPaymentRequired( : requiredCumulative const restore = () => parameters.restoreCumulative(snapshot.channelId, cumulativeBeforeVoucher) let retry: Response + let dispatched = false try { const credential = await parameters.createSessionCredential(challenge, { action: 'voucher', @@ -705,12 +706,14 @@ export async function retryHttpPaymentRequired( descriptor: currentChannel.descriptor, cumulativeAmountRaw: cumulativeAmount.toString(), }) - retry = await parameters.fetch( - parameters.input, - requestInitWithAuthorization(parameters.input, parameters.init, credential), - ) + const retryInit = requestInitWithAuthorization(parameters.input, parameters.init, credential) + const signal = + retryInit.signal ?? + (parameters.input instanceof Request ? parameters.input.signal : undefined) + dispatched = !signal?.aborted + retry = await parameters.fetch(parameters.input, retryInit) } catch (error) { - await restore() + if (!dispatched) await restore() throw error } if (!retry.ok && !retry.headers.get(Constants.Headers.paymentReceipt)) {