sdk/js: a join that hangs says where, and stops what it started

sdk/js/src/auth.ts

@@ -60,9 +60,10 @@ function framed(stream: Stream) {
 /**
  * Client side: presents frame on a new /mesh/auth/1.0.0 stream and returns
  * the peer's verified credential. trustedKeys are the control plane keys.
+ * The handshake has a timeout of its own; outer, when given, can end it sooner.
  */
-export async function authenticateWithPeer(conn: Connection, frame: Uint8Array, trustedKeys: Uint8Array[]): Promise<VerifiedBiscuit> {
-  const signal = AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS);
+export async function authenticateWithPeer(conn: Connection, frame: Uint8Array, trustedKeys: Uint8Array[], outer?: AbortSignal): Promise<VerifiedBiscuit> {
+  const signal = outer === undefined ? AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS) : AbortSignal.any([outer, AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS)]);
   // A peer behind a router is reached over a relayed, limited connection.
   const stream = await conn.newStream(AUTH_PROTOCOL, { signal, runOnLimitedConnection: true });
   try {

sdk/js/src/controlplane.ts

@@ -257,17 +257,17 @@ export class ControlPlaneClient {
   }
 
   /** GET /info: OIDC settings, router addresses and the ban list. Unauthenticated. */
-  async info(): Promise<ControlPlaneInfoResponse> {
-    const body = await this.#request("GET", "/info");
+  async info(signal?: AbortSignal): Promise<ControlPlaneInfoResponse> {
+    const body = await this.#request("GET", "/info", undefined, {}, signal);
     return fromBinary(ControlPlaneInfoResponseSchema, body);
   }
 
   /**
    * GET /keys: every signing key the control plane currently trusts,
    * verified against a key the caller already trusts (the enrollment key).
    */
-  async keys(trusted: Uint8Array[]): Promise<Uint8Array[]> {
-    const body = await this.#request("GET", "/keys");
+  async keys(trusted: Uint8Array[], signal?: AbortSignal): Promise<Uint8Array[]> {
+    const body = await this.#request("GET", "/keys", undefined, {}, signal);
     return verifyKeysResponse(fromBinary(KeysResponseSchema, body), trusted, this.#now());
   }
 
@@ -426,14 +426,15 @@ export class ControlPlaneClient {
     path: string,
     body?: Uint8Array,
     headers: Record<string, string> = {},
+    signal?: AbortSignal,
   ): Promise<{ ok: boolean; status: number; buf: Uint8Array; dateHeader: string | null }> {
-    const controller = new AbortController();
-    const timer = setTimeout(() => controller.abort(new Error(`control plane ${path}: timed out after ${this.#timeoutMs}ms`)), this.#timeoutMs);
+    // The caller's signal cancels the request itself, not only the wait for it.
+    const timeout = AbortSignal.timeout(this.#timeoutMs);
     try {
       const init: RequestInit = {
         method,
         headers: { Accept: PROTOBUF_CONTENT_TYPE, ...headers },
-        signal: controller.signal,
+        signal: signal === undefined ? timeout : AbortSignal.any([signal, timeout]),
       };
       if (body !== undefined) {
         init.body = new Uint8Array(body);
@@ -445,13 +446,22 @@ export class ControlPlaneClient {
         throw new Error(`control plane ${path}: response of ${buf.length} bytes exceeds the ${MAX_RESPONSE_BYTES} byte limit`);
       }
       return { ok: resp.ok, status: resp.status, buf, dateHeader: resp.headers.get("date") };
-    } finally {
-      clearTimeout(timer);
+    } catch (err) {
+      if (timeout.aborted && signal?.aborted !== true) {
+        throw new Error(`control plane ${path}: timed out after ${this.#timeoutMs}ms`);
+      }
+      throw err;
     }
   }
 
-  async #request(method: "GET" | "POST", path: string, body?: Uint8Array, headers: Record<string, string> = {}): Promise<Uint8Array> {
-    const outcome = await this.#fetchOnce(method, path, body, headers);
+  async #request(
+    method: "GET" | "POST",
+    path: string,
+    body?: Uint8Array,
+    headers: Record<string, string> = {},
+    signal?: AbortSignal,
+  ): Promise<Uint8Array> {
+    const outcome = await this.#fetchOnce(method, path, body, headers, signal);
     if (!outcome.ok) {
       throw new ControlPlaneError(path, outcome.status, new TextDecoder().decode(outcome.buf));
     }

sdk/js/src/discovery.test.ts

@@ -99,3 +99,25 @@ test("a lookup one router answers with nothing is put to the next; an announceme
 
   await assert.rejects(request([{ peerId: "nobody", connection: () => undefined }], find), DirectoryError);
 });
+
+test("a request no router answered names every router and what each said", async () => {
+  const key = await serviceKey("mcp", "calculator");
+  const find = create(DirectoryRequestSchema, { request: { case: "findProviders", value: { key, limit: 20 } } });
+  answers.set("a", () => create(DirectoryResponseSchema, { error: "routing table empty" }));
+  answers.set("b", () => create(DirectoryResponseSchema, { error: "rate limited" }));
+  try {
+    await assert.rejects(request([held(routerA), held(routerB)], find), (err: Error) => {
+      assert.ok(err instanceof DirectoryError, `got ${err.name}, want DirectoryError`);
+      for (const [router, said] of [
+        [routerA, "routing table empty"],
+        [routerB, "rate limited"],
+      ] as const) {
+        assert.ok(err.message.includes(`${router.peerId.toString()}: ${said}`), `got ${JSON.stringify(err.message)}, want ${router.peerId.toString()} and what it said`);
+      }
+      return true;
+    });
+  } finally {
+    answers.delete("a");
+    answers.delete("b");
+  }
+});

sdk/js/src/discovery.ts

@@ -110,7 +110,7 @@ async function ask(conn: Connection, req: DirectoryRequest, signal: AbortSignal)
  */
 export async function request(routers: DirectoryRouter[], req: DirectoryRequest, signal?: AbortSignal): Promise<DirectoryResponse> {
   const lookup = req.request.case === "findProviders" || req.request.case === "findPeer";
-  let last: unknown;
+  const failures: string[] = [];
   let asked = 0;
   let empty: DirectoryResponse | undefined;
   for (const router of routers) {
@@ -128,7 +128,7 @@ export async function request(routers: DirectoryRouter[], req: DirectoryRequest,
       }
       return resp;
     } catch (err) {
-      last = err;
+      failures.push(`${router.peerId}: ${err instanceof Error ? err.message : String(err)}`);
       if (signal?.aborted === true) {
         break;
       }
@@ -140,7 +140,7 @@ export async function request(routers: DirectoryRouter[], req: DirectoryRequest,
   if (asked === 0) {
     throw new DirectoryError("no held router to ask");
   }
-  throw new DirectoryError(`no held router answered: ${last instanceof Error ? last.message : String(last)}`);
+  throw new DirectoryError(`no held router answered: ${failures.join("; ")}`);
 }
 
 /**

sdk/js/src/index.ts

@@ -54,6 +54,7 @@ export {
 } from "./challenges.ts";
 export {
   DEFAULT_ROUTERS,
+  JOIN_TIMEOUT_MS,
   MeshSession,
   ROUTER_REDIAL_BACKOFFS_MS,
   candidatesFromRouterInfos,

sdk/js/src/mesh.ts

@@ -310,12 +310,13 @@ export class AgentMesh {
    * Trades the current biscuit for a fresh one and persists it. The control
    * plane redeems only the last biscuit it issued, so a lost refresh result
    * means re-enrolling; persisting before returning keeps that rare.
-   * Concurrent calls wait for each other.
+   * Concurrent calls wait for each other. A refresh whose signal has
+   * aborted by the time its turn comes does not start.
    */
-  async refresh(): Promise<MeshCredential> {
+  async refresh(signal?: AbortSignal): Promise<MeshCredential> {
     const run = this.#refreshChain.then(
-      () => this.#refreshOnce(),
-      () => this.#refreshOnce(),
+      () => this.#refreshOnce(signal),
+      () => this.#refreshOnce(signal),
     );
     this.#refreshChain = run.then(
       () => undefined,
@@ -324,7 +325,8 @@ export class AgentMesh {
     return run;
   }
 
-  async #refreshOnce(): Promise<MeshCredential> {
+  async #refreshOnce(signal?: AbortSignal): Promise<MeshCredential> {
+    signal?.throwIfAborted();
     let jwt: string | undefined;
     if (this.#jwtSource !== undefined) {
       try {
@@ -336,6 +338,10 @@ export class AgentMesh {
         // Best effort: fall back to session-only refresh if the token source is temporarily unavailable.
       }
     }
+    // The last point to stop. The /refresh request itself is not cut short:
+    // the control plane spends the old biscuit when it answers, so only a
+    // refresh that runs to the end can save the new one.
+    signal?.throwIfAborted();
     const result = await this.controlPlane.refresh({
       identity: this.identity,
       biscuit: this.#credential.biscuit,
@@ -379,33 +385,37 @@ export class AgentMesh {
    * addresses and the ban set. Each part is attempted even when another
    * fails; the errors are reported together. Gossip events only bring this
    * forward; they are never the only way state arrives.
+   *
+   * An aborted signal cancels the request in flight and throws its reason:
+   * nothing more is sent and nothing is written to the state directory.
    */
-  async syncControlPlane(): Promise<ControlPlaneSync> {
+  async syncControlPlane(signal?: AbortSignal): Promise<ControlPlaneSync> {
     const errors: string[] = [];
     let keysChanged = false;
     let refreshed = false;
     try {
-      const keys = await this.controlPlane.keys(this.#credential.controlPlaneKeys);
+      const keys = await this.controlPlane.keys(this.#credential.controlPlaneKeys, signal);
       if (keys.length === 0) {
         throw new Error("/keys returned no keys");
       }
       keysChanged = !sameKeySet(keys, this.#credential.controlPlaneKeys);
       this.#credential = withCredentialMethods({ ...this.#credential, controlPlaneKeys: keys });
       if (credentialPredatesRotation(this.#credential)) {
-        await this.refresh();
+        await this.refresh(signal);
         refreshed = true;
       }
     } catch (err) {
       errors.push(`keys: ${err instanceof Error ? err.message : String(err)}`);
     }
+    signal?.throwIfAborted();
 
     let bannedPeerIds: string[] | undefined;
     let routers: RouterInfo[] | undefined;
     // Taken before the request: a ban recorded after this instant cannot be
     // in the answer, so its absence must not be read as an unban.
     const fetchedAt = new Date();
     try {
-      const info = await this.controlPlane.info();
+      const info = await this.controlPlane.info(signal);
       routers = info.routers;
       this.#routers = info.routers;
       const discovered =
@@ -419,6 +429,7 @@ export class AgentMesh {
     } catch (err) {
       errors.push(`info: ${err instanceof Error ? err.message : String(err)}`);
     }
+    signal?.throwIfAborted();
     if (keysChanged || bannedPeerIds !== undefined) {
       await this.save();
     }