fix(tempo): preserve dispatched session vouchers

.changeset/quiet-sessions-recover.md

@@ -0,0 +1,5 @@
+---
+'mppx': patch
+---
+
+Preserved dispatched session voucher authorization after ambiguous HTTP response failures.

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'])
   })
 })

src/client/internal/Fetch.ts

@@ -363,17 +363,24 @@ export function from<const methods extends readonly Method.AnyClient[]>(
           }),
         )
 
-        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)
 }
 

src/client/internal/MethodResponse.ts

@@ -8,7 +8,7 @@ const attempts = new WeakMap<object, Attempt>()
 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)
 }

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
 }
 

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 })
   })
 

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

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({

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
   }

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)
     })