feat(mesh): route mesh traffic over Distributed API remotes (#1048)

* refactor(mesh): extract shared TCP stream switchboard

Pull the `tcp_open` / `tcp_open_ack` / `tcp_open_reverse` / `tcp_close`
+ `TcpData` handling out of the pilot agent into a reusable module so a
second caller (the upcoming proxy-mode WS handler) can run the same
frame parser, stream allocator, idle timers, and Compose-label
resolver. One parser, two callers, zero drift on a
security-sensitive protocol surface.

The pilot agent keeps its public `openMeshTcpStream` API and delegates
to a per-connection switchboard constructed in `connect()` and torn
down in `cleanupAfterDisconnect`. Existing reverse-stream and resolver
tests are rewritten to exercise the shared module directly.

* feat(mesh): route mesh traffic over Distributed API remotes

Sencho Mesh now works against remotes in Distributed API mode in
addition to Pilot Agent mode. Central opens a short-lived WebSocket
tunnel to the remote's new `/api/mesh/proxy-tunnel` endpoint on demand
and tears it down after 5 minutes of idle (configurable via
`SENCHO_MESH_PROXY_TUNNEL_IDLE_MS`). Mesh dispatch is mode-agnostic at
the routing layer; `PilotTunnelManager.ensureBridge` resolves an
existing tunnel or asks the new `MeshProxyTunnelDialer` to dial.

The Routing tab badges now use a `reachableMode` classifier: `★ Local`
for local, `pilot offline` red badge for a pilot with a down tunnel,
and a new `unreachable` red badge surfacing the specific reason
(missing token, scope not full-admin, TLS failure, remote does not
support proxy mesh). Distributed API remotes with valid credentials
show no negative badge; the tunnel opens on first dial.

Auth: the proxy-tunnel WS upgrade requires an `Authorization: Bearer`
API token with the `full-admin` scope. Lower-scoped tokens are
rejected at upgrade time so a leaked read-only token cannot reach the
mesh data plane.

Bidirectional: the proxy-mode WS handler registers itself as the
local `MeshService` reverse dialer (compare-and-swap), so meshed
containers on a Distributed API remote can dial cross-node aliases
via `tcp_open_reverse` over the same tunnel. Cross-node relays
(`PilotTunnelBridge.acceptReverseRelay`) await `ensureBridge` so
proxy-to-proxy mesh works without standing tunnels.

* docs(mesh): describe Distributed API mesh and unreachable troubleshooting

Refresh the user-facing mesh documentation to reflect that mesh works
over both Pilot Agent and Distributed API remotes. Adds a
Troubleshooting accordion entry for the new `unreachable` Routing tab
badge so operators can match a tooltip reason ("api token rejected
(scope must be full-admin)", "remote does not support proxy mesh",
"TLS handshake failed", "api_url not set", "api token missing") to a
concrete fix.

* refactor(mesh): apply /simplify review findings

Five cleanups surfaced by the post-implementation code review pass.
No behaviour change for fleets running on `main`; only internal
structure improves.

- **Deterministic container IP shared.** Extract the compose-default
  network preference (`pickContainerIp`) and the conventional-name
  fast path (`lookupContainerIp`) into `backend/src/mesh/containerLookup.ts`.
  Both `MeshService.resolveContainerIp` (existing same-node fast
  path) and the switchboard's `resolveByComposeLabels` (proxy/pilot
  inbound dial) now use the same logic. Earlier the switchboard
  helper grabbed the first `Object.values(Networks)` IP, which
  could flip across daemon versions on multi-network containers;
  fixed.
- **WS URL upgrade extracted.** New `backend/src/utils/wsUrl.ts`
  exposes `httpUrlToWs(baseUrl)` that maps `http://` to `ws://` and
  `https://` to `wss://`. Used in both `pilot/agent.ts` and the
  proxy-tunnel dialer. The previous inline `replace(/^http/, 'ws')`
  silently downgraded `https://` to `ws://` (cleartext).
- **`computeReachable` takes the pre-fetched node.** `getStatus`
  already iterated `db.getNodes()`; the helper previously re-queried
  by id per node (N+1 reads). Pass the row in.
- **Reachable-reason ternary -> const map.** Replace the nested
  ternary in `MeshService.computeReachable` with a
  `Record<DialFailureCode, string>` lookup keyed on the dialer's
  failure code.
- **Activity-type ternary -> const map.** Same flattening inside
  `MeshProxyTunnelDialer.logActivity`.

All mesh tests pass (85/85). Full backend suite green (2126/2126).

* fix(mesh): cache proxy dial failures and contain WS handshake errors

`ensureBridge` now consults the recent-failure cache before dialing so a
continuous mesh workload against a misconfigured proxy-mode remote does
not produce one upgrade attempt per cross-node TCP open. `recordFailure`
emits at most one activity-log entry per cache window per (nodeId, code)
so a connect-loop on a single bad remote cannot flush the ring buffer.
Failure messages run through `redactSensitiveText` before reaching the
log so any embedded Bearer / JWT / inline-URL credentials are scrubbed.

`stop()` clears the inflight map and `dial()` checks `this.stopped`
between awaits so a shutdown does not leak a half-opened bridge.

When `awaitOpen` rejects (401, 4403, TLS), the dialer attaches a noop
'error' listener before calling `ws.close()`. Without it the ws library
emits a tail 'error' on a still-CONNECTING socket that propagates as an
unhandled exception. Surfaced by a live-network test against a real
proxy-mode peer.

Adds `mesh-proxy-tunnel-live.test.ts` (skipped unless MESH_AUDIT_URL
and MESH_AUDIT_TOKEN_FILE env vars are set) covering both the happy
path and the auth-rejected path against an actual remote Sencho.

* feat(mesh): instrument proxy tunnel observability

Adds three counters to `PilotMetrics`:
- `proxy_bridges_total` (incremented on `registerProxyBridge` success)
- `proxy_dials_failed` (every failed dial attempt, not deduped)
- `proxy_idle_closes` (idle-sweep teardowns)

All three surface automatically via `GET /api/system/pilot-tunnels`
since the route returns the full `Counters` snapshot.

Gates two pre-existing always-on `console.warn` calls in
`tcpStreamSwitchboard.ts` (mid-stream socket errors and Docker resolve
failures) behind `isDebugEnabled()`. Both fire on per-stream events and
would otherwise violate the diagnostic-log safety rule under load.

Adds `mesh-tcp-stream-switchboard.test.ts` covering the forward
`tcp_open` path: real localhost dial, resolver errors (no_target,
denied), per-tunnel cap saturation, frame-routing fall-through
invariants, `tcp_close` socket teardown.

Drops `MESH_CONNECT_TIMEOUT_MS` from the public surface; only the
switchboard itself uses it.

* fix(mesh): tighten Routing tab unreachable handling

`RoutingNodeCard.tsx`: the **Add stack to mesh** button now disables
when `reachableMode === 'unreachable'`, matching the existing
TogglePill behavior. Without this, an operator on an unreachable node
could open the opt-in sheet, confirm, and watch the redeploy proceed
against a target whose mesh data plane will silently fail to route.
Also drops the leftover `status.nodeId !== -1` guard on the pilot-
offline badge.

`MeshService.ts`: renames `isMeshReachable` to `isMeshConfigured` with
the new predicate that returns true for proxy-mode remotes whose creds
are valid (the tunnel is opened on demand). `getRouteDiagnostic` now
distinguishes "routable" (configured) from "pilotLive" (live tunnel
state, only meaningful for pilot mode); without the split, every idle
proxy-mode route would report `tunnel down`.

`MeshService.setReverseDialer` warns when the unconditional install
path silently overwrites a non-null current dialer. By topology a
Sencho is either pilot or central, so the branch flags a misconfigured
deployment rather than an expected race.

Drops the dead `export { ReverseTcpStreamHandle }` re-export from
`pilot/agent.ts`; its only consumer imports straight from
`mesh/tcpStreamSwitchboard.ts`. Fixes a stale doc comment in
`MeshService.ts` that referenced the wrong source file.

Adds `mesh-proxy-tunnel-handler.test.ts` covering the WS handler
lifecycle: pilot-mode 404 rejection, post-upgrade reverse-dialer
install, concurrent-upgrade 1013 rejection (single-tenant slot), and
error-path teardown.

Updates the troubleshooting accordion in the user docs to mention the
disabled-state behavior.

* fix(mesh): close ESLint and CodeQL findings on the proxy-tunnel diag logs

Remove the unused `reverseDialer` local in `meshProxyTunnel.ts`; the
closure target is `localDialer` above and this assignment was always
dead. Strip line breaks inline on the three `MeshProxyTunnelDialer`
diag log lines so CodeQL `js/log-injection` data flow recognises the
sanitisation that `sanitizeForLog` already performs.

No behaviour change; the diag logs render the same characters they
do today.
This commit is contained in:
Anso
2026-05-14 20:51:42 -04:00
committed by GitHub
parent 7c985238e1
commit a38a3e0226
24 changed files with 2225 additions and 561 deletions
@@ -0,0 +1,101 @@
/**
* MeshProxyTunnelDialer: central-side dialer that opens an on-demand
* `/api/mesh/proxy-tunnel` WebSocket to a Distributed API remote.
*
* These tests cover the deterministic surface: failure caching when no
* proxy target is configured, idle-close behavior, and recent-failure
* lookup. End-to-end TLS / handshake paths are covered by the manual
* production verification recipe rather than by network-bound tests.
*/
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
let tmpDir: string;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
let MeshProxyTunnelDialer: typeof import('../services/MeshProxyTunnelDialer').MeshProxyTunnelDialer;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ DatabaseService } = await import('../services/DatabaseService'));
({ MeshProxyTunnelDialer } = await import('../services/MeshProxyTunnelDialer'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
beforeEach(() => {
DatabaseService.getInstance().getDb().prepare("DELETE FROM nodes WHERE name LIKE 'proxy-test-%'").run();
});
describe('MeshProxyTunnelDialer', () => {
it('returns null and caches a no_target failure when getProxyTarget yields no node', async () => {
const dialer = MeshProxyTunnelDialer.resetForTest(0); // idle-close disabled
const result = await dialer.ensureBridge(9999); // unknown nodeId
expect(result).toBeNull();
const failure = dialer.getRecentFailure(9999);
expect(failure?.code).toBe('no_target');
});
it('returns null and caches no_target when the node has no api_token', async () => {
const dialer = MeshProxyTunnelDialer.resetForTest(0);
const db = DatabaseService.getInstance();
const nodeId = db.addNode({
name: 'proxy-test-no-token',
type: 'remote',
compose_dir: '',
is_default: false,
mode: 'proxy',
api_url: 'https://proxy-test.invalid',
api_token: '',
});
const result = await dialer.ensureBridge(nodeId);
expect(result).toBeNull();
const failure = dialer.getRecentFailure(nodeId);
expect(failure?.code).toBe('no_target');
});
it('hasBridge reports false before the first successful dial', async () => {
const dialer = MeshProxyTunnelDialer.resetForTest(0);
const db = DatabaseService.getInstance();
const nodeId = db.addNode({
name: 'proxy-test-hasbridge',
type: 'remote',
compose_dir: '',
is_default: false,
mode: 'proxy',
api_url: '',
api_token: '',
});
expect(dialer.hasBridge(nodeId)).toBe(false);
await dialer.ensureBridge(nodeId);
expect(dialer.hasBridge(nodeId)).toBe(false);
});
it('expires the recent-failure cache entry after the cache TTL window', async () => {
const dialer = MeshProxyTunnelDialer.resetForTest(0);
const result = await dialer.ensureBridge(8888);
expect(result).toBeNull();
const cached = dialer.getRecentFailure(8888);
expect(cached).not.toBeNull();
type FailureMap = Map<number, { code: string; message?: string; ts: number }>;
const cacheRef = (dialer as unknown as { recentFailures: FailureMap }).recentFailures;
const entry = cacheRef.get(8888);
if (!entry) throw new Error('cache entry missing');
entry.ts = Date.now() - 90_000; // 90s old, well past the 60s TTL
expect(dialer.getRecentFailure(8888)).toBeNull();
expect(cacheRef.has(8888)).toBe(false);
});
it('stop() tears down the singleton and the idle-check timer', () => {
const dialer = MeshProxyTunnelDialer.resetForTest(60_000);
const timerRef = (dialer as unknown as { idleCheckTimer: NodeJS.Timeout | null }).idleCheckTimer;
expect(timerRef).not.toBeNull();
dialer.stop();
const timerAfter = (dialer as unknown as { idleCheckTimer: NodeJS.Timeout | null }).idleCheckTimer;
expect(timerAfter).toBeNull();
});
});
@@ -0,0 +1,168 @@
/**
* `meshProxyTunnel.ts` is the server-side ingress for Distributed API
* (proxy-mode) mesh tunnels. Auth + scope gating live one layer up in
* `upgradeHandler.ts`; this file covers the handler's own invariants:
*
* - It refuses to serve in pilot-mode deployments (returns 404).
* - On a successful upgrade it installs the shared switchboard AND
* CAS-installs itself as the local MeshService reverse dialer.
* - On WS close the reverse dialer slot is cleared so a subsequent
* dialer can install.
* - A concurrent second upgrade is rejected (1013) while the slot is
* held — this prevents two central instances from racing for the
* same remote's reverse-dial slot.
*/
import http from 'http';
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import WebSocket from 'ws';
import type { AddressInfo } from 'net';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
let tmpDir: string;
let handleMeshProxyTunnel: typeof import('../websocket/meshProxyTunnel').handleMeshProxyTunnel;
let MeshService: typeof import('../services/MeshService').MeshService;
interface ServerHandle {
server: http.Server;
port: number;
close: () => Promise<void>;
}
/**
* Stand up a bare http.Server that routes `/api/mesh/proxy-tunnel`
* upgrades through the handler. Skips the `upgradeHandler.ts` auth
* pipeline because that path is exercised by other integration tests;
* here the contract being verified is what the handler does AFTER
* auth has already accepted the upgrade.
*/
async function startServer(): Promise<ServerHandle> {
const server = http.createServer();
server.on('upgrade', (req, socket, head) => {
if (req.url === '/api/mesh/proxy-tunnel') {
void handleMeshProxyTunnel(req, socket, head);
} else {
socket.destroy();
}
});
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', () => resolve()));
const port = (server.address() as AddressInfo).port;
return {
server,
port,
close: () => new Promise<void>((resolve) => server.close(() => resolve())),
};
}
function dialTunnel(port: number): Promise<WebSocket> {
return new Promise((resolve, reject) => {
const ws = new WebSocket(`ws://127.0.0.1:${port}/api/mesh/proxy-tunnel`);
ws.once('open', () => resolve(ws));
ws.once('error', reject);
});
}
beforeAll(async () => {
tmpDir = await setupTestDb();
({ handleMeshProxyTunnel } = await import('../websocket/meshProxyTunnel'));
({ MeshService } = await import('../services/MeshService'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
beforeEach(() => {
delete process.env.SENCHO_MODE;
// Defensive: clear any reverse dialer left over from a prior test.
MeshService.getInstance().setReverseDialer(null);
});
describe('handleMeshProxyTunnel', () => {
it('rejects with 404 when SENCHO_MODE=pilot (pilot-mode Sencho receives mesh via the pilot tunnel only)', async () => {
process.env.SENCHO_MODE = 'pilot';
const srv = await startServer();
try {
await new Promise<void>((resolve, reject) => {
const ws = new WebSocket(`ws://127.0.0.1:${srv.port}/api/mesh/proxy-tunnel`);
ws.once('open', () => reject(new Error('upgrade should not have succeeded')));
ws.once('unexpected-response', (_req, res) => {
expect(res.statusCode).toBe(404);
res.resume();
resolve();
});
ws.once('error', () => { /* expected */ });
});
} finally {
await srv.close();
}
});
it('on successful upgrade, installs a reverse dialer that exposes openMeshTcpStream', async () => {
const srv = await startServer();
try {
const ws = await dialTunnel(srv.port);
// setReverseDialer is the load-bearing post-upgrade side effect; the
// mesh service's dispatch path consults it on every cross-node TCP
// open. The exact dialer instance is internal, but the presence of
// `openMeshTcpStream` is the contract MeshService.dialMeshTcpStream
// depends on.
// Allow the handler's microtasks to install the dialer.
await new Promise((r) => setTimeout(r, 20));
const dialer = (MeshService.getInstance() as unknown as { reverseDialer: { openMeshTcpStream?: unknown } | null }).reverseDialer;
expect(dialer).not.toBeNull();
expect(typeof dialer?.openMeshTcpStream).toBe('function');
ws.close(1000, 'test cleanup');
// Wait for the close handler's CAS-uninstall.
await new Promise((r) => setTimeout(r, 30));
const after = (MeshService.getInstance() as unknown as { reverseDialer: unknown }).reverseDialer;
expect(after).toBeNull();
} finally {
await srv.close();
}
});
it('a concurrent second upgrade is closed (the reverse dialer slot is single-tenant)', async () => {
const srv = await startServer();
try {
const wsA = await dialTunnel(srv.port);
await new Promise((r) => setTimeout(r, 20));
// The second upgrade succeeds at the WS layer (the handler
// accepted the upgrade before discovering the slot was taken)
// but the handler then closes the socket with code 1013.
const wsB = new WebSocket(`ws://127.0.0.1:${srv.port}/api/mesh/proxy-tunnel`);
const closeInfo = await new Promise<{ code: number; reason: string }>((resolve, reject) => {
wsB.once('close', (code, reason) => resolve({ code, reason: reason.toString() }));
wsB.once('error', reject);
});
expect(closeInfo.code).toBe(1013);
// The first tunnel's dialer is still installed.
const dialer = (MeshService.getInstance() as unknown as { reverseDialer: unknown }).reverseDialer;
expect(dialer).not.toBeNull();
wsA.close(1000, 'test cleanup');
await new Promise((r) => setTimeout(r, 30));
} finally {
await srv.close();
}
});
it('on WS error the reverse dialer slot is cleared (defensive: matches close-path teardown)', async () => {
const srv = await startServer();
try {
const ws = await dialTunnel(srv.port);
await new Promise((r) => setTimeout(r, 20));
// Force-terminate without a clean close; the handler's `error`
// listener (line 133-138) should still run teardown.
ws.terminate();
await new Promise((r) => setTimeout(r, 30));
const after = (MeshService.getInstance() as unknown as { reverseDialer: unknown }).reverseDialer;
expect(after).toBeNull();
} finally {
await srv.close();
}
});
});
@@ -0,0 +1,105 @@
/**
* Live-network test for the proxy-mode mesh tunnel against a real remote
* Sencho. SKIPPED unless `MESH_AUDIT_URL` and `MESH_AUDIT_TOKEN_FILE`
* env vars are set; run by hand from a developer machine, never in CI.
*
* The token file path is read indirectly so the credential never appears
* in the test source, environment dump, or vitest reporter output.
*
* Run manually:
* MESH_AUDIT_URL=http://<remote>:1852 \
* MESH_AUDIT_TOKEN_FILE=/tmp/mesh-audit-token.txt \
* npx vitest run --no-coverage src/__tests__/mesh-proxy-tunnel-live.test.ts
*/
import fs from 'fs';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
const REMOTE_URL = process.env.MESH_AUDIT_URL;
const TOKEN_FILE = process.env.MESH_AUDIT_TOKEN_FILE;
const enabled = !!(REMOTE_URL && TOKEN_FILE);
const describeLive = enabled ? describe : describe.skip;
let tmpDir: string;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
let MeshProxyTunnelDialer: typeof import('../services/MeshProxyTunnelDialer').MeshProxyTunnelDialer;
let PilotMetrics: typeof import('../services/PilotMetrics').PilotMetrics;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ DatabaseService } = await import('../services/DatabaseService'));
({ MeshProxyTunnelDialer } = await import('../services/MeshProxyTunnelDialer'));
({ PilotMetrics } = await import('../services/PilotMetrics'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
describeLive('MeshProxyTunnelDialer (live)', () => {
it('dials a real Sencho remote, registers a bridge, dedupes a second call, and tears down', async () => {
const token = fs.readFileSync(TOKEN_FILE!, 'utf8').trim();
const dialer = MeshProxyTunnelDialer.resetForTest(0); // disable idle close
const db = DatabaseService.getInstance();
const nodeId = db.addNode({
name: 'audit-mesh-live',
type: 'remote',
compose_dir: '',
is_default: false,
mode: 'proxy',
api_url: REMOTE_URL!,
api_token: token,
});
try {
const before = PilotMetrics.snapshot();
const t0 = Date.now();
const bridge = await dialer.ensureBridge(nodeId);
const elapsed = Date.now() - t0;
expect(bridge, `dial returned null after ${elapsed}ms`).not.toBeNull();
expect(elapsed).toBeLessThan(15_000);
// Dedupe: second call hits the cached bridge instantly.
const t1 = Date.now();
const second = await dialer.ensureBridge(nodeId);
expect(second).toBe(bridge);
expect(Date.now() - t1).toBeLessThan(50);
const after = PilotMetrics.snapshot();
expect(after.proxy_bridges_total).toBeGreaterThan(before.proxy_bridges_total);
dialer.closeBridge(nodeId, 'audit teardown');
await new Promise((r) => setTimeout(r, 200));
expect(dialer.hasBridge(nodeId)).toBe(false);
} finally {
db.getDb().prepare('DELETE FROM nodes WHERE id = ?').run(nodeId);
}
}, 30_000);
it('records auth_failed and a single deduped activity entry when the token is rejected', async () => {
const dialer = MeshProxyTunnelDialer.resetForTest(0);
const db = DatabaseService.getInstance();
const nodeId = db.addNode({
name: 'audit-mesh-live-bad',
type: 'remote',
compose_dir: '',
is_default: false,
mode: 'proxy',
api_url: REMOTE_URL!,
api_token: 'this-is-not-a-real-jwt',
});
try {
const result = await dialer.ensureBridge(nodeId);
expect(result).toBeNull();
const failure = dialer.getRecentFailure(nodeId);
expect(failure?.code).toBe('auth_failed');
// Second call should short-circuit via the recent-failure cache.
const t0 = Date.now();
const again = await dialer.ensureBridge(nodeId);
expect(again).toBeNull();
expect(Date.now() - t0).toBeLessThan(20);
} finally {
db.getDb().prepare('DELETE FROM nodes WHERE id = ?').run(nodeId);
}
}, 20_000);
});
@@ -0,0 +1,210 @@
/**
* TcpStreamSwitchboard forward path: `tcp_open` → resolveTarget → dial
* a local TCP target → splice bytes through `TcpData` frames. The
* reverse path is covered by `pilot-agent-reverse-stream.test.ts`; this
* file exercises the inbound side both the pilot agent and the proxy-
* mode WS handler share.
*
* Critical because forward `tcp_open` is the trust boundary on every
* mesh-receiving Sencho: the WS upgrade authenticates the credential,
* but every per-stream dial must resolve via Compose labels and never
* touch a non-mesh-opted target. Resolution policy lives in the
* caller's `resolveTarget` callback; the switchboard's job is to honor
* it without leaking on errors.
*/
import net from 'net';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import {
AGENT_REVERSE_ID_BASE,
BinaryFrameType,
decodeBinaryFrame,
decodeJsonFrame,
encodeBinaryFrame,
} from '../pilot/protocol';
let tmpDir: string;
let attachTcpStreamSwitchboard: typeof import('../mesh/tcpStreamSwitchboard').attachTcpStreamSwitchboard;
interface CapturedSend {
raw: string | Buffer;
binary: boolean;
}
function makeSwitchboard(opts: {
resolve?: typeof import('../mesh/tcpStreamSwitchboard').resolveByComposeLabels;
extraStreamCount?: () => number;
} = {}): {
switchboard: ReturnType<typeof attachTcpStreamSwitchboard>;
sent: CapturedSend[];
mockWs: { readyState: number; send: (data: unknown, opts?: { binary?: boolean }) => void };
} {
const sent: CapturedSend[] = [];
const mockWs = {
readyState: 1,
send(data: unknown, sendOpts?: { binary?: boolean }) {
const isBinary = sendOpts?.binary === true;
sent.push({
raw: isBinary ? (Buffer.isBuffer(data) ? data : Buffer.from(data as Uint8Array)) : String(data),
binary: isBinary,
});
},
};
const switchboard = attachTcpStreamSwitchboard({
ws: mockWs as unknown as import('ws').WebSocket,
resolveTarget: opts.resolve ?? (async () => ({ ok: false, err: 'no_target' })),
extraStreamCount: opts.extraStreamCount,
logLabel: 'Test',
});
return { switchboard, sent, mockWs };
}
/** Spin up a one-shot localhost TCP server to serve as a dial target. */
function startEchoServer(): Promise<{ port: number; server: net.Server; firstConn: Promise<net.Socket> }> {
return new Promise((resolve) => {
let resolveFirst!: (s: net.Socket) => void;
const firstConn = new Promise<net.Socket>((r) => { resolveFirst = r; });
const server = net.createServer((sock) => {
resolveFirst(sock);
});
server.listen(0, '127.0.0.1', () => {
const addr = server.address();
if (!addr || typeof addr === 'string') throw new Error('no addr');
resolve({ port: addr.port, server, firstConn });
});
});
}
beforeAll(async () => {
tmpDir = await setupTestDb();
({ attachTcpStreamSwitchboard } = await import('../mesh/tcpStreamSwitchboard'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
describe('TcpStreamSwitchboard.onTcpOpen (forward path)', () => {
it('rejects with no_target when resolveTarget returns no_target, and never opens a socket', async () => {
const { switchboard, sent } = makeSwitchboard({
resolve: async () => ({ ok: false, err: 'no_target' }),
});
switchboard.handleJsonFrame({ t: 'tcp_open', s: 1, stack: 'api', service: 'db', port: 5432 });
await new Promise((r) => setImmediate(r));
const text = sent.find((s) => !s.binary);
expect(text).toBeDefined();
const decoded = decodeJsonFrame(text!.raw as string);
if (decoded.t !== 'tcp_open_ack') throw new Error('expected ack');
expect(decoded.s).toBe(1);
expect(decoded.ok).toBe(false);
expect(decoded.err).toBe('no_target');
expect(switchboard.tcpStreamCount()).toBe(0);
});
it('rejects with the resolver-supplied err code (denied) without dialing', async () => {
const { switchboard, sent } = makeSwitchboard({
resolve: async () => ({ ok: false, err: 'denied' }),
});
switchboard.handleJsonFrame({ t: 'tcp_open', s: 7, stack: 'api', service: 'db', port: 5432 });
await new Promise((r) => setImmediate(r));
const ack = decodeJsonFrame(sent.find((s) => !s.binary)!.raw as string);
if (ack.t !== 'tcp_open_ack') throw new Error('expected ack');
expect(ack.err).toBe('denied');
});
it('happy path: resolves, dials a local TCP target, acks ok=true, splices echoed bytes back as TcpData frames', async () => {
const { port, server, firstConn } = await startEchoServer();
try {
const { switchboard, sent } = makeSwitchboard({
resolve: async () => ({ ok: true, host: '127.0.0.1', port }),
});
switchboard.handleJsonFrame({ t: 'tcp_open', s: 42, stack: 'api', service: 'db', port: 5432 });
const targetSocket = await firstConn;
// Wait for the ack frame to land in `sent`.
await new Promise((r) => setTimeout(r, 25));
const ack = sent.map((s) => s.binary ? null : decodeJsonFrame(s.raw as string)).find((d) => d?.t === 'tcp_open_ack');
if (!ack || ack.t !== 'tcp_open_ack') throw new Error('no ack');
expect(ack.ok).toBe(true);
expect(ack.s).toBe(42);
expect(switchboard.tcpStreamCount()).toBe(1);
// Server writes some bytes; switchboard should encode them as a
// TcpData binary frame on the WS keyed on the same streamId.
targetSocket.write('pong');
await new Promise((r) => setTimeout(r, 25));
const dataFrame = sent
.filter((s) => s.binary)
.map((s) => decodeBinaryFrame(s.raw as Buffer))
.find((d) => d.type === BinaryFrameType.TcpData && d.streamId === 42);
expect(dataFrame).toBeDefined();
expect(dataFrame!.payload.toString()).toBe('pong');
// Inbound TcpData should land on the target socket.
const echoed = new Promise<Buffer>((r) => targetSocket.once('data', r));
const buf = encodeBinaryFrame(BinaryFrameType.TcpData, 42, Buffer.from('ping'));
switchboard.handleBinaryFrame(decodeBinaryFrame(buf));
const received = await echoed;
expect(received.toString()).toBe('ping');
switchboard.cleanup('test');
} finally {
server.close();
}
});
it('refuses to dial when the per-tunnel cap (with extraStreamCount) is already saturated', async () => {
const { switchboard, sent } = makeSwitchboard({
resolve: async () => ({ ok: true, host: '127.0.0.1', port: 1 }),
// Pretend we already have MAX_STREAMS_PER_TUNNEL non-mesh streams.
extraStreamCount: () => 1024,
});
switchboard.handleJsonFrame({ t: 'tcp_open', s: 99, stack: 's', service: 'svc', port: 80 });
await new Promise((r) => setImmediate(r));
const ack = decodeJsonFrame(sent.find((s) => !s.binary)!.raw as string);
if (ack.t !== 'tcp_open_ack') throw new Error('expected ack');
expect(ack.ok).toBe(false);
expect(ack.err).toBe('agent_error');
expect(switchboard.tcpStreamCount()).toBe(0);
});
it('returns false from handleJsonFrame for non-TCP frame types so the outer dispatcher can route them', () => {
const { switchboard } = makeSwitchboard();
const consumed = switchboard.handleJsonFrame({ t: 'http_req', s: 1, method: 'GET', path: '/', headers: {} });
expect(consumed).toBe(false);
});
it('returns false for tcp_open_ack with a forward-range id (low half) so the caller routes it elsewhere', () => {
const { switchboard } = makeSwitchboard();
const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: 5, ok: true });
expect(consumed).toBe(false);
});
it('consumes tcp_open_ack with a reverse-range id (high half) even when no matching handle exists', () => {
const { switchboard } = makeSwitchboard();
const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: AGENT_REVERSE_ID_BASE + 1, ok: true });
// Owned id-space; switchboard signals it consumed the frame even when
// the matching reverse handle was already torn down (race-safe).
expect(consumed).toBe(true);
});
it('tcp_close on a forward stream destroys the socket and drops the entry', async () => {
const { port, server, firstConn } = await startEchoServer();
try {
const { switchboard } = makeSwitchboard({
resolve: async () => ({ ok: true, host: '127.0.0.1', port }),
});
switchboard.handleJsonFrame({ t: 'tcp_open', s: 11, stack: 's', service: 'svc', port: 80 });
const sock = await firstConn;
await new Promise((r) => setTimeout(r, 25));
expect(switchboard.tcpStreamCount()).toBe(1);
const closed = new Promise<void>((r) => sock.once('close', () => r()));
switchboard.handleJsonFrame({ t: 'tcp_close', s: 11 });
await closed;
expect(switchboard.tcpStreamCount()).toBe(0);
} finally {
server.close();
}
});
});
@@ -1,53 +1,36 @@
/**
* F7 regression: pilot agent's mesh dial path no longer denies based on the
* pilot's local `mesh_stacks` table. Central is the sole authority for
* mesh opt-in (state lives in central's SQLite); the pilot resolves the
* target container by Compose labels and dials it directly. The tunnel JWT
* authenticates the caller, so per-stack gating on the pilot would only
* deny legitimate central-issued dials whenever Phase D's central-only
* state model is in effect.
* Compose-label container resolver shared by the pilot agent and the
* proxy-mode WS handler. The resolver runs the conventional compose name
* fast path first (`<stack>-<service>-1`), then falls back to a
* label-filtered `listContainers` call. The deterministic IP preference
* lives in `pickContainerIp` and is exercised here through the wrapper.
*
* Central is the sole authority for mesh opt-in (state lives in central's
* SQLite); the tunnel WS authentication is the trust boundary, so this
* resolver has no per-stack gating.
*/
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
const listContainersMock = vi.fn();
const inspectMock = vi.fn();
const getContainerMock = vi.fn(() => ({ inspect: inspectMock }));
vi.mock('dockerode', () => {
function Docker(this: unknown) {
(this as { listContainers: typeof listContainersMock }).listContainers = listContainersMock;
const self = this as { listContainers: typeof listContainersMock; getContainer: typeof getContainerMock };
self.listContainers = listContainersMock;
self.getContainer = getContainerMock;
}
return { default: Docker };
});
let tmpDir: string;
let PilotAgent: typeof import('../pilot/agent').PilotAgent;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
interface ResolveResult {
ok: boolean;
host?: string;
port?: number;
err?: string;
}
function makeAgent(): import('../pilot/agent').PilotAgent {
return new PilotAgent({
primaryUrl: 'http://primary.invalid',
loopbackPort: 1,
initialToken: 'irrelevant',
enrolling: false,
});
}
function callResolve(agent: import('../pilot/agent').PilotAgent, stack: string, service: string, port: number): Promise<ResolveResult> {
const fn = (agent as unknown as { resolveMeshTarget: (s: string, sv: string, p: number) => Promise<ResolveResult> }).resolveMeshTarget.bind(agent);
return fn(stack, service, port);
}
let resolveByComposeLabels: typeof import('../mesh/tcpStreamSwitchboard').resolveByComposeLabels;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ PilotAgent } = await import('../pilot/agent'));
({ DatabaseService } = await import('../services/DatabaseService'));
({ resolveByComposeLabels } = await import('../mesh/tcpStreamSwitchboard'));
});
afterAll(() => {
@@ -56,32 +39,41 @@ afterAll(() => {
afterEach(() => {
listContainersMock.mockReset();
DatabaseService.getInstance().getDb().prepare('DELETE FROM mesh_stacks').run();
inspectMock.mockReset();
getContainerMock.mockClear();
});
describe('PilotAgent.resolveMeshTarget (F7: trust central)', () => {
it('returns the container IP when mesh_stacks is empty (central is the gate)', async () => {
listContainersMock.mockResolvedValue([
{ NetworkSettings: { Networks: { sencho_mesh: { IPAddress: '172.30.0.5' } } } },
]);
describe('resolveByComposeLabels (mesh target resolver)', () => {
it('uses the conventional name fast path when it succeeds', async () => {
inspectMock.mockResolvedValueOnce({
NetworkSettings: { Networks: { sencho_mesh: { IPAddress: '172.30.0.5' } } },
});
const agent = makeAgent();
const result = await callResolve(agent, 'audit-mesh-pilot', 'echo', 9001);
const result = await resolveByComposeLabels('audit-mesh-pilot', 'echo', 9001);
expect(result.err).not.toBe('denied');
expect(result.ok).toBe(true);
if (!result.ok) throw new Error('narrowing');
expect(result.host).toBe('172.30.0.5');
expect(result.port).toBe(9001);
// Fast path hit; the label-filtered listContainers call was not needed.
expect(getContainerMock).toHaveBeenCalledWith('audit-mesh-pilot-echo-1');
expect(listContainersMock).not.toHaveBeenCalled();
});
it('queries dockerode with the Compose project + service label filter', async () => {
listContainersMock.mockResolvedValue([
{ NetworkSettings: { Networks: { bridge: { IPAddress: '10.0.0.7' } } } },
]);
it('falls back to a label-filtered listContainers when the fast path returns no IP', async () => {
// Fast inspect returns a container with no usable IP.
inspectMock.mockResolvedValueOnce({ NetworkSettings: { Networks: {} } });
listContainersMock.mockResolvedValueOnce([{ Id: 'abc' }]);
// Fallback inspect resolves the IP.
inspectMock.mockResolvedValueOnce({
NetworkSettings: { Networks: { bridge: { IPAddress: '10.0.0.7' } } },
});
const agent = makeAgent();
await callResolve(agent, 'api', 'db', 5432);
const result = await resolveByComposeLabels('api', 'db', 5432);
expect(result.ok).toBe(true);
if (!result.ok) throw new Error('narrowing');
expect(result.host).toBe('10.0.0.7');
expect(listContainersMock).toHaveBeenCalledTimes(1);
const args = listContainersMock.mock.calls[0][0] as { filters: { label: string[] } };
expect(args.filters.label).toEqual([
@@ -90,23 +82,44 @@ describe('PilotAgent.resolveMeshTarget (F7: trust central)', () => {
]);
});
it('returns no_target (not denied) when dockerode finds no matching container', async () => {
listContainersMock.mockResolvedValue([]);
it('returns no_target when both the fast path and label fallback find nothing', async () => {
inspectMock.mockResolvedValueOnce(null);
listContainersMock.mockResolvedValueOnce([]);
const agent = makeAgent();
const result = await callResolve(agent, 'missing', 'svc', 8080);
const result = await resolveByComposeLabels('missing', 'svc', 8080);
expect(result.ok).toBe(false);
if (result.ok) throw new Error('narrowing');
expect(result.err).toBe('no_target');
});
it('returns agent_error (not denied) when dockerode throws', async () => {
listContainersMock.mockRejectedValue(new Error('docker daemon unreachable'));
it('returns agent_error when dockerode rejects on the listContainers call', async () => {
inspectMock.mockResolvedValueOnce(null);
listContainersMock.mockRejectedValueOnce(new Error('docker daemon unreachable'));
const agent = makeAgent();
const result = await callResolve(agent, 'api', 'db', 5432);
const result = await resolveByComposeLabels('api', 'db', 5432);
expect(result.ok).toBe(false);
if (result.ok) throw new Error('narrowing');
expect(result.err).toBe('agent_error');
});
it('prefers the compose default network over any other attached network', async () => {
// Container is attached to both bridge and stack_default; pickContainerIp
// must return the stack_default IP regardless of object key order.
inspectMock.mockResolvedValueOnce({
NetworkSettings: {
Networks: {
bridge: { IPAddress: '10.0.0.7' },
api_default: { IPAddress: '172.30.0.42' },
},
},
});
const result = await resolveByComposeLabels('api', 'db', 5432);
expect(result.ok).toBe(true);
if (!result.ok) throw new Error('narrowing');
expect(result.host).toBe('172.30.0.42');
});
});
@@ -1,10 +1,12 @@
/**
* Phase B: PilotAgent's outbound reverse mesh stream. The agent's
* MeshForwarder hands off cross-node connections to PilotAgent via
* `openMeshTcpStream`, which sends a `tcp_open_reverse` frame and returns
* a stream handle. The test exercises the wire shape and the lifecycle
* dispatchers (open / data / close) without needing a real WebSocket or
* the full Sencho boot.
* Reverse mesh streams: the switchboard's `openReverseStream` is the
* outbound side of the protocol. Exercises wire-shape (the JSON
* `tcp_open_reverse` frame and the binary `TcpData` envelope) and the
* stream-handle lifecycle dispatchers (open / data / close).
*
* Both the pilot agent and the proxy-mode WS handler delegate to the
* switchboard's `openReverseStream`; the lifecycle assertions here cover
* the shared logic both callers depend on.
*/
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
@@ -17,15 +19,19 @@ import {
} from '../pilot/protocol';
let tmpDir: string;
let PilotAgent: typeof import('../pilot/agent').PilotAgent;
let ReverseTcpStreamHandle: typeof import('../pilot/agent').ReverseTcpStreamHandle;
let attachTcpStreamSwitchboard: typeof import('../mesh/tcpStreamSwitchboard').attachTcpStreamSwitchboard;
let ReverseTcpStreamHandle: typeof import('../mesh/tcpStreamSwitchboard').ReverseTcpStreamHandle;
interface CapturedSend {
raw: string | Buffer;
binary: boolean;
}
function makeMockAgent(): { agent: import('../pilot/agent').PilotAgent; sent: CapturedSend[]; mockWs: { readyState: number; send: (data: unknown, opts?: { binary?: boolean }) => void } } {
function makeSwitchboard(): {
switchboard: ReturnType<typeof attachTcpStreamSwitchboard>;
sent: CapturedSend[];
mockWs: { readyState: number; send: (data: unknown, opts?: { binary?: boolean }) => void };
} {
const sent: CapturedSend[] = [];
const mockWs = {
readyState: 1, // WebSocket.OPEN
@@ -37,32 +43,29 @@ function makeMockAgent(): { agent: import('../pilot/agent').PilotAgent; sent: Ca
});
},
};
const agent = new PilotAgent({
primaryUrl: 'http://primary.invalid',
loopbackPort: 1,
initialToken: 'irrelevant',
enrolling: false,
// The switchboard accepts the `ws` type structurally; the cast is
// narrow and only used in this test harness.
const switchboard = attachTcpStreamSwitchboard({
ws: mockWs as unknown as import('ws').WebSocket,
resolveTarget: async () => ({ ok: false, err: 'no_target' }),
logLabel: 'Test',
});
// Inject the mock ws into the private slot. The agent's
// openMeshTcpStream checks readyState and calls send(); the mock
// captures both for assertions without needing real connect/handshake.
(agent as unknown as { ws: typeof mockWs }).ws = mockWs;
return { agent, sent, mockWs };
return { switchboard, sent, mockWs };
}
beforeAll(async () => {
tmpDir = await setupTestDb();
({ PilotAgent, ReverseTcpStreamHandle } = await import('../pilot/agent'));
({ attachTcpStreamSwitchboard, ReverseTcpStreamHandle } = await import('../mesh/tcpStreamSwitchboard'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
describe('PilotAgent.openMeshTcpStream (Phase B)', () => {
describe('TcpStreamSwitchboard.openReverseStream', () => {
it('allocates an id in the agent-reverse range and emits a tcp_open_reverse frame with the target', () => {
const { agent, sent } = makeMockAgent();
const handle = agent.openMeshTcpStream({
const { switchboard, sent } = makeSwitchboard();
const handle = switchboard.openReverseStream({
nodeId: 12,
stack: 'api',
service: 'db',
@@ -84,15 +87,15 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => {
});
it('returns null when the tunnel is not OPEN', () => {
const { agent, mockWs } = makeMockAgent();
const { switchboard, mockWs } = makeSwitchboard();
mockWs.readyState = 0; // CONNECTING
const handle = agent.openMeshTcpStream({ nodeId: 1, stack: 'a', service: 'b', port: 1 });
const handle = switchboard.openReverseStream({ nodeId: 1, stack: 'a', service: 'b', port: 1 });
expect(handle).toBeNull();
});
it('handle.write encodes a TcpData binary frame with the agent-allocated streamId', () => {
const { agent, sent } = makeMockAgent();
const handle = agent.openMeshTcpStream({ nodeId: 2, stack: 's', service: 'svc', port: 80 });
it('handle.write encodes a TcpData binary frame with the allocated streamId', () => {
const { switchboard, sent } = makeSwitchboard();
const handle = switchboard.openReverseStream({ nodeId: 2, stack: 's', service: 'svc', port: 80 });
if (!handle) throw new Error('handle should exist');
sent.length = 0; // clear the open frame
@@ -106,10 +109,11 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => {
expect(decoded.payload.toString()).toBe('hello');
});
it('handle.end sends a tcp_close JSON frame and removes the stream from the agent map', () => {
const { agent, sent } = makeMockAgent();
const handle = agent.openMeshTcpStream({ nodeId: 3, stack: 's', service: 'svc', port: 80 });
it('handle.end sends a tcp_close JSON frame and drops the stream count', () => {
const { switchboard, sent } = makeSwitchboard();
const handle = switchboard.openReverseStream({ nodeId: 3, stack: 's', service: 'svc', port: 80 });
if (!handle) throw new Error('handle should exist');
expect(switchboard.tcpStreamCount()).toBe(1);
sent.length = 0;
handle.end();
@@ -121,31 +125,25 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => {
if (decoded.t !== 'tcp_close') throw new Error('narrowing');
expect(decoded.s).toBe(handle.streamId);
const internalMap = (agent as unknown as { reverseTcpStreams: Map<number, unknown> }).reverseTcpStreams;
expect(internalMap.has(handle.streamId)).toBe(false);
expect(switchboard.tcpStreamCount()).toBe(0);
});
it('inbound tcp_open_ack {ok: true} fires the open event on the matching handle', () => {
const { agent, mockWs } = makeMockAgent();
const handle = agent.openMeshTcpStream({ nodeId: 4, stack: 's', service: 'svc', port: 80 });
const { switchboard } = makeSwitchboard();
const handle = switchboard.openReverseStream({ nodeId: 4, stack: 's', service: 'svc', port: 80 });
if (!handle) throw new Error('handle should exist');
let opened = false;
handle.on('open', () => { opened = true; });
// Simulate an inbound ack frame: invoke the agent's private json-
// dispatch path with a synthetic frame.
const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent);
onTcpOpenAckReverse({ t: 'tcp_open_ack', s: handle.streamId, ok: true });
const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: handle.streamId, ok: true });
expect(consumed).toBe(true);
expect(opened).toBe(true);
// Sanity: mockWs's readyState wasn't touched.
expect(mockWs.readyState).toBe(1);
});
it('inbound tcp_open_ack {ok: false} emits error and close, then drops the handle', () => {
const { agent } = makeMockAgent();
const handle = agent.openMeshTcpStream({ nodeId: 5, stack: 's', service: 'svc', port: 80 });
const { switchboard } = makeSwitchboard();
const handle = switchboard.openReverseStream({ nodeId: 5, stack: 's', service: 'svc', port: 80 });
if (!handle) throw new Error('handle should exist');
let errMessage: string | undefined;
@@ -153,46 +151,60 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => {
handle.on('error', (err: Error) => { errMessage = err.message; });
handle.on('close', () => { closed = true; });
const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent);
onTcpOpenAckReverse({ t: 'tcp_open_ack', s: handle.streamId, ok: false, err: 'unreachable' });
switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: handle.streamId, ok: false, err: 'unreachable' });
expect(errMessage).toBe('unreachable');
expect(closed).toBe(true);
const internalMap = (agent as unknown as { reverseTcpStreams: Map<number, unknown> }).reverseTcpStreams;
expect(internalMap.has(handle.streamId)).toBe(false);
expect(switchboard.tcpStreamCount()).toBe(0);
});
it('forward-direction tcp_open_ack ids (low half) do not match reverse handles', () => {
const { agent } = makeMockAgent();
const handle = agent.openMeshTcpStream({ nodeId: 6, stack: 's', service: 'svc', port: 80 });
it('forward-direction tcp_open_ack ids (low half) are not consumed by the switchboard', () => {
const { switchboard } = makeSwitchboard();
const handle = switchboard.openReverseStream({ nodeId: 6, stack: 's', service: 'svc', port: 80 });
if (!handle) throw new Error('handle should exist');
let opened = false;
handle.on('open', () => { opened = true; });
// Primary-direction id (< AGENT_REVERSE_ID_BASE) must not bleed into
// the reverse map.
const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent);
onTcpOpenAckReverse({ t: 'tcp_open_ack', s: 5, ok: true });
const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: 5, ok: true });
// Low-half ack is left for the caller to route (it's the
// primary-allocated forward-stream ack, irrelevant to the
// switchboard's reverse map).
expect(consumed).toBe(false);
expect(opened).toBe(false);
});
it('inbound TcpData binary frame (encoded against the wire) emits data on the handle', () => {
const { agent } = makeMockAgent();
const handle = agent.openMeshTcpStream({ nodeId: 7, stack: 's', service: 'svc', port: 80 });
const { switchboard } = makeSwitchboard();
const handle = switchboard.openReverseStream({ nodeId: 7, stack: 's', service: 'svc', port: 80 });
if (!handle) throw new Error('handle should exist');
const received: Buffer[] = [];
handle.on('data', (chunk: Buffer) => received.push(chunk));
// Drive the agent's binary-frame handler with a wire-encoded TcpData
// frame to also exercise the routing branch added in Phase B.
const buf = encodeBinaryFrame(BinaryFrameType.TcpData, handle.streamId, Buffer.from('echo!'));
const decoded = decodeBinaryFrame(buf);
const handleBinaryFrame = (agent as unknown as { handleBinaryFrame: (frame: unknown) => void }).handleBinaryFrame.bind(agent);
handleBinaryFrame(decoded);
const consumed = switchboard.handleBinaryFrame(decoded);
expect(consumed).toBe(true);
expect(Buffer.concat(received).toString()).toBe('echo!');
});
it('cleanup() emits error + close on every outstanding reverse handle and resets the count', () => {
const { switchboard } = makeSwitchboard();
const h1 = switchboard.openReverseStream({ nodeId: 8, stack: 's', service: 'a', port: 1 })!;
const h2 = switchboard.openReverseStream({ nodeId: 9, stack: 's', service: 'b', port: 2 })!;
const closed: number[] = [];
const errors: string[] = [];
h1.on('close', () => closed.push(h1.streamId));
h2.on('close', () => closed.push(h2.streamId));
h1.on('error', (e: Error) => errors.push(e.message));
h2.on('error', (e: Error) => errors.push(e.message));
switchboard.cleanup('mock disconnect');
expect(closed).toContain(h1.streamId);
expect(closed).toContain(h2.streamId);
expect(errors).toEqual(expect.arrayContaining(['mock disconnect', 'mock disconnect']));
expect(switchboard.tcpStreamCount()).toBe(0);
});
});
@@ -0,0 +1,112 @@
/**
* PilotTunnelManager.ensureBridge: dispatcher used by MeshService to
* resolve a mesh-capable bridge for any nodeId without caring whether the
* node runs in pilot-agent mode (long-lived tunnel) or Distributed API
* mode (Phase C on-demand proxy tunnel).
*
* Covers:
* - returns the existing bridge when one is registered.
* - delegates to MeshProxyTunnelDialer.ensureBridge when none exists.
* - registerProxyBridge refuses to shadow a pre-existing bridge.
*/
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
let tmpDir: string;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
let PilotTunnelManager: typeof import('../services/PilotTunnelManager').PilotTunnelManager;
let MeshProxyTunnelDialer: typeof import('../services/MeshProxyTunnelDialer').MeshProxyTunnelDialer;
let PilotTunnelBridge: typeof import('../services/PilotTunnelBridge').PilotTunnelBridge;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ DatabaseService } = await import('../services/DatabaseService'));
({ PilotTunnelManager } = await import('../services/PilotTunnelManager'));
({ MeshProxyTunnelDialer } = await import('../services/MeshProxyTunnelDialer'));
({ PilotTunnelBridge } = await import('../services/PilotTunnelBridge'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
beforeEach(() => {
DatabaseService.getInstance().getDb().prepare("DELETE FROM nodes WHERE name LIKE 'pilot-mgr-test-%'").run();
MeshProxyTunnelDialer.resetForTest(0); // idle-close disabled; clears any prior bridges
// Drop any bridges held over from a prior test in the same file.
type BridgeMap = Map<number, unknown>;
const bridges = (PilotTunnelManager.getInstance() as unknown as { bridges: BridgeMap }).bridges;
bridges.clear();
});
function makeFakeBridge(nodeId: number): { bridge: import('../services/PilotTunnelBridge').PilotTunnelBridge; close: () => void } {
// Bridges hold a WS reference but only the surface we touch in this
// test (openTcpStream, getBufferedAmount, close emission) ever runs.
const mockWs = {
readyState: 1,
bufferedAmount: 0,
on() { return this; },
off() { return this; },
once() { return this; },
send() { /* ignore */ },
close() { /* ignore */ },
} as unknown as import('ws').WebSocket;
const bridge = new PilotTunnelBridge(nodeId, mockWs);
return { bridge, close: () => bridge.close(1000, 'test cleanup') };
}
describe('PilotTunnelManager.ensureBridge', () => {
it('returns the existing bridge when one is registered for the node', async () => {
const mgr = PilotTunnelManager.getInstance();
const { bridge, close } = makeFakeBridge(101);
try {
mgr.registerProxyBridge(101, bridge);
const handle = await mgr.ensureBridge(101);
expect(handle).toBe(bridge);
} finally {
close();
}
});
it('delegates to MeshProxyTunnelDialer when no bridge is registered', async () => {
const mgr = PilotTunnelManager.getInstance();
// No bridge, no node configured -> dialer returns null and caches no_target.
const handle = await mgr.ensureBridge(404);
expect(handle).toBeNull();
const cached = MeshProxyTunnelDialer.getInstance().getRecentFailure(404);
expect(cached?.code).toBe('no_target');
});
it('registerProxyBridge refuses to shadow a node that already has a bridge', () => {
const mgr = PilotTunnelManager.getInstance();
const { bridge: bridgeA, close: closeA } = makeFakeBridge(202);
const { bridge: bridgeB, close: closeB } = makeFakeBridge(202);
try {
mgr.registerProxyBridge(202, bridgeA);
expect(() => mgr.registerProxyBridge(202, bridgeB)).toThrow(/already registered/);
expect(mgr.getBridge(202)).toBe(bridgeA);
} finally {
closeA();
closeB();
}
});
it('registerProxyBridge emits proxy-bridge-up and proxy-bridge-down on lifecycle', async () => {
const mgr = PilotTunnelManager.getInstance();
const up: number[] = [];
const down: number[] = [];
mgr.on('proxy-bridge-up', (id: number) => up.push(id));
mgr.on('proxy-bridge-down', (id: number) => down.push(id));
const { bridge } = makeFakeBridge(303);
mgr.registerProxyBridge(303, bridge);
expect(up).toContain(303);
bridge.close(1000, 'test teardown');
await new Promise((resolve) => setImmediate(resolve));
expect(down).toContain(303);
mgr.off('proxy-bridge-up', () => {});
mgr.off('proxy-bridge-down', () => {});
});
});
+84
View File
@@ -0,0 +1,84 @@
/**
* Shared Compose-container lookup for mesh routing. Both
* `MeshService.resolveContainerIp` (same-node fast path) and the
* `TcpStreamSwitchboard` (inbound `tcp_open` handler) need to translate
* a stack/service pair to a single deterministic IPv4 address; their
* earlier implementations diverged on the network-preference rule,
* which caused flaky resolution on containers attached to multiple
* networks (compose `_default` plus `sencho_mesh`, for example).
*
* `pickContainerIp` is the canonical preference rule. `lookupContainerIp`
* runs the conventional-name fast path first (`<stack>-<service>-1`)
* before falling back to a compose-label container list.
*/
interface ContainerInspectInfo {
NetworkSettings?: {
Networks?: Record<string, { IPAddress?: string } | undefined>;
IPAddress?: string;
};
}
interface ContainerListItem {
Id: string;
}
interface DockerodeLike {
getContainer(id: string): { inspect(): Promise<ContainerInspectInfo> };
listContainers(opts: unknown): Promise<unknown[]>;
}
/**
* Pick a deterministic IPv4 for a container. Prefer the compose default
* network (`<stack>_default` or any network named `<stack>_*`), then any
* other attached network, then the legacy `NetworkSettings.IPAddress`.
* Without this preference order, `Object.values(Networks)` ordering on
* multi-network containers varies across daemon versions and can make
* same-node forwarding flaky on a redeploy.
*/
export function pickContainerIp(stackName: string, info: ContainerInspectInfo): string | null {
const networks = info.NetworkSettings?.Networks ?? {};
const composeDefault = networks[`${stackName}_default`];
if (composeDefault?.IPAddress) return composeDefault.IPAddress;
for (const [name, net] of Object.entries(networks)) {
if (name.startsWith(`${stackName}_`) && net?.IPAddress) return net.IPAddress;
}
for (const net of Object.values(networks)) {
if (net?.IPAddress) return net.IPAddress;
}
return info.NetworkSettings?.IPAddress || null;
}
/**
* Resolve a stack + service to a container IPv4. Tries the conventional
* compose name first, then falls back to a label-filtered list. Returns
* null when no container matches or the matching container has no IP on
* any attached network.
*
* Errors propagate. Callers that only need a single string-or-null
* surface should wrap with a try/catch (`MeshService.resolveContainerIp`
* does this); callers that need to distinguish "Docker errored" from
* "no container matched" use the thrown error to pick the right code.
*/
export async function lookupContainerIp(
docker: DockerodeLike,
stack: string,
service: string,
): Promise<string | null> {
const conventional = `${stack}-${service}-1`;
const fast = await docker.getContainer(conventional).inspect().catch(() => null);
const fastIp = fast ? pickContainerIp(stack, fast) : null;
if (fastIp) return fastIp;
const containers = (await docker.listContainers({
all: true,
filters: {
label: [
`com.docker.compose.project=${stack}`,
`com.docker.compose.service=${service}`,
],
},
})) as ContainerListItem[];
if (containers.length === 0) return null;
const info = await docker.getContainer(containers[0].Id).inspect().catch(() => null);
return info ? pickContainerIp(stack, info) : null;
}
+415
View File
@@ -0,0 +1,415 @@
import net from 'net';
import { EventEmitter } from 'events';
import WebSocket from 'ws';
import {
AGENT_REVERSE_ID_BASE,
BinaryFrameType,
DecodedBinaryFrame,
JsonFrame,
MAX_STREAMS_PER_TUNNEL,
MeshErrCode,
STREAM_IDLE_TIMEOUT_MS,
StreamIdAllocator,
encodeBinaryFrame,
encodeJsonFrame,
} from '../pilot/protocol';
import { lookupContainerIp } from './containerLookup';
import { sanitizeForLog } from '../utils/safeLog';
import { isDebugEnabled } from '../utils/debug';
/**
* Sencho Mesh TCP stream switchboard.
*
* Owns the "agent side" of the mesh frame protocol: accept `tcp_open`,
* resolve a target by Compose labels, dial a local container, and splice
* bytes via `TcpData` frames in both directions. Also owns the outbound
* `tcp_open_reverse` allocator so the local MeshForwarder can emit reverse
* streams when its meshed containers dial cross-node aliases.
*
* Single source of truth for the protocol's "agent" behavior. Two callers:
*
* - `backend/src/pilot/agent.ts` — pilot-mode WS client. Mesh ws is the
* long-lived pilot tunnel; the agent threads its HTTP and WebSocket
* stream count into `extraStreamCount` so the per-tunnel cap is shared.
*
* - `backend/src/websocket/meshProxyTunnel.ts` — proxy-mode WS server.
* Mesh ws is a short-lived central-initiated tunnel that only carries
* TCP frames; `extraStreamCount` defaults to 0.
*
* The switchboard does not touch authentication or lifecycle; callers
* authenticate the WS upgrade themselves and decide when to call
* `cleanup()` on disconnect. State held here (`tcpStreams`,
* `reverseTcpStreams`, `reverseStreamIds`, per-stream idle timers) is
* scoped to a single WS instance — recreate the switchboard on reconnect.
*/
const MESH_CONNECT_TIMEOUT_MS = 10_000;
export type MeshResolveResult =
| { ok: true; host: string; port: number }
| { ok: false; err: MeshErrCode };
export type ResolveTarget = (stack: string, service: string, port: number) => Promise<MeshResolveResult>;
export interface SwitchboardCtx {
/** Per-connection WebSocket. The switchboard sends frames on this and never reassigns it; recreate the switchboard on reconnect. */
ws: WebSocket;
/** Resolve `stack`+`service`+`port` to a TCP target. Implementations typically query Dockerode by Compose labels. */
resolveTarget: ResolveTarget;
/** Non-mesh streams sharing the per-tunnel cap (e.g., the pilot's HTTP + WS multiplex). 0 for pure-TCP tunnels. */
extraStreamCount?: () => number;
/** Diagnostic prefix for log lines emitted from this switchboard. */
logLabel?: string;
}
interface ForwardTcpStream {
socket: net.Socket;
accepted: boolean;
}
/**
* Handle returned by `openReverseStream` to MeshService. Mirrors the
* surface of `PilotTunnelBridge.TcpStream` (write/end/destroy +
* 'open'/'data'/'error'/'close' events) so MeshService.openCrossNode can
* splice bytes against it without caring whether the underlying tunnel is
* a pilot tunnel or a proxy-mode tunnel.
*/
export class ReverseTcpStreamHandle extends EventEmitter {
public readonly streamId: number;
private readonly sendData: (streamId: number, payload: Buffer) => void;
private readonly sendClose: (streamId: number) => void;
private closed = false;
constructor(
streamId: number,
sendData: (streamId: number, payload: Buffer) => void,
sendClose: (streamId: number) => void,
) {
super();
this.streamId = streamId;
this.sendData = sendData;
this.sendClose = sendClose;
}
public write(chunk: Buffer): boolean {
if (this.closed) return false;
this.sendData(this.streamId, chunk);
return true;
}
public end(): void {
if (this.closed) return;
this.closed = true;
this.sendClose(this.streamId);
}
public destroy(): void { this.end(); }
/** @internal Called by the switchboard on inbound `tcp_open_ack { ok: true }`. */
public _dispatchOpen(): void { this.emit('open'); }
/** @internal Called by the switchboard on inbound `TcpData` for this stream. */
public _dispatchData(chunk: Buffer): void { this.emit('data', chunk); }
/** @internal Called by the switchboard on tunnel-side error or rejection. */
public _dispatchError(err: Error): void { this.emit('error', err); }
/** @internal Called by the switchboard on tunnel-side close. */
public _dispatchClose(): void {
if (this.closed) return;
this.closed = true;
this.emit('close');
}
}
export class TcpStreamSwitchboard {
private readonly ctx: SwitchboardCtx;
private readonly logLabel: string;
private readonly tcpStreams = new Map<number, ForwardTcpStream>();
private readonly reverseTcpStreams = new Map<number, ReverseTcpStreamHandle>();
private reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE);
private readonly idleTimers = new Map<number, NodeJS.Timeout>();
constructor(ctx: SwitchboardCtx) {
this.ctx = ctx;
this.logLabel = ctx.logLabel || 'Mesh';
}
/** Active mesh stream count (forward + reverse) owned by this switchboard. */
public tcpStreamCount(): number {
return this.tcpStreams.size + this.reverseTcpStreams.size;
}
private totalStreamCount(): number {
return this.tcpStreamCount() + (this.ctx.extraStreamCount?.() ?? 0);
}
private refreshIdleTimer(streamId: number): void {
const existing = this.idleTimers.get(streamId);
if (existing) clearTimeout(existing);
const timer = setTimeout(() => this.onStreamIdle(streamId), STREAM_IDLE_TIMEOUT_MS);
this.idleTimers.set(streamId, timer);
}
private clearIdleTimer(streamId: number): void {
const timer = this.idleTimers.get(streamId);
if (timer) {
clearTimeout(timer);
this.idleTimers.delete(streamId);
}
}
private onStreamIdle(streamId: number): void {
this.idleTimers.delete(streamId);
const ws = this.ctx.ws;
const fwd = this.tcpStreams.get(streamId);
if (fwd) {
try { fwd.socket.destroy(); } catch { /* ignore */ }
this.tcpStreams.delete(streamId);
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ }
return;
}
const reverse = this.reverseTcpStreams.get(streamId);
if (reverse) {
this.reverseTcpStreams.delete(streamId);
reverse._dispatchError(new Error('mesh idle timeout'));
reverse._dispatchClose();
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ }
}
}
/**
* Dispatch a JSON frame. Returns true when the frame matched one of the
* TCP-related types and was handled; false otherwise so the caller's
* outer dispatcher can route HTTP / WS / control frames.
*/
public handleJsonFrame(frame: JsonFrame): boolean {
switch (frame.t) {
case 'tcp_open':
void this.onTcpOpen(frame);
return true;
case 'tcp_open_ack':
if (frame.s < AGENT_REVERSE_ID_BASE) return false;
this.onTcpOpenAckReverse(frame);
return true;
case 'tcp_close':
this.onTcpClose(frame.s);
return true;
default:
return false;
}
}
/**
* Dispatch a binary frame. Returns true iff the frame is `TcpData` (the
* only binary type owned by the switchboard).
*/
public handleBinaryFrame(frame: DecodedBinaryFrame): boolean {
if (frame.type !== BinaryFrameType.TcpData) return false;
if (frame.streamId >= AGENT_REVERSE_ID_BASE) {
const reverse = this.reverseTcpStreams.get(frame.streamId);
if (!reverse) return true;
reverse._dispatchData(frame.payload);
this.refreshIdleTimer(frame.streamId);
return true;
}
const fwd = this.tcpStreams.get(frame.streamId);
if (!fwd) return true;
try { fwd.socket.write(frame.payload); } catch { /* ignore */ }
this.refreshIdleTimer(frame.streamId);
return true;
}
private async onTcpOpen(frame: { s: number; stack: string; service: string; port: number }): Promise<void> {
const ws = this.ctx.ws;
if (this.totalStreamCount() >= MAX_STREAMS_PER_TUNNEL) {
try {
ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: 'agent_error' }));
} catch { /* ignore */ }
return;
}
const target = await this.ctx.resolveTarget(frame.stack, frame.service, frame.port);
if (!target.ok) {
try {
ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: target.err }));
} catch { /* ignore */ }
return;
}
const socket = net.createConnection({ host: target.host, port: target.port });
socket.setTimeout(MESH_CONNECT_TIMEOUT_MS);
const entry: ForwardTcpStream = { socket, accepted: false };
this.tcpStreams.set(frame.s, entry);
this.refreshIdleTimer(frame.s);
const sendAck = (ok: boolean, err?: MeshErrCode) => {
try { ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok, err })); } catch { /* ignore */ }
};
socket.once('connect', () => {
entry.accepted = true;
socket.setTimeout(0);
sendAck(true);
this.refreshIdleTimer(frame.s);
});
socket.on('data', (chunk: Buffer) => {
try {
ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, frame.s, chunk), { binary: true });
} catch { /* ignore */ }
this.refreshIdleTimer(frame.s);
});
socket.on('timeout', () => {
if (entry.accepted) return;
entry.accepted = true;
sendAck(false, 'unreachable');
this.tcpStreams.delete(frame.s);
this.clearIdleTimer(frame.s);
try { socket.destroy(); } catch { /* ignore */ }
});
socket.on('error', (err) => {
if (!entry.accepted) {
entry.accepted = true;
sendAck(false, 'unreachable');
this.tcpStreams.delete(frame.s);
this.clearIdleTimer(frame.s);
return;
}
if (isDebugEnabled()) {
console.warn(`[${this.logLabel}:diag] tcp stream error:`, sanitizeForLog(err.message));
}
if (this.tcpStreams.delete(frame.s)) {
this.clearIdleTimer(frame.s);
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
}
});
socket.on('close', () => {
if (this.tcpStreams.delete(frame.s)) {
this.clearIdleTimer(frame.s);
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
}
});
}
private onTcpClose(streamId: number): void {
if (streamId >= AGENT_REVERSE_ID_BASE) {
const handle = this.reverseTcpStreams.get(streamId);
if (!handle) return;
this.reverseTcpStreams.delete(streamId);
this.clearIdleTimer(streamId);
handle._dispatchClose();
return;
}
const entry = this.tcpStreams.get(streamId);
if (!entry) return;
this.tcpStreams.delete(streamId);
this.clearIdleTimer(streamId);
try { entry.socket.destroy(); } catch { /* ignore */ }
}
private onTcpOpenAckReverse(frame: { s: number; ok: boolean; err?: MeshErrCode }): void {
const handle = this.reverseTcpStreams.get(frame.s);
if (!handle) return;
if (frame.ok) {
handle._dispatchOpen();
this.refreshIdleTimer(frame.s);
} else {
this.reverseTcpStreams.delete(frame.s);
this.clearIdleTimer(frame.s);
handle._dispatchError(new Error(frame.err ?? 'tcp_open_reverse rejected'));
handle._dispatchClose();
}
}
/**
* Allocate a reverse stream id, send `tcp_open_reverse`, and return a
* handle MeshService can splice bytes through. Returns null if the WS
* is not OPEN or the per-tunnel cap is reached.
*/
public openReverseStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null {
const ws = this.ctx.ws;
if (ws.readyState !== WebSocket.OPEN) return null;
if (this.totalStreamCount() >= MAX_STREAMS_PER_TUNNEL) return null;
const streamId = this.reverseStreamIds.allocate();
const handle = new ReverseTcpStreamHandle(
streamId,
(sid, payload) => {
if (ws.readyState !== WebSocket.OPEN) return;
try { ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, sid, payload), { binary: true }); } catch { /* ignore */ }
this.refreshIdleTimer(sid);
},
(sid) => {
if (!this.reverseTcpStreams.has(sid)) return;
this.reverseTcpStreams.delete(sid);
this.clearIdleTimer(sid);
if (ws.readyState !== WebSocket.OPEN) return;
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: sid })); } catch { /* ignore */ }
},
);
this.reverseTcpStreams.set(streamId, handle);
this.refreshIdleTimer(streamId);
try {
ws.send(encodeJsonFrame({
t: 'tcp_open_reverse',
s: streamId,
targetNodeId: target.nodeId,
stack: target.stack,
service: target.service,
port: target.port,
}));
} catch (err) {
this.reverseTcpStreams.delete(streamId);
this.clearIdleTimer(streamId);
handle._dispatchError(err as Error);
handle._dispatchClose();
return null;
}
return handle;
}
/**
* Tear down all stream state. Call on WS disconnect; the next
* connection should construct a fresh switchboard.
*/
public cleanup(reason = 'mesh tunnel closed'): void {
for (const [, entry] of this.tcpStreams) {
try { entry.socket.destroy(); } catch { /* ignore */ }
}
this.tcpStreams.clear();
for (const [, handle] of this.reverseTcpStreams) {
try {
handle._dispatchError(new Error(reason));
handle._dispatchClose();
} catch { /* ignore */ }
}
this.reverseTcpStreams.clear();
// Reset the allocator so a long-lived caller that reconnects many
// times doesn't drift up the id range and approach the wrap point.
this.reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE);
for (const [, timer] of this.idleTimers) clearTimeout(timer);
this.idleTimers.clear();
}
}
export function attachTcpStreamSwitchboard(ctx: SwitchboardCtx): TcpStreamSwitchboard {
return new TcpStreamSwitchboard(ctx);
}
/**
* Resolve a Compose-managed container's IP address by stack name + service
* name. Honors the deterministic network preference shared with
* `MeshService.resolveContainerIp` so a redeploy that adds or reorders
* networks does not flip which IP is returned. Used by both the pilot
* agent and the proxy-mode WS handler.
*/
export async function resolveByComposeLabels(stack: string, service: string, port: number): Promise<MeshResolveResult> {
try {
const dockerodeMod = await import('dockerode');
const Docker = (dockerodeMod as { default: new (opts?: unknown) => Parameters<typeof lookupContainerIp>[0] }).default;
const docker = new Docker();
const ip = await lookupContainerIp(docker, stack, service);
if (!ip) return { ok: false, err: 'no_target' };
return { ok: true, host: ip, port };
} catch (err) {
if (isDebugEnabled()) {
console.warn('[Mesh:diag] resolveByComposeLabels failed:', sanitizeForLog((err as Error).message));
}
return { ok: false, err: 'agent_error' };
}
}
+34 -318
View File
@@ -1,22 +1,17 @@
import fs from 'fs';
import net from 'net';
import path from 'path';
import http from 'http';
import { EventEmitter } from 'events';
import jwt from 'jsonwebtoken';
import WebSocket from 'ws';
import { getSenchoVersion } from '../services/CapabilityRegistry';
import { DatabaseService } from '../services/DatabaseService';
import { NodeRegistry } from '../services/NodeRegistry';
import {
AGENT_REVERSE_ID_BASE,
BinaryFrameType,
MAX_FRAME_SIZE_BYTES,
MAX_STREAMS_PER_TUNNEL,
MeshErrCode,
PROTOCOL_VERSION,
STREAM_IDLE_TIMEOUT_MS,
StreamIdAllocator,
decodeBinaryFrame,
decodeJsonFrame,
encodeBinaryFrame,
@@ -24,7 +19,14 @@ import {
wsDataToBuffer,
wsDataToString,
} from './protocol';
import {
ReverseTcpStreamHandle,
TcpStreamSwitchboard,
attachTcpStreamSwitchboard,
resolveByComposeLabels,
} from '../mesh/tcpStreamSwitchboard';
import { sanitizeForLog } from '../utils/safeLog';
import { httpUrlToWs } from '../utils/wsUrl';
import { isDebugEnabled } from '../utils/debug';
const RECONNECT_MIN_MS = 1_000;
@@ -90,11 +92,9 @@ export class PilotAgent {
private reconnectTimer?: NodeJS.Timeout;
private readonly httpStreams = new Map<number, { req: http.ClientRequest }>();
private readonly wsStreams = new Map<number, WebSocket>();
private readonly tcpStreams = new Map<number, MeshTcpStream>();
/** Reverse mesh streams the agent itself initiated via `tcp_open_reverse`. Keyed on the agent-allocated id. Disjoint from `tcpStreams` because those are primary-allocated and live in the lower id half. */
private readonly reverseTcpStreams = new Map<number, ReverseTcpStreamHandle>();
/** Allocator is recreated on every disconnect (`cleanupAfterDisconnect`) so a long-lived agent that reconnects many times doesn't drift up the id range. */
private reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE);
/** Per-connection mesh frame handler. Created on `connect()`, cleaned up on disconnect. */
private switchboard: TcpStreamSwitchboard | null = null;
/** Idle timers for HTTP and WebSocket streams only; mesh streams keep their own timers inside the switchboard. */
private readonly idleTimers = new Map<number, NodeJS.Timeout>();
private shuttingDown = false;
private readonly agentVersion: string;
@@ -170,7 +170,7 @@ export class PilotAgent {
private connect(): void {
if (this.shuttingDown) return;
const wsUrl = this.options.primaryUrl.replace(/^http/, 'ws').replace(/\/$/, '') + '/api/pilot/tunnel';
const wsUrl = httpUrlToWs(this.options.primaryUrl) + '/api/pilot/tunnel';
const ws = new WebSocket(wsUrl, {
headers: {
Authorization: `Bearer ${this.token}`,
@@ -180,13 +180,19 @@ export class PilotAgent {
maxPayload: MAX_FRAME_SIZE_BYTES,
// Self-signed deployments can supply an internal CA bundle via
// SENCHO_PILOT_CA_FILE; rejectUnauthorized stays true. There is
// intentionally no env var to disable TLS verification — that
// intentionally no env var to disable TLS verification, which
// would defeat the entire trust model of the tunnel credential.
// The bundle is read once at agent construction (this.customCa);
// rotate by restarting the container.
...(this.customCa ? { ca: this.customCa } : {}),
});
this.ws = ws;
this.switchboard = attachTcpStreamSwitchboard({
ws,
resolveTarget: resolveByComposeLabels,
extraStreamCount: () => this.httpStreams.size + this.wsStreams.size,
logLabel: 'Pilot',
});
ws.on('open', () => {
// Backoff intentionally NOT reset here: a TCP-level connect that
@@ -234,27 +240,16 @@ export class PilotAgent {
try { ws.close(1006, 'tunnel closed'); } catch { /* ignore */ }
}
this.wsStreams.clear();
for (const [, stream] of this.tcpStreams) {
try { stream.socket.destroy(); } catch { /* ignore */ }
if (this.switchboard) {
this.switchboard.cleanup('pilot tunnel closed');
this.switchboard = null;
}
this.tcpStreams.clear();
for (const [, handle] of this.reverseTcpStreams) {
try {
handle._dispatchError(new Error('pilot tunnel closed'));
handle._dispatchClose();
} catch { /* ignore */ }
}
this.reverseTcpStreams.clear();
// Reset the reverse allocator so a long-lived agent that
// reconnects many times doesn't drift up the id range and
// approach the wrap point unnecessarily.
this.reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE);
for (const [, timer] of this.idleTimers) clearTimeout(timer);
this.idleTimers.clear();
}
private streamCount(): number {
return this.httpStreams.size + this.wsStreams.size + this.tcpStreams.size + this.reverseTcpStreams.size;
return this.httpStreams.size + this.wsStreams.size + (this.switchboard?.tcpStreamCount() ?? 0);
}
private refreshIdleTimer(streamId: number): void {
@@ -291,25 +286,6 @@ export class PilotAgent {
if (ws) {
try { ws.send(encodeJsonFrame({ t: 'ws_close', s: streamId, code: 1001, reason: 'idle' })); } catch { /* ignore */ }
}
return;
}
const tcpEntry = this.tcpStreams.get(streamId);
if (tcpEntry) {
try { tcpEntry.socket.destroy(); } catch { /* ignore */ }
this.tcpStreams.delete(streamId);
if (ws) {
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ }
}
return;
}
const reverseEntry = this.reverseTcpStreams.get(streamId);
if (reverseEntry) {
this.reverseTcpStreams.delete(streamId);
reverseEntry._dispatchError(new Error('agent idle timeout'));
reverseEntry._dispatchClose();
if (ws) {
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ }
}
}
}
@@ -346,6 +322,7 @@ export class PilotAgent {
private handleJsonFrame(frame: ReturnType<typeof decodeJsonFrame>): void {
const ws = this.ws;
if (!ws) return;
if (this.switchboard?.handleJsonFrame(frame)) return;
switch (frame.t) {
case 'hello': {
if (frame.version !== PROTOCOL_VERSION) {
@@ -374,9 +351,6 @@ export class PilotAgent {
case 'ws_open': this.onWsOpen(frame); break;
case 'ws_msg_text': this.onWsMsgText(frame.s, frame.data); break;
case 'ws_close': this.onWsClose(frame.s, frame.code, frame.reason); break;
case 'tcp_open': this.onTcpOpen(frame); break;
case 'tcp_open_ack': this.onTcpOpenAckReverse(frame); break;
case 'tcp_close': this.onTcpClose(frame.s); break;
default:
// Other frame types are primary-bound only; agent ignores.
break;
@@ -384,6 +358,7 @@ export class PilotAgent {
}
private handleBinaryFrame(frame: ReturnType<typeof decodeBinaryFrame>): void {
if (this.switchboard?.handleBinaryFrame(frame)) return;
switch (frame.type) {
case BinaryFrameType.HttpReqBody: {
const entry = this.httpStreams.get(frame.streamId);
@@ -399,20 +374,6 @@ export class PilotAgent {
this.refreshIdleTimer(frame.streamId);
break;
}
case BinaryFrameType.TcpData: {
if (frame.streamId >= AGENT_REVERSE_ID_BASE) {
const reverse = this.reverseTcpStreams.get(frame.streamId);
if (!reverse) return;
reverse._dispatchData(frame.payload);
this.refreshIdleTimer(frame.streamId);
return;
}
const stream = this.tcpStreams.get(frame.streamId);
if (!stream) return;
try { stream.socket.write(frame.payload); } catch { /* ignore */ }
this.refreshIdleTimer(frame.streamId);
break;
}
default:
break;
}
@@ -560,272 +521,27 @@ export class PilotAgent {
this.clearIdleTimer(streamId);
}
// --- Sencho Mesh TCP dispatch (tunnel -> Compose service container) ---
// --- Sencho Mesh TCP dispatch ---
//
// Central is the sole authority for mesh opt-in (state lives in central's
// SQLite mesh_stacks table). The pilot resolves a target by Compose
// container labels and dials directly. The tunnel JWT (scope
// 'pilot_tunnel') gates the WS upgrade itself, so any tcp_open frame on
// an open tunnel is trusted to originate from central.
private async onTcpOpen(frame: Extract<ReturnType<typeof decodeJsonFrame>, { t: 'tcp_open' }>): Promise<void> {
const ws = this.ws;
if (!ws) return;
if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) {
try {
ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: 'agent_error' }));
} catch { /* ignore */ }
return;
}
const target = await this.resolveMeshTarget(frame.stack, frame.service, frame.port);
if (!target.ok) {
try {
ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: target.err }));
} catch { /* ignore */ }
return;
}
const socket = net.createConnection({ host: target.host, port: target.port });
socket.setTimeout(MESH_CONNECT_TIMEOUT_MS);
const entry: MeshTcpStream = { socket, accepted: false };
this.tcpStreams.set(frame.s, entry);
this.refreshIdleTimer(frame.s);
const sendAck = (ok: boolean, err?: MeshErrCode) => {
try { ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok, err })); } catch { /* ignore */ }
};
socket.once('connect', () => {
entry.accepted = true;
socket.setTimeout(0);
sendAck(true);
this.refreshIdleTimer(frame.s);
});
socket.on('data', (chunk: Buffer) => {
try {
ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, frame.s, chunk), { binary: true });
} catch { /* ignore */ }
this.refreshIdleTimer(frame.s);
});
socket.on('timeout', () => {
if (entry.accepted) return;
entry.accepted = true;
sendAck(false, 'unreachable');
this.tcpStreams.delete(frame.s);
this.clearIdleTimer(frame.s);
try { socket.destroy(); } catch { /* ignore */ }
});
socket.on('error', (err) => {
if (!entry.accepted) {
entry.accepted = true;
sendAck(false, 'unreachable');
this.tcpStreams.delete(frame.s);
this.clearIdleTimer(frame.s);
return;
}
console.warn('[Pilot] tcp stream error:', sanitizeForLog(err.message));
if (this.tcpStreams.delete(frame.s)) {
this.clearIdleTimer(frame.s);
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
}
});
socket.on('close', () => {
if (this.tcpStreams.delete(frame.s)) {
this.clearIdleTimer(frame.s);
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
}
});
}
private onTcpClose(streamId: number): void {
// Reverse stream (agent-initiated): primary is closing the
// upstream half. Tear down the local socket the agent's
// MeshForwarder handed us via openMeshTcpStream.
if (streamId >= AGENT_REVERSE_ID_BASE) {
const handle = this.reverseTcpStreams.get(streamId);
if (!handle) return;
this.reverseTcpStreams.delete(streamId);
this.clearIdleTimer(streamId);
handle._dispatchClose();
return;
}
const entry = this.tcpStreams.get(streamId);
if (!entry) return;
this.tcpStreams.delete(streamId);
this.clearIdleTimer(streamId);
try { entry.socket.destroy(); } catch { /* ignore */ }
}
// Mesh frame handling lives in `backend/src/mesh/tcpStreamSwitchboard.ts`
// and is shared with the proxy-mode WS handler. The agent owns a
// switchboard per WS connection and delegates the public reverse-dial
// surface used by MeshService.
/**
* Inbound `tcp_open_ack` for an agent-initiated reverse stream. The
* primary acknowledges (or rejects) the dial; emit 'open' or 'error' on
* the local handle so MeshService's splice setup can either start
* piping bytes or tear down the source socket.
*/
private onTcpOpenAckReverse(frame: Extract<ReturnType<typeof decodeJsonFrame>, { t: 'tcp_open_ack' }>): void {
if (frame.s < AGENT_REVERSE_ID_BASE) return; // forward-direction acks are primary-bound; ignore.
const handle = this.reverseTcpStreams.get(frame.s);
if (!handle) return;
if (frame.ok) {
handle._dispatchOpen();
this.refreshIdleTimer(frame.s);
} else {
this.reverseTcpStreams.delete(frame.s);
this.clearIdleTimer(frame.s);
handle._dispatchError(new Error(frame.err ?? 'tcp_open_reverse rejected'));
handle._dispatchClose();
}
}
/**
* Public entry point for the agent's mesh forwarder. Allocates a
* reverse stream id, sends `tcp_open_reverse`, and returns a handle
* MeshService can splice bytes through. Returns null if the tunnel is
* not currently open or the per-tunnel stream cap is reached.
* Allocates a reverse stream id, sends `tcp_open_reverse`, and returns
* a handle MeshService can splice bytes through. Returns null if the
* tunnel is not currently open or the per-tunnel stream cap is
* reached. Delegates to the per-connection switchboard.
*/
public openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null {
const ws = this.ws;
if (!ws || ws.readyState !== WebSocket.OPEN) return null;
if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) return null;
const streamId = this.reverseStreamIds.allocate();
const handle = new ReverseTcpStreamHandle(
streamId,
(sid, payload) => {
if (this.ws?.readyState !== WebSocket.OPEN) return;
try { this.ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, sid, payload), { binary: true }); } catch { /* ignore */ }
this.refreshIdleTimer(sid);
},
(sid) => {
if (!this.reverseTcpStreams.has(sid)) return;
this.reverseTcpStreams.delete(sid);
this.clearIdleTimer(sid);
if (this.ws?.readyState !== WebSocket.OPEN) return;
try { this.ws.send(encodeJsonFrame({ t: 'tcp_close', s: sid })); } catch { /* ignore */ }
},
);
this.reverseTcpStreams.set(streamId, handle);
this.refreshIdleTimer(streamId);
try {
ws.send(encodeJsonFrame({
t: 'tcp_open_reverse',
s: streamId,
targetNodeId: target.nodeId,
stack: target.stack,
service: target.service,
port: target.port,
}));
} catch (err) {
this.reverseTcpStreams.delete(streamId);
this.clearIdleTimer(streamId);
handle._dispatchError(err as Error);
handle._dispatchClose();
return null;
}
return handle;
}
/**
* Resolves a mesh target by Compose container labels and returns the
* container's first usable IP. Central has already validated that the
* target stack is opted in before issuing the dial; the tunnel JWT
* (scope 'pilot_tunnel') authenticates the caller, so this handler
* does no per-stack gating of its own.
*/
private async resolveMeshTarget(
stack: string,
service: string,
port: number,
): Promise<MeshResolveResult> {
try {
const dockerodeMod = await import('dockerode');
const Docker = (dockerodeMod as { default: new (opts?: unknown) => { listContainers: (opts?: unknown) => Promise<unknown[]> } }).default;
const docker = new Docker();
const containers = (await docker.listContainers({
filters: { label: [`com.docker.compose.project=${stack}`, `com.docker.compose.service=${service}`] },
})) as Array<{ NetworkSettings?: { Networks?: Record<string, { IPAddress?: string }> } }>;
for (const c of containers) {
const networks = c.NetworkSettings?.Networks ?? {};
for (const net of Object.values(networks)) {
if (net.IPAddress) return { ok: true, host: net.IPAddress, port };
}
}
return { ok: false, err: 'no_target' };
} catch (err) {
console.warn('[Pilot] resolveMeshTarget failed:', sanitizeForLog((err as Error).message));
return { ok: false, err: 'agent_error' };
}
return this.switchboard?.openReverseStream(target) ?? null;
}
}
const MESH_CONNECT_TIMEOUT_MS = 10_000;
interface MeshTcpStream {
socket: net.Socket;
accepted: boolean;
}
/**
* Handle returned by `PilotAgent.openMeshTcpStream` to MeshService. Mirrors
* the surface of `PilotTunnelBridge.TcpStream` (write/end/destroy +
* 'open'/'data'/'error'/'close' events) so MeshService.openCrossNode can
* splice bytes against it without caring whether it's running on central
* or on a pilot. The agent owns the per-stream WS plumbing through the
* `sendData` and `sendClose` callbacks; this class is a thin EventEmitter
* facade.
*/
export class ReverseTcpStreamHandle extends EventEmitter {
public readonly streamId: number;
private readonly sendData: (streamId: number, payload: Buffer) => void;
private readonly sendClose: (streamId: number) => void;
private closed = false;
constructor(
streamId: number,
sendData: (streamId: number, payload: Buffer) => void,
sendClose: (streamId: number) => void,
) {
super();
this.streamId = streamId;
this.sendData = sendData;
this.sendClose = sendClose;
}
public write(chunk: Buffer): boolean {
if (this.closed) return false;
this.sendData(this.streamId, chunk);
return true;
}
public end(): void {
if (this.closed) return;
this.closed = true;
this.sendClose(this.streamId);
}
public destroy(): void { this.end(); }
/** @internal Called by PilotAgent on inbound `tcp_open_ack { ok: true }`. */
public _dispatchOpen(): void { this.emit('open'); }
/** @internal Called by PilotAgent on inbound `TcpData` for this stream. */
public _dispatchData(chunk: Buffer): void { this.emit('data', chunk); }
/** @internal Called by PilotAgent on tunnel-side error or rejection. */
public _dispatchError(err: Error): void { this.emit('error', err); }
/** @internal Called by PilotAgent on tunnel-side close. */
public _dispatchClose(): void {
if (this.closed) return;
this.closed = true;
this.emit('close');
}
}
type MeshResolveResult =
| { ok: true; host: string; port: number }
| { ok: false; err: MeshErrCode };
/**
* Read the persisted long-lived tunnel token from disk if present. ENOENT is
* the normal first-boot case and stays silent. Any other error class
@@ -0,0 +1,371 @@
import { EventEmitter } from 'events';
import WebSocket from 'ws';
import { MAX_FRAME_SIZE_BYTES } from '../pilot/protocol';
import { PilotTunnelBridge, type MeshTunnelHandle } from './PilotTunnelBridge';
import { PilotTunnelManager } from './PilotTunnelManager';
import { NodeRegistry } from './NodeRegistry';
import { redactSensitiveText, sanitizeForLog } from '../utils/safeLog';
import { httpUrlToWs } from '../utils/wsUrl';
import { isDebugEnabled } from '../utils/debug';
import { PilotMetrics } from './PilotMetrics';
import type { MeshActivityType } from './MeshService';
/**
* Central-side dialer for proxy-mode mesh tunnels.
*
* Today's pilot tunnels are agent-initiated: the pilot dials central and
* the resulting WS is registered in `PilotTunnelManager`. Proxy-mode
* remotes do not dial; central reaches them via the existing HTTP proxy
* using the long-lived `api_token`. To carry streaming TCP mesh traffic to
* a proxy-mode remote, central opens a WebSocket to the remote's new
* `/api/mesh/proxy-tunnel` endpoint on demand and registers the resulting
* `PilotTunnelBridge` in the manager under the same nodeId. The rest of
* the mesh code path (alias dispatch, openTcpStream, reverse-stream relay)
* is mode-agnostic from there.
*
* Lifecycle:
* - `ensureBridge(nodeId)` dials if not connected; concurrent callers
* dedupe through an in-flight promise map.
* - A periodic check tears down bridges that have seen no active streams
* for `SENCHO_MESH_PROXY_TUNNEL_IDLE_MS` (default 5 minutes). Setting
* the env var to `0` disables idle close (tunnel persists until the
* remote drops it or central shuts down).
* - Recent failures are cached for 60 seconds so a misconfigured remote
* does not cause a continuous redial storm; `MeshService.getStatus`
* consults the cache to surface a `reachableReason` to the UI.
*/
const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000;
const IDLE_CHECK_INTERVAL_MS = 60 * 1000;
const HANDSHAKE_TIMEOUT_MS = 15_000;
const FAILURE_CACHE_TTL_MS = 60 * 1000;
export type DialFailureCode =
| 'no_target'
| 'endpoint_not_found'
| 'auth_failed'
| 'tls_failed'
| 'network_error';
/**
* Activity-log reason. Wider than `DialFailureCode` because some failures
* share a wire-level code (e.g., `network_error`) but deserve a more
* specific label in the operator-facing log to distinguish a network-
* layer failure from a post-handshake bridge failure or a manager
* rejection.
*/
type DialFailureReason = DialFailureCode | 'bridge_start_failed' | 'manager_rejected';
type ProxyTunnelEvent = 'open.ok' | 'open.fail' | 'close';
const ACTIVITY_TYPE: Record<ProxyTunnelEvent, MeshActivityType> = {
'open.ok': 'proxy-tunnel.open.ok',
'open.fail': 'proxy-tunnel.open.fail',
'close': 'proxy-tunnel.close',
};
export interface DialFailure {
code: DialFailureCode;
message?: string;
ts: number;
}
export class MeshProxyTunnelDialer extends EventEmitter {
private static instance: MeshProxyTunnelDialer | null = null;
private readonly bridges = new Map<number, PilotTunnelBridge>();
private readonly inflight = new Map<number, Promise<MeshTunnelHandle | null>>();
private readonly idleSince = new Map<number, number>();
private readonly recentFailures = new Map<number, DialFailure>();
private readonly idleTtlMs: number;
private idleCheckTimer: NodeJS.Timeout | null = null;
private stopped = false;
private constructor(idleTtlOverrideMs?: number) {
super();
if (typeof idleTtlOverrideMs === 'number') {
this.idleTtlMs = idleTtlOverrideMs;
} else {
const raw = process.env.SENCHO_MESH_PROXY_TUNNEL_IDLE_MS;
const parsed = raw === undefined ? Number.NaN : Number(raw);
this.idleTtlMs = Number.isFinite(parsed) && parsed >= 0 ? parsed : DEFAULT_IDLE_TTL_MS;
}
this.startIdleCheck();
}
public static getInstance(): MeshProxyTunnelDialer {
if (!this.instance) this.instance = new MeshProxyTunnelDialer();
return this.instance;
}
/** Test hook: reset the singleton with an optional idle-TTL override. */
public static resetForTest(idleTtlOverrideMs?: number): MeshProxyTunnelDialer {
if (this.instance) this.instance.stop();
this.instance = new MeshProxyTunnelDialer(idleTtlOverrideMs);
return this.instance;
}
public hasBridge(nodeId: number): boolean {
return this.bridges.has(nodeId);
}
public getBridge(nodeId: number): MeshTunnelHandle | null {
return this.bridges.get(nodeId) ?? null;
}
public getRecentFailure(nodeId: number): DialFailure | null {
const entry = this.recentFailures.get(nodeId);
if (!entry) return null;
if (Date.now() - entry.ts > FAILURE_CACHE_TTL_MS) {
this.recentFailures.delete(nodeId);
return null;
}
return entry;
}
/**
* Dial-if-needed. Concurrent callers dedupe through `inflight`; a
* cached recent failure short-circuits the dial so a misconfigured
* remote does not see one upgrade attempt per cross-node TCP open.
*/
public async ensureBridge(nodeId: number): Promise<MeshTunnelHandle | null> {
const existing = this.bridges.get(nodeId);
if (existing) {
this.idleSince.set(nodeId, Date.now());
return existing;
}
if (this.getRecentFailure(nodeId)) return null;
const inflight = this.inflight.get(nodeId);
if (inflight) return inflight;
const dial = this.dial(nodeId).finally(() => {
this.inflight.delete(nodeId);
});
this.inflight.set(nodeId, dial);
return dial;
}
/** Force-close a bridge (e.g., on node deletion or scope change). */
public closeBridge(nodeId: number, reason = 'closed by dialer'): void {
const bridge = this.bridges.get(nodeId);
if (!bridge) return;
this.bridges.delete(nodeId);
this.idleSince.delete(nodeId);
try { bridge.close(1000, reason); } catch { /* ignore */ }
}
/** Stop the idle-check timer and tear down every open bridge. */
public stop(): void {
this.stopped = true;
if (this.idleCheckTimer) {
clearInterval(this.idleCheckTimer);
this.idleCheckTimer = null;
}
for (const [nodeId, bridge] of this.bridges) {
this.bridges.delete(nodeId);
try { bridge.close(1000, 'dialer shutdown'); } catch { /* ignore */ }
}
this.idleSince.clear();
this.recentFailures.clear();
this.inflight.clear();
}
/** Test hook: count active bridges. */
public get activeBridgeCount(): number {
return this.bridges.size;
}
private async dial(nodeId: number): Promise<MeshTunnelHandle | null> {
const target = NodeRegistry.getInstance().getProxyTarget(nodeId);
if (!target || !target.apiToken) {
this.recordFailure(nodeId, 'no_target', 'no proxy target configured');
return null;
}
if (this.stopped) return null;
const dialStartedAt = Date.now();
if (isDebugEnabled()) {
console.log(`[MeshProxyDialer:diag] dialing node=${nodeId} url=${sanitizeForLog(target.apiUrl)}`.replace(/[\n\r]/g, ''));
}
const wsUrl = httpUrlToWs(target.apiUrl) + '/api/mesh/proxy-tunnel';
let ws: WebSocket;
try {
ws = new WebSocket(wsUrl, {
headers: { Authorization: `Bearer ${target.apiToken}` },
handshakeTimeout: HANDSHAKE_TIMEOUT_MS,
maxPayload: MAX_FRAME_SIZE_BYTES,
});
} catch (err) {
this.recordFailure(nodeId, 'network_error', (err as Error).message);
return null;
}
try {
await this.awaitOpen(ws);
} catch (err) {
// `awaitOpen` removes the 'error' listener on reject; calling
// `close()` on a still-CONNECTING socket emits a tail 'error'
// ('WebSocket was closed before the connection was established')
// that would otherwise propagate as an unhandled exception.
ws.on('error', () => { /* swallow tail error */ });
try { ws.close(); } catch { /* ignore */ }
const failure = classifyDialError(err);
this.recordFailure(nodeId, failure.code, failure.message);
return null;
}
if (this.stopped) {
try { ws.close(1001, 'dialer shutdown'); } catch { /* ignore */ }
return null;
}
const bridge = new PilotTunnelBridge(nodeId, ws);
try {
await bridge.start();
} catch (err) {
this.recordFailure(nodeId, 'network_error', (err as Error).message, 'bridge_start_failed');
try { bridge.close(1011, 'bridge start failed'); } catch { /* ignore */ }
return null;
}
if (this.stopped) {
try { bridge.close(1001, 'dialer shutdown'); } catch { /* ignore */ }
return null;
}
try {
PilotTunnelManager.getInstance().registerProxyBridge(nodeId, bridge);
} catch (err) {
// Cap hit or a pilot tunnel concurrently claimed this nodeId.
this.recordFailure(nodeId, 'network_error', (err as Error).message, 'manager_rejected');
try { bridge.close(1013, 'manager rejected'); } catch { /* ignore */ }
return null;
}
// Mirror the registration in the dialer's own map so the idle
// sweeper can call `getActiveStreamCount()` (not on the narrow
// MeshTunnelHandle interface) without poking into the manager.
bridge.once('closed', () => {
if (this.bridges.get(nodeId) === bridge) {
this.bridges.delete(nodeId);
this.idleSince.delete(nodeId);
void this.logActivity(nodeId, 'close', { reason: 'remote-closed' });
this.emit('proxy-bridge-down', nodeId);
}
});
this.bridges.set(nodeId, bridge);
this.idleSince.set(nodeId, Date.now());
this.recentFailures.delete(nodeId);
void this.logActivity(nodeId, 'open.ok', {});
this.emit('proxy-bridge-up', nodeId);
if (isDebugEnabled()) {
console.log(`[MeshProxyDialer:diag] dial ok node=${nodeId} elapsedMs=${Date.now() - dialStartedAt}`.replace(/[\n\r]/g, ''));
}
return bridge;
}
private awaitOpen(ws: WebSocket): Promise<void> {
return new Promise<void>((resolve, reject) => {
const cleanup = () => {
ws.removeAllListeners('open');
ws.removeAllListeners('error');
ws.removeAllListeners('unexpected-response');
};
ws.once('open', () => { cleanup(); resolve(); });
ws.once('error', (err) => { cleanup(); reject(err); });
ws.once('unexpected-response', (_req, res) => {
cleanup();
const err = new Error(`upgrade failed: HTTP ${res.statusCode}`) as Error & { httpStatus?: number };
err.httpStatus = res.statusCode ?? 0;
try { res.resume(); } catch { /* ignore */ }
reject(err);
});
});
}
/**
* Cache a dial failure and emit at most one activity-log entry per
* cache window per `(nodeId, code)`. The dedupe bounds operator log
* noise when a meshed container retries cross-node TCP opens against
* a misconfigured remote.
*/
private recordFailure(nodeId: number, code: DialFailureCode, rawMessage?: string, reasonOverride?: DialFailureReason): void {
const message = rawMessage ? sanitizeForLog(redactSensitiveText(rawMessage)) : undefined;
const previous = this.recentFailures.get(nodeId);
const reason = reasonOverride ?? code;
const isFresh = !previous
|| Date.now() - previous.ts > FAILURE_CACHE_TTL_MS
|| previous.code !== code;
this.recentFailures.set(nodeId, { code, message, ts: Date.now() });
PilotMetrics.increment('proxy_dials_failed');
if (isFresh) {
void this.logActivity(nodeId, 'open.fail', message ? { reason, message } : { reason });
}
if (isDebugEnabled()) {
console.warn(`[MeshProxyDialer:diag] dial failure node=${nodeId} code=${code} reason=${reason}${message ? ` message=${message}` : ''}`.replace(/[\n\r]/g, ''));
}
}
private startIdleCheck(): void {
if (this.idleCheckTimer || this.stopped) return;
if (this.idleTtlMs <= 0) return; // 0 disables idle close
this.idleCheckTimer = setInterval(() => this.runIdleCheck(), IDLE_CHECK_INTERVAL_MS);
this.idleCheckTimer.unref?.();
}
private runIdleCheck(): void {
const now = Date.now();
for (const [nodeId, bridge] of this.bridges) {
if (bridge.getActiveStreamCount() > 0) {
this.idleSince.set(nodeId, now);
continue;
}
const last = this.idleSince.get(nodeId) ?? now;
if (now - last >= this.idleTtlMs) {
this.bridges.delete(nodeId);
this.idleSince.delete(nodeId);
try { bridge.close(1000, 'idle timeout'); } catch { /* ignore */ }
PilotMetrics.increment('proxy_idle_closes');
void this.logActivity(nodeId, 'close', { reason: 'idle' });
this.emit('proxy-bridge-down', nodeId);
if (isDebugEnabled()) {
console.log(`[MeshProxyDialer:diag] idle close node=${nodeId} idleMs=${now - last}`);
}
}
}
}
private async logActivity(
nodeId: number,
event: ProxyTunnelEvent,
details: Record<string, unknown>,
): Promise<void> {
try {
// Lazy import to avoid a circular dependency with MeshService.
const { MeshService } = await import('./MeshService');
const message = typeof details.message === 'string'
? details.message
: `proxy tunnel ${event} for node ${nodeId}`;
MeshService.getInstance().logActivity({
source: 'mesh',
level: event === 'open.fail' ? 'error' : 'info',
type: ACTIVITY_TYPE[event],
nodeId,
message,
details,
});
} catch {
// Activity logging is best-effort; never let it propagate.
}
}
}
function classifyDialError(err: unknown): { code: DialFailureCode; message: string } {
const message = sanitizeForLog((err as Error).message || String(err));
const httpStatus = (err as Error & { httpStatus?: number }).httpStatus;
if (httpStatus === 404) return { code: 'endpoint_not_found', message: 'remote does not expose /api/mesh/proxy-tunnel' };
if (httpStatus === 401 || httpStatus === 403) return { code: 'auth_failed', message: 'api token rejected by remote' };
const tlsCodes = new Set(['CERT_HAS_EXPIRED', 'UNABLE_TO_VERIFY_LEAF_SIGNATURE', 'DEPTH_ZERO_SELF_SIGNED_CERT', 'SELF_SIGNED_CERT_IN_CHAIN']);
const errno = (err as NodeJS.ErrnoException).code;
if (errno && tlsCodes.has(errno)) return { code: 'tls_failed', message };
return { code: 'network_error', message };
}
+141 -82
View File
@@ -12,7 +12,9 @@ import { PROXY_TIER_HEADER, PROXY_VARIANT_HEADER } from './license-headers';
import { MeshForwarder, type MeshForwarderHost } from './MeshForwarder';
import { NodeRegistry } from './NodeRegistry';
import { PilotTunnelManager } from './PilotTunnelManager';
import { MeshProxyTunnelDialer, type DialFailureCode } from './MeshProxyTunnelDialer';
import { generateOverrideYaml, MeshAlias, SENCHO_MESH_NETWORK } from './MeshComposeOverride';
import { lookupContainerIp } from '../mesh/containerLookup';
import { sanitizeForLog } from '../utils/safeLog';
import { isPathWithinBase, isValidStackName } from '../utils/validation';
import { PORT as SENCHO_LISTEN_PORT } from '../helpers/constants';
@@ -24,6 +26,14 @@ const PROBE_TIMEOUT_MS = 5_000;
const SLOW_PROBE_THRESHOLD_MS = 500;
const DEFAULT_MESH_SUBNET = '172.30.0.0/24';
const REACHABLE_REASON: Record<DialFailureCode, string> = {
auth_failed: 'api token rejected (scope must be full-admin)',
endpoint_not_found: 'remote does not support proxy mesh',
tls_failed: 'TLS handshake failed',
no_target: 'proxy target missing',
network_error: 'remote unreachable',
};
/**
* Returns the static IPv4 address Sencho will pin itself to on the mesh
* Docker network: `<network address> + 2`. The Docker daemon assigns
@@ -60,7 +70,8 @@ export type MeshActivityType =
| 'mesh.enable' | 'mesh.disable'
| 'mesh.override.preserved'
| 'probe.ok' | 'probe.fail'
| 'forwarder.listen' | 'forwarder.unlisten' | 'forwarder.error';
| 'forwarder.listen' | 'forwarder.unlisten' | 'forwarder.error'
| 'proxy-tunnel.open.ok' | 'proxy-tunnel.open.fail' | 'proxy-tunnel.close';
export interface MeshActivityEvent {
ts: number;
@@ -105,13 +116,38 @@ export interface MeshRegenSummary {
reason?: string;
}
/**
* How a node participates in mesh routing right now:
* - `local`: the Sencho serving this request.
* - `pilot`: a remote with a pilot agent. Live-tunnel state is captured
* separately in `pilotConnected`.
* - `proxy`: a remote that central reaches via the long-lived api_token.
* Mesh tunnels are opened on demand by `MeshProxyTunnelDialer`; the
* operator sees no badge while the configuration is sound.
* - `unreachable`: configuration or runtime problem keeps mesh traffic
* from flowing. `reachableReason` carries an actionable hint.
*/
export type MeshReachableMode = 'local' | 'pilot' | 'proxy' | 'unreachable';
export interface MeshNodeStatus {
nodeId: number;
nodeName: string;
enabled: boolean;
/** Forwarder state for the LOCAL node (the Sencho instance answering this request). Always `null` for any non-local node — fetching the remote forwarder state requires a cross-node call which lands in Phase B. */
localForwarderListening: boolean | null;
/**
* True iff a pilot tunnel is currently registered for this node. Only
* meaningful when `reachableMode === 'pilot'`. Kept for diagnostic
* surfaces; the Routing tab badge logic consumes `reachableMode` and
* ignores this field for proxy / local nodes.
* TODO: collapse into `reachableMode` (introduce `pilot_offline` value)
* once no remaining caller reads `pilotConnected` directly.
*/
pilotConnected: boolean;
/** Canonical reachability classification consumed by the Routing tab. */
reachableMode: MeshReachableMode;
/** Short, operator-facing reason when `reachableMode === 'unreachable'`. Null otherwise. */
reachableReason: string | null;
optedInStacks: string[];
activeStreamCount: number;
}
@@ -1284,7 +1320,7 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
if (target.nodeId === selfNodeId) {
await this.openSameNode(target, src);
} else {
this.openCrossNode(target, src);
await this.openCrossNode(target, src);
}
}
@@ -1340,28 +1376,9 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
* central.
*/
public async resolveContainerIp(target: { stack: string; service: string }): Promise<string | null> {
const docker = DockerController.getInstance().getDocker() as unknown as Parameters<typeof lookupContainerIp>[0];
try {
const docker = DockerController.getInstance().getDocker();
// Compose default container name pattern; -1 is the first replica.
const conventionalName = `${target.stack}-${target.service}-1`;
const info = await docker.getContainer(conventionalName).inspect().catch(() => null);
const fromInspect = info ? this.extractContainerIp(target.stack, info) : null;
if (fromInspect) return fromInspect;
// Fallback: filter by compose labels in case of a non-conventional
// container name (operator overrode `container_name` or compose
// project).
const containers = await docker.listContainers({
all: true,
filters: {
label: [
`com.docker.compose.project=${target.stack}`,
`com.docker.compose.service=${target.service}`,
],
},
});
if (containers.length === 0) return null;
const fallbackInfo = await docker.getContainer(containers[0].Id).inspect().catch(() => null);
return fallbackInfo ? this.extractContainerIp(target.stack, fallbackInfo) : null;
return await lookupContainerIp(docker, target.stack, target.service);
} catch (err) {
console.warn('[MeshService] container IP lookup failed:', sanitizeForLog((err as Error).message));
return null;
@@ -1369,43 +1386,39 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
}
/**
* Pick a deterministic IP. Prefer the compose default network
* (`<stack>_default` or any network whose name starts with `<stack>_`),
* then any other declared network, then the legacy bridge `IPAddress`.
* Without this preference order, `Object.values(Networks)` ordering on
* containers attached to multiple networks varies across daemon
* versions and can make same-node forwarding flaky on a redeploy.
*/
private extractContainerIp(
stackName: string,
info: { NetworkSettings?: { Networks?: Record<string, { IPAddress?: string }>; IPAddress?: string } },
): string | null {
const networks = info.NetworkSettings?.Networks ?? {};
const composeDefault = networks[`${stackName}_default`];
if (composeDefault?.IPAddress) return composeDefault.IPAddress;
for (const [name, net] of Object.entries(networks)) {
if (name.startsWith(`${stackName}_`) && net?.IPAddress) return net.IPAddress;
}
for (const net of Object.values(networks)) {
if (net?.IPAddress) return net.IPAddress;
}
return info.NetworkSettings?.IPAddress || null;
}
/**
* Pluggable reverse dialer. Set by `PilotAgent` on a pilot host; left
* null on a central host. When set, `openCrossNode` routes outbound
* mesh dials through the agent's `tcp_open_reverse` path; when unset,
* `openCrossNode` uses the central-side `PilotTunnelManager.getBridge`
* directly. Lets the same MeshService code work on both sides.
* Pluggable reverse dialer. Set by `PilotAgent` on a pilot host or by
* the proxy-mode WS handler when a `/api/mesh/proxy-tunnel` upgrade
* comes in; left null on a central host with no inbound mesh tunnel.
* When set, `openCrossNode` routes outbound mesh dials through the
* dialer's `tcp_open_reverse` path; when unset, `openCrossNode` uses
* the central-side `PilotTunnelManager.getBridge` directly. Lets the
* same MeshService code work on both sides.
*/
private reverseDialer: ReverseMeshDialer | null = null;
public setReverseDialer(dialer: ReverseMeshDialer | null): void {
/**
* Install or clear the reverse dialer. Supports compare-and-swap via
* the optional `expected` argument so a caller (e.g., the proxy-mode
* WS handler) can install on a null slot and uninstall only if its
* own dialer is still the active one. Without the `expected` arg the
* operation is unconditional (used by the pilot agent at boot).
*
* Returns true when the swap happened, false when the CAS rejected
* because `expected` did not match the current dialer.
*/
public setReverseDialer(dialer: ReverseMeshDialer | null, expected?: ReverseMeshDialer | null): boolean {
if (expected !== undefined && this.reverseDialer !== expected) return false;
// Loud warn instead of silent overwrite: pilot and proxy modes are
// mutually exclusive by topology, so this branch indicates a
// misconfigured deployment rather than an expected race.
if (expected === undefined && dialer !== null && this.reverseDialer !== null && this.reverseDialer !== dialer) {
console.warn('[MeshService] reverse dialer overwritten without CAS; pilot/proxy mode race or duplicate install');
}
this.reverseDialer = dialer;
return true;
}
private dialMeshTcpStream(target: MeshTarget): MeshTcpStreamLike | null {
private async dialMeshTcpStream(target: MeshTarget): Promise<MeshTcpStreamLike | null> {
if (this.reverseDialer) {
return this.reverseDialer.openMeshTcpStream({
nodeId: target.nodeId,
@@ -1415,13 +1428,16 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
});
}
const ptm = PilotTunnelManager.getInstance();
if (!ptm.hasActiveTunnel(target.nodeId)) return null;
const bridge = ptm.getBridge(target.nodeId);
// ensureBridge resolves an existing pilot tunnel, an existing
// proxy-mode tunnel, or dials a fresh proxy-mode tunnel on demand.
// Returns null for unreachable nodes (no api_url/api_token, scope
// insufficient, remote pre-Phase-C, network error).
const bridge = await ptm.ensureBridge(target.nodeId);
if (!bridge) return null;
return bridge.openTcpStream({ stack: target.stack, service: target.service, port: target.port });
}
private openCrossNode(target: MeshTarget, src: net.Socket): void {
private async openCrossNode(target: MeshTarget, src: net.Socket): Promise<void> {
// Log every dispatch entry. Same-node logs route.resolve.ok on
// its TCP `connect` event; cross-node only logs route.resolve.ok
// once tcp_open_ack arrives from the agent. Without this entry
@@ -1433,14 +1449,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
message: `cross-node dispatch to ${target.alias} on node ${target.nodeId}`,
});
const tcpStream = this.dialMeshTcpStream(target);
const tcpStream = await this.dialMeshTcpStream(target);
if (!tcpStream) {
this.logActivity({
source: 'pilot', level: 'error', type: 'tunnel.fail',
nodeId: target.nodeId, alias: target.alias,
message: this.reverseDialer
? `cannot open reverse mesh stream to node ${target.nodeId}`
: `no active pilot tunnel to node ${target.nodeId}`,
: `no mesh tunnel reachable for node ${target.nodeId}`,
});
try { src.destroy(); } catch { /* ignore */ }
return;
@@ -1597,19 +1613,45 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
// --- Diagnostics ---
/**
* Whether mesh traffic to this node can flow. Local nodes are always
* reachable because mesh uses the same-node fast path on localhost.
* Remote nodes are reachable only when a pilot tunnel is registered. The
* literal `hasActiveTunnel(localNodeId)` would always be false (local
* nodes do not establish tunnels to themselves), so a direct call would
* render every local alias as `tunnel down` in the UI even on a working
* route.
* Whether mesh CAN route to a node based on current configuration.
* Distinct from "live tunnel up right now": for proxy-mode remotes
* the tunnel is opened on demand, so a caller that demanded a
* current tunnel would misreport every idle proxy-mode route as
* `tunnel down`. Live pilot-tunnel state is surfaced separately.
*/
private isMeshReachable(nodeId: number): boolean {
const node = DatabaseService.getInstance().getNode(nodeId);
private isNodeMeshConfigured(node: ReturnType<typeof DatabaseService.prototype.getNode>): boolean {
if (!node) return false;
if (node.type !== 'remote') return true;
return PilotTunnelManager.getInstance().hasActiveTunnel(nodeId);
if (node.mode === 'pilot_agent') return true;
if (node.mode === 'proxy') return !!node.api_url && !!node.api_token;
return false;
}
/**
* Compute the reachability classification surfaced in `MeshNodeStatus`
* and consumed by the Routing tab. See `MeshReachableMode` for the
* meaning of each value. Callers pass in the already-fetched node row
* (and the local nodeId) so this helper is free of DB I/O when
* `getStatus` iterates the fleet.
*/
private computeReachable(node: ReturnType<typeof DatabaseService.prototype.getNode>, localNodeId: number): { mode: MeshReachableMode; reason: string | null } {
if (!node) return { mode: 'unreachable', reason: 'unknown node' };
if (node.id === localNodeId || node.type !== 'remote') return { mode: 'local', reason: null };
if (node.mode === 'pilot_agent') return { mode: 'pilot', reason: null };
if (node.mode === 'proxy') {
if (!node.api_url) return { mode: 'unreachable', reason: 'api_url not set' };
if (!node.api_token) return { mode: 'unreachable', reason: 'api token missing' };
// Recent-failure cache surfaces the last failed dial so the
// operator sees a clear reason without triggering a redial
// storm.
const failure = MeshProxyTunnelDialer.getInstance().getRecentFailure(node.id);
if (failure) {
const reason = REACHABLE_REASON[failure.code] ?? failure.message ?? 'remote unreachable';
return { mode: 'unreachable', reason };
}
return { mode: 'proxy', reason: null };
}
return { mode: 'unreachable', reason: 'unknown node mode' };
}
public async getRouteDiagnostic(alias: string): Promise<MeshRouteDiagnostic> {
@@ -1621,14 +1663,20 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
return { alias, target: null, pilot: { connected: false, lastSeen: null }, lastError, lastProbeMs, state: 'not authorized' };
}
const pilotConnected = this.isMeshReachable(target.nodeId);
const node = DatabaseService.getInstance().getNode(target.nodeId);
const routable = this.isNodeMeshConfigured(node);
// Pilot-mode routes surface live tunnel state; proxy-mode routes
// fall back to `routable` because the tunnel is opened on demand
// and a quiescent state is normal.
const pilotLive = node?.type === 'remote' && node.mode === 'pilot_agent'
? PilotTunnelManager.getInstance().hasActiveTunnel(target.nodeId)
: routable;
const lastSeen = node?.pilot_last_seen ?? null;
const optedIn = DatabaseService.getInstance().isMeshStackEnabled(target.nodeId, target.stackName);
let state: MeshRouteDiagnostic['state'];
if (!optedIn) state = 'not authorized';
else if (!pilotConnected) state = 'tunnel down';
else if (!routable || !pilotLive) state = 'tunnel down';
else if (lastError && Date.now() - lastError.ts < 60_000) state = 'unreachable';
else if (lastProbeMs !== null && lastProbeMs > SLOW_PROBE_THRESHOLD_MS) state = 'degraded';
else state = 'healthy';
@@ -1642,7 +1690,7 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
port: target.port,
alias,
},
pilot: { connected: pilotConnected, lastSeen },
pilot: { connected: pilotLive, lastSeen },
lastError,
lastProbeMs,
state,
@@ -1689,15 +1737,26 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
const nodes = db.getNodes();
const localNodeId = NodeRegistry.getInstance().getDefaultNodeId();
const localListening = this.isLocalForwarderActive();
return nodes.map((node) => ({
nodeId: node.id,
nodeName: node.name,
enabled: db.getNodeMeshEnabled(node.id),
localForwarderListening: node.id === localNodeId ? localListening : null,
pilotConnected: this.isMeshReachable(node.id),
optedInStacks: db.listMeshStacks(node.id).map((s) => s.stack_name),
activeStreamCount: this.activeStreams.size,
}));
const ptm = PilotTunnelManager.getInstance();
return nodes.map((node) => {
const reach = this.computeReachable(node, localNodeId);
return {
nodeId: node.id,
nodeName: node.name,
enabled: db.getNodeMeshEnabled(node.id),
localForwarderListening: node.id === localNodeId ? localListening : null,
// `pilotConnected` stays at its original meaning: a pilot
// tunnel is currently registered for this node. The
// Routing tab now derives badge state from `reachableMode`
// and reads `pilotConnected` only for the pilot-offline
// sub-state.
pilotConnected: node.type !== 'remote' || ptm.hasActiveTunnel(node.id),
reachableMode: reach.mode,
reachableReason: reach.reason,
optedInStacks: db.listMeshStacks(node.id).map((s) => s.stack_name),
activeStreamCount: this.activeStreams.size,
};
});
}
/** True when the local Sencho's forwarder is started and bound to at least one alias port. */
@@ -1731,7 +1790,7 @@ export class MeshError extends Error {
/**
* Common surface of an outbound mesh TCP stream as MeshService consumes
* it. Both the central-side `PilotTunnelBridge.TcpStream` and the
* pilot-side `ReverseTcpStreamHandle` (from `pilot/agent.ts`) implement
* pilot-side `ReverseTcpStreamHandle` (from `mesh/tcpStreamSwitchboard.ts`) implement
* this shape structurally so MeshService.openCrossNode can splice bytes
* against either without caring which side initiated the stream.
*/
+9
View File
@@ -16,6 +16,12 @@ interface Counters {
tunnels_rejected_capacity: number;
enroll_acks: number;
frame_decode_errors: number;
/** Successful Distributed API mesh proxy-tunnel registrations. Counterpart to `tunnels_total` for pilot-mode tunnels; the two together cover every mesh-capable bridge the manager ever accepted. */
proxy_bridges_total: number;
/** Failed proxy-tunnel dial attempts (all reasons). Pair with `proxy_bridges_total` for an attempt/success ratio; pair with `proxy_idle_closes` for retention. */
proxy_dials_failed: number;
/** Proxy-tunnel teardowns initiated by the dialer's idle sweep (zero active streams for the configured TTL). */
proxy_idle_closes: number;
}
class PilotMetricsImpl {
@@ -25,6 +31,9 @@ class PilotMetricsImpl {
tunnels_rejected_capacity: 0,
enroll_acks: 0,
frame_decode_errors: 0,
proxy_bridges_total: 0,
proxy_dials_failed: 0,
proxy_idle_closes: 0,
};
public increment<K extends keyof Counters>(name: K): void {
+6 -1
View File
@@ -203,6 +203,8 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
public getLoopbackUrl(): string { return this.loopbackUrl; }
public getConnectedAt(): number { return this.connectedAt; }
public getBufferedAmount(): number { return this.tunnelWs.bufferedAmount; }
/** Total active stream count across HTTP, WebSocket, forward TCP, and reverse TCP. Used by the proxy-tunnel dialer to detect idle bridges eligible for teardown. */
public getActiveStreamCount(): number { return this.streams.size; }
public isOpen(): boolean { return !this.closed && this.tunnelWs.readyState === WebSocket.OPEN; }
/**
@@ -823,7 +825,10 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
private async acceptReverseRelay(s: number, targetNodeId: number, target: { stack: string; service: string; port: number }): Promise<void> {
const { PilotTunnelManager } = await import('./PilotTunnelManager');
const targetBridge = PilotTunnelManager.getInstance().getBridge(targetNodeId);
// ensureBridge resolves either an existing pilot tunnel or a fresh
// proxy-mode tunnel dialed on demand, so proxy <-> proxy relays
// succeed even when neither remote has a pilot agent.
const targetBridge = await PilotTunnelManager.getInstance().ensureBridge(targetNodeId);
if (!targetBridge) {
this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' });
return;
+81 -8
View File
@@ -34,17 +34,36 @@ export class PilotTunnelCapacityError extends Error {
}
/**
* PilotTunnelManager: singleton registry of active pilot tunnels.
* PilotTunnelManager: singleton registry of active mesh-capable bridges.
*
* Each enrolled pilot-agent node holds one outbound WebSocket to the primary.
* For every such tunnel we spin up a local loopback HTTP server that demuxes
* requests into frames. Remote-proxy code paths (http-proxy-middleware and the
* WebSocket upgrade handler) can then treat pilot nodes identically to standard
* proxy nodes by pointing at the loopback URL.
* Two flavors of bridge live in the same `bridges` map, keyed by nodeId:
*
* - **Pilot-agent tunnels** (the original use case): long-lived,
* agent-initiated. The pilot dials central; `registerTunnel` accepts
* the WS, starts a loopback HTTP server, and emits `tunnel-up` so
* downstream observers (capability cache, status badges) refresh. A
* pilot-agent bridge stays open for the agent's lifetime and supports
* HTTP, WebSocket, and TCP multiplexing.
*
* - **Proxy-mode tunnels** (Phase C): short-lived, central-initiated.
* `ensureBridge` delegates to `MeshProxyTunnelDialer`, which opens a
* WebSocket to the remote's `/api/mesh/proxy-tunnel` endpoint using
* the long-lived `api_token`. Carries only TCP mesh frames; the
* loopback HTTP server is left running on the bridge but unused
* because proxy-mode HTTP traffic flows through the existing
* `remoteNodeProxy`. Idle close after a configurable TTL.
*
* Mesh dispatch is mode-agnostic: `MeshService.dialMeshTcpStream` awaits
* `ensureBridge(nodeId)` and consumes the resulting `MeshTunnelHandle`.
*
* Events:
* - 'tunnel-up' (nodeId: number) after a tunnel is accepted
* - 'tunnel-down' (nodeId: number) after a tunnel closes (for any reason)
* - 'tunnel-up' (nodeId) when a pilot-agent tunnel is accepted (NOT
* emitted for proxy-mode bridges, which are opened on demand and
* should not trigger pilot-specific listeners like the F9 capability
* cache invalidation).
* - 'tunnel-down' (nodeId) when a pilot-agent tunnel closes.
* - 'proxy-bridge-up' / 'proxy-bridge-down' (nodeId) for observability
* on proxy-mode bridge lifecycle. No current consumer.
*/
export class PilotTunnelManager extends EventEmitter {
private static instance: PilotTunnelManager;
@@ -168,6 +187,60 @@ export class PilotTunnelManager extends EventEmitter {
return this.bridges.get(nodeId) ?? null;
}
/**
* Dial-if-needed: return the existing pilot or proxy bridge, or open
* a new proxy-mode bridge on demand. Used by `MeshService` so cross-
* node TCP dispatch works for both pilot-agent remotes (long-lived
* tunnel) and proxy-mode remotes (on-demand tunnel) without any
* mode-specific branching at the call site.
*
* Returns null if the node has no active pilot tunnel AND cannot be
* dialed as a proxy-mode remote (missing api_url / api_token, scope
* insufficient, remote offline, or remote pre-Phase-C).
*/
public async ensureBridge(nodeId: number): Promise<MeshTunnelHandle | null> {
const existing = this.bridges.get(nodeId);
if (existing) return existing;
// Lazy import to avoid a cycle: MeshProxyTunnelDialer imports
// PilotTunnelBridge, which imports PilotTunnelManager via the
// existing tcp_open_reverse relay path.
const { MeshProxyTunnelDialer } = await import('./MeshProxyTunnelDialer');
return MeshProxyTunnelDialer.getInstance().ensureBridge(nodeId);
}
/**
* Register a central-initiated proxy-mode bridge for an existing
* remote. Distinct from `registerTunnel`: skips the pilot-only side
* effects (DB node-status update, `pilot_last_seen` write,
* `tunnel-up` event, replacement of any prior pilot tunnel). Still
* honors the hard tunnel cap so a dial storm cannot exhaust gateway
* memory.
*
* Throws `PilotTunnelCapacityError` when the cap is reached.
*/
public registerProxyBridge(nodeId: number, bridge: PilotTunnelBridge): void {
const existing = this.bridges.get(nodeId);
if (existing) {
// A pilot tunnel for this node already exists. Proxy bridges
// should not silently shadow them; refuse the registration so
// the dialer can surface a clear error.
throw new Error(`pilot tunnel already registered for node ${nodeId}; proxy bridge refused`);
}
if (this.bridges.size >= PILOT_TUNNEL_HARD_LIMIT) {
PilotMetrics.increment('tunnels_rejected_capacity');
throw new PilotTunnelCapacityError(PILOT_TUNNEL_HARD_LIMIT);
}
bridge.once('closed', () => {
if (this.bridges.get(nodeId) === bridge) {
this.bridges.delete(nodeId);
this.emit('proxy-bridge-down', nodeId);
}
});
this.bridges.set(nodeId, bridge);
PilotMetrics.increment('proxy_bridges_total');
this.emit('proxy-bridge-up', nodeId);
}
/**
* Force-close a tunnel (e.g., on node deletion).
*/
+15
View File
@@ -0,0 +1,15 @@
/**
* Convert an HTTP(S) base URL to a WebSocket URL by upgrading the scheme.
* Maps `http://` to `ws://` and `https://` to `wss://`; trims a trailing
* slash so the caller can append a path. Other schemes pass through
* unchanged because the caller may already have a `ws://` or `wss://`
* URL.
*
* Existed previously as inline regex replacements in several places that
* either dropped the trailing slash unevenly or used `replace(/^http/, 'ws')`
* which silently downgrades `https://` to `ws://` (cleartext). This helper
* is the single correct form.
*/
export function httpUrlToWs(baseUrl: string): string {
return baseUrl.replace(/\/$/, '').replace(/^https?/, (m) => (m === 'https' ? 'wss' : 'ws'));
}
+137
View File
@@ -0,0 +1,137 @@
import type { IncomingMessage } from 'http';
import type { Duplex } from 'stream';
import { WebSocketServer, type WebSocket } from 'ws';
import { MAX_FRAME_SIZE_BYTES, decodeBinaryFrame, decodeJsonFrame, wsDataToBuffer, wsDataToString } from '../pilot/protocol';
import {
TcpStreamSwitchboard,
attachTcpStreamSwitchboard,
resolveByComposeLabels,
type ReverseTcpStreamHandle,
} from '../mesh/tcpStreamSwitchboard';
import { sanitizeForLog } from '../utils/safeLog';
import { isDebugEnabled } from '../utils/debug';
import { rejectUpgrade as reject } from './reject';
/**
* Mesh proxy-tunnel ingress.
*
* The remote side of a Phase C proxy-mode mesh tunnel. Central dials
* `WSS <api_url>/api/mesh/proxy-tunnel` using the long-lived `api_token`
* as a Bearer credential; this handler upgrades the connection and wires
* the shared `TcpStreamSwitchboard` to handle `tcp_open` / `tcp_open_ack`
* / `tcp_open_reverse` / `tcp_close` and `TcpData` frames.
*
* Auth + scope gating happens in `upgradeHandler.ts` before this handler
* runs (require `full-admin` api_token scope). The handler itself trusts
* the upgrade; the WS credential is the only trust boundary.
*
* Bidirectional: when the tunnel opens, the handler registers itself as
* the reverse-dialer on the local MeshService so meshed containers on
* this Sencho can dial cross-node aliases via `tcp_open_reverse` over
* the same WS. On disconnect the reverse-dialer registration is
* cleared.
*/
const wss = new WebSocketServer({ noServer: true, maxPayload: MAX_FRAME_SIZE_BYTES });
interface SwitchboardReverseDialer {
openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null;
}
export async function handleMeshProxyTunnel(req: IncomingMessage, socket: Duplex, head: Buffer): Promise<void> {
// Mesh service is only available on the central deployment (SENCHO_MODE
// unset or 'central'). Pilot-mode Sencho receives mesh traffic via the
// pilot tunnel and has no use for the proxy-mode WS path.
if (process.env.SENCHO_MODE === 'pilot') {
return reject(socket, 404, 'Not Found');
}
await new Promise<void>((resolve) => {
wss.handleUpgrade(req, socket as Parameters<typeof wss.handleUpgrade>[1], head, (ws) => {
void attachSwitchboard(ws).finally(resolve);
});
});
}
async function attachSwitchboard(ws: WebSocket): Promise<void> {
let switchboard: TcpStreamSwitchboard | null = null;
let meshServiceCleanup: (() => void) | null = null;
try {
switchboard = attachTcpStreamSwitchboard({
ws,
resolveTarget: resolveByComposeLabels,
logLabel: 'MeshProxy',
});
// Register a reverse dialer so this side's MeshForwarder can dial
// cross-node aliases via `tcp_open_reverse` over the same WS. The
// CAS swap refuses to overwrite a dialer that another caller
// (a concurrent proxy-tunnel upgrade, or a pilot agent in a
// misconfigured deployment) has already installed.
const { MeshService } = await import('../services/MeshService');
const meshService = MeshService.getInstance();
const localSwitchboard = switchboard;
const localDialer: SwitchboardReverseDialer = {
openMeshTcpStream(target) {
return localSwitchboard.openReverseStream(target);
},
};
const installed = meshService.setReverseDialer(localDialer, null);
if (!installed) {
console.warn('[MeshProxy] reverse dialer already installed; rejecting concurrent tunnel');
try { ws.close(1013, 'reverse dialer already installed'); } catch { /* ignore */ }
switchboard.cleanup('reverse dialer already installed');
switchboard = null;
return;
}
meshServiceCleanup = () => {
meshService.setReverseDialer(null, localDialer);
};
} catch (err) {
if (isDebugEnabled()) {
console.warn('[MeshProxy:diag] failed to attach switchboard:', sanitizeForLog((err as Error).message));
}
try { ws.close(1011, 'switchboard attach failed'); } catch { /* ignore */ }
return;
}
const onMessage = (data: unknown, isBinary: boolean): void => {
if (!switchboard) return;
try {
if (isBinary) {
const buf = wsDataToBuffer(data);
if (!buf) return;
switchboard.handleBinaryFrame(decodeBinaryFrame(buf));
} else {
const text = wsDataToString(data);
if (text == null) return;
switchboard.handleJsonFrame(decodeJsonFrame(text));
}
} catch (err) {
if (isDebugEnabled()) {
console.warn('[MeshProxy:diag] malformed frame:', sanitizeForLog((err as Error).message));
}
}
};
const teardown = (): void => {
ws.off('message', onMessage);
if (switchboard) {
switchboard.cleanup('mesh proxy-tunnel closed');
switchboard = null;
}
if (meshServiceCleanup) {
meshServiceCleanup();
meshServiceCleanup = null;
}
};
ws.on('message', onMessage);
ws.once('close', teardown);
ws.once('error', (err) => {
if (isDebugEnabled()) {
console.warn('[MeshProxy:diag] ws error:', sanitizeForLog(err.message));
}
teardown();
});
}
+19 -5
View File
@@ -7,6 +7,7 @@ import { DatabaseService, type UserRole } from '../services/DatabaseService';
import { NodeRegistry } from '../services/NodeRegistry';
import { COOKIE_NAME } from '../helpers/constants';
import { handlePilotTunnel } from './pilotTunnel';
import { handleMeshProxyTunnel } from './meshProxyTunnel';
import { handleNotificationsWs } from './notifications';
import { handleRemoteForwarder } from './remoteForwarder';
import { handleLogsWs } from './logs';
@@ -32,11 +33,12 @@ function parseCookies(req: IncomingMessage): Record<string, string> {
* 1. `/api/pilot/tunnel` -> handlePilotTunnel (own auth, own wss)
* 2. shared cookie/Bearer auth + JWT verify (rejects unauthenticated)
* 3. API token scope gate (read-only / deploy-only restricted to logs + notifications)
* 4. `/ws/notifications` local -> handleNotificationsWs
* 5. remote nodeId path -> handleRemoteForwarder
* 6. `/api/stacks/:name/logs` -> handleLogsWs
* 7. `/api/system/host-console` -> handleHostConsoleWs
* 8. fallback -> handleGenericWs (`/ws` exec + stats)
* 4. `/api/mesh/proxy-tunnel` -> handleMeshProxyTunnel (requires full-admin api_token scope)
* 5. `/ws/notifications` local -> handleNotificationsWs
* 6. remote nodeId path -> handleRemoteForwarder
* 7. `/api/stacks/:name/logs` -> handleLogsWs
* 8. `/api/system/host-console` -> handleHostConsoleWs
* 9. fallback -> handleGenericWs (`/ws` exec + stats)
*/
export function attachUpgrade(
server: http.Server,
@@ -120,6 +122,18 @@ export function attachUpgrade(
}
}
// Mesh proxy-tunnel ingress: a sibling Sencho is dialing this node
// to carry mesh TCP traffic. Require an api_token Bearer with the
// full-admin scope; mesh manipulates traffic and must not be
// reachable under a session cookie or a node_proxy JWT.
if (pathname === '/api/mesh/proxy-tunnel') {
if (wsApiTokenScope !== 'full-admin') {
return reject(socket, 403, 'Forbidden');
}
await handleMeshProxyTunnel(req, socket, head);
return;
}
const nodeIdParam = parsedUrl.searchParams.get('nodeId');
const nodeId = nodeIdParam ? parseInt(nodeIdParam, 10) : NodeRegistry.getInstance().getDefaultNodeId();
const node = NodeRegistry.getInstance().getNode(nodeId);