diff --git a/CHANGELOG.md b/CHANGELOG.md index 99c9bb380..bddf04179 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ This project uses selective package publishing. Each release entry lists the pub ### Added +- Seller shutdown now stops new work and drains active requests, final payment authorizations, and outgoing transport buffers before closing connections. Configure the default 60-second drain budget with `antseed seller start --shutdown-drain-timeout-ms` or the SDK's `shutdownDrainTimeoutMs` option. + - Desktop telemetry's `app_connect` / `app_disconnect` user actions now carry which app was connected, as an `app` property drawn from a fixed local taxonomy (the packaged profile names plus the Telegram bot); user-added custom apps report as `custom`, so raw app names never leave the device. Connect events are also attributed to the specific app being connected rather than firing on every profile-set restart (profile switches and custom-app removals no longer emit spurious `app_connect`). - Desktop now finds T3 Code installed under any release channel — the launch-target lookup previously only checked for "T3 Code (Alpha)", so stable/Beta/Nightly installs got no app icon, no default "Open with" application, and no restart action. All channel variants are now probed (stable first), and the T3 Code rows fall back to the official t3.codes icon instead of the generic mark when the app isn't installed. - Desktop's Home screen keeps the "Use AntSeed on your favorite app" pills visible after connecting a tool — previously connecting anything hid the whole list. The pitch now disappears only once the user has chats, and an already-connected app's pill shows as connected (green dot, green-tinted border) and opens the Apps page instead of reconnecting. diff --git a/apps/cli/README.md b/apps/cli/README.md index f2f484127..0a9ac4c7b 100644 --- a/apps/cli/README.md +++ b/apps/cli/README.md @@ -272,6 +272,22 @@ antseed buyer start --disable-metadata-v2-services For production sellers, prefer a dedicated Base JSON-RPC endpoint over public defaults. You can set it durably with `payments.crypto.rpcUrl`, at runtime with `ANTSEED_BASE_RPC_URL`, or for one run with `antseed seller start --base-rpc-url `. +### Graceful seller shutdown + +On SIGINT or SIGTERM, the seller stops advertising and rejects new requests with +503 while allowing running requests, final payment authorizations, and outgoing +transport buffers to drain. The default drain budget is 60 seconds: + +```bash +antseed seller start --shutdown-drain-timeout-ms 120000 +``` + +The SDK equivalent is `new AntseedNode({ role: 'seller', shutdownDrainTimeoutMs: 120000 })`. +Set the timeout to `0` to skip waiting. Requests still running at the deadline are +disconnected; forced termination (such as SIGKILL) cannot be drained. A service +manager's termination timeout should exceed the drain budget and leave time for +settlement and cleanup. + ### Metadata v12 rollout This release announces metadata v12. Buyers supporting only older metadata versions drop v12 sellers from discovery, while updated buyers continue accepting older v10/v11 sellers. Upgrade buyer CLIs and desktop apps before upgrading sellers. Removing capability or unit-billing fields does not downgrade the metadata version; rollback requires running the older seller binary. diff --git a/apps/cli/src/cli/commands/seller/start.ts b/apps/cli/src/cli/commands/seller/start.ts index 017f24e90..3cb5fc7db 100644 --- a/apps/cli/src/cli/commands/seller/start.ts +++ b/apps/cli/src/cli/commands/seller/start.ts @@ -377,6 +377,11 @@ export function registerSellerStartCommand(sellerCmd: Command): void { .option('-r, --reserve ', 'runtime-only reserve floor override (does not write config file)', parseFloat) .option('--input-usd-per-million ', 'runtime-only input pricing override in USD per 1M tokens', parseFloat) .option('--output-usd-per-million ', 'runtime-only output pricing override in USD per 1M tokens', parseFloat) + .option('--shutdown-drain-timeout-ms ', 'maximum wait for active requests and final payment authorizations on shutdown', (value: string) => { + const timeout = Number(value) + if (!Number.isSafeInteger(timeout) || timeout < 0 || timeout > 2_147_483_647 || value.trim() === '') throw new Error('Shutdown drain timeout must be an integer between 0 and 2147483647') + return timeout + }, 60_000) .option('--dht-port ', 'UDP port for DHT (default: 6881)', parseInt) .option('--signaling-port ', 'TCP port for P2P signaling (default: 6882)', parseInt) .option('--min-settle-delta ', 'minimum unsettled delta (USDC decimal, e.g. 0.002) before idle settle submits a tx') @@ -692,6 +697,7 @@ export function registerSellerStartCommand(sellerCmd: Command): void { const node = new AntseedNode({ role: 'seller', + shutdownDrainTimeoutMs: options.shutdownDrainTimeoutMs as number, displayName: config.identity.displayName, ...(config.seller.publicAddress ? { publicAddress: config.seller.publicAddress } : {}), ...(effectiveSellerConfig.verifications ? { verifications: effectiveSellerConfig.verifications } : {}), diff --git a/packages/node/src/discovery/announcer.ts b/packages/node/src/discovery/announcer.ts index 2aa614aab..d1b9a1f94 100644 --- a/packages/node/src/discovery/announcer.ts +++ b/packages/node/src/discovery/announcer.ts @@ -123,6 +123,7 @@ export class PeerAnnouncer { async announce(): Promise { this._latestMetadata = await this._buildSignedMetadata(true); + if (this.stopped) return; const failures = await this._announceTopics(); if (failures > 0) { diff --git a/packages/node/src/node.ts b/packages/node/src/node.ts index 94c89e9a4..94a1aebf8 100644 --- a/packages/node/src/node.ts +++ b/packages/node/src/node.ts @@ -221,6 +221,7 @@ export interface NodeVerificationConfig { export interface NodeConfig { role: 'seller' | 'buyer'; + shutdownDrainTimeoutMs?: number; displayName?: string; /** Publicly reachable seller address override ("host:port") announced in metadata. */ publicAddress?: string; @@ -330,6 +331,7 @@ export class AntseedNode extends EventEmitter { private _provers: Prover[] = []; private _router: Router | null = null; private _started = false; + private _stopPromise: Promise | null = null; private _announcer: PeerAnnouncer | null = null; /** Set while advertising is paused (e.g. seller wallet out of gas). */ private _advertisingPausedReason: string | null = null; @@ -394,6 +396,9 @@ export class AntseedNode extends EventEmitter { constructor(config: NodeConfig) { super(); + if (config.shutdownDrainTimeoutMs !== undefined && (!Number.isSafeInteger(config.shutdownDrainTimeoutMs) || config.shutdownDrainTimeoutMs < 0 || config.shutdownDrainTimeoutMs > 2_147_483_647)) { + throw new Error('shutdownDrainTimeoutMs must be an integer between 0 and 2147483647'); + } this._config = config; } @@ -434,6 +439,7 @@ export class AntseedNode extends EventEmitter { /** Resume DHT announcements after `pauseAdvertising` (announces immediately). */ resumeAdvertising(): void { + if (this._stopPromise) return; if (this._advertisingPausedReason === null) return; this._advertisingPausedReason = null; this._announcer?.startPeriodicAnnounce(); @@ -573,11 +579,44 @@ export class AntseedNode extends EventEmitter { this.emit("started"); } - async stop(): Promise { + stop(): Promise { + if (!this._started) return Promise.resolve(); + if (!this._stopPromise) { + this._stopPromise = this._stop().finally(() => { this._stopPromise = null; }); + } + return this._stopPromise; + } + + private async _stop(): Promise { if (!this._started) { return; } + if (this._sellerHandler) { + const timeoutMs = this._config.shutdownDrainTimeoutMs ?? 60_000; + const deadline = Date.now() + timeoutMs; + this._sellerPaymentManager?.beginDrain(); + const draining = this._sellerHandler.drain(timeoutMs); + void this.pauseAdvertising('shutting-down').catch((err: unknown) => { + debugWarn(`[Node] Could not refresh shutdown metadata: ${String(err)}`); + }); + const completed = await draining; + if (!completed) { + debugWarn(`[Node] Seller drain timed out after ${timeoutMs}ms; closing remaining requests`); + } else { + await this._sellerPaymentManager?.drainPendingPayments(Math.max(0, deadline - Date.now())); + } + await Promise.all([...this._connectionManager?.connections.values() ?? []].map(async (connection) => { + try { + if (!await connection.drainOutgoing(Math.max(0, deadline - Date.now()))) { + debugWarn('[Node] Shutdown deadline reached before outgoing transport buffers drained'); + } + } catch (err) { + debugWarn(`[Node] Could not drain outgoing transport: ${String(err)}`); + } + })); + } + // Give in-transit NeedAuth messages time to arrive on the DataChannel, // then wait for their handlers to finish. This ensures the seller has a // valid SpendingAuth for settlement before we close the connection. diff --git a/packages/node/src/p2p/connection-manager.ts b/packages/node/src/p2p/connection-manager.ts index ab12006a4..477160629 100644 --- a/packages/node/src/p2p/connection-manager.ts +++ b/packages/node/src/p2p/connection-manager.ts @@ -295,6 +295,19 @@ export class PeerConnection extends EventEmitter { } } + async drainOutgoing(timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (true) { + const buffered = this._dataChannel?.isOpen() + ? this._dataChannel.bufferedAmount() + : this._rawSocket?.writableLength ?? 0; + if (buffered === 0) return true; + const remaining = deadline - Date.now(); + if (remaining <= 0) return false; + await new Promise((resolve) => setTimeout(resolve, Math.min(10, remaining))); + } + } + /** Send a message through the active transport. */ send(data: Uint8Array): void { if (this._state !== ConnectionState.Open && this._state !== ConnectionState.Authenticated) { diff --git a/packages/node/src/payments/seller-payment-manager.ts b/packages/node/src/payments/seller-payment-manager.ts index 14df013f8..a29bff6a8 100644 --- a/packages/node/src/payments/seller-payment-manager.ts +++ b/packages/node/src/payments/seller-payment-manager.ts @@ -87,6 +87,7 @@ interface LatestAuth { * The seller tracks spending locally and settles/closes via the contract at session end. */ export class SellerPaymentManager { + private _draining = false; private readonly _signer: AbstractSigner; private readonly _channelsClient: ChannelsClient; private readonly _config: SellerPaymentConfig; @@ -501,6 +502,7 @@ export class SellerPaymentManager { const channelsDomain = makeChannelsDomain(this._config.chainId, channelsAddr); if (existingCumulative === undefined) { + if (this._draining) return 'rejected'; const hasReserveFields = payload.reserveSalt != null || payload.reserveMaxAmount != null || payload.reserveDeadline != null; @@ -1524,6 +1526,32 @@ export class SellerPaymentManager { }; } + async drainPendingPayments(timeoutMs: number): Promise { + let timer: ReturnType | undefined; + try { + const complete = await Promise.race([ + (async () => { + const deadline = Date.now() + timeoutMs; + await Promise.all([...this._buyerLocks.values()]); + const results = await Promise.all(this._channelStore.getActiveChannels(CHANNEL_ROLE.SELLER).map(async (channel) => { + const reached = await this.awaitAcceptedAtLeast(channel.sessionId, this.getCumulativeSpend(channel.sessionId), Math.max(0, deadline - Date.now())); + await this.waitForPendingAuths(channel.peerId); + return reached; + })); + return results.every(Boolean); + })(), + new Promise((resolve) => { timer = setTimeout(() => resolve(false), timeoutMs); }), + ]); + if (!complete) debugWarn('[SellerPayment] Shutdown deadline reached before all final spending authorizations arrived'); + } finally { + clearTimeout(timer); + } + } + + beginDrain(): void { + this._draining = true; + } + // ── Buyer-requested cooperative close ───────────────────────── /** diff --git a/packages/node/src/seller-request-handler.ts b/packages/node/src/seller-request-handler.ts index 3e449942a..834ad6433 100644 --- a/packages/node/src/seller-request-handler.ts +++ b/packages/node/src/seller-request-handler.ts @@ -84,11 +84,29 @@ export class SellerRequestHandler { private readonly _providerLoadCounts = new Map(); private readonly _attestRateWindows = new Map(); private _metadataRefreshTimer: ReturnType | null = null; + private _draining = false; + private _aborted = false; + private readonly _pendingRequests = new Set>(); constructor(deps: SellerRequestHandlerDeps) { this._deps = deps; } + async drain(timeoutMs: number): Promise { + this._draining = true; + let timer: ReturnType | undefined; + try { + const completed = await Promise.race([ + Promise.allSettled([...this._pendingRequests]).then(() => true), + new Promise((resolve) => { timer = setTimeout(() => resolve(false), timeoutMs); }), + ]); + this._aborted = !completed; + return completed; + } finally { + clearTimeout(timer); + } + } + private _allowAttest(buyerPeerId: string): boolean { const now = Date.now(); const win = this._attestRateWindows.get(buyerPeerId); @@ -123,7 +141,7 @@ export class SellerRequestHandler { maxUploadBodyBytes: this._deps.maxUploadBodyBytes, }); - mux.onProxyRequest(async (request: SerializedHttpRequest) => { + const processRequest = async (request: SerializedHttpRequest): Promise => { debugLog(`[SellerHandler] Received request: ${request.method} ${request.path} (reqId=${request.requestId.slice(0, 8)})`); // Handle /v1/models locally — free metadata endpoint, no payment required. @@ -183,6 +201,7 @@ export class SellerRequestHandler { headers: request.headers, body: request.body, }); + if (this._aborted) return; mux.sendProxyResponse({ requestId: request.requestId, statusCode: resp.statusCode, @@ -190,6 +209,7 @@ export class SellerRequestHandler { body: resp.body, }); } catch (err) { + if (this._aborted) return; const message = err instanceof Error ? err.message : String(err); mux.sendProxyResponse({ requestId: request.requestId, @@ -293,6 +313,7 @@ export class SellerRequestHandler { // that has queued later auths behind its per-buyer mutex) so we don't // 402 against a stale accepted cumulative. await spm.waitForPendingAuths(buyerPeerId); + if (this._aborted) return; // Re-read after the await — the session may have been evicted (timeout // checker, disconnect) while the on-chain top-up was confirming. const session = spm.getChannelByPeer(buyerPeerId); @@ -458,6 +479,7 @@ export class SellerRequestHandler { ...request, headers: { ...request.headers }, }; + if (this._aborted) return; // Track active seller session at request start this._deps.sessionTracker?.getOrCreateSession(buyerPeerId, provider.name); @@ -500,6 +522,7 @@ export class SellerRequestHandler { try { const response = await this._executeRequest(provider, request, { onResponseStart: (streamResponseStart) => { + if (this._aborted) return; streamedResponseStarted = true; responseStartedAt = Date.now(); statusCode = streamResponseStart.statusCode; @@ -508,6 +531,7 @@ export class SellerRequestHandler { mux.sendProxyResponse(streamResponseStart); }, onResponseChunk: (chunk) => { + if (this._aborted) return; if (!streamedResponseStarted) return; // Hold the done chunk — send it after usage is parsed so we can append cost trailer if (chunk.done) { @@ -517,6 +541,7 @@ export class SellerRequestHandler { mux.sendProxyChunk(chunk); }, }); + if (this._aborted) return; statusCode = response.statusCode; responseBody = response.body ?? new Uint8Array(0); responseForAuth = response; @@ -547,6 +572,7 @@ export class SellerRequestHandler { }); } } catch (err) { + if (this._aborted) return; const message = err instanceof Error ? err.message : "Internal error"; debugWarn(`[SellerHandler] Provider exception: provider="${provider.name}" model="${requestedModel}" buyer=${buyerPeerId.slice(0, 12)}... (${Date.now() - startTime}ms) ${message}`); responseBody = new TextEncoder().encode(message); @@ -611,6 +637,7 @@ export class SellerRequestHandler { providerUsage: responseUsage, }); } + if (this._aborted) return; // Record spend and send NeedAuth with cost data after every request. // The buyer validates the cost independently and responds with SpendingAuth. @@ -678,6 +705,25 @@ export class SellerRequestHandler { this.adjustProviderLoad(provider.name, -1); if (isBillable) spm!.endBillableRequest(buyerPeerId); } + }; + + mux.onProxyRequest(async (request: SerializedHttpRequest) => { + if (this._draining) { + mux.sendProxyResponse({ + requestId: request.requestId, + statusCode: 503, + headers: { 'content-type': 'application/json', 'retry-after': '2' }, + body: new TextEncoder().encode(JSON.stringify({ error: 'seller_shutting_down' })), + }); + return; + } + const operation = processRequest(request); + this._pendingRequests.add(operation); + try { + await operation; + } finally { + this._pendingRequests.delete(operation); + } }); return { mux }; @@ -974,7 +1020,7 @@ export class SellerRequestHandler { } private _scheduleMetadataRefresh(): void { - if (!this._deps.announcer || this._metadataRefreshTimer) { + if (this._draining || !this._deps.announcer || this._metadataRefreshTimer) { return; } diff --git a/packages/node/tests/seller-graceful-shutdown.test.ts b/packages/node/tests/seller-graceful-shutdown.test.ts new file mode 100644 index 000000000..81bf35252 --- /dev/null +++ b/packages/node/tests/seller-graceful-shutdown.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, it, vi } from 'vitest'; +import { SellerRequestHandler } from '../src/seller-request-handler.js'; +import { AntseedNode } from '../src/node.js'; +import type { Provider, ProviderStreamCallbacks } from '../src/interfaces/seller-provider.js'; +import { encodeHttpRequest, decodeHttpResponse } from '../src/proxy/request-codec.js'; +import { decodeFrame } from '../src/p2p/message-protocol.js'; +import { MessageType } from '../src/types/protocol.js'; +import { PeerConnection } from '../src/p2p/connection-manager.js'; + +function harness() { + let finish!: () => void; + let callbacks!: ProviderStreamCallbacks; + const body = new TextEncoder().encode(JSON.stringify({ usage: { prompt_tokens: 2, completion_tokens: 3 } })); + const provider: Provider = { + name: 'openai', services: ['model'], maxConcurrency: 10, + pricing: { defaults: { inputUsdPerMillion: 1, outputUsdPerMillion: 2 } }, + getCapacity: () => ({ current: 0, max: 10 }), + handleRequest: vi.fn(), + handleRequestStream: vi.fn(async (request, streamCallbacks) => { + callbacks = streamCallbacks; + callbacks.onResponseStart({ requestId: request.requestId, statusCode: 200, headers: { 'content-type': 'text/event-stream' }, body: new Uint8Array() }); + await new Promise((resolve) => { finish = resolve; }); + callbacks.onResponseChunk({ requestId: request.requestId, data: body, done: true }); + return { requestId: request.requestId, statusCode: 200, headers: {}, body }; + }), + }; + const frames: Uint8Array[] = []; + const recordSpend = vi.fn(); + const sendNeedAuth = vi.fn(); + const payments = { + hasSession: () => true, getChannelByPeer: () => ({ sessionId: 'channel', authMax: '1000000' }), + getAcceptedCumulative: () => 0n, getCumulativeSpend: () => 0n, getEffectiveReserveMax: () => 1_000_000n, + waitForPendingAuths: async () => {}, isChannelBlocked: () => false, hasClosingChannel: () => false, + beginBillableRequest: vi.fn(), endBillableRequest: vi.fn(), recordSpend, + }; + const handler = new SellerRequestHandler({ + identity: { peerId: 'a'.repeat(40) } as any, providers: [provider], + sellerPaymentManager: payments as any, channelsClient: {} as any, sessionTracker: null, announcer: null, emit: () => false, + }); + const { mux } = handler.handleConnection({ send: (frame: Uint8Array) => frames.push(frame), hasRemoteCapability: () => false } as any, + 'b'.repeat(40), { sendNeedAuth } as any, {} as any); + const send = (requestId: string) => mux.handleFrame({ + type: MessageType.HttpRequest, messageId: 1, + payload: encodeHttpRequest({ requestId, method: 'POST', path: '/v1/chat/completions', headers: { 'content-type': 'application/json' }, body: new TextEncoder().encode(JSON.stringify({ model: 'model', stream: true })) }), + }); + return { provider, handler, send, frames, recordSpend, sendNeedAuth, finish: () => finish(), started: () => Boolean(finish) }; +} + +describe('seller graceful shutdown', () => { + it('finishes an active stream while rejecting new requests during drain', async () => { + const state = harness(); + state.send('running'); + await vi.waitFor(() => expect(state.started()).toBe(true)); + let drained = false; + const draining = state.handler.drain(1000).then((completed) => { drained = completed; }); + state.send('rejected'); + await vi.waitFor(() => expect(state.frames.length).toBe(2)); + expect(decodeHttpResponse(decodeFrame(state.frames[1]!)!.message.payload).statusCode).toBe(503); + expect(drained).toBe(false); + state.finish(); + await draining; + expect(drained).toBe(true); + expect(state.recordSpend).toHaveBeenCalledWith('channel', 8n); + expect(state.sendNeedAuth).toHaveBeenCalledOnce(); + expect(state.provider.handleRequestStream).toHaveBeenCalledOnce(); + expect(state.frames).toHaveLength(3); + }); + + it('times out a stuck provider without allowing late frames or billing', async () => { + const state = harness(); + state.send('stuck'); + await vi.waitFor(() => expect(state.started()).toBe(true)); + expect(await state.handler.drain(5)).toBe(false); + const count = state.frames.length; + state.finish(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(state.frames).toHaveLength(count); + expect(state.recordSpend).not.toHaveBeenCalled(); + }); + + it('rejects invalid shutdown deadlines', () => { + for (const timeout of [-1, NaN, Infinity, 1.5]) { + expect(() => new AntseedNode({ role: 'seller', shutdownDrainTimeoutMs: timeout })).toThrow(); + } + }); + + it('orders node shutdown after requests, payment authorizations, and transport drain', async () => { + const node = new AntseedNode({ role: 'seller', shutdownDrainTimeoutMs: 1000 }); + const events: string[] = []; + let finishRequest!: () => void; + Object.assign(node, { + _started: true, + _sellerHandler: { + drain: async () => { + events.push('requests'); + await new Promise((resolve) => { finishRequest = resolve; }); + return true; + }, + clearMetadataRefreshTimer: () => {}, + }, + _sellerPaymentManager: { beginDrain: () => events.push('reject-reserves'), drainPendingPayments: async () => { events.push('payments'); } }, + _announcer: { stopPeriodicAnnounce: () => events.push('stop-advertising'), refreshMetadata: async () => {} }, + _sessionTracker: { finalizeAllSessions: async () => { events.push('finalize'); }, clearTimers: () => {} }, + _connectionManager: { + connections: new Map([['buyer', { drainOutgoing: async () => { events.push('transport'); return true; } }]]), + closeAll: () => events.push('close'), + }, + }); + const stopping = node.stop(); + expect(node.stop()).toBe(stopping); + expect(events).not.toContain('close'); + expect(events).toContain('stop-advertising'); + finishRequest(); + await stopping; + expect(events.indexOf('payments')).toBeLessThan(events.indexOf('transport')); + expect(events.indexOf('transport')).toBeLessThan(events.indexOf('finalize')); + expect(events.indexOf('finalize')).toBeLessThan(events.indexOf('close')); + }); + + it('waits for outgoing bytes rather than just provider completion', async () => { + const connection = new PeerConnection({ remotePeerId: 'b'.repeat(40) as any, isInitiator: false }); + let buffered = 100; + connection.attachDataChannel({ isOpen: () => true, bufferedAmount: () => buffered, onOpen: () => {}, onClosed: () => {}, onError: () => {}, onMessage: () => {} } as any); + expect(await connection.drainOutgoing(0)).toBe(false); + const draining = connection.drainOutgoing(1000); + buffered = 0; + expect(await draining).toBe(true); + }); + +}); diff --git a/packages/node/tests/seller-payment-manager.test.ts b/packages/node/tests/seller-payment-manager.test.ts index 6e219b265..4e081209b 100644 --- a/packages/node/tests/seller-payment-manager.test.ts +++ b/packages/node/tests/seller-payment-manager.test.ts @@ -189,6 +189,24 @@ describe('SellerPaymentManager', () => { expect(manager.hasSession(buyerIdentity.peerId)).toBe(true); }); + it('waits for final spending authorization during drain, but respects the deadline', async () => { + const channelId = makeChannelId(72); + const payload = await buildSpendingAuth(buyerIdentity, sellerIdentity, channelId, { isReserve: true }); + await manager.handleSpendingAuth(buyerIdentity.peerId, payload, mux); + manager.recordSpend(channelId, 100n); + let drained = false; + const draining = manager.drainPendingPayments(1000).then(() => { drained = true; }); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(drained).toBe(false); + const auth = await buildSpendingAuth(buyerIdentity, sellerIdentity, channelId, { cumulativeAmount: 100n }); + await manager.handleSpendingAuth(buyerIdentity.peerId, auth, mux); + await draining; + expect(drained).toBe(true); + manager.recordSpend(channelId, 1n); + await manager.drainPendingPayments(5); + expect(manager.getCumulativeSpend(channelId)).toBe(101n); + }); + it('retries overlapping initial reserves after delegated account transaction backpressure', async () => { vi.useFakeTimers(); try {