gsadmin/sencho
1
0
Fork 0
mirror of https://github.com/Studio-Saelix/sencho.git synced 2026-08-06 00:47:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mesh): peer-initiated callback bridge installs reverse dialer (#1071)

Browse Source 
* fix(mesh): peer-initiated callback bridge installs reverse dialer

The R1-A2 symmetric-dial work added a peer-side dialer that opens a
callback WS to central at `/api/mesh/proxy-tunnel-from-peer`. The WS
upgrade, JWT validation, and `mesh_centrals` persistence all worked,
but the peer-side handler put the wrong object on its end of the pipe:
a `PilotTunnelBridge` (the central-role object) rather than a
`TcpStreamSwitchboard` with a registered `reverseDialer`.

The design doc states: "Central retains PilotTunnelBridge ownership;
peer retains TcpStreamSwitchboard + reverseDialer ownership." The
central-initiated handler at `meshProxyTunnel.ts:115-163` honors this.
The peer-initiated handler at `PeerToCentralMeshSessionDialer.attachBridge`
did not. It created a `PilotTunnelBridge(0, ws)` and stored it in a
private `currentSession` field. Result: `MeshService.reverseDialer`
stayed null, `dialMeshTcpStream` fell through to
`PilotTunnelManager.ensureBridge(target.nodeId)`, and `NodeRegistry.
getProxyTarget` on the peer returned null because proxy-mode peers
do not enroll their central. Cross-fleet dispatch from a peer
container failed with `proxy-tunnel.open.fail nodeId=<central>
reason=no_target` even though the callback bridge was up and healthy
(`bridgeOpen: true`, `last_used_at` populated).

Replace the peer-side `PilotTunnelBridge` with the same wiring the
central-initiated handler uses:

  - `attachTcpStreamSwitchboard` with `resolveByComposeLabels` (allows
    inbound `tcp_open` from central if it ever uses the callback
    bridge for a reverse-direction dispatch; matches the central-
    initiated pipe's resolver).
  - A `SwitchboardReverseDialer` that delegates `openMeshTcpStream`
    to `switchboard.openReverseStream`.
  - `MeshService.setReverseDialer(localDialer, null)` with the same
    CAS guard the central-initiated handler uses, so a concurrent
    central-initiated tunnel does not get silently overwritten.
  - `ws.on('message', onMessage)` dispatches JSON / binary frames to
    the switchboard.
  - `ws.once('close'/'error', teardown)` clears the switchboard, the
    reverseDialer, and the `currentSession` reference idempotently.

The `currentSession` field type changes from `PilotTunnelBridge | null`
to `TcpStreamSwitchboard | null`. Callers in `MeshService.openCrossNode`
only use the return value for truthiness; the cast is harmless. The
public `hasSession()` signature is unchanged.

A new `currentWs` field holds the WS reference so `resetForTest` can
close the connection on teardown (the switchboard's own `cleanup`
does not close its WS).

Test update mirrors the production wiring: the existing
"marks the row used on successful WS open" case now asserts the
return value is a `TcpStreamSwitchboard`, that `MeshService.reverseDialer`
is non-null after `ensureSession`, and explicitly closes the WS in
the test's finally so `wss.close` can resolve.

Pre-existing test logic that asserted `bridge instanceof PilotTunnelBridge`
on the same path is removed in favor of the new switchboard assertion.

Discovered during the v0.81.1 live verification: the peer-side
`route.dispatch cross-node` event always paired with
`proxy-tunnel.open.fail no_target` and `tunnel.fail`, even though
`centralCallback.bridgeOpen: true` and `mesh_centrals.last_used_at`
were populated. The callback path opened cleanly in isolation but
the dispatch path never consumed it. This fix makes the dispatch
path consume it.

Tests:
  - cd backend && npx tsc --noEmit          clean
  - cd backend && npx vitest run mesh        170/170 green
  - cd backend && npx vitest run             2259/2268 green; only
    failures are the pre-existing Windows EBUSY flake in
    filesystem-backup.test.ts (documented in prior handoff)

* fix(mesh): drop unused imports and align reverse-dialer interface name

CI lint failures on the previous push were two @typescript-eslint/no-unused-vars
errors in the test file:

  24:76  'vi' is defined but never used
  31:5   'TcpStreamSwitchboardCtor' is assigned a value but only used as a type

Both were collateral from a local-debugging simplification: vi.waitFor was
swapped for a synchronous assertion when the test's wss.close hang was traced
to an unclosed client WS; the toBeInstanceOf assertion that used the runtime
binding was removed at the same time. Restore the instanceof assertion next
to the existing not.toBeNull check so the runtime binding is in use and the
test guards against future regressions where the wrong end-of-pipe object
type is returned on the callback bridge (the bug this PR fixes). Drop the
now-unused `vi` symbol from the vitest import.

Also fold in a /simplify convergent finding: rename the local
CallbackReverseDialer interface to SwitchboardReverseDialer to match the
sibling declaration in meshProxyTunnel.ts:72. Same shape, same name; readers
of either handler now see the same mental model. Cross-file extraction of
the interface to tcpStreamSwitchboard.ts is the cleaner long-term shape but
touches a file outside this PR's scope; filing as a follow-up.

Considered-and-deferred findings from the /simplify pass (file as separate
PRs to keep one-branch-one-concern):
  - extract shared attachSwitchboard helper (~40 duplicated lines between
    meshProxyTunnel.ts and PeerToCentralMeshSessionDialer.ts)
  - public TcpStreamSwitchboard.getWs accessor + drop currentWs field
  - public MeshService.hasReverseDialer for the test bracket-cast
  - shared close-reason constants (drift risk across the two handlers)

Tests:
  - cd backend && npx tsc --noEmit          clean
  - cd backend && npx eslint <changed>      clean (lint_exit=0)
  - cd backend && npx vitest run            mesh 170/170, dialer 8/8 green
This commit is contained in:
Anso
2026-05-16 20:11:31 -04:00
committed by GitHub
parent 6354cb3387
commit 0b4cf90bf1
2 changed files with 146 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/__tests__/peer-to-central-mesh-session-dialer.test.ts
+37 -12
View File
@@ -20,15 +20,16 @@
*/
import http from 'http';
import type { AddressInfo } from 'net';
import { WebSocketServer } from 'ws';
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { WebSocketServer, type WebSocket as WsClient } from 'ws';
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
let tmpDir: string;
let PeerToCentralMeshSessionDialer: typeof import('../services/PeerToCentralMeshSessionDialer').PeerToCentralMeshSessionDialer;
let MeshCentralRegistry: typeof import('../services/MeshCentralRegistry').MeshCentralRegistry;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
let PilotTunnelBridge: typeof import('../services/PilotTunnelBridge').PilotTunnelBridge;
let TcpStreamSwitchboardCtor: typeof import('../mesh/tcpStreamSwitchboard').TcpStreamSwitchboard;
let MeshService: typeof import('../services/MeshService').MeshService;
interface RejectingServer {
server: http.Server;
@@ -79,7 +80,8 @@ beforeAll(async () => {
({ PeerToCentralMeshSessionDialer } = await import('../services/PeerToCentralMeshSessionDialer'));
({ MeshCentralRegistry } = await import('../services/MeshCentralRegistry'));
({ DatabaseService } = await import('../services/DatabaseService'));
({ PilotTunnelBridge } = await import('../services/PilotTunnelBridge'));
({ TcpStreamSwitchboard: TcpStreamSwitchboardCtor } = await import('../mesh/tcpStreamSwitchboard'));
({ MeshService } = await import('../services/MeshService'));
});
afterAll(() => {
@@ -207,7 +209,7 @@ describe('PeerToCentralMeshSessionDialer', () => {
}
});
it('marks the row used on successful WS open', async () => {
it('marks the row used on successful WS open and installs a reverseDialer', async () => {
const wss = new WebSocketServer({ noServer: true });
const srv = http.createServer();
srv.on('upgrade', (req, socket, head) => {
@@ -220,7 +222,10 @@ describe('PeerToCentralMeshSessionDialer', () => {
await new Promise<void>((resolve) => srv.listen(0, '127.0.0.1', () => resolve()));
const port = (srv.address() as AddressInfo).port;
const url = `http://127.0.0.1:${port}`;
let bridge: InstanceType<typeof PilotTunnelBridge> | null = null;
// Start with a clean reverseDialer slot on the local MeshService so
// setReverseDialer's CAS swap succeeds inside attachSwitchboard.
MeshService.getInstance().setReverseDialer(null);
let switchboard: InstanceType<typeof TcpStreamSwitchboardCtor> | null = null;
try {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-success',
@@ -229,14 +234,34 @@ describe('PeerToCentralMeshSessionDialer', () => {
jwtIssuedAt: 1,
jwtExpiresAt: 9999999999,
});
bridge = await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
expect(bridge).toBeInstanceOf(PilotTunnelBridge);
await vi.waitFor(() => {
expect(MeshCentralRegistry.getInstance().getActive()?.lastUsedAt ?? 0).toBeGreaterThan(0);
});
switchboard = await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
expect(switchboard).not.toBeNull();
// The R1-A2 design puts a TcpStreamSwitchboard on the peer end of
// this WS, not a PilotTunnelBridge (which is central's role). A
// future regression that returns the wrong shape would surface
// here before the more abstract reverseDialer-installed check.
expect(switchboard).toBeInstanceOf(TcpStreamSwitchboardCtor);
expect(PeerToCentralMeshSessionDialer.getInstance().hasSession()).toBe(true);
// markUsed is called synchronously inside attachSwitchboard before
// ensureSession resolves, so the DB row reflects it immediately.
expect(MeshCentralRegistry.getInstance().getActive()?.lastUsedAt ?? 0).toBeGreaterThan(0);
// The R1-A2 wiring under test: MeshService.reverseDialer must be
// populated so MeshService.dialMeshTcpStream routes peer-side
// cross-fleet traffic through this callback bridge instead of
// falling through to PilotTunnelManager.ensureBridge(centralId),
// which has no record for central on a proxy peer.
const meshSvc = MeshService.getInstance() as unknown as { reverseDialer: unknown };
expect(meshSvc.reverseDialer).not.toBeNull();
} finally {
try { bridge?.close(1000, 'test done'); } catch { /* ignore */ }
try { switchboard?.cleanup('test done'); } catch { /* ignore */ }
// The dialer owns the client-side WS and the switchboard doesn't
// close it on cleanup; tear it down explicitly so wss.close can
// resolve (it waits for all client connections to disconnect).
try {
const inst = PeerToCentralMeshSessionDialer.getInstance() as unknown as { currentWs: WsClient | null };
inst.currentWs?.close(1000, 'test done');
} catch { /* ignore */ }
MeshService.getInstance().setReverseDialer(null);
await new Promise<void>((resolve) => wss.close(() => resolve()));
await new Promise<void>((resolve) => srv.close(() => resolve()));
}
backend/src/services/PeerToCentralMeshSessionDialer.ts
+109 -17
View File
@@ -21,13 +21,28 @@
*/
import { EventEmitter } from 'events';
import WebSocket from 'ws';
import { MAX_FRAME_SIZE_BYTES } from '../pilot/protocol';
import {
MAX_FRAME_SIZE_BYTES,
decodeBinaryFrame,
decodeJsonFrame,
wsDataToBuffer,
wsDataToString,
} from '../pilot/protocol';
import { MeshCentralRegistry } from './MeshCentralRegistry';
import { PilotTunnelBridge } from './PilotTunnelBridge';
import {
attachTcpStreamSwitchboard,
resolveByComposeLabels,
type TcpStreamSwitchboard,
type ReverseTcpStreamHandle,
} from '../mesh/tcpStreamSwitchboard';
import { PilotMetrics } from './PilotMetrics';
import { httpUrlToWs } from '../utils/wsUrl';
import { sanitizeForLog } from '../utils/safeLog';
interface SwitchboardReverseDialer {
openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null;
}
const HANDSHAKE_TIMEOUT_MS = 15_000;
const RATE_LIMIT_WINDOW_MS = 60_000;
const RATE_LIMIT_MAX = 5;
@@ -59,8 +74,9 @@ interface DialError extends Error {
export class PeerToCentralMeshSessionDialer extends EventEmitter {
private static instance: PeerToCentralMeshSessionDialer | null = null;
private currentSession: PilotTunnelBridge | null = null;
private inflight: Promise<PilotTunnelBridge | null> | null = null;
private currentSession: TcpStreamSwitchboard | null = null;
private currentWs: WebSocket | null = null;
private inflight: Promise<TcpStreamSwitchboard | null> | null = null;
private recentDials: number[] = [];
private endpointUnavailableUntil = 0;
@@ -73,7 +89,8 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter {
public static resetForTest(): void {
if (this.instance) {
try { this.instance.currentSession?.close(1000, 'test reset'); } catch { /* ignore */ }
try { this.instance.currentSession?.cleanup('test reset'); } catch { /* ignore */ }
try { this.instance.currentWs?.close(1000, 'test reset'); } catch { /* ignore */ }
}
this.instance = null;
}
@@ -82,7 +99,7 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter {
return this.currentSession !== null;
}
public async ensureSession(): Promise<PilotTunnelBridge | null> {
public async ensureSession(): Promise<TcpStreamSwitchboard | null> {
if (this.currentSession) return this.currentSession;
if (Date.now() < this.endpointUnavailableUntil) return null;
if (this.isRateLimited()) return null;
@@ -97,7 +114,7 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter {
return this.recentDials.length >= RATE_LIMIT_MAX;
}
private async dial(): Promise<PilotTunnelBridge | null> {
private async dial(): Promise<TcpStreamSwitchboard | null> {
const material = MeshCentralRegistry.getInstance().getActive();
if (!material) return null;
this.recentDials.push(Date.now());
@@ -115,25 +132,100 @@ export class PeerToCentralMeshSessionDialer extends EventEmitter {
this.handleDialFailure(err, material.centralInstanceId);
return null;
}
return this.attachBridge(ws, material.centralInstanceId);
return this.attachSwitchboard(ws, material.centralInstanceId);
}
private async attachBridge(ws: WebSocket, instanceId: string): Promise<PilotTunnelBridge | null> {
const bridge = new PilotTunnelBridge(0, ws);
/**
* Wire the peer-initiated callback WS into the local MeshService. The
* R1-A2 design puts the peer end of the bridge in TcpStreamSwitchboard
* mode (peer multiplexes streams; central side runs PilotTunnelBridge).
* Without this wiring the WS opens cleanly but MeshService.reverseDialer
* stays null, so MeshService.dialMeshTcpStream falls through to
* PilotTunnelManager.ensureBridge(centralNodeId), which has no record
* for central on a proxy-mode peer (peers do not enroll central) and
* fails with proxy-tunnel.open.fail reason=no_target. End state matches
* the v0.78.1 reverse-direction failure even though the callback bridge
* is alive.
*
* Wiring symmetric to the central-initiated handler at
* `meshProxyTunnel.ts:115-163`:
* - attachTcpStreamSwitchboard with the same compose-label resolver
* - SwitchboardReverseDialer that delegates to switchboard.openReverseStream
* - setReverseDialer(localDialer, null) with CAS so a concurrent
* central-initiated tunnel does not get silently overwritten
* - ws.on('message') dispatches JSON/binary frames to the switchboard
* - ws.on('close'/'error') tears down switchboard + clears reverseDialer
*/
private async attachSwitchboard(ws: WebSocket, instanceId: string): Promise<TcpStreamSwitchboard | null> {
let switchboard: TcpStreamSwitchboard;
try {
await bridge.start();
} catch {
try { bridge.close(1011, 'bridge start failed'); } catch { /* ignore */ }
switchboard = attachTcpStreamSwitchboard({
ws,
resolveTarget: resolveByComposeLabels,
logLabel: 'MeshCallback',
});
} catch (err) {
try { ws.close(1011, 'switchboard attach failed'); } catch { /* ignore */ }
PilotMetrics.increment('mesh_callback_dials_failed_total');
console.warn(`[PeerToCentralMeshSessionDialer] attach failed: ${sanitizeForLog((err as Error).message)}`);
return null;
}
const { MeshService } = await import('./MeshService');
const meshService = MeshService.getInstance();
const localDialer: SwitchboardReverseDialer = {
openMeshTcpStream(target) {
return switchboard.openReverseStream(target);
},
};
const installed = meshService.setReverseDialer(localDialer, null);
if (!installed) {
console.warn('[PeerToCentralMeshSessionDialer] reverse dialer already installed; rejecting concurrent callback bridge');
switchboard.cleanup('reverse dialer already installed');
try { ws.close(1013, 'reverse dialer already installed'); } catch { /* ignore */ }
PilotMetrics.increment('mesh_callback_dials_failed_total');
return null;
}
bridge.once('closed', () => {
if (this.currentSession === bridge) this.currentSession = null;
const onMessage = (data: unknown, isBinary: boolean): void => {
try {
if (isBinary) {
const buf = wsDataToBuffer(data);
if (!buf) return;
switchboard.handleBinaryFrame(decodeBinaryFrame(buf));
return;
}
const text = wsDataToString(data);
if (text == null) return;
switchboard.handleJsonFrame(decodeJsonFrame(text));
} catch (err) {
console.warn(`[PeerToCentralMeshSessionDialer] malformed frame: ${sanitizeForLog((err as Error).message)}`);
}
};
let tornDown = false;
const teardown = (): void => {
if (tornDown) return;
tornDown = true;
ws.off('message', onMessage);
try { switchboard.cleanup('mesh callback bridge closed'); } catch { /* ignore */ }
meshService.setReverseDialer(null, localDialer);
if (this.currentSession === switchboard) this.currentSession = null;
if (this.currentWs === ws) this.currentWs = null;
};
ws.on('message', onMessage);
ws.once('close', teardown);
ws.once('error', (err) => {
console.warn(`[PeerToCentralMeshSessionDialer] ws error: ${sanitizeForLog(err.message)}`);
teardown();
});
this.currentSession = bridge;
this.currentSession = switchboard;
this.currentWs = ws;
PilotMetrics.increment('mesh_central_bootstraps_total');
MeshCentralRegistry.getInstance().markUsed(instanceId);
return bridge;
return switchboard;
}
private awaitOpen(ws: WebSocket): Promise<void> {