diff --git a/sdk/js/src/auth.ts b/sdk/js/src/auth.ts index 835deb09..15aea2c6 100644 --- a/sdk/js/src/auth.ts +++ b/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 { - const signal = AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS); +export async function authenticateWithPeer(conn: Connection, frame: Uint8Array, trustedKeys: Uint8Array[], outer?: AbortSignal): Promise { + 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 { diff --git a/sdk/js/src/controlplane.ts b/sdk/js/src/controlplane.ts index 636ed346..fd52126f 100644 --- a/sdk/js/src/controlplane.ts +++ b/sdk/js/src/controlplane.ts @@ -257,8 +257,8 @@ export class ControlPlaneClient { } /** GET /info: OIDC settings, router addresses and the ban list. Unauthenticated. */ - async info(): Promise { - const body = await this.#request("GET", "/info"); + async info(signal?: AbortSignal): Promise { + const body = await this.#request("GET", "/info", undefined, {}, signal); return fromBinary(ControlPlaneInfoResponseSchema, body); } @@ -266,8 +266,8 @@ export class ControlPlaneClient { * 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 { - const body = await this.#request("GET", "/keys"); + async keys(trusted: Uint8Array[], signal?: AbortSignal): Promise { + 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 = {}, + 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 = {}): Promise { - const outcome = await this.#fetchOnce(method, path, body, headers); + async #request( + method: "GET" | "POST", + path: string, + body?: Uint8Array, + headers: Record = {}, + signal?: AbortSignal, + ): Promise { + const outcome = await this.#fetchOnce(method, path, body, headers, signal); if (!outcome.ok) { throw new ControlPlaneError(path, outcome.status, new TextDecoder().decode(outcome.buf)); } diff --git a/sdk/js/src/discovery.test.ts b/sdk/js/src/discovery.test.ts index 1f2d4826..10bd289d 100644 --- a/sdk/js/src/discovery.test.ts +++ b/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"); + } +}); diff --git a/sdk/js/src/discovery.ts b/sdk/js/src/discovery.ts index 711a7784..37dbf94a 100644 --- a/sdk/js/src/discovery.ts +++ 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 { 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("; ")}`); } /** diff --git a/sdk/js/src/index.ts b/sdk/js/src/index.ts index 947fd3ba..3d32b898 100644 --- a/sdk/js/src/index.ts +++ b/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, diff --git a/sdk/js/src/mesh.ts b/sdk/js/src/mesh.ts index 606ccc21..7567bd00 100644 --- a/sdk/js/src/mesh.ts +++ b/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 { + async refresh(signal?: AbortSignal): Promise { 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 { + async #refreshOnce(signal?: AbortSignal): Promise { + 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,25 +385,29 @@ 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 { + async syncControlPlane(signal?: AbortSignal): Promise { 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; @@ -405,7 +415,7 @@ export class AgentMesh { // 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(); } diff --git a/sdk/js/src/session.test.ts b/sdk/js/src/session.test.ts index db5442ed..9f3633b4 100644 --- a/sdk/js/src/session.test.ts +++ b/sdk/js/src/session.test.ts @@ -36,7 +36,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { after, before, test } from "node:test"; -import { AUTH_HANDLER_OPTIONS, AUTH_PROTOCOL, AuthRejectedError, GOAWAY_MAX_RETRY_AFTER_MS, GOAWAY_PROTOCOL, ROUTER_SHUN_DURATION_MS, authenticateWithPeer, authStreamHandler, clampGoAwayRetryAfterMs } from "./auth.ts"; +import { AUTH_HANDLER_OPTIONS, AUTH_HANDSHAKE_TIMEOUT_MS, AUTH_PROTOCOL, AuthRejectedError, GOAWAY_MAX_RETRY_AFTER_MS, GOAWAY_PROTOCOL, ROUTER_SHUN_DURATION_MS, authenticateWithPeer, authStreamHandler, clampGoAwayRetryAfterMs } from "./auth.ts"; import { BiscuitVerificationError, ROLE_ROUTER, loadBiscuit, verifyPeerBiscuit } from "./biscuit.ts"; import { ROLE_NODE } from "./controlplane.ts"; import { BootstrapEnrollRequestSchema, BootstrapEnrollResponseSchema, ControlPlaneInfoResponseSchema, EnrollmentStatus, KeysResponseSchema, AuthFrameSchema, DirectoryRequestSchema, DirectoryResponseSchema, PeerRecordSchema, RouterGoAwaySchema, RouterGoAway_Reason, RouterInfoSchema, TokenRefreshRequestSchema, TokenRefreshResponseSchema, type RouterInfo } from "./gen/agentmesh_pb.ts"; @@ -44,7 +44,7 @@ import { Identity } from "./identity.ts"; import { HTTP_HANDLER_OPTIONS, HTTP_PROTOCOL, a2aEndpoint, httpIngressHandler } from "./libp2p-http.ts"; import { LabelsNotSatisfiedError } from "./mcp.ts"; import { AgentMesh } from "./mesh.ts"; -import { MeshSession, candidatesFromRouterInfos, selectRouters, type JoinOptions } from "./session.ts"; +import { JOIN_SYNC_TIMEOUT_MS, JOIN_TIMEOUT_MS, MeshSession, candidatesFromRouterInfos, joinMesh, selectRouters, type JoinOptions } from "./session.ts"; type Wasm = Awaited>; type KeyPair = InstanceType; @@ -127,6 +127,52 @@ function fakeControlPlane(routerAddresses: string[], routers: RouterInfo[] = []) }) as typeof fetch; } +/** + * A control plane whose /keys and /info, once stall() is called, take the + * request and never answer until it is aborted. paths lists every request + * made after stall(), in order. + */ +function stallingControlPlane(routes: typeof fetch): { fetch: typeof fetch; stall(): void; paths: string[] } { + let stalled = false; + const paths: string[] = []; + const stalling = (async (input: Parameters[0], init?: RequestInit) => { + const path = new URL(new Request(input, init).url).pathname; + if (!stalled) { + return routes(input, init); + } + paths.push(path); + if (path !== "/keys" && path !== "/info") { + return routes(input, init); + } + return new Promise((_, reject) => { + const signal = init?.signal; + if (signal === undefined || signal === null) { + reject(new Error(`${path} was sent without a signal and would never end`)); + return; + } + if (signal.aborted) { + reject(signal.reason); + return; + } + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + }) as typeof fetch; + return { fetch: stalling, stall: () => { stalled = true; }, paths }; +} + +/** A libp2p host that takes the auth stream and never writes back: a router whose handshake neither completes nor fails. */ +async function silentRouter(): Promise { + const node = await createLibp2p({ + addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [tls()], + streamMuxers: [yamux()], + services: { identify: identify() }, + }); + await node.handle(AUTH_PROTOCOL, () => {}); + return node; +} + before(async () => { wasm = await loadBiscuit(); cpKeyPair = biscuitKeyPair(cpIdentity); @@ -605,6 +651,117 @@ test("join reports a router that refuses the handshake", async () => { } }); +test("a join bounded by its signal names the router whose handshake never answered", async () => { + const silent = await silentRouter(); + try { + const addr = (silent.getMultiaddrs()[0] as ReturnType).toString(); + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([addr]) }); + const started = Date.now(); + await assert.rejects(mesh.join({ signal: AbortSignal.timeout(500) }), (err: Error) => { + assert.match(err.message, new RegExp(`${silent.peerId.toString()}: auth handshake: `), `got ${JSON.stringify(err.message)}, want the router and the step it hung in`); + return true; + }); + const took = Date.now() - started; + assert.ok(took < AUTH_HANDSHAKE_TIMEOUT_MS, `join took ${took}ms, want the join's signal to end it before the ${AUTH_HANDSHAKE_TIMEOUT_MS}ms handshake timeout`); + } finally { + await silent.stop(); + } +}); + +test("a join stopped by its signal while syncing sends nothing more to the control plane", async () => { + const cp = stallingControlPlane(fakeControlPlane([routerAddr])); + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: cp.fetch }); + cp.stall(); + const started = Date.now(); + await assert.rejects(mesh.join({ signal: AbortSignal.timeout(500) }), (err: Error) => { + assert.match(err.message, /^join stopped while syncing with the control plane: /, `got ${JSON.stringify(err.message)}, want the step the join stopped in`); + assert.doesNotMatch(err.message, /127\.0\.0\.1:1/, `got ${JSON.stringify(err.message)}, want no control plane URL in the error`); + return true; + }); + const took = Date.now() - started; + assert.ok(took < JOIN_SYNC_TIMEOUT_MS, `join took ${took}ms, want the caller's signal to end it before the ${JOIN_SYNC_TIMEOUT_MS}ms sync timeout`); + // The aborted /keys request ends the sync: no /info, no refresh. + assert.deepEqual(cp.paths, ["/keys"], "requests after the control plane stalled"); +}); + +test("a join goes on with the stored credential when the control plane does not answer", async () => { + const cp = stallingControlPlane(fakeControlPlane([routerAddr])); + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: cp.fetch }); + cp.stall(); + const started = Date.now(); + const session = await mesh.join({ refreshLeadMs: 0, reserveRelay: false, controlPlaneSyncIntervalMs: 0 }); + try { + const took = Date.now() - started; + assert.ok(took < JOIN_TIMEOUT_MS, `join took ${took}ms, want the sync to give up after about ${JOIN_SYNC_TIMEOUT_MS}ms`); + assert.ok(session.isHeldRouter(router.peerId.toString()), `held routers ${session.routers.map((r) => r.peerId).join(",")}, want the router from the stored credential`); + } finally { + await session.close(); + } +}); + +test("a top-up whose signal has aborted tries no router and shuns none", async () => { + const second = await createLibp2p({ + addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [tls()], + streamMuxers: [yamux()], + services: { identify: identify(), relay: circuitRelayServer() }, + connectionManager: { inboundConnectionThreshold: Infinity }, + }); + const secondId = second.peerId.toString(); + const secondBiscuit = mint(secondId, ROLE_ROUTER); + await second.handle(AUTH_PROTOCOL, authStreamHandler({ ownBiscuit: () => secondBiscuit, trustedKeys: () => [cpKey] })); + const secondAddr = (second.getMultiaddrs()[0] as ReturnType).toString(); + const listed = [routerAddr]; + try { + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane(listed) }); + const session = await mesh.join({ refreshLeadMs: 0, reserveRelay: false, controlPlaneSyncIntervalMs: 0 }); + try { + // The control plane now lists a router the session has not tried. + listed.push(secondAddr); + await mesh.syncControlPlane(); + + const failures = await session.topUpRouters(AbortSignal.abort(new Error("join over"))); + assert.deepEqual(failures, [], "an aborted top-up reports no router it did not try"); + assert.equal(session.shunnedRouters.has(secondId), false, `${secondId} was shunned by a top-up that never tried it`); + + // The router was eligible all along: the same top-up without the signal holds it. + assert.deepEqual(await session.topUpRouters(), []); + assert.ok(session.isHeldRouter(secondId), `held routers ${session.routers.map((r) => r.peerId).join(",")}, want ${secondId} among them`); + } finally { + await session.close(); + } + } finally { + await second.stop(); + } +}); + +test("a join that runs out of its own time keeps the router that admitted it", async () => { + const silent = await silentRouter(); + try { + const silentId = silent.peerId.toString(); + const silentAddr = (silent.getMultiaddrs()[0] as ReturnType).toString(); + // The idle router is tried first, the silent one second. + const routers = [ + create(RouterInfoSchema, { peerId: router.peerId.toString(), addresses: [routerAddr], connections: 0, connectionLimit: 100 }), + create(RouterInfoSchema, { peerId: silentId, addresses: [silentAddr], connections: 90, connectionLimit: 100 }), + ]; + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([routerAddr, silentAddr], routers) }); + const started = Date.now(); + const session = await joinMesh(mesh, { refreshLeadMs: 0, reserveRelay: false, controlPlaneSyncIntervalMs: 0 }, 2_000); + try { + const took = Date.now() - started; + assert.ok(took < AUTH_HANDSHAKE_TIMEOUT_MS, `join took ${took}ms, want its own timeout to end it before the ${AUTH_HANDSHAKE_TIMEOUT_MS}ms handshake timeout`); + assert.deepEqual(session.routers.map((r) => r.peerId), [router.peerId.toString()], "held routers"); + assert.equal(session.shunnedRouters.has(silentId), false, `${silentId} was shunned for a handshake the join's timeout cut off`); + } finally { + await session.close(); + } + } finally { + await silent.stop(); + } +}); + test("an explicit rejection is reported with its reason", async () => { const reason = new AuthRejectedError("12D3KooWtest", "peer is revoked"); assert.match(reason.message, /12D3KooWtest.*peer is revoked/); diff --git a/sdk/js/src/session.ts b/sdk/js/src/session.ts index 3e3fe5d1..54c04377 100644 --- a/sdk/js/src/session.ts +++ b/sdk/js/src/session.ts @@ -125,7 +125,12 @@ export interface JoinOptions extends MeshHostOptions { * requiredLabels. Held on every call, MCP and HTTP alike; no call waives it. */ egressRequireLabels?: Record; - /** Bounds the whole join. */ + /** + * Bounds the whole join: the control plane sync, and the dial, auth + * handshake and relay reservation with every router tried. Without one the + * join is bounded by JOIN_TIMEOUT_MS (60s). A join that stops throws an + * error naming each router tried and the step it was in. + */ signal?: AbortSignal; } @@ -198,6 +203,17 @@ const DEFAULT_RELAY_CHECK_MS = 30 * 1000; const DEFAULT_ANNOUNCE_MS = 5 * 60 * 1000; /** What close() waits for the member's record to be withdrawn. */ const WITHDRAW_TIMEOUT_MS = 2 * 1000; +/** + * Bounds a join whose caller passed no signal: each step has a timeout of + * its own, but a join tries them in turn over every router it is offered. + */ +export const JOIN_TIMEOUT_MS = 60_000; +/** + * Bounds the control plane sync a join starts with. The sync is best effort, + * so a control plane that does not answer in time leaves the join to go on + * with the stored credential instead of using up the whole join. + */ +export const JOIN_SYNC_TIMEOUT_MS = 5_000; /** How long a provider's positive egress verdict is kept; agentmesh-node's labelGateTTL. */ const EGRESS_VERDICT_TTL_MS = 5 * 60 * 1000; @@ -394,7 +410,7 @@ export class MeshSession { * the mesh's record TTL. A member with no relayed address announces * nothing: there is nothing to reach it at. */ - async announce(): Promise { + async announce(signal?: AbortSignal): Promise { if (this.#closed || this.#isTaskView) { return; } @@ -402,7 +418,7 @@ export class MeshSession { if (addrs.length === 0) { return; } - await provide(this.#directoryRouters(), await peerKey(this.peerId), addrs); + await provide(this.#directoryRouters(), await peerKey(this.peerId), addrs, signal); this.#announced = addrs; } @@ -554,14 +570,15 @@ export class MeshSession { /** * Runs the handshake on a router connection once: the connection:open * listener and an explicit admission both call this, and the second waits - * for the first instead of running a second handshake. + * for the first instead of running a second handshake. The signal applies + * to a handshake this call starts; one already running keeps its own. */ - #authenticateRouter(conn: Connection): Promise { + #authenticateRouter(conn: Connection, signal?: AbortSignal): Promise { const pending = this.#routerHandshakes.get(conn.id); if (pending !== undefined) { return pending; } - const run = authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys).finally(() => { + const run = authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys, signal).finally(() => { this.#routerHandshakes.delete(conn.id); }); this.#routerHandshakes.set(conn.id, run); @@ -681,6 +698,8 @@ export class MeshSession { */ async topUpRouters(signal?: AbortSignal): Promise { const failures: string[] = []; + // A function, not a narrowed value: the signal aborts across the awaits below. + const aborted = () => signal?.aborted === true; const want = this.#wantRouters(); const candidates = this.#candidatesForHolding(); const tried = new Set(); @@ -698,21 +717,36 @@ export class MeshSession { } let progress = false; for (const cand of picks) { + // A router not dialed yet is not tried under a signal that has + // already aborted: it would fail at once and be shunned for nothing. + if (aborted()) { + return failures; + } tried.add(cand.peerId); + // Each step has a timeout of its own; the failure names the one the + // router was in, so a join that hangs says where. + let step = `dial ${cand.addresses.map(String).join(", ")}`; try { const conn = await this.node.dial(cand.addresses, signal !== undefined ? { signal } : {}); - const credential = await this.#authenticateRouter(conn); + step = "auth handshake"; + const credential = await this.#authenticateRouter(conn, signal); requireRole(credential, ROLE_ROUTER); const addr = connectedAddress(conn, cand.peerId); if (this.#reserveRelay) { + step = "relay reservation"; await this.#reserveOnRouter(cand.peerId, addr); } this.#upsertHeldRouter({ peerId: cand.peerId, addr, credential }); progress = true; } catch (err) { - this.#shunRouter(cand.peerId, ROUTER_SHUN_DURATION_MS); + failures.push(`${cand.peerId}: ${step}: ${errorText(err)}`); await this.node.hangUp(peerIdFromString(cand.peerId)).catch(() => {}); - failures.push(`${cand.peerId}: ${err instanceof Error ? err.message : String(err)}`); + // A router cut off by the signal did not fail on its own, so it is + // not shunned. + if (aborted()) { + return failures; + } + this.#shunRouter(cand.peerId, ROUTER_SHUN_DURATION_MS); } } if (!progress && picks.length < need) { @@ -1408,8 +1442,14 @@ export function selectRouters(candidates: readonly RouterCandidate[], k: number, return scored.slice(0, k).map((s) => s.candidate); } -/** Implements AgentMesh.join(); lives here to keep mesh.ts free of libp2p. */ -export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Promise { +/** + * Implements AgentMesh.join(); lives here to keep mesh.ts free of libp2p. + * timeoutMs bounds a join whose caller passed no signal. + */ +export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}, timeoutMs = JOIN_TIMEOUT_MS): Promise { + const signal = options.signal ?? AbortSignal.timeout(timeoutMs); + // Why the join stopped: the caller's reason, or its own timeout. + const stopReason = () => (options.signal !== undefined ? errorText(signal.reason) : `timed out after ${timeoutMs}ms`); // The pull agentmesh-node makes before it starts: a member resuming from its // state directory after a key rotation would otherwise verify the routers, // which already hold credentials under the new key, against the keys it @@ -1417,9 +1457,13 @@ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Prom // once joined. Best effort; the stored credential serves when the control // plane cannot be reached. try { - await mesh.syncControlPlane(); + await mesh.syncControlPlane(AbortSignal.any([signal, AbortSignal.timeout(JOIN_SYNC_TIMEOUT_MS)])); } catch { - // Joining goes on with what the credential holds. + // Joining goes on with what the credential holds, unless the caller + // stopped the join while the control plane kept it waiting. + if (signal.aborted) { + throw new Error(`join stopped while syncing with the control plane: ${stopReason()}`); + } } const rawAddrs = options.routerAddresses ?? mesh.credential.routerAddresses; if (rawAddrs.length === 0 && mesh.routers.length === 0) { @@ -1455,14 +1499,18 @@ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Prom ); session = new MeshSession(mesh, node, [], authenticatedPeers, banned, options); - const failures = await session.topUpRouters(options.signal); - if (session.routers.length === 0) { + const failures = await session.topUpRouters(signal); + // One admitted router is enough to be on the mesh; only the caller's own + // signal turns that into a failure. + if (options.signal?.aborted === true || session.routers.length === 0) { const detail = failures.length > 0 ? `:\n ${failures.join("\n ")}` : ""; - throw new Error(`no router admitted this member${detail}`); + throw new Error(signal.aborted ? `join stopped: ${stopReason()}${detail}` : `no router admitted this member${detail}`); } // Best effort: a router that does not take the record is a warning the - // next announcement retries; the member is on the mesh regardless. - await session.announce().catch(() => {}); + // next announcement retries; the member is on the mesh regardless. Under + // the caller's signal only: a join that ran out of its own time still + // announces the routers it holds. + await session.announce(options.signal).catch(() => {}); return session; } catch (err) { if (session !== undefined) { @@ -1474,6 +1522,10 @@ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Prom } } +function errorText(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + /** The peer a multiaddr ends at, in canonical form: its trailing `/p2p/`, or undefined for a relay address with no target yet. */ function targetPeerOf(ma: Multiaddr): string | undefined { const last = ma.getComponents().at(-1);