Merge pull request #49 from MatrixAI/feat-monitor-fixes

Feat monitor fixes

Author: tegefaulkes

package-lock.json

@@ -39,7 +39,7 @@
     "@matrixai/async-cancellable": "^1.1.1",
     "@matrixai/async-init": "^1.8.4",
     "@matrixai/async-locks": "^4.0.0",
-    "@matrixai/contexts": "^1.1.0",
+    "@matrixai/contexts": "^1.2.0",
     "@matrixai/errors": "^1.1.7",
     "@matrixai/logger": "^3.1.0",
     "@matrixai/resources": "^1.1.5",

package.json

@@ -16,7 +16,7 @@ import { quiche } from './native';
 import * as utils from './utils';
 import * as errors from './errors';
 import * as events from './events';
-import { clientDefault } from './config';
+import { clientDefault, minIdleTimeout } from './config';
 import QUICSocket from './QUICSocket';
 import QUICConnection from './QUICConnection';
 import QUICConnectionId from './QUICConnectionId';
@@ -87,7 +87,7 @@ class QUICClient extends EventTarget {
     },
     ctx?: Partial<ContextTimedInput>,
   ): PromiseCancellable<QUICClient>;
-  @timedCancellable(true, Infinity, errors.ErrorQUICClientCreateTimeOut)
+  @timedCancellable(true, minIdleTimeout, errors.ErrorQUICClientCreateTimeOut)
   public static async createQUICClient(
     {
       host,

src/QUICClient.ts

@@ -24,7 +24,9 @@ import {
 import Logger from '@matrixai/logger';
 import { Timer } from '@matrixai/timer';
 import { context, timedCancellable } from '@matrixai/contexts/dist/decorators';
-import { buildQuicheConfig } from './config';
+import { withF } from '@matrixai/resources';
+import { utils as contextsUtils } from '@matrixai/contexts';
+import { buildQuicheConfig, minIdleTimeout } from './config';
 import QUICStream from './QUICStream';
 import { quiche } from './native';
 import * as events from './events';
@@ -235,7 +237,11 @@ class QUICConnection extends EventTarget {
         },
     ctx?: Partial<ContextTimedInput>,
   ): PromiseCancellable<QUICConnection>;
-  @timedCancellable(true, Infinity, errors.ErrorQUICConnectionStartTimeOut)
+  @timedCancellable(
+    true,
+    minIdleTimeout,
+    errors.ErrorQUICConnectionStartTimeOut,
+  )
   public static async createQUICConnection(
     args:
       | {
@@ -265,40 +271,20 @@ class QUICConnection extends EventTarget {
         },
     @context ctx: ContextTimed,
   ): Promise<QUICConnection> {
-    const timeoutTime = ctx.timer.getTimeout();
-    if (timeoutTime !== Infinity && timeoutTime >= args.config.maxIdleTimeout) {
-      throw new errors.ErrorQUICConnectionInvalidConfig(
-        'connection timeout timer must be strictly less than maxIdleTimeout',
-      );
-    }
     ctx.signal.throwIfAborted();
     const abortProm = promise<never>();
     const abortHandler = () => {
       abortProm.rejectP(ctx.signal.reason);
     };
     ctx.signal.addEventListener('abort', abortHandler);
     const connection = new this(args);
-    const startProm = connection.start().then(async () => {
-      // If this is a server connection, we need to process the packet that created it
-      if (args.type === 'server') {
-        await utils.withMonitor(
-          undefined,
-          connection.lockbox,
-          RWLockWriter,
-          async (mon) => {
-            await mon.withF(connection.lockCode, async (mon) => {
-              await connection.recv(args.data, args.remoteInfo, mon);
-              await connection.send(mon);
-            });
-          },
-        );
-      }
-    });
+    // If it's a server connection we want to pass the initial packet
+    const data = args.type === 'server' ? args.data : undefined;
     // This ensures that TLS has been established and verified on both sides
     try {
       await Promise.race([
         Promise.all([
-          startProm,
+          connection.start(data),
           connection.establishedP,
           connection.secureEstablishedP,
         ]),
@@ -471,12 +457,25 @@ class QUICConnection extends EventTarget {
 
   /**
    * This will set up the connection initiate sending
+   * @param data - the initial packet that triggered the creation of the connection.
    */
-  public async start(): Promise<void> {
+  public async start(data?: Uint8Array): Promise<void> {
     this.logger.info(`Start ${this.constructor.name}`);
     // Set the connection up
     this.socket.connectionMap.set(this.connectionId, this);
-    await this.send();
+    await withF(
+      [contextsUtils.monitor(this.lockbox, RWLockWriter)],
+      async ([mon]) => {
+        if (data != null) {
+          await this.recv(
+            data,
+            { host: this._remoteHost, port: this._remotePort },
+            mon,
+          );
+        }
+        await this.send(mon);
+      },
+    );
     this.logger.info(`Started ${this.constructor.name}`);
   }
 
@@ -535,7 +534,7 @@ class QUICConnection extends EventTarget {
     this.stopKeepAliveIntervalTimer();
 
     // Trigger closing connection in the background and await close later.
-    void utils.withMonitor(mon, this.lockbox, RWLockWriter, async (mon) => {
+    void this.withMonitor(mon, this.lockbox, RWLockWriter, async (mon) => {
       await mon.withF(this.lockCode, async (mon) => {
         // If this is already closed, then `Done` will be thrown
         // Otherwise it can send `CONNECTION_CLOSE` frame
@@ -665,13 +664,15 @@ class QUICConnection extends EventTarget {
   public async recv(
     data: Uint8Array,
     remoteInfo: RemoteInfo,
-    mon: Monitor<RWLockWriter>,
+    mon?: Monitor<RWLockWriter>,
   ): Promise<void> {
-    if (!mon.isLocked(this.lockCode)) {
-      return mon.withF(this.lockCode, async (mon) => {
-        return this.recv(data, remoteInfo, mon);
-      });
-    }
+    await this.withMonitor(mon, this.lockbox, RWLockWriter, async (mon) => {
+      if (!mon.isLocked(this.lockCode)) {
+        return mon.withF(this.lockCode, async (mon) => {
+          return this.recv(data, remoteInfo, mon);
+        });
+      }
+    });
 
     try {
       // The remote information may be changed on each receive
@@ -775,7 +776,7 @@ class QUICConnection extends EventTarget {
    * @internal
    */
   public async send(mon?: Monitor<RWLockWriter>): Promise<void> {
-    await utils.withMonitor(mon, this.lockbox, RWLockWriter, async (mon) => {
+    await this.withMonitor(mon, this.lockbox, RWLockWriter, async (mon) => {
       if (!mon.isLocked(this.lockCode)) {
         return mon.withF(this.lockCode, async (mon) => {
           return this.send(mon);
@@ -1083,6 +1084,26 @@ class QUICConnection extends EventTarget {
       return quicStream;
     });
   }
+
+  /**
+   * Used as a clean way to create a new monitor if it doesn't exist, otherwise uses the existing one.
+   */
+  protected async withMonitor<T>(
+    mon: Monitor<RWLockWriter> | undefined,
+    lockBox: LockBox<RWLockWriter>,
+    lockConstructor: { new (): RWLockWriter },
+    f: (mon: Monitor<RWLockWriter>) => Promise<T>,
+    locksPending?: Map<string, { count: number }>,
+  ): Promise<T> {
+    if (mon == null) {
+      return await withF(
+        [contextsUtils.monitor(lockBox, lockConstructor, locksPending)],
+        ([mon]) => f(mon),
+      );
+    } else {
+      return f(mon);
+    }
+  }
 }
 
 export default QUICConnection;

src/QUICConnection.ts

@@ -53,6 +53,7 @@ class QUICServer extends EventTarget {
   protected codeToReason: StreamCodeToReason | undefined;
   protected verifyCallback: VerifyCallback | undefined;
   protected connectionMap: QUICConnectionMap;
+  protected minIdleTimeout: number | undefined;
   // Used to track address string for logging ONLY
   protected address: string;
 
@@ -109,6 +110,7 @@ class QUICServer extends EventTarget {
     reasonToCode,
     codeToReason,
     verifyCallback,
+    minIdleTimeout,
     logger,
   }: {
     crypto: {
@@ -124,6 +126,7 @@ class QUICServer extends EventTarget {
     reasonToCode?: StreamReasonToCode;
     codeToReason?: StreamCodeToReason;
     verifyCallback?: VerifyCallback;
+    minIdleTimeout?: number;
     logger?: Logger;
   }) {
     super();
@@ -151,6 +154,7 @@ class QUICServer extends EventTarget {
     this.reasonToCode = reasonToCode;
     this.codeToReason = codeToReason;
     this.verifyCallback = verifyCallback;
+    this.minIdleTimeout = minIdleTimeout;
   }
 
   @ready(new errors.ErrorQUICServerNotRunning())
@@ -355,23 +359,28 @@ class QUICServer extends EventTarget {
       `Accepting new connection from QUIC packet from ${remoteInfo.host}:${remoteInfo.port}`,
     );
     const clientConnRef = Buffer.from(header.scid).toString('hex').slice(32);
-    const connectionProm = QUICConnection.createQUICConnection({
-      type: 'server',
-      scid: newScid,
-      dcid: dcidOriginal,
-      socket: this.socket,
-      remoteInfo,
-      data,
-      config: this.config,
-      reasonToCode: this.reasonToCode,
-      codeToReason: this.codeToReason,
-      verifyCallback: this.verifyCallback,
-      logger: this.logger.getChild(
-        `${QUICConnection.name} ${scid.toString().slice(32)}-${clientConnRef}`,
-      ),
-    });
+    let connection: QUICConnection;
     try {
-      await connectionProm;
+      connection = await QUICConnection.createQUICConnection(
+        {
+          type: 'server',
+          scid: newScid,
+          dcid: dcidOriginal,
+          socket: this.socket,
+          remoteInfo,
+          data,
+          config: this.config,
+          reasonToCode: this.reasonToCode,
+          codeToReason: this.codeToReason,
+          verifyCallback: this.verifyCallback,
+          logger: this.logger.getChild(
+            `${QUICConnection.name} ${scid
+              .toString()
+              .slice(32)}-${clientConnRef}`,
+          ),
+        },
+        { timer: this.minIdleTimeout },
+      );
     } catch (e) {
       // Ignoring any errors here as a failure to connect
       this.dispatchEvent(
@@ -383,7 +392,6 @@ class QUICServer extends EventTarget {
       );
       return;
     }
-    const connection = await connectionProm;
     // Handling connection events
     connection.addEventListener(
       'connectionError',

src/QUICServer.ts

@@ -8,6 +8,8 @@ import { running } from '@matrixai/async-init';
 import { StartStop, ready } from '@matrixai/async-init/dist/StartStop';
 import { RWLockWriter } from '@matrixai/async-locks';
 import { status } from '@matrixai/async-init/dist/utils';
+import { withF } from '@matrixai/resources';
+import { utils as contextsUtils } from '@matrixai/contexts';
 import QUICConnectionId from './QUICConnectionId';
 import QUICConnectionMap from './QUICConnectionMap';
 import { quiche } from './native';
@@ -108,11 +110,9 @@ class QUICSocket extends EventTarget {
     // Acquire the conn lock, this ensures mutual exclusion
     // for state changes on the internal connection
     try {
-      await utils.withMonitor(
-        undefined,
-        connection.lockbox,
-        RWLockWriter,
-        async (mon) => {
+      await withF(
+        [contextsUtils.monitor(connection.lockbox, RWLockWriter)],
+        async ([mon]) => {
           await mon.withF(connection.lockCode, async (mon) => {
             // Even if we are `stopping`, the `quiche` library says we need to
             // continue processing any packets.

src/QUICSocket.ts

@@ -23,12 +23,26 @@ const sigalgs = [
   'ed25519',
 ].join(':');
 
+/**
+ * Usually we would create separate timeouts for connecting vs idling.
+ * Unfortunately quiche only has 1 config option that controls both.
+ * And it is not possible to mutate this option after connecting.
+ * Therefore, this option is just a way to set a shorter connecting timeout
+ * compared to the idling timeout.
+ * If this is the larger than the `maxIdleTimeout` (remember that `0` is `Infinity`) for `maxIdleTimeout`, then this has no effect.
+ * This only has an effect if this is set to a number less than `maxIdleTimeout`.
+ * Thus, it is the "minimum boundary" of the timeout during connecting.
+ * While the `maxIdleTimeout` is still the "maximum boundary" during connecting.
+ */
+const minIdleTimeout = Infinity;
+
 const clientDefault: QUICConfig = {
   sigalgs,
   verifyPeer: true,
   verifyAllowFail: false,
   grease: true,
-  maxIdleTimeout: 1 * 60 * 1000,
+  keepAliveIntervalTime: undefined,
+  maxIdleTimeout: 0,
   maxRecvUdpPayloadSize: quiche.MAX_DATAGRAM_SIZE, // 65527
   maxSendUdpPayloadSize: quiche.MIN_CLIENT_INITIAL_LEN, // 1200,
   initialMaxData: 10 * 1024 * 1024,
@@ -48,7 +62,8 @@ const serverDefault: QUICConfig = {
   verifyPeer: false,
   verifyAllowFail: false,
   grease: true,
-  maxIdleTimeout: 1 * 60 * 1000,
+  keepAliveIntervalTime: undefined,
+  maxIdleTimeout: 0,
   maxRecvUdpPayloadSize: quiche.MAX_DATAGRAM_SIZE, // 65527
   maxSendUdpPayloadSize: quiche.MIN_CLIENT_INITIAL_LEN, // 1200
   initialMaxData: 10 * 1024 * 1024,
@@ -188,4 +203,4 @@ function buildQuicheConfig(config: QUICConfig): QuicheConfig {
   return quicheConfig;
 }
 
-export { clientDefault, serverDefault, buildQuicheConfig };
+export { minIdleTimeout, clientDefault, serverDefault, buildQuicheConfig };