autoconfig: abort slower routers after fastest routers are connected

Author: yoursunny

pkg/autoconfig/package.json

@@ -32,14 +32,15 @@
     "@ndn/packet": "workspace:*",
     "@ndn/quic-transport": "workspace:*",
     "@ndn/ws-transport": "workspace:*",
+    "@ndn/util": "workspace:*",
     "default-gateway": "^7.2.2",
-    "tslib": "^2.8.1"
+    "p-event": "^7.1.0",
+    "tslib": "^2.8.1",
+    "type-fest": "^5.6.0"
   },
   "devDependencies": {
-    "@ndn/util": "workspace:*",
     "@types/default-gateway": "^7.2.2",
     "@types/koa": "^3.0.3",
-    "koa": "^3.2.1",
-    "type-fest": "^5.6.0"
+    "koa": "^3.2.1"
   }
 }

pkg/autoconfig/src/network.ts

@@ -39,6 +39,13 @@ export interface ConnectNetworkOptions extends ConnectRouterOptions {
    * @defaultValue 1
    */
   fastest?: number;
+
+  /***
+   * After `fastest` faces have been created and finished testConnection,
+   * how much longer to wait for other faces, in milliseconds.
+   * @defaultValue 1000
+   */
+  waitAfterFastest?: number;
 }
 
 /** Connect to an NDN network. */
@@ -49,7 +56,16 @@ export async function connectToNetwork(opts: ConnectNetworkOptions = {}): Promis
     tryDefaultGateway = true,
     fallback = [],
     fastest = 1,
+    waitAfterFastest = 1000,
+    signal: parentSignal,
   } = opts;
+  parentSignal?.throwIfAborted();
+
+  const fastestAbort = new AbortController();
+  const routerOpts: ConnectRouterOptions = {
+    ...opts,
+    signal: parentSignal ? AbortSignal.any([parentSignal, fastestAbort.signal]) : fastestAbort.signal,
+  };
 
   const connected: ConnectRouterResult[] = [];
   const errs: Record<string, unknown> = {};
@@ -75,7 +91,13 @@ export async function connectToNetwork(opts: ConnectNetworkOptions = {}): Promis
   ) {
     await Promise.all(routers.map(async (router) => {
       try {
-        connected.push(await connectToRouter(router, opts));
+        connected.push(await connectToRouter(router, routerOpts));
+        if (connected.length === fastest) {
+          setTimeout(
+            () => fastestAbort.abort(new Error("sufficient number of faces have been established")),
+            waitAfterFastest,
+          );
+        }
       } catch (err: unknown) {
         errs[router] = err;
       }

pkg/autoconfig/src/router.ts

@@ -2,6 +2,9 @@ import { consume, type ConsumerOptions } from "@ndn/endpoint";
 import { type Forwarder, type FwFace, TapFace } from "@ndn/fw";
 import { Interest, Name, type NameLike } from "@ndn/packet";
 import type { H3Transport } from "@ndn/quic-transport";
+import { assert } from "@ndn/util";
+import { pEvent } from "p-event";
+import type { Arrayable } from "type-fest";
 
 import { createFace } from "./platform_node";
 
@@ -50,7 +53,7 @@ export interface ConnectRouterOptions {
    *   If string ends with "/*", it's replaced with a random component.
    * - function: execute the custom tester function.
    */
-  testConnection?: false | TestConnectionPacket | TestConnectionPacket[] |
+  testConnection?: false | Arrayable<TestConnectionPacket> |
     ((face: FwFace) => Promise<unknown>);
 
   /**
@@ -66,7 +69,10 @@ export interface ConnectRouterOptions {
    * Routes to be added on the created face.
    * @defaultValue `["/"]`
    */
-  addRoutes?: NameLike[];
+  addRoutes?: readonly NameLike[];
+
+  /** AbortSignal that allows canceling the attempt via AbortController. */
+  signal?: AbortSignal;
 }
 
 /** {@link connectToRouter} result. */
@@ -83,14 +89,34 @@ export interface ConnectRouterResult {
 
 /** Connect to a router and test the connection. */
 export async function connectToRouter(router: string, opts: ConnectRouterOptions = {}): Promise<ConnectRouterResult> {
-  const face = await createFace(router, opts);
+  const { signal } = opts;
+  let face: FwFace | undefined;
+  const promises: Array<Promise<void>> = [
+    (async () => {
+      // createFace does not take AbortSignal, but clear it to protect against future changes
+      face = await createFace(router, { ...opts, signal: undefined });
+    })(),
+  ];
+  if (signal) {
+    promises.push((async () => {
+      if (!signal.aborted) {
+        await pEvent(signal, "abort");
+      }
+    })());
+  }
+  await Promise.race(promises);
+  if (!face) {
+    assert(signal?.aborted);
+    throw signal.reason; // eslint-disable-line @typescript-eslint/only-throw-error
+  }
 
   const testConnectionStart = performance.now();
   let testConnectionDuration: number;
   let testConnectionResult: unknown;
   try {
     testConnectionResult = await testConnection(face, opts);
     testConnectionDuration = performance.now() - testConnectionStart;
+    signal?.throwIfAborted();
   } catch (err: unknown) {
     face.close();
     throw err;
@@ -103,8 +129,11 @@ async function testConnection(
     {
       testConnection: tc = new Name("/localhop/nfd/rib/list"),
       testConnectionTimeout = 2000,
+      signal: parentSignal,
     }: ConnectRouterOptions,
 ): Promise<unknown> {
+  parentSignal?.throwIfAborted();
+
   if (tc === false) {
     return undefined;
   }
@@ -117,20 +146,23 @@ async function testConnection(
 
   const tapFace = TapFace.create(face);
   tapFace.addRoute("/");
-  const abort = new AbortController();
-  const cOpts: ConsumerOptions = { fw: tapFace.fw, signal: abort.signal };
+  const raceAbort = new AbortController();
+  const cOpts: ConsumerOptions = {
+    fw: tapFace.fw,
+    signal: parentSignal ? AbortSignal.any([parentSignal, raceAbort.signal]) : raceAbort.signal,
+  };
   try {
-    await Promise.any(tc.map((pkt) => {
+    return await Promise.any(tc.map(async (pkt, i) => {
       if (typeof pkt === "string" && pkt.endsWith("/*")) {
         pkt = new Name(pkt.slice(0, -2)).append(Math.trunc(Math.random() * 1e8).toString().padStart(8, "0"));
       }
       const interest = pkt instanceof Interest ? pkt :
         new Interest(pkt, Interest.CanBePrefix, Interest.Lifetime(testConnectionTimeout));
-      return consume(interest, cOpts);
+      const data = await consume(interest, cOpts);
+      return { i, interest, data };
     }));
   } finally {
-    abort.abort();
+    raceAbort.abort();
     tapFace.close();
   }
-  return undefined;
 }

pkg/autoconfig/tests/network.t.ts

@@ -44,13 +44,18 @@ test("connectToNetwork", async () => {
     ...await addClosedServers(),
   ];
 
+  const t0 = Date.now();
   const faces = await connectToNetwork({
     fch: false,
     tryDefaultGateway: false,
     fallback: servers,
+    connectTimeout: 5000,
     testConnection: "/localhop/test-connection/*",
-    testConnectionTimeout: 1500,
+    testConnectionTimeout: 5000,
+    waitAfterFastest: 100,
   });
+  const t1 = Date.now();
+  expect(t1 - t0).toBeLessThanOrEqual(2000);
   closers.push(...faces);
   expect(faces).toHaveLength(1);
   expect(faces[0]!.toString()).toContain(servers[1]);
@@ -62,6 +67,7 @@ test("connectToNetwork", async () => {
     tryDefaultGateway: false,
     fallback: servers,
     fastest: 2,
+    connectTimeout: 1500,
     testConnection: ["/localhop/test-connection/*", new Name("/unreachable")],
     testConnectionTimeout: 1500,
   });

pkg/endpoint/src/consumer.ts

@@ -81,6 +81,8 @@ function makeConsumer(
   const retxGen = makeRetxGenerator(retx)(interest.lifetime)[Symbol.iterator]();
 
   const promise = new Promise<Data>((resolve, reject) => {
+    signal?.throwIfAborted();
+
     const rx = pushable<FwPacket>();
 
     let timer: NodeJS.Timeout | number | undefined;