make passive heartbeat case respond async

package.json

@@ -1,7 +1,7 @@
 {
   "name": "@replit/river",
   "description": "It's like tRPC but... with JSON Schema Support, duplex streaming and support for service multiplexing. Transport agnostic!",
-  "version": "0.207.0",
+  "version": "0.207.1",
   "type": "module",
   "exports": {
     ".": {

transport/sessionStateMachine/SessionConnected.ts

@@ -226,7 +226,13 @@ export class SessionConnected<
     // if we are not actively heartbeating, we are in passive
     // heartbeat mode and should send a response to the ack
     if (!this.isActivelyHeartbeating) {
-      this.sendHeartbeat();
+      // purposefully make this async to avoid weird browser behavior
+      // where _some_ browsers will decide that it is ok to interrupt fully
+      // synchronous code execution (e.g. an existing .send) to receive a
+      // websocket message and hit this codepath
+      void Promise.resolve().then(() => {
+        this.sendHeartbeat();
+      });
     }
   };
 

transport/sessionStateMachine/common.ts

@@ -281,6 +281,10 @@ export abstract class IdentifiedSession extends CommonSession {
   constructMsg<Payload>(
     partialMsg: PartialTransportMessage<Payload>,
   ): TransportMessage<Payload> {
+    if (this._isConsumed) {
+      throw new Error(ERR_CONSUMED);
+    }
+
     const msg = {
       ...partialMsg,
       id: generateId(),

transport/sessionStateMachine/stateMachine.test.ts

@@ -1958,7 +1958,9 @@ describe('session state machine', () => {
       );
 
       // make sure the session acks the heartbeat
-      expect(conn.send).toHaveBeenCalledTimes(1);
+      await waitFor(() => {
+        expect(conn.send).toHaveBeenCalledTimes(1);
+      });
     });
 
     test('does not dispatch acks', async () => {