transport: add opt-in receiver ACKs and per-stream backpressure

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.

__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<Array<boolean>> = [];
+    const service = createProtoService().define(TestService, {
+      countUp: ({ request, resWritable }) => {
+        const results: Array<boolean> = [];
+        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 });
+    }
+  });
+});

__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<typeof services>(
+      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 });
+    }
+  },
+);

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,

protobuf/server.ts

@@ -371,6 +371,7 @@ class ProtobufServer<
     const sessionBackpressure = this.transport.getSessionBackpressure(
       from,
       sessionId,
+      streamId,
     );
 
     const deferredCleanups: Array<() => void | Promise<void>> = [];

router/client.ts

@@ -342,13 +342,13 @@ function handleProc<RejectionCodeSchema extends CustomHandshakeErrorCodeSchema>(
     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,

router/server.ts

@@ -464,6 +464,7 @@ class RiverServer<
     const sessionBackpressure = this.transport.getSessionBackpressure(
       from,
       sessionId,
+      streamId,
     );
 
     const cancelStream = (

router/streams.ts

@@ -92,15 +92,15 @@ export interface Writable<T> {
   /**
    * {@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
    * node's `stream.Writable.write`.
    */
   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.
    *

transport/sessionStateMachine/SessionConnected.ts

@@ -60,15 +60,27 @@ export class SessionConnected<
   private isActivelyHeartbeating = false;
   private rehandshakeTimer?: ReturnType<typeof setTimeout> | undefined;
   private credentialExpiry?: number | undefined;
+  private receiverAckTimer?: ReturnType<typeof setTimeout>;
+  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 {