sdk/js: cancel an abandoned join's work instead of waiting past it

From review on #739.

The examples go back to what they were: they are embedded in the user
guides, and printing the control plane URL could leak credentials or
query parameters in it. A join that stops already says where in the
error Node prints, so the probe's output shows it without them. The
error no longer names the control plane URL either.

untilAborted only stopped waiting: a control plane sync the join gave
up on went on in the background, could still call /info and refresh(),
and write credential.json. The signal now goes into the requests
themselves. ControlPlaneClient.info() and keys() and
authenticateWithPeer take one and combine it with their own timeout,
as discovery does, and syncControlPlane stops before its next request
and before saving once it aborts. A /refresh already sent is left to
finish, since the control plane spends the old biscuit when it answers.

The sync before a join is best effort, so it gets JOIN_SYNC_TIMEOUT_MS
(5s) rather than the whole join: an unreachable control plane leaves
the join to go on with the stored credential.

A join that holds a router succeeds even if its own timeout fires while
a second router is still being tried; only the caller's signal turns
that into a failure. A router cut off by the signal is no longer
shunned.

Refs #732

Author: saurav-shakya

sdk/js/README.md

@@ -61,43 +61,32 @@ if (argv[0] !== undefined && !argv[0].includes("://")) {
 }
 const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv;
 
-// Each stage goes to stderr as it is reached, so a run that hangs shows the
-// last one it got to; stdout carries only what callers parse.
-const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com";
-console.error(`enrolling with ${controlPlaneUrl}`);
 const mesh = await AgentMesh.enroll({
-  controlPlaneUrl,
+  controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com",
   bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH,
   jwtPath: process.env.AGENTMESH_JWT_PATH,
   stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`,
   // A plaintext http:// control plane is otherwise accepted only on loopback.
   allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true",
 });
-console.error(`enrolled ${mesh.peerId}`);
 const session = await mesh.join();
-console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`);
 console.log(`on the mesh as ${session.peerId}`);
 
 // The peer to call: an agent named by ID, or a provider of a published service.
 let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] };
 if (peerId !== undefined) {
   // An agent is not in the discovery table; the SDK finds the path to its
   // peer ID through the routers.
-  console.error(`dialing ${peerId}`);
   await session.connect(peerId);
 } else {
   const providers = await session.discover(service);
-  for (const p of providers) {
-    console.error(`provider ${p.peerId}`);
-  }
   if (providers.length === 0) {
     throw new Error(`no member of the mesh serves ${service}`);
   }
   // A provider record can outlive its member; the first that answers is used.
   let reached = false;
   for (const candidate of providers) {
     try {
-      console.error(`dialing ${candidate.peerId}`);
       await session.connect(candidate);
       provider = candidate;
       reached = true;

sdk/js/examples/a2a-agent.ts

@@ -20,21 +20,15 @@ import { AgentEvent, DefaultRequestHandler, InMemoryTaskStore, type AgentExecuto
 import { UserBuilder, agentCardHandler, jsonRpcHandler } from "@a2a-js/sdk/server/express";
 import { AgentMesh, MeshSession } from "@agentmesh-p2p/sdk";
 
-// Each stage goes to stderr as it is reached, so a run that hangs shows the
-// last one it got to; stdout carries only what callers parse.
-const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com";
-console.error(`enrolling with ${controlPlaneUrl}`);
 const mesh = await AgentMesh.enroll({
-  controlPlaneUrl,
+  controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com",
   bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH,
   jwtPath: process.env.AGENTMESH_JWT_PATH,
   stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-agent`,
   // A plaintext http:// control plane is otherwise accepted only on loopback.
   allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true",
 });
-console.error(`enrolled ${mesh.peerId}`);
 const session = await mesh.join();
-console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`);
 
 // The card names the agent as the mesh reaches it: the URL an A2A client on
 // the mesh gives its fetch, http://mesh/mesh/<peer-id>/a2a/agent.

sdk/js/examples/a2a-call.ts

@@ -21,21 +21,15 @@ if (peerId === undefined) {
   throw new Error("usage: a2a-call.js <peer-id> [text]");
 }
 
-// Each stage goes to stderr as it is reached, so a run that hangs shows the
-// last one it got to; stdout carries only what callers parse.
-const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com";
-console.error(`enrolling with ${controlPlaneUrl}`);
 const mesh = await AgentMesh.enroll({
-  controlPlaneUrl,
+  controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com",
   bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH,
   jwtPath: process.env.AGENTMESH_JWT_PATH,
   stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-caller`,
   // A plaintext http:// control plane is otherwise accepted only on loopback.
   allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true",
 });
-console.error(`enrolled ${mesh.peerId}`);
 const session = await mesh.join();
-console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`);
 console.log(`on the mesh as ${session.peerId}`);
 
 // The agent's URL on the mesh, and a fetch that carries requests to it.
@@ -46,7 +40,6 @@ const factory = new ClientFactory({
   cardResolver: new DefaultAgentCardResolver({ fetchImpl }),
 });
 // The card lives under the agent's URL; the trailing slash keeps the last segment when the resolver appends the well-known path.
-console.error(`dialing ${peerId}`);
 const client = await factory.createFromUrl(agentURL + "/");
 const card = await client.getAgentCard();
 console.log(`agent: ${card.name}, ${card.description}`);

sdk/js/examples/call.ts

@@ -22,43 +22,32 @@ if (argv[0] !== undefined && !argv[0].includes("://")) {
 }
 const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv;
 
-// Each stage goes to stderr as it is reached, so a run that hangs shows the
-// last one it got to; stdout carries only what callers parse.
-const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com";
-console.error(`enrolling with ${controlPlaneUrl}`);
 const mesh = await AgentMesh.enroll({
-  controlPlaneUrl,
+  controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com",
   bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH,
   jwtPath: process.env.AGENTMESH_JWT_PATH,
   stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`,
   // A plaintext http:// control plane is otherwise accepted only on loopback.
   allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true",
 });
-console.error(`enrolled ${mesh.peerId}`);
 const session = await mesh.join();
-console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`);
 console.log(`on the mesh as ${session.peerId}`);
 
 // The peer to call: an agent named by ID, or a provider of a published service.
 let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] };
 if (peerId !== undefined) {
   // An agent is not in the discovery table; the SDK finds the path to its
   // peer ID through the routers.
-  console.error(`dialing ${peerId}`);
   await session.connect(peerId);
 } else {
   const providers = await session.discover(service);
-  for (const p of providers) {
-    console.error(`provider ${p.peerId}`);
-  }
   if (providers.length === 0) {
     throw new Error(`no member of the mesh serves ${service}`);
   }
   // A provider record can outlive its member; the first that answers is used.
   let reached = false;
   for (const candidate of providers) {
     try {
-      console.error(`dialing ${candidate.peerId}`);
       await session.connect(candidate);
       provider = candidate;
       reached = true;

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/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();
     }