SyncNode.pair(ticket) swaps the tunnel listener and restarts the embedded Jazz server on the SAME port, so browser clients reconnect to an unchanged URL. Ticket resolution happens before any teardown; concurrent pair() calls are rejected. Tested: unpaired leaf pairs to a live root, keeps its port, root observes the new inbound tunnel.
- commit
- 8c7e9edbf680d5aba9f079947fa5a10d39ef85cf
- parent
- f45b5adb2037256f75568af6c8ef5cca74daf0f6
- author
- Dami <dami.etoile@gmail.com>
| M | src/ | 62 | +++++++++++++++++++++++------- |
| M | tests/ | 35 | +++++++++++++++++ |
@@ -52,12 +52,16 @@ export interface SyncNode {
5252 appId: string;5353 /** Pairing ticket for this node; throws MeshUnavailableError if mesh is down. */5454 ticket(): string;55+ /** Re-elect the upstream to a peer AT RUNTIME: swaps the tunnel and56+ * restarts the embedded Jazz server on the SAME port, so browser clients57+ * simply reconnect. Throws MeshUnavailableError if the mesh is down. */58+ pair(peerTicket: string): Promise<void>;5559 status(): SyncNodeStatus;5660 stop(): Promise<void>;5761 }5862 5963 export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode> {60- const upstream: UpstreamConfig = options.upstream ?? "none";64+ let upstream: UpstreamConfig = options.upstream ?? "none";6165 const meshMode = options.mesh ?? "auto";6266 const inMemory = options.inMemory ?? options.dataDir === undefined;6367 @@ -88,35 +92,44 @@ export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode
8892 }8993 9094 // 2. Resolve the upstream election to a concrete URL for JazzServer.91- let upstreamUrl: string | undefined;9295 let tunnelListener: TunnelListener | null = null;93- if (upstream !== "none" && "url" in (upstream as object)) {94- upstreamUrl = (upstream as { url: string }).url;95- } else if (upstream !== "none" && "peer" in (upstream as object)) {96- const ticket = (upstream as { peer: string }).peer;96+ function resolveUpstream(97+ election: UpstreamConfig,98+ ): { url: string | undefined; listener: TunnelListener | null } {99+ if (election === "none") return { url: undefined, listener: null };100+ if ("url" in (election as object)) {101+ return { url: (election as { url: string }).url, listener: null };102+ }103+ const ticket = (election as { peer: string }).peer;97104 if (!irohNode) {98105 const reason = mesh.state === "unavailable" ? mesh.reason : "mesh is off";99106 throw new MeshUnavailableError(`peer upstream requires the mesh: ${reason}`);100107 }101- tunnelListener = startTunnelListener(irohNode, ticket);102- upstreamUrl = `ws://127.0.0.1:${tunnelListener.port}`;108+ const listener = startTunnelListener(irohNode, ticket);109+ return { url: `ws://127.0.0.1:${listener.port}`, listener };103110 }104111 105- // 3. Jazz server.106- let jazz: JazzHandle;107- try {108- jazz = await startJazz({112+ function startJazzFor(upstreamUrl: string | undefined, port: number | undefined) {113+ return startJazz({109114 appId: options.appId,110115 backendSecret: options.backendSecret,111116 adminSecret: options.adminSecret,112- port: options.listen?.port,117+ port,113118 dataDir: options.dataDir ? `${options.dataDir}/jazz` : undefined,114119 inMemory,115120 upstreamUrl,116121 allowLocalFirstAuth: options.allowLocalFirstAuth,117122 });123+ }124+125+ // 3. Jazz server.126+ let jazz: JazzHandle;127+ const initial = resolveUpstream(upstream);128+ tunnelListener = initial.listener;129+ try {130+ jazz = await startJazzFor(initial.url, options.listen?.port);118131 } catch (e) {119- await tunnelListener?.close();132+ await initial.listener?.close();120133 await irohNode?.close();121134 throw e;122135 }@@ -128,6 +141,7 @@ export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode
128141 }129142 130143 let stopped: Promise<void> | null = null;144+ let pairing: Promise<void> | null = null;131145 return {132146 url: jazz.url,133147 port: jazz.port,@@ -139,6 +153,26 @@ export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode
139153 }140154 return ticketString;141155 },156+ pair: (peerTicket: string) => {157+ if (stopped) return Promise.reject(new Error("node is stopped"));158+ if (pairing) return Promise.reject(new Error("a pair() is already in progress"));159+ pairing = (async () => {160+ try {161+ const keptPort = jazz.port;162+ // resolveUpstream throws (MeshUnavailableError) BEFORE teardown.163+ const next = resolveUpstream({ peer: peerTicket });164+ const oldListener = tunnelListener;165+ tunnelListener = next.listener;166+ await oldListener?.close();167+ await jazz.stop();168+ jazz = await startJazzFor(next.url, keptPort);169+ upstream = { peer: peerTicket };170+ } finally {171+ pairing = null;172+ }173+ })();174+ return pairing;175+ },142176 status: () => ({143177 appId: options.appId,144178 jazz: { url: jazz.url, port: jazz.port, storage: inMemory ? "memory" : "persistent" },@@ -68,3 +68,38 @@ Deno.test({
6868 await root.stop();6969 },7070 });71+72+Deno.test({73+ name: "runtime pair(): late upstream election without losing the port",74+ ignore: !available,75+ sanitizeOps: false,76+ sanitizeResources: false,77+ fn: async () => {78+ const shared = {79+ appId: crypto.randomUUID(),80+ backendSecret: "lofi_backend_latepair",81+ adminSecret: "lofi_admin_latepair",82+ };83+ const root = await createSyncNode({ ...shared, inMemory: true });84+ const leaf = await createSyncNode({ ...shared, inMemory: true }); // unpaired85+ const portBefore = leaf.port;86+ assertEquals(leaf.status().upstream, "none");87+88+ await leaf.pair(root.ticket());89+90+ assertEquals(leaf.port, portBefore, "port survives re-pairing");91+ assert("peer" in (leaf.status().upstream as object), "upstream re-elected");92+ assert(await healthy(leaf.url), "leaf healthy on the same URL after pair");93+94+ // Jazz eagerly dials the new upstream through the tunnel.95+ await new Promise((r) => setTimeout(r, 3000));96+ const rootMesh = root.status().mesh;97+ assert(98+ rootMesh.state === "up" && rootMesh.connections.some((c) => c.direction === "in"),99+ "root sees the newly paired leaf's tunnel",100+ );101+102+ await leaf.stop();103+ await root.stop();104+ },105+});