make client-side heartbeat timeout instead of interval based (#302)

## Why

- https://developer.chrome.com/blog/timer-throttling-in-chrome-88 causes
intervals to be heavily throttled, prefer to use a single timeout

## What changed

- refactor to use a single timeout instead of heartbeatsUntilDead *
intervals
- fix some mock transport harness logic
- remove stack tracking code, it didnt tell us much 😭 

## Versioning

- [ ] Breaking protocol change
- [ ] Breaking ts/js API change

<!-- Kind reminder to add tests and updated documentation if needed -->

Author: jackyzha0

testUtil/duplex/duplexPair.ts

@@ -56,6 +56,17 @@ export function duplexPair(): [DuplexSide, DuplexSide] {
   const side1 = new DuplexSide();
   side0[kInitOtherSide](side1);
   side1[kInitOtherSide](side0);
+  side0.on('close', () => {
+    setImmediate(() => {
+      side1.destroy();
+    });
+  });
+
+  side1.on('close', () => {
+    setImmediate(() => {
+      side0.destroy();
+    });
+  });
 
   return [side0, side1];
 }

testUtil/fixtures/mockTransport.ts

@@ -166,14 +166,14 @@ export function createMockTransportNetwork(
 
       // kill all connections while we're at it
       for (const conn of Object.values(connections.get())) {
-        conn.serverToClient.end();
-        conn.clientToServer.end();
+        conn.serverToClient.destroy();
+        conn.clientToServer.destroy();
       }
     },
     cleanup() {
       for (const conn of Object.values(connections.get())) {
-        conn.serverToClient.end();
-        conn.clientToServer.end();
+        conn.serverToClient.destroy();
+        conn.clientToServer.destroy();
       }
     },
   };

transport/sessionStateMachine/SessionConnected.ts

@@ -28,13 +28,6 @@ export interface SessionConnectedProps<ConnType extends Connection>
   listeners: SessionConnectedListeners;
 }
 
-interface TrackedMsg {
-  id: string;
-  seq: number;
-  streamId: string;
-  stack?: string;
-}
-
 /*
  * A session that is connected and can send and receive messages.
  * See transitions.ts for valid transitions.
@@ -47,29 +40,18 @@ export class SessionConnected<
   listeners: SessionConnectedListeners;
 
   private heartbeatHandle?: ReturnType<typeof setInterval> | undefined;
-  private heartbeatMisses = 0;
-  isActivelyHeartbeating: boolean;
-
-  private lastConstructedMsgs: Array<TrackedMsg> = [];
-  private pushLastConstructedMsgs = (msg: OpaqueTransportMessage) => {
-    const trackedMsg = {
-      id: msg.id,
-      seq: msg.seq,
-      streamId: msg.streamId,
-      stack: new Error().stack,
-    };
-
-    this.lastConstructedMsgs.push(trackedMsg);
-
-    if (this.lastConstructedMsgs.length > 10) {
-      this.lastConstructedMsgs.shift();
-    }
-  };
+  private heartbeatMissTimeout?: ReturnType<typeof setTimeout> | undefined;
+  private isActivelyHeartbeating = false;
 
   updateBookkeeping(ack: number, seq: number) {
     this.sendBuffer = this.sendBuffer.filter((unacked) => unacked.seq >= ack);
     this.ack = seq + 1;
-    this.heartbeatMisses = 0;
+
+    if (this.heartbeatMissTimeout) {
+      clearTimeout(this.heartbeatMissTimeout);
+    }
+
+    this.startMissingHeartbeatTimeout();
   }
 
   private assertSendOrdering(constructedMsg: TransportMessage) {
@@ -79,9 +61,6 @@ export class SessionConnected<
         ...this.loggingMetadata,
         transportMessage: constructedMsg,
         tags: ['invariant-violation'],
-        extras: {
-          lastConstructedMsgs: this.lastConstructedMsgs,
-        },
       });
 
       throw new Error(msg);
@@ -90,7 +69,6 @@ export class SessionConnected<
 
   send(msg: PartialTransportMessage): string {
     const constructedMsg = this.constructMsg(msg);
-    this.pushLastConstructedMsgs(constructedMsg);
     this.assertSendOrdering(constructedMsg);
     this.sendBuffer.push(constructedMsg);
     this.conn.send(this.options.codec.toBuffer(constructedMsg));
@@ -109,6 +87,8 @@ export class SessionConnected<
     this.conn.addErrorListener(this.listeners.onConnectionErrored);
 
     // send any buffered messages
+    // dont explicity clear the buffer, we'll just filter out old messages
+    // when we receive an ack
     if (this.sendBuffer.length > 0) {
       this.log?.info(
         `sending ${
@@ -124,41 +104,7 @@ export class SessionConnected<
       }
     }
 
-    // dont explicity clear the buffer, we'll just filter out old messages
-    // when we receive an ack
-
-    // setup heartbeat
-    this.isActivelyHeartbeating = false;
-    this.heartbeatHandle = setInterval(() => {
-      const misses = this.heartbeatMisses;
-      const missDuration = misses * this.options.heartbeatIntervalMs;
-      if (misses >= this.options.heartbeatsUntilDead) {
-        this.log?.info(
-          `closing connection to ${this.to} due to inactivity (missed ${misses} heartbeats which is ${missDuration}ms)`,
-          this.loggingMetadata,
-        );
-        this.telemetry.span.addEvent('closing connection due to inactivity');
-
-        // it is OK to close this even on the client when we can't trust the client timer
-        // due to browser throttling or hibernation
-        // at worst, this interval will fire later than what the server expects and the server
-        // will have already closed the connection
-        // this just helps us in cases where we have a proxying setup where the server has closed
-        // the connection but the proxy hasn't synchronized the server-side close to the client so
-        // the client isn't stuck with a pseudo-dead connection forever
-        this.conn.close();
-        clearInterval(this.heartbeatHandle);
-        this.heartbeatHandle = undefined;
-
-        return;
-      }
-
-      if (this.isActivelyHeartbeating) {
-        this.sendHeartbeat();
-      }
-
-      this.heartbeatMisses++;
-    }, this.options.heartbeatIntervalMs);
+    this.startMissingHeartbeatTimeout();
   }
 
   get loggingMetadata() {
@@ -168,8 +114,27 @@ export class SessionConnected<
     };
   }
 
+  startMissingHeartbeatTimeout() {
+    const maxMisses = this.options.heartbeatsUntilDead;
+    const missDuration = maxMisses * this.options.heartbeatIntervalMs;
+    this.heartbeatMissTimeout = setTimeout(() => {
+      this.log?.info(
+        `closing connection to ${this.to} due to inactivity (missed ${maxMisses} heartbeats which is ${missDuration}ms)`,
+        this.loggingMetadata,
+      );
+      this.telemetry.span.addEvent(
+        'closing connection due to missing heartbeat',
+      );
+
+      this.conn.close();
+    }, missDuration);
+  }
+
   startActiveHeartbeat() {
     this.isActivelyHeartbeating = true;
+    this.heartbeatHandle = setInterval(() => {
+      this.sendHeartbeat();
+    }, this.options.heartbeatIntervalMs);
   }
 
   private sendHeartbeat() {
@@ -246,13 +211,7 @@ 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) {
-      // 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();
-      });
+      this.sendHeartbeat();
     }
   };
 
@@ -266,6 +225,11 @@ export class SessionConnected<
       clearInterval(this.heartbeatHandle);
       this.heartbeatHandle = undefined;
     }
+
+    if (this.heartbeatMissTimeout) {
+      clearTimeout(this.heartbeatMissTimeout);
+      this.heartbeatMissTimeout = undefined;
+    }
   }
 
   _handleClose(): void {

transport/sessionStateMachine/SessionConnecting.ts

@@ -83,16 +83,16 @@ export class SessionConnecting<
 
   _handleStateExit(): void {
     super._handleStateExit();
-
     if (this.connectionTimeout) {
       clearTimeout(this.connectionTimeout);
       this.connectionTimeout = undefined;
     }
   }
 
   _handleClose(): void {
+    super._handleClose();
+
     // close the pending connection if it resolves
     this.bestEffortClose();
-    super._handleClose();
   }
 }

transport/sessionStateMachine/common.ts

@@ -281,10 +281,6 @@ 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(),