diff --git a/PROTOCOL.md b/PROTOCOL.md index 453b50e2..fc23b77d 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -669,6 +669,23 @@ When sending messages to the other side, the session associated with the connect When receiving messages, the transport MUST ensure that the only messages it validates are those that have a `seq` that is exactly equal to the `ack` of the session. This ensures exactly-once delivery semantics. +#### Optional receiver-driven acknowledgements + +A receiver MAY send an explicit ACK after a configured number of inbound +non-ACK messages or a delay: `receiverAck: { maxMessages, maxDelayMs }`. +This is disabled by default. Piggybacked ACKs cancel pending batches; ACK +frames do not schedule batches. These are existing heartbeat ACK frames, so +either peer can opt in without negotiation or a protocol version change. +Existing active/passive heartbeat behavior and reconnect guarantees remain. + +#### Optional per-stream advisory backpressure + +`streamSendBufferHighWaterMark` makes router writers use their own unacknowledged +message count instead of the shared session watermark. It is disabled by default. +ACKs release slots; reconnect preserves pending waiters and close releases them. +Writes remain advisory and accepted. This is not a memory cap: aggregate occupancy +can exceed the session watermark. Unscoped accessors retain session behavior. + #### On disconnect Handling close events is detailed in the 'Creating Connections and Sessions' section above. diff --git a/__tests__/protobufBackpressure.test.ts b/__tests__/protobufBackpressure.test.ts new file mode 100644 index 00000000..f6664721 --- /dev/null +++ b/__tests__/protobufBackpressure.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, test, vi } from 'vitest'; +import { + createClient, + createProtoService, + createServer, + Ok, + ProtoCodec, +} from '../protobuf'; +import { cleanupTransports, waitFor } from '../testUtil/fixtures/cleanup'; +import { TestService } from '../testUtil/fixtures/protobuf'; +import { transports } from '../testUtil/fixtures/transports'; +import { isAck } from '../transport/message'; + +describe('protobuf stream isolation', () => { + test('request and response writers use their own stream watermarks', async () => { + vi.useRealTimers(); + const options = { + codec: ProtoCodec, + sendBufferHighWaterMark: 4, + streamSendBufferHighWaterMark: 4, + receiverAck: { maxMessages: 3, maxDelayMs: 20 }, + heartbeatIntervalMs: 30_000, + sessionDisconnectGraceMs: 90_000, + }; + const setup = await transports[0].setup({ + client: options, + server: options, + }); + const clientTransport = setup.getClientTransport('client'); + const serverTransport = setup.getServerTransport(); + const responsePressure: Array> = []; + const service = createProtoService().define(TestService, { + countUp: ({ request, resWritable }) => { + const results: Array = []; + for (let value = 1; value <= request.limit; value++) { + results.push(resWritable.write(Ok({ value }))); + } + responsePressure.push(results); + resWritable.close(); + }, + sum: async ({ reqReadable }) => { + let total = 0; + for await (const result of reqReadable) { + if (result.ok) total += result.payload.value; + } + + return Ok({ total }); + }, + }); + createServer(serverTransport, [service]); + const client = createClient( + TestService, + clientTransport, + serverTransport.clientId, + ); + try { + const heavyRequest = client.sum(); + for (let i = 0; i < 6; i++) heavyRequest.reqWritable.write({ value: i }); + const quietRequest = client.sum(); + expect(quietRequest.reqWritable.write({ value: 1 })).toBe(true); + heavyRequest.reqWritable.close(); + quietRequest.reqWritable.close(); + expect(await heavyRequest.finalize()).toMatchObject({ + ok: true, + payload: { total: 15 }, + }); + expect(await quietRequest.finalize()).toMatchObject({ + ok: true, + payload: { total: 1 }, + }); + + const heavyResponse = client.countUp({ limit: 6 }); + const quietResponse = client.countUp({ limit: 1 }); + expect(await heavyResponse.collect()).toHaveLength(6); + expect(await quietResponse.collect()).toHaveLength(1); + expect(responsePressure[0][5]).toBe(false); + expect(responsePressure[1][0]).toBe(true); + await waitFor(() => { + const session = [...serverTransport.sessions.values()][0]; + expect( + session.sendBuffer.filter((msg) => !isAck(msg.msg.controlFlags)), + ).toHaveLength(0); + }); + } finally { + await cleanupTransports([clientTransport, serverTransport]); + await setup.cleanup(); + vi.useFakeTimers({ shouldAdvanceTime: true }); + } + }); +}); diff --git a/__tests__/receiverAck.test.ts b/__tests__/receiverAck.test.ts new file mode 100644 index 00000000..bc71d68c --- /dev/null +++ b/__tests__/receiverAck.test.ts @@ -0,0 +1,62 @@ +import { expect, test, vi } from 'vitest'; +import { createClient } from '../router/client'; +import { createServer } from '../router/server'; +import { closeAllConnections, readNextResult } from '../testUtil'; +import { cleanupTransports } from '../testUtil/fixtures/cleanup'; +import { TestServiceSchema } from '../testUtil/fixtures/services'; +import { transports } from '../testUtil/fixtures/transports'; + +test.each(['client', 'server'])( + 'stream isolation and reconnect work with only the %s opting in', + async (enabled) => { + vi.useRealTimers(); + const options = { + streamSendBufferHighWaterMark: 4, + receiverAck: { maxMessages: 3, maxDelayMs: 20 }, + }; + const setup = await transports[0].setup({ + client: enabled === 'client' ? options : {}, + server: enabled === 'server' ? options : {}, + }); + const clientTransport = setup.getClientTransport('client'); + const serverTransport = setup.getServerTransport(); + const services = { test: TestServiceSchema }; + createServer(serverTransport, services); + const client = createClient( + clientTransport, + serverTransport.clientId, + ); + try { + const heavy = client.test.echo.stream({}); + heavy.reqWritable.write({ msg: 'establish', ignore: false }); + await readNextResult(heavy.resReadable); + closeAllConnections(clientTransport); + const writes = Array.from({ length: 6 }, (_, i) => + heavy.reqWritable.write({ msg: String(i), ignore: false }), + ); + if (enabled === 'client') expect(writes[5]).toBe(false); + const ready = heavy.reqWritable.waitForWriteReady(); + const light = client.test.echo.stream({}); + expect(light.reqWritable.write({ msg: 'light', ignore: false })).toBe( + true, + ); + expect(await readNextResult(light.resReadable)).toMatchObject({ + ok: true, + payload: { response: 'light' }, + }); + await ready; + for (let i = 0; i < 6; i++) { + expect(await readNextResult(heavy.resReadable)).toMatchObject({ + ok: true, + payload: { response: String(i) }, + }); + } + heavy.reqWritable.close(); + light.reqWritable.close(); + } finally { + await cleanupTransports([clientTransport, serverTransport]); + await setup.cleanup(); + vi.useFakeTimers({ shouldAdvanceTime: true }); + } + }, +); diff --git a/protobuf/client.ts b/protobuf/client.ts index 00947531..91fcc74c 100644 --- a/protobuf/client.ts +++ b/protobuf/client.ts @@ -274,12 +274,12 @@ function startMethodCall< serverId, session.id, ); + const streamId = generateId(); const sessionBackpressure = transport.getSessionBackpressure( serverId, session.id, + streamId, ); - - const streamId = generateId(); const { span, ctx } = createProcTelemetryInfo( transport.tracer, session, diff --git a/protobuf/server.ts b/protobuf/server.ts index e21af017..898f50c5 100644 --- a/protobuf/server.ts +++ b/protobuf/server.ts @@ -371,6 +371,7 @@ class ProtobufServer< const sessionBackpressure = this.transport.getSessionBackpressure( from, sessionId, + streamId, ); const deferredCleanups: Array<() => void | Promise> = []; diff --git a/router/client.ts b/router/client.ts index a14d4454..e64966cc 100644 --- a/router/client.ts +++ b/router/client.ts @@ -342,13 +342,13 @@ function handleProc( serverId, session.id, ); + const procClosesWithInit = procType === 'rpc' || procType === 'subscription'; + const streamId = generateId(); const sessionBackpressure = transport.getSessionBackpressure( serverId, session.id, + streamId, ); - - const procClosesWithInit = procType === 'rpc' || procType === 'subscription'; - const streamId = generateId(); const { span, ctx } = createProcTelemetryInfo( transport.tracer, session, diff --git a/router/server.ts b/router/server.ts index b8ba4616..f29f9552 100644 --- a/router/server.ts +++ b/router/server.ts @@ -464,6 +464,7 @@ class RiverServer< const sessionBackpressure = this.transport.getSessionBackpressure( from, sessionId, + streamId, ); const cancelStream = ( diff --git a/router/streams.ts b/router/streams.ts index f81ed27f..93cd6bd5 100644 --- a/router/streams.ts +++ b/router/streams.ts @@ -92,7 +92,7 @@ export interface Writable { /** * {@link write} writes a value to the pipe. An error is thrown if writing to a closed {@link Writable}. * - * Returns `false` if the underlying session's send buffer is at or above + * Returns `false` if the underlying session or stream send buffer is at or above * its high-water mark, signalling that the producer should stop writing * until {@link waitForWriteReady} resolves. This is purely advisory — * the value is still buffered and will be delivered, exactly like @@ -100,7 +100,7 @@ export interface Writable { */ write(value: T): boolean; /** - * {@link waitForWriteReady} resolves once the underlying session's send + * {@link waitForWriteReady} resolves once the underlying session or stream send * buffer has drained back below its high-water mark and it is productive * to call {@link write} again. * diff --git a/transport/sessionStateMachine/SessionConnected.ts b/transport/sessionStateMachine/SessionConnected.ts index f0ce6a8b..7c495106 100644 --- a/transport/sessionStateMachine/SessionConnected.ts +++ b/transport/sessionStateMachine/SessionConnected.ts @@ -60,15 +60,27 @@ export class SessionConnected< private isActivelyHeartbeating = false; private rehandshakeTimer?: ReturnType | undefined; private credentialExpiry?: number | undefined; + private receiverAckTimer?: ReturnType; + private pendingReceiverAckMessages = 0; updateBookkeeping(ack: number, seq: number) { - this.sendBuffer = this.sendBuffer.filter((unacked) => unacked.seq >= ack); + this.sendBuffer = this.sendBuffer.filter((msg) => { + if (msg.seq >= ack) return true; + const stream = this.streamSendBuffers.get(msg.msg.streamId); + if (stream) { + stream.size--; + if (!this.isSendBufferFull(msg.msg.streamId)) { + stream.waiter?.resolve(); + stream.waiter = undefined; + } + if (stream.size === 0) this.streamSendBuffers.delete(msg.msg.streamId); + } + + return false; + }); this.ack = seq + 1; this.lastInboundAt = Date.now(); - - if (this.sendBuffer.length < this.options.sendBufferHighWaterMark) { - this.notifySendBufferDrain(); - } + if (!this.isSendBufferFull()) this.notifySendBufferDrain(); } private assertSendOrdering(encodedMsg: EncodedTransportMessage) { @@ -91,8 +103,11 @@ export class SessionConnected< const encodedMsg = encodeResult.value; this.assertSendOrdering(encodedMsg); - this.sendBuffer.push(encodedMsg); + this.bufferMessage(encodedMsg); + // A synchronous connection can deliver more inbound messages during send. + // Only clear the batch covered by the ACK we just encoded. + this.clearReceiverAck(); const sent = this.conn.send(encodedMsg.data); if (!sent) { const reason = 'failed to send message'; @@ -218,6 +233,32 @@ export class SessionConnected< this.send(heartbeat); } + private clearReceiverAck(): void { + if (this.receiverAckTimer !== undefined) { + clearTimeout(this.receiverAckTimer); + this.receiverAckTimer = undefined; + } + this.pendingReceiverAckMessages = 0; + } + + private scheduleReceiverAck(): void { + const options = this.options.receiverAck; + if (!options) { + return; + } + + this.pendingReceiverAckMessages++; + if (this.pendingReceiverAckMessages >= options.maxMessages) { + this.sendHeartbeat(); + + return; + } + this.receiverAckTimer ??= setTimeout(() => { + this.receiverAckTimer = undefined; + this.sendHeartbeat(); + }, options.maxDelayMs); + } + /** * Schedules the next proactive re-handshake from the credential's expiry. The * server calls this after each (re)validation, mirroring {@link startActiveHeartbeat}: @@ -351,6 +392,7 @@ export class SessionConnected< // dispatch directly if its not an explicit ack if (!isAck(parsedMsg.controlFlags)) { + this.scheduleReceiverAck(); // re-handshake frames ride a reserved stream and are consumed by the // transport, never surfaced to the router (same as the acks handled below) if (parsedMsg.streamId === RehandshakeStreamId) { @@ -390,6 +432,7 @@ export class SessionConnected< this.clearHeartbeatWatchdog(); this.clearRehandshakeTimer(); + this.clearReceiverAck(); } _handleClose(): void { diff --git a/transport/sessionStateMachine/common.ts b/transport/sessionStateMachine/common.ts index 96a06607..927950c8 100644 --- a/transport/sessionStateMachine/common.ts +++ b/transport/sessionStateMachine/common.ts @@ -146,6 +146,10 @@ export interface SessionOptions { * session close. */ sendBufferHighWaterMark: number; + /** Opt-in per-stream advisory watermark instead of the session watermark. */ + streamSendBufferHighWaterMark?: number; + /** Opt-in ACK batching for inbound non-ACK messages. */ + receiverAck?: { maxMessages: number; maxDelayMs: number }; /** * The codec to use for encoding/decoding messages over the wire */ @@ -203,6 +207,11 @@ export interface IdentifiedSessionListeners { ) => void; } +interface StreamSendBuffer { + size: number; + waiter?: PromiseWithResolvers; +} + // all sessions where we know the other side's client id export interface IdentifiedSessionProps extends CommonSessionProps { id: SessionId; @@ -212,6 +221,7 @@ export interface IdentifiedSessionProps extends CommonSessionProps { seqSent: number; sendBuffer: Array; sendBufferDrainWaiter: PromiseWithResolvers | undefined; + streamSendBuffers?: Map; telemetry: TelemetryInfo; protocolVersion: ProtocolVersion; listeners: IdentifiedSessionListeners; @@ -246,6 +256,7 @@ export abstract class IdentifiedSession extends CommonSession { * Carried across session state transitions alongside {@link sendBuffer}. */ sendBufferDrainWaiter: PromiseWithResolvers | undefined; + readonly streamSendBuffers: Map; constructor(props: IdentifiedSessionProps) { const { @@ -268,6 +279,18 @@ export abstract class IdentifiedSession extends CommonSession { this.ack = ack; this.sendBuffer = sendBuffer; this.sendBufferDrainWaiter = sendBufferDrainWaiter; + this.streamSendBuffers = + props.streamSendBuffers ?? new Map(); + if ( + !props.streamSendBuffers && + this.options.streamSendBufferHighWaterMark + ) { + for (const { msg } of sendBuffer) { + const stream = this.streamSendBuffers.get(msg.streamId) ?? { size: 0 }; + stream.size++; + this.streamSendBuffers.set(msg.streamId, stream); + } + } this.telemetry = telemetry; this.log = log; this.protocolVersion = protocolVersion; @@ -332,7 +355,17 @@ export abstract class IdentifiedSession extends CommonSession { return this.sendBuffer.length > 0 ? this.sendBuffer[0].seq : this.seq; } - isSendBufferFull(): boolean { + isSendBufferFull(streamId?: string): boolean { + if ( + streamId !== undefined && + this.options.streamSendBufferHighWaterMark !== undefined + ) { + return ( + (this.streamSendBuffers.get(streamId)?.size ?? 0) >= + this.options.streamSendBufferHighWaterMark + ); + } + return this.sendBuffer.length >= this.options.sendBufferHighWaterMark; } @@ -342,11 +375,19 @@ export abstract class IdentifiedSession extends CommonSession { * already is. Also resolves (never rejects) when the session closes so * waiting producers don't hang forever. */ - waitForSendBufferDrain(): Promise { - if (!this.isSendBufferFull()) { + waitForSendBufferDrain(streamId?: string): Promise { + if (!this.isSendBufferFull(streamId)) { return Promise.resolve(); } + const stream = + streamId === undefined ? undefined : this.streamSendBuffers.get(streamId); + if (stream) { + stream.waiter ??= createPromiseWithResolvers(); + + return stream.waiter.promise; + } + this.sendBufferDrainWaiter ??= createPromiseWithResolvers(); return this.sendBufferDrainWaiter.promise; @@ -357,13 +398,24 @@ export abstract class IdentifiedSession extends CommonSession { this.sendBufferDrainWaiter = undefined; } + protected bufferMessage(msg: EncodedTransportMessage): void { + this.sendBuffer.push(msg); + if (this.options.streamSendBufferHighWaterMark !== undefined) { + const stream = this.streamSendBuffers.get(msg.msg.streamId) ?? { + size: 0, + }; + stream.size++; + this.streamSendBuffers.set(msg.msg.streamId, stream); + } + } + send(msg: PartialTransportMessage): SendResult { const encodeResult = this.encodeMsg(msg); if (!encodeResult.ok) { return encodeResult; } - this.sendBuffer.push(encodeResult.value); + this.bufferMessage(encodeResult.value); return { ok: true, @@ -378,6 +430,9 @@ export abstract class IdentifiedSession extends CommonSession { _handleClose(): void { // zero out the buffer this.sendBuffer.length = 0; + for (const stream of this.streamSendBuffers.values()) + stream.waiter?.resolve(); + this.streamSendBuffers.clear(); // wake any producers waiting for drain so they don't hang forever, // they should check isWritable/isSendBufferFull before writing again this.notifySendBufferDrain(); diff --git a/transport/sessionStateMachine/flowControl.test.ts b/transport/sessionStateMachine/flowControl.test.ts new file mode 100644 index 00000000..bbd1d94b --- /dev/null +++ b/transport/sessionStateMachine/flowControl.test.ts @@ -0,0 +1,153 @@ +import { afterEach, describe, expect, test, vi } from 'vitest'; +import { CodecMessageAdapter } from '../../codec'; +import { createSessionTelemetryInfo, getTracer } from '../../tracing'; +import { Connection } from '../connection'; +import { ControlFlags, currentProtocolVersion } from '../message'; +import { defaultTransportOptions, ProvidedTransportOptions } from '../options'; +import { SessionConnected } from './SessionConnected'; + +class RecordingConnection extends Connection { + send = vi.fn((_data: Uint8Array) => true); + close = vi.fn(); +} + +function connected(options: ProvidedTransportOptions = {}) { + const conn = new RecordingConnection(); + const tracer = getTracer(); + const session = new SessionConnected({ + from: 'receiver', + to: 'sender', + id: 'session', + seq: 0, + seqSent: 0, + ack: 0, + sendBuffer: [], + sendBufferDrainWaiter: undefined, + options: { ...defaultTransportOptions, ...options }, + codec: new CodecMessageAdapter(defaultTransportOptions.codec), + protocolVersion: currentProtocolVersion, + tracer, + log: undefined, + telemetry: createSessionTelemetryInfo( + tracer, + 'session', + 'sender', + 'receiver', + ), + conn, + listeners: { + onMessage: vi.fn(), + onRehandshake: vi.fn(), + onConnectionErrored: vi.fn(), + onConnectionClosed: vi.fn(), + onInvalidMessage: vi.fn(), + onMessageSendFailure: vi.fn(), + }, + }); + const receive = (controlFlags = 0) => { + const data = session.codec.toBuffer({ + id: `inbound-${session.ack}`, + from: 'sender', + to: 'receiver', + streamId: controlFlags ? 'heartbeat' : 'data', + controlFlags, + payload: controlFlags ? { type: 'ACK' } : {}, + seq: session.ack, + ack: 0, + }); + if (!data.ok) throw new Error(data.reason); + session.onMessageData(data.value); + }; + + return { session, conn, receive }; +} + +afterEach(() => vi.clearAllTimers()); + +describe('receiver-driven ACKs', () => { + test('count threshold sends a cumulative ACK without waiting for a heartbeat', () => { + const { session, conn, receive } = connected({ + receiverAck: { maxMessages: 3, maxDelayMs: 50 }, + }); + receive(); + receive(); + expect(conn.send).not.toHaveBeenCalled(); + receive(); + expect(conn.send).toHaveBeenCalledTimes(1); + expect(session.codec.fromBuffer(conn.send.mock.calls[0][0])).toMatchObject({ + ok: true, + value: { + ack: 3, + controlFlags: ControlFlags.AckBit, + payload: { type: 'ACK' }, + }, + }); + vi.advanceTimersByTime(50); + expect(conn.send).toHaveBeenCalledTimes(1); + session._handleClose(); + }); + + test('delay is bounded from the first message, not reset by later arrivals', () => { + const { session, conn, receive } = connected({ + receiverAck: { maxMessages: 10, maxDelayMs: 50 }, + }); + receive(); + vi.advanceTimersByTime(40); + receive(); + vi.advanceTimersByTime(10); + expect(conn.send).toHaveBeenCalledTimes(1); + session._handleClose(); + }); + + test('piggybacking cancels a batch and connected-state exit cancels its timer', () => { + const { session, conn, receive } = connected({ + receiverAck: { maxMessages: 10, maxDelayMs: 50 }, + }); + receive(); + session.send({ streamId: 'reply', controlFlags: 0, payload: {} }); + vi.advanceTimersByTime(50); + expect(conn.send).toHaveBeenCalledTimes(1); + receive(); + session._handleStateExit(); + vi.advanceTimersByTime(50); + expect(conn.send).toHaveBeenCalledTimes(1); + session._handleClose(); + }); + + test('ACKs do not arm batches; default mode does not send unsolicited ACKs', () => { + const enabled = connected({ + receiverAck: { maxMessages: 1, maxDelayMs: 50 }, + }); + enabled.session.startActiveHeartbeat(); + enabled.receive(ControlFlags.AckBit); + vi.advanceTimersByTime(50); + expect(enabled.conn.send).not.toHaveBeenCalled(); + enabled.session._handleClose(); + + const disabled = connected(); + disabled.receive(); + vi.advanceTimersByTime(50); + expect(disabled.conn.send).not.toHaveBeenCalled(); + disabled.session._handleClose(); + }); + + test('inbound messages arriving during a synchronous send keep their ACK deadline', () => { + const { session, conn, receive } = connected({ + receiverAck: { maxMessages: 10, maxDelayMs: 50 }, + }); + receive(); + conn.send.mockImplementationOnce(() => { + receive(); + + return true; + }); + session.send({ streamId: 'reply', controlFlags: 0, payload: {} }); + vi.advanceTimersByTime(50); + expect(conn.send).toHaveBeenCalledTimes(2); + expect(session.codec.fromBuffer(conn.send.mock.calls[1][0])).toMatchObject({ + ok: true, + value: { ack: 2 }, + }); + session._handleClose(); + }); +}); diff --git a/transport/sessionStateMachine/transitions.ts b/transport/sessionStateMachine/transitions.ts index 41df4313..c390ef8a 100644 --- a/transport/sessionStateMachine/transitions.ts +++ b/transport/sessionStateMachine/transitions.ts @@ -60,6 +60,7 @@ function inheritSharedSession( seqSent: session.seqSent, sendBuffer: session.sendBuffer, sendBufferDrainWaiter: session.sendBufferDrainWaiter, + streamSendBuffers: session.streamSendBuffers, telemetry: session.telemetry, options: session.options, log: session.log, diff --git a/transport/transport.ts b/transport/transport.ts index 390f5d47..122b89cd 100644 --- a/transport/transport.ts +++ b/transport/transport.ts @@ -118,6 +118,26 @@ export abstract class Transport< providedOptions?: ProvidedTransportOptions, ) { this.options = { ...defaultTransportOptions, ...providedOptions }; + const streamMark = this.options.streamSendBufferHighWaterMark; + if ( + streamMark !== undefined && + (!Number.isSafeInteger(streamMark) || streamMark < 1) + ) { + throw new Error( + 'streamSendBufferHighWaterMark must be a positive integer', + ); + } + const receiverAck = this.options.receiverAck; + if ( + receiverAck && + (!Number.isSafeInteger(receiverAck.maxMessages) || + receiverAck.maxMessages < 1 || + !Number.isFinite(receiverAck.maxDelayMs) || + receiverAck.maxDelayMs <= 0 || + receiverAck.maxDelayMs > 2_147_483_647) + ) { + throw new Error('receiverAck requires positive message and delay limits'); + } this.eventDispatcher = new EventDispatcher< EventTypes, HandshakeFailureCode @@ -414,6 +434,7 @@ export abstract class Transport< getSessionBackpressure( to: TransportClientId, sessionId: SessionId, + streamId?: string, ): SessionBackpressure { const currentSession = () => { const session = this.sessions.get(to); @@ -425,9 +446,10 @@ export abstract class Transport< }; return { - isSendBufferFull: () => currentSession()?.isSendBufferFull() ?? false, + isSendBufferFull: () => + currentSession()?.isSendBufferFull(streamId) ?? false, waitForSendBufferDrain: () => - currentSession()?.waitForSendBufferDrain() ?? Promise.resolve(), + currentSession()?.waitForSendBufferDrain(streamId) ?? Promise.resolve(), }; } }