mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-08-12 03:36:59 +00:00
fix(mesh): route peer→central traffic over the existing forward WS (#1094)
* fix(mesh): route peer→central traffic over the existing forward WS The reverse mesh callback path (`/api/mesh/proxy-tunnel-from-peer`) needed SENCHO_PRIMARY_URL on central plus a publicly reachable origin from the peer's perspective. In a typical homelab where central sits behind NAT, peer→central dispatch silently failed at the dialer's short-circuit and the headline "call any service on any node by hostname" worked one way only. The forward WS at `/api/mesh/proxy-tunnel` is already bidirectional end to end. Make the bridge a persistent control-plane primitive: dial every mesh-enabled proxy peer at startup, reconcile every 60 s, never idle-close. Peer→central traffic multiplexes over the same WS via `tcp_open_reverse`. Removed: - `meshProxyTunnelFromPeer.ts` WS handler and dispatch - `MeshCentralRegistry`, `PeerToCentralMeshSessionDialer` - `mesh_handshake` first-frame state machine in `meshProxyTunnel.ts` - `maybeSendBootstrap`, `buildHandshakeFrame` in the dialer - `mesh_proxy_callback_bootstrap` capability and `maybeWarnUnsetPrimaryUrl` - `mesh_centrals` table (drop migration; greenfield, no users) - `PilotTunnelManager.replaceOrRegisterProxyBridge` (dead after handler removal) - twelve associated unit/integration tests plus the peer-recovery branch in `MeshService.openCrossNode` Added: - `MeshService.proactiveBridgeFanout` selects every mesh-enabled proxy peer (no longer gated on `mesh_stacks` rows) - `startBridgeReconcileLoop` runs the fanout every 60 s (override via `SENCHO_MESH_RECONCILE_INTERVAL_MS`) - `MeshProxyTunnelDialer` default idle TTL is now `0` and exposes `isDialing(nodeId)` for the status surface - `MeshNodeStatus.reverseCallbackStatus` discriminator (`connected | connecting | unavailable | not_applicable`) surfaced via `/api/mesh/status` and rendered as a pill in the Routing tab - `openCrossNode` error message distinguishes "no proxy target" from "waiting for central to dial the reverse bridge" - New tests: `mesh-service-proxy-tunnel-reconcile`, `mesh-status-reverse-callback`, `mesh-proxy-tunnel-dialer-no-idle-close` SENCHO_PRIMARY_URL is no longer required for any mesh function. * fix(mesh): rewrite proxy-tunnel reconcile test contents The previous commit renamed the file but the rewritten test bodies stayed unstaged on top of the rename. This commit lands the actual rewrite: the fanout assertion now requires every mesh-enabled proxy peer to be dialed, not just those with `mesh_stacks` rows, and adds a reconcile-tick repeated-call test.
This commit is contained in:
Anso
@@ -34,7 +34,6 @@ export const CAPABILITIES = [
|
||||
'registries',
|
||||
'self-update',
|
||||
'vulnerability-scanning',
|
||||
'mesh_proxy_callback_bootstrap',
|
||||
] as const;
|
||||
|
||||
export type Capability = (typeof CAPABILITIES)[number];
|
||||
|
||||
@@ -1489,23 +1489,13 @@ export class DatabaseService {
|
||||
} catch (e) {
|
||||
console.warn('[DatabaseService] mesh_stacks migration:', (e as Error).message);
|
||||
}
|
||||
try {
|
||||
this.db.prepare(`
|
||||
CREATE TABLE IF NOT EXISTS mesh_centrals (
|
||||
central_instance_id TEXT PRIMARY KEY,
|
||||
central_api_url TEXT NOT NULL,
|
||||
callback_jwt TEXT NOT NULL,
|
||||
jwt_issued_at INTEGER NOT NULL,
|
||||
jwt_expires_at INTEGER NOT NULL,
|
||||
last_bootstrap_at INTEGER NOT NULL,
|
||||
last_used_at INTEGER,
|
||||
last_rejected_at INTEGER,
|
||||
last_reject_reason TEXT
|
||||
)
|
||||
`).run();
|
||||
} catch (e) {
|
||||
console.warn('[DatabaseService] Could not create mesh_centrals:', (e as Error).message);
|
||||
}
|
||||
// mesh_centrals was the peer-side cache of the reverse-callback JWT
|
||||
// (central → peer bootstrap material). Peer→central traffic now
|
||||
// multiplexes over the existing forward WS via `tcp_open_reverse`,
|
||||
// so the table is no longer written or read. Drop it on every boot;
|
||||
// idempotent.
|
||||
try { this.db.prepare('DROP TABLE IF EXISTS mesh_centrals').run(); }
|
||||
catch (e) { console.warn('[DatabaseService] Could not drop mesh_centrals:', (e as Error).message); }
|
||||
this.tryAddColumn('nodes', 'mesh_enabled', 'INTEGER NOT NULL DEFAULT 0');
|
||||
}
|
||||
|
||||
|
||||
@@ -1,119 +0,0 @@
|
||||
import { EventEmitter } from 'events';
|
||||
import { DatabaseService } from './DatabaseService';
|
||||
|
||||
export interface MeshCentralMaterial {
|
||||
centralInstanceId: string;
|
||||
centralApiUrl: string;
|
||||
callbackJwt: string;
|
||||
jwtIssuedAt: number;
|
||||
jwtExpiresAt: number;
|
||||
}
|
||||
|
||||
export interface MeshCentralRow extends MeshCentralMaterial {
|
||||
lastBootstrapAt: number;
|
||||
lastUsedAt: number | null;
|
||||
lastRejectedAt: number | null;
|
||||
lastRejectReason: string | null;
|
||||
}
|
||||
|
||||
export class MeshCentralRegistry extends EventEmitter {
|
||||
private static instance: MeshCentralRegistry | null = null;
|
||||
private warnedMultiRow = false;
|
||||
|
||||
private constructor() { super(); }
|
||||
|
||||
public static getInstance(): MeshCentralRegistry {
|
||||
if (!this.instance) this.instance = new MeshCentralRegistry();
|
||||
return this.instance;
|
||||
}
|
||||
|
||||
public static resetForTest(): void {
|
||||
this.instance = null;
|
||||
}
|
||||
|
||||
public upsert(material: MeshCentralMaterial): void {
|
||||
const db = DatabaseService.getInstance().getDb();
|
||||
const prior = this.getActive();
|
||||
db.prepare(`
|
||||
INSERT INTO mesh_centrals (
|
||||
central_instance_id, central_api_url, callback_jwt,
|
||||
jwt_issued_at, jwt_expires_at, last_bootstrap_at,
|
||||
last_used_at, last_rejected_at, last_reject_reason
|
||||
) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
|
||||
ON CONFLICT(central_instance_id) DO UPDATE SET
|
||||
central_api_url = excluded.central_api_url,
|
||||
callback_jwt = excluded.callback_jwt,
|
||||
jwt_issued_at = excluded.jwt_issued_at,
|
||||
jwt_expires_at = excluded.jwt_expires_at,
|
||||
last_bootstrap_at = excluded.last_bootstrap_at
|
||||
`).run(
|
||||
material.centralInstanceId,
|
||||
material.centralApiUrl,
|
||||
material.callbackJwt,
|
||||
material.jwtIssuedAt,
|
||||
material.jwtExpiresAt,
|
||||
Date.now(),
|
||||
);
|
||||
|
||||
const isInstanceChange = prior && prior.centralInstanceId !== material.centralInstanceId;
|
||||
if (isInstanceChange) {
|
||||
this.emit('central-instance-changed', {
|
||||
previousInstanceId: prior!.centralInstanceId,
|
||||
newInstanceId: material.centralInstanceId,
|
||||
});
|
||||
}
|
||||
this.emit('central-bootstrap', { centralInstanceId: material.centralInstanceId });
|
||||
}
|
||||
|
||||
public getActive(): MeshCentralRow | null {
|
||||
const db = DatabaseService.getInstance().getDb();
|
||||
const rows = db.prepare(`
|
||||
SELECT * FROM mesh_centrals ORDER BY last_bootstrap_at DESC
|
||||
`).all() as Array<{
|
||||
central_instance_id: string;
|
||||
central_api_url: string;
|
||||
callback_jwt: string;
|
||||
jwt_issued_at: number;
|
||||
jwt_expires_at: number;
|
||||
last_bootstrap_at: number;
|
||||
last_used_at: number | null;
|
||||
last_rejected_at: number | null;
|
||||
last_reject_reason: string | null;
|
||||
}>;
|
||||
if (rows.length === 0) return null;
|
||||
if (rows.length > 1 && !this.warnedMultiRow) {
|
||||
console.warn(`[MeshCentralRegistry] multiple central rows present (${rows.length}), multi-central unsupported in v0.79`);
|
||||
this.warnedMultiRow = true;
|
||||
}
|
||||
const r = rows[0];
|
||||
return {
|
||||
centralInstanceId: r.central_instance_id,
|
||||
centralApiUrl: r.central_api_url,
|
||||
callbackJwt: r.callback_jwt,
|
||||
jwtIssuedAt: r.jwt_issued_at,
|
||||
jwtExpiresAt: r.jwt_expires_at,
|
||||
lastBootstrapAt: r.last_bootstrap_at,
|
||||
lastUsedAt: r.last_used_at,
|
||||
lastRejectedAt: r.last_rejected_at,
|
||||
lastRejectReason: r.last_reject_reason,
|
||||
};
|
||||
}
|
||||
|
||||
public clearForInstance(centralInstanceId: string): void {
|
||||
DatabaseService.getInstance().getDb()
|
||||
.prepare(`DELETE FROM mesh_centrals WHERE central_instance_id = ?`)
|
||||
.run(centralInstanceId);
|
||||
}
|
||||
|
||||
public markUsed(centralInstanceId: string): void {
|
||||
DatabaseService.getInstance().getDb()
|
||||
.prepare(`UPDATE mesh_centrals SET last_used_at = ? WHERE central_instance_id = ?`)
|
||||
.run(Date.now(), centralInstanceId);
|
||||
}
|
||||
|
||||
public markRejected(centralInstanceId: string, reason: string): void {
|
||||
DatabaseService.getInstance().getDb()
|
||||
.prepare(`UPDATE mesh_centrals SET last_rejected_at = ?, last_reject_reason = ? WHERE central_instance_id = ?`)
|
||||
.run(Date.now(), reason, centralInstanceId);
|
||||
}
|
||||
}
|
||||
@@ -1,12 +1,9 @@
|
||||
import { EventEmitter } from 'events';
|
||||
import { createHash } from 'crypto';
|
||||
import jwt from 'jsonwebtoken';
|
||||
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 { DatabaseService } from './DatabaseService';
|
||||
import { redactSensitiveText, sanitizeForLog } from '../utils/safeLog';
|
||||
import { httpUrlToWs } from '../utils/wsUrl';
|
||||
import { isDebugEnabled } from '../utils/debug';
|
||||
@@ -29,15 +26,16 @@ import type { MeshActivityType } from './MeshService';
|
||||
* 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).
|
||||
* - The bridge is a persistent bidirectional control-plane channel:
|
||||
* central → peer `tcp_open` frames and peer → central `tcp_open_reverse`
|
||||
* frames flow over the same WS. The default idle TTL is 0 (no idle
|
||||
* close); `SENCHO_MESH_PROXY_TUNNEL_IDLE_MS` overrides the default for
|
||||
* operators who still want stream-scoped tunnels.
|
||||
* - 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 DEFAULT_IDLE_TTL_MS = 0;
|
||||
const IDLE_CHECK_INTERVAL_MS = 60 * 1000;
|
||||
const HANDSHAKE_TIMEOUT_MS = 15_000;
|
||||
const FAILURE_CACHE_TTL_MS = 60 * 1000;
|
||||
@@ -148,6 +146,11 @@ export class MeshProxyTunnelDialer extends EventEmitter {
|
||||
return this.bridges.get(nodeId) ?? null;
|
||||
}
|
||||
|
||||
/** True when a dial for this node is currently in flight. */
|
||||
public isDialing(nodeId: number): boolean {
|
||||
return this.inflight.has(nodeId);
|
||||
}
|
||||
|
||||
public getRecentFailure(nodeId: number): DialFailure | null {
|
||||
const entry = this.recentFailures.get(nodeId);
|
||||
if (!entry) return null;
|
||||
@@ -262,18 +265,6 @@ export class MeshProxyTunnelDialer extends EventEmitter {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Capability-gated mesh_handshake: pushes a signed `mesh_tunnel` JWT
|
||||
// and the central origin so the peer can dial central back for
|
||||
// reverse cross-fleet TCP streams (R2). Skipped silently when the
|
||||
// peer lacks the capability or `SENCHO_PRIMARY_URL` is unset so
|
||||
// legacy peers keep their forward-only TCP path untouched.
|
||||
await this.maybeSendBootstrap(nodeId, ws);
|
||||
|
||||
if (this.stopped) {
|
||||
try { ws.close(1001, 'dialer shutdown'); } catch { /* ignore */ }
|
||||
return null;
|
||||
}
|
||||
|
||||
const bridge = new PilotTunnelBridge(nodeId, ws);
|
||||
try {
|
||||
await bridge.start();
|
||||
@@ -451,47 +442,6 @@ export class MeshProxyTunnelDialer extends EventEmitter {
|
||||
this.redialTimers.set(nodeId, timer);
|
||||
}
|
||||
|
||||
/**
|
||||
* If the peer advertises `mesh_proxy_callback_bootstrap` and central has
|
||||
* a canonical origin (`SENCHO_PRIMARY_URL`), mint a `mesh_tunnel`-scoped
|
||||
* JWT bound to the peer's `api_token` fingerprint and push it as the
|
||||
* first text frame on the freshly-opened WS. The peer's
|
||||
* `meshProxyTunnel` first-frame state machine persists the bootstrap
|
||||
* material before yielding to TCP traffic. Every other path is a silent
|
||||
* skip so legacy peers continue to receive raw TCP frames immediately.
|
||||
*/
|
||||
private async maybeSendBootstrap(nodeId: number, ws: WebSocket): Promise<void> {
|
||||
const canonicalOrigin = (process.env.SENCHO_PRIMARY_URL ?? '').replace(/\/+$/, '');
|
||||
if (!canonicalOrigin) return;
|
||||
|
||||
let meta;
|
||||
try { meta = await NodeRegistry.getInstance().fetchMetaForNode(nodeId); }
|
||||
catch { return; }
|
||||
if (!meta?.online) return;
|
||||
if (!meta.capabilities?.includes('mesh_proxy_callback_bootstrap')) return;
|
||||
|
||||
const db = DatabaseService.getInstance();
|
||||
const node = db.getNode(nodeId);
|
||||
if (!node || node.type !== 'remote' || node.mode !== 'proxy' || !node.api_token) return;
|
||||
|
||||
const authSecret = db.getGlobalSettings().auth_jwt_secret;
|
||||
const centralInstanceId = db.getSystemState('instance_id');
|
||||
if (!authSecret || !centralInstanceId) return;
|
||||
|
||||
const built = buildHandshakeFrame(nodeId, node.api_token, canonicalOrigin, centralInstanceId, authSecret);
|
||||
try {
|
||||
ws.send(JSON.stringify(built.frame));
|
||||
void this.logActivity(nodeId, 'open.ok', {
|
||||
bootstrapSent: true,
|
||||
centralApiUrl: redactSensitiveText(canonicalOrigin),
|
||||
kid: 'v1',
|
||||
jwtExpiresAt: built.expSec,
|
||||
});
|
||||
} catch (err) {
|
||||
console.warn(`[MeshProxyDialer] mesh_handshake send failed for node ${nodeId}: ${(err as Error).message}`.replace(/[\n\r]/g, ''));
|
||||
}
|
||||
}
|
||||
|
||||
private async logActivity(
|
||||
nodeId: number,
|
||||
event: ProxyTunnelEvent,
|
||||
@@ -517,58 +467,6 @@ export class MeshProxyTunnelDialer extends EventEmitter {
|
||||
}
|
||||
}
|
||||
|
||||
interface HandshakeFrame {
|
||||
t: 'mesh_handshake';
|
||||
v: 1;
|
||||
peerNodeId: number;
|
||||
centralInstanceId: string;
|
||||
centralApiUrl: string;
|
||||
meshTunnelJwt: string;
|
||||
jwtExpiresAt: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the mesh_handshake frame and the associated JWT. Pure helper so the
|
||||
* dialer's `maybeSendBootstrap` stays under the 30-line ceiling.
|
||||
*/
|
||||
function buildHandshakeFrame(
|
||||
nodeId: number,
|
||||
apiToken: string,
|
||||
canonicalOrigin: string,
|
||||
centralInstanceId: string,
|
||||
authSecret: string,
|
||||
): { frame: HandshakeFrame; expSec: number } {
|
||||
const peerTokenFp = createHash('sha256').update(apiToken).digest('hex').slice(0, 16);
|
||||
const nowSec = Math.floor(Date.now() / 1000);
|
||||
const expSec = nowSec + 365 * 24 * 3600;
|
||||
const meshTunnelJwt = jwt.sign(
|
||||
{
|
||||
sub: String(nodeId),
|
||||
iss: centralInstanceId,
|
||||
aud: canonicalOrigin,
|
||||
scope: 'mesh_tunnel',
|
||||
iat: nowSec,
|
||||
exp: expSec,
|
||||
kid: 'v1',
|
||||
peer_token_fp: peerTokenFp,
|
||||
},
|
||||
authSecret,
|
||||
{ algorithm: 'HS256' },
|
||||
);
|
||||
return {
|
||||
frame: {
|
||||
t: 'mesh_handshake',
|
||||
v: 1,
|
||||
peerNodeId: nodeId,
|
||||
centralInstanceId,
|
||||
centralApiUrl: canonicalOrigin,
|
||||
meshTunnelJwt,
|
||||
jwtExpiresAt: expSec,
|
||||
},
|
||||
expSec,
|
||||
};
|
||||
}
|
||||
|
||||
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;
|
||||
|
||||
+107
-109
@@ -4,7 +4,7 @@ import fs from 'fs/promises';
|
||||
import { EventEmitter } from 'events';
|
||||
import * as YAML from 'yaml';
|
||||
import { ComposeService } from './ComposeService';
|
||||
import { DatabaseService } from './DatabaseService';
|
||||
import { DatabaseService, type NodeMode } from './DatabaseService';
|
||||
import DockerController from './DockerController';
|
||||
import { FileSystemService } from './FileSystemService';
|
||||
import { LicenseService } from './LicenseService';
|
||||
@@ -14,7 +14,6 @@ import { NodeRegistry } from './NodeRegistry';
|
||||
import { PilotTunnelManager } from './PilotTunnelManager';
|
||||
import { MeshProxyTunnelDialer, type DialFailureCode } from './MeshProxyTunnelDialer';
|
||||
import { generateOverrideYaml, MeshAlias, SENCHO_MESH_NETWORK } from './MeshComposeOverride';
|
||||
import { disableCapability, enableCapability } from './CapabilityRegistry';
|
||||
import { lookupContainerIp } from '../mesh/containerLookup';
|
||||
import { sanitizeForLog } from '../utils/safeLog';
|
||||
import { isPathWithinBase, isValidStackName } from '../utils/validation';
|
||||
@@ -97,7 +96,7 @@ export type MeshActivityType =
|
||||
| 'forwarder.listen' | 'forwarder.unlisten' | 'forwarder.error'
|
||||
| 'proxy-tunnel.open.ok' | 'proxy-tunnel.open.fail' | 'proxy-tunnel.close'
|
||||
| 'mesh.proxy_tunnel.identify'
|
||||
| 'mesh_handshake.received';
|
||||
| 'mesh.reconcile.fail';
|
||||
|
||||
export interface MeshActivityEvent {
|
||||
ts: number;
|
||||
@@ -148,13 +147,28 @@ export interface MeshRegenSummary {
|
||||
* - `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.
|
||||
* Central maintains a persistent bidirectional WS to each mesh-enabled
|
||||
* proxy peer, reconciled periodically; the operator sees no badge while
|
||||
* the bridge is up.
|
||||
* - `unreachable`: configuration or runtime problem keeps mesh traffic
|
||||
* from flowing. `reachableReason` carries an actionable hint.
|
||||
*/
|
||||
export type MeshReachableMode = 'local' | 'pilot' | 'proxy' | 'unreachable';
|
||||
|
||||
/**
|
||||
* State of the peer→central reverse path. The forward WS to a proxy-mode
|
||||
* peer is bidirectional; peer→central traffic flows over the same WS via
|
||||
* `tcp_open_reverse`. This discriminator surfaces whether that bridge is
|
||||
* currently usable so the Routing tab can show a transient pill while the
|
||||
* dialer is reconnecting.
|
||||
* - `connected`: forward WS is open; peer can dispatch reverse streams.
|
||||
* - `connecting`: dial in flight; transient.
|
||||
* - `unavailable`: no bridge and no dial in flight (peer just rebooted, or
|
||||
* last dial cached a failure).
|
||||
* - `not_applicable`: not a proxy-mode peer, or mesh disabled on this node.
|
||||
*/
|
||||
export type MeshReverseCallbackStatus = 'connected' | 'connecting' | 'unavailable' | 'not_applicable';
|
||||
|
||||
export interface MeshNodeStatus {
|
||||
nodeId: number;
|
||||
nodeName: string;
|
||||
@@ -174,6 +188,8 @@ export interface MeshNodeStatus {
|
||||
reachableMode: MeshReachableMode;
|
||||
/** Short, operator-facing reason when `reachableMode === 'unreachable'`. Null otherwise. */
|
||||
reachableReason: string | null;
|
||||
/** Peer→central reverse path state. `not_applicable` for non-proxy peers. */
|
||||
reverseCallbackStatus: MeshReverseCallbackStatus;
|
||||
optedInStacks: string[];
|
||||
activeStreamCount: number;
|
||||
}
|
||||
@@ -245,6 +261,7 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
private activity: MeshActivityEvent[] = [];
|
||||
private activeStreams = new Map<number, ActiveStreamRecord>();
|
||||
private aliasRefreshTimer?: NodeJS.Timeout;
|
||||
private bridgeReconcileTimer?: NodeJS.Timeout;
|
||||
private routeErrorMap = new Map<string, { ts: number; message: string }>();
|
||||
private routeLatencyMap = new Map<string, number>();
|
||||
private activityListeners = new Set<(e: MeshActivityEvent) => void>();
|
||||
@@ -335,8 +352,6 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
|
||||
this.selfCentralNodeId = this.resolveSelfCentralNodeId();
|
||||
|
||||
this.maybeWarnUnsetPrimaryUrl();
|
||||
|
||||
await this.setupMeshNetwork();
|
||||
try {
|
||||
await this.refreshAliasCache();
|
||||
@@ -355,10 +370,12 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
});
|
||||
}
|
||||
await this.regenerateAllOverrides();
|
||||
// Trigger 3: proactively re-establish central->peer bridges so any
|
||||
// peer that is online and capable receives bootstrap material on
|
||||
// startup. Fire-and-forget so start() does not block on remote I/O.
|
||||
void this.proactiveBootstrapFanout();
|
||||
// Proactively dial every mesh-enabled proxy peer so the forward WS
|
||||
// (which also carries peer→central reverse traffic) is up before any
|
||||
// user request hits it. Fire-and-forget so start() does not block on
|
||||
// remote I/O.
|
||||
void this.proactiveBridgeFanout();
|
||||
this.startBridgeReconcileLoop();
|
||||
this.aliasRefreshTimer = setInterval(() => {
|
||||
void (async () => {
|
||||
try {
|
||||
@@ -388,49 +405,25 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
clearInterval(this.aliasRefreshTimer);
|
||||
this.aliasRefreshTimer = undefined;
|
||||
}
|
||||
this.stopBridgeReconcileLoop();
|
||||
await this.forwarder.shutdown();
|
||||
}
|
||||
|
||||
/**
|
||||
* Operator-facing preflight: if `SENCHO_PRIMARY_URL` is unset at
|
||||
* startup and the central has at least one mesh-enabled proxy-mode
|
||||
* node, warn that the Task 8 capability-gated handshake will fall
|
||||
* back to an inferred origin for the callback bootstrap. The matching
|
||||
* fail-safe in `MeshProxyTunnelDialer` silently skips the bootstrap
|
||||
* when the env is unset; this warn makes the consequence visible.
|
||||
* Walk every mesh-enabled proxy-mode peer and call
|
||||
* `MeshProxyTunnelDialer.ensureBridge(nodeId)` on each. The bridge is
|
||||
* the persistent bidirectional control-plane channel: central→peer
|
||||
* `tcp_open` and peer→central `tcp_open_reverse` both flow over the
|
||||
* same WS. Bounded concurrency 4 with a 250 ms stagger; failures are
|
||||
* logged and never abort the fan-out. Called both at startup and on
|
||||
* every reconcile tick. `ensureBridge` short-circuits on already-open
|
||||
* bridges, so steady-state cost is one Map lookup per peer.
|
||||
*/
|
||||
private maybeWarnUnsetPrimaryUrl(): void {
|
||||
if (process.env.SENCHO_PRIMARY_URL) return;
|
||||
const row = DatabaseService.getInstance().getDb().prepare(`
|
||||
SELECT COUNT(*) as n FROM nodes
|
||||
WHERE type='remote' AND mode='proxy' AND mesh_enabled=1
|
||||
`).get() as { n: number };
|
||||
if (row.n > 0) {
|
||||
console.warn(
|
||||
'[Mesh] SENCHO_PRIMARY_URL is unset; mesh callback bootstrap will use ' +
|
||||
'inferred origin. Set SENCHO_PRIMARY_URL to make peer-initiated mesh ' +
|
||||
'callbacks robust against reverse-proxy edge cases.'
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Trigger 3: at central startup, walk every mesh-enabled proxy-mode
|
||||
* peer that already has at least one `mesh_stacks` row and call
|
||||
* `MeshProxyTunnelDialer.ensureBridge(nodeId)`. The dialer's
|
||||
* capability-gated handshake then mints and ships bootstrap material
|
||||
* to peers that just came online or just got upgraded to a build that
|
||||
* advertises `mesh_proxy_callback_bootstrap`. Bounded concurrency 4
|
||||
* with a 250ms stagger; failures are logged and never abort the
|
||||
* fan-out.
|
||||
*/
|
||||
private async proactiveBootstrapFanout(): Promise<void> {
|
||||
private async proactiveBridgeFanout(): Promise<void> {
|
||||
const rows = DatabaseService.getInstance().getDb().prepare(`
|
||||
SELECT DISTINCT n.id AS id
|
||||
FROM nodes n
|
||||
INNER JOIN mesh_stacks ms ON ms.node_id = n.id
|
||||
WHERE n.type = 'remote' AND n.mode = 'proxy' AND n.mesh_enabled = 1
|
||||
ORDER BY n.id
|
||||
SELECT id FROM nodes
|
||||
WHERE type = 'remote' AND mode = 'proxy' AND mesh_enabled = 1
|
||||
ORDER BY id
|
||||
`).all() as Array<{ id: number }>;
|
||||
|
||||
const queue = rows.map((r) => r.id);
|
||||
@@ -445,8 +438,8 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
this.logActivity({
|
||||
source: 'mesh', level: 'warn', type: 'proxy-tunnel.open.fail',
|
||||
nodeId,
|
||||
message: `boot proactive bootstrap failed: ${sanitizeForLog((err as Error).message)}`,
|
||||
details: { trigger: 'startup_fanout' },
|
||||
message: `proxy-tunnel reconcile dial failed: ${sanitizeForLog((err as Error).message)}`,
|
||||
details: { trigger: 'reconcile' },
|
||||
});
|
||||
}
|
||||
await new Promise((r) => setTimeout(r, 250));
|
||||
@@ -456,6 +449,35 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
await Promise.all(Array.from({ length: workerCount }, () => worker()));
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule the proxy-tunnel reconcile tick. Interval is overridable via
|
||||
* `SENCHO_MESH_RECONCILE_INTERVAL_MS` (default 60_000 ms) so an operator
|
||||
* can tune the peer-reboot detection window. Idempotent: a second call
|
||||
* is a no-op while the timer is live.
|
||||
*/
|
||||
private startBridgeReconcileLoop(): void {
|
||||
if (this.bridgeReconcileTimer) return;
|
||||
const raw = process.env.SENCHO_MESH_RECONCILE_INTERVAL_MS;
|
||||
const parsed = raw === undefined ? Number.NaN : Number(raw);
|
||||
const intervalMs = Number.isFinite(parsed) && parsed >= 1000 ? parsed : 60_000;
|
||||
this.bridgeReconcileTimer = setInterval(() => {
|
||||
void this.proactiveBridgeFanout().catch((err) => {
|
||||
this.logActivity({
|
||||
source: 'mesh', level: 'error', type: 'mesh.reconcile.fail',
|
||||
message: `bridge reconcile threw: ${sanitizeForLog((err as Error).message)}`,
|
||||
});
|
||||
});
|
||||
}, intervalMs);
|
||||
this.bridgeReconcileTimer.unref?.();
|
||||
}
|
||||
|
||||
private stopBridgeReconcileLoop(): void {
|
||||
if (this.bridgeReconcileTimer) {
|
||||
clearInterval(this.bridgeReconcileTimer);
|
||||
this.bridgeReconcileTimer = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
public getSenchoIp(): string | null {
|
||||
return this.senchoIp;
|
||||
}
|
||||
@@ -481,7 +503,6 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
/**
|
||||
* Single recording path for every mesh-setup failure. Keeps the legacy
|
||||
* `networkSetupError` string in sync, sets the typed `dataPlaneStatus`,
|
||||
* strips `mesh_proxy_callback_bootstrap` from advertised capabilities,
|
||||
* and emits a `mesh.disable` activity entry. Callers pass `level: 'warn'`
|
||||
* for expected conditions (`not_in_docker` in dev mode) and `'error'` for
|
||||
* real failures.
|
||||
@@ -500,7 +521,6 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
this.networkSetupError = message;
|
||||
this.dataPlaneStatus = { ok: false, reason, message, subnet };
|
||||
this.senchoIp = null;
|
||||
disableCapability('mesh_proxy_callback_bootstrap');
|
||||
this.logActivity({
|
||||
source: 'mesh',
|
||||
level,
|
||||
@@ -580,7 +600,6 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
|
||||
this.networkSetupError = null;
|
||||
this.dataPlaneStatus = { ok: true, reason: 'ok', message: null, subnet };
|
||||
enableCapability('mesh_proxy_callback_bootstrap');
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -856,15 +875,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
source: 'mesh', level: 'info', type: 'mesh.enable',
|
||||
nodeId, message: `mesh enabled on node ${nodeId}`,
|
||||
});
|
||||
// Trigger 1: proactive bootstrap. If the peer is a proxy-mode remote,
|
||||
// dial the mesh callback bridge immediately so the capability-gated
|
||||
// handshake can ship `mesh_handshake` material without waiting for
|
||||
// the next forwarder dial. Fire-and-forget; failures are logged by
|
||||
// the dialer.
|
||||
// When mesh is enabled on a proxy peer, dial the persistent bridge
|
||||
// immediately so the next forward (or peer-initiated reverse)
|
||||
// request has the WS already up. Fire-and-forget; failures are
|
||||
// logged by the dialer.
|
||||
const node = DatabaseService.getInstance().getNode(nodeId);
|
||||
if (node && node.type === 'remote' && node.mode === 'proxy') {
|
||||
void MeshProxyTunnelDialer.getInstance().ensureBridge(nodeId).catch((err) => {
|
||||
console.warn(`[Mesh] proactive bootstrap on mesh-enable failed for node ${nodeId}: ${(err as Error).message}`);
|
||||
console.warn(`[Mesh] proxy-tunnel dial on mesh-enable failed for node ${nodeId}: ${(err as Error).message}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -1812,59 +1830,23 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
message: `cross-node dispatch to ${target.alias} on node ${target.nodeId}`,
|
||||
});
|
||||
|
||||
// Peer-side recovery: when this Sencho is acting as a proxy-mode peer
|
||||
// (mesh_centrals has a row from a prior bootstrap) and the
|
||||
// reverseDialer is not currently installed, the inbound bridge from
|
||||
// central is either cold-start (peer just rebooted) or has been torn
|
||||
// down (idle close, central restart). Kick the symmetric-dial path
|
||||
// so the peer re-opens its callback WS to central before attempting
|
||||
// the cross-node dispatch.
|
||||
//
|
||||
// Central instances never have a mesh_centrals row (central is not a
|
||||
// peer of itself), so this branch is correctly skipped on central.
|
||||
// Central falls straight through to dialMeshTcpStream which uses its
|
||||
// own PilotTunnelManager + MeshProxyTunnelDialer to reach the target
|
||||
// peer. Without this gate, central enters the branch on every
|
||||
// forward dispatch, finds no session, and destroys the inbound
|
||||
// socket with route.resolve.fail forward-from-peer no_session.
|
||||
if (!this.reverseDialer) {
|
||||
const { MeshCentralRegistry } = await import('./MeshCentralRegistry');
|
||||
const isProxyPeer = MeshCentralRegistry.getInstance().getActive() !== null;
|
||||
if (isProxyPeer) {
|
||||
try {
|
||||
const { PeerToCentralMeshSessionDialer } = await import('./PeerToCentralMeshSessionDialer');
|
||||
const session = await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
|
||||
if (!session) {
|
||||
this.logActivity({
|
||||
source: 'mesh', level: 'warn', type: 'route.resolve.fail',
|
||||
nodeId: target.nodeId, alias: target.alias,
|
||||
message: `peer cross-node dispatch failed: no central callback session available`,
|
||||
details: { direction: 'forward-from-peer', reason: 'no_session' },
|
||||
});
|
||||
try { src.destroy(); } catch { /* ignore */ }
|
||||
return;
|
||||
}
|
||||
} catch (err) {
|
||||
this.logActivity({
|
||||
source: 'mesh', level: 'warn', type: 'route.resolve.fail',
|
||||
nodeId: target.nodeId, alias: target.alias,
|
||||
message: `peer cross-node dispatch failed: bootstrap threw`,
|
||||
details: { direction: 'forward-from-peer', reason: 'bootstrap_threw' },
|
||||
});
|
||||
try { src.destroy(); } catch { /* ignore */ }
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const tcpStream = await this.dialMeshTcpStream(target);
|
||||
if (!tcpStream) {
|
||||
// Three failure shapes on this path:
|
||||
// - central with reverseDialer somehow installed (unusual; relay edge case)
|
||||
// - central without reverseDialer (normal): ensureBridge could not reach the target peer
|
||||
// - peer side: target nodeId is central's id, which the peer's NodeRegistry does
|
||||
// not know as a remote proxy target, so ensureBridge returns null. The actionable
|
||||
// condition is "central has not dialed the bridge yet"; tell the operator.
|
||||
const targetIsLocalKnownRemote = NodeRegistry.getInstance().getNode(target.nodeId)?.type === 'remote';
|
||||
const message = this.reverseDialer
|
||||
? `cannot open reverse mesh stream to node ${target.nodeId}`
|
||||
: targetIsLocalKnownRemote
|
||||
? `no mesh tunnel reachable for node ${target.nodeId}`
|
||||
: `peer cross-node dispatch deferred: waiting for central to dial the reverse bridge`;
|
||||
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 mesh tunnel reachable for node ${target.nodeId}`,
|
||||
nodeId: target.nodeId, alias: target.alias, message,
|
||||
});
|
||||
try { src.destroy(); } catch { /* ignore */ }
|
||||
return;
|
||||
@@ -2148,12 +2130,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
const localNodeId = NodeRegistry.getInstance().getDefaultNodeId();
|
||||
const localListening = this.isLocalForwarderActive();
|
||||
const ptm = PilotTunnelManager.getInstance();
|
||||
const dialer = MeshProxyTunnelDialer.getInstance();
|
||||
return nodes.map((node) => {
|
||||
const reach = this.computeReachable(node, localNodeId);
|
||||
const meshEnabled = db.getNodeMeshEnabled(node.id);
|
||||
return {
|
||||
nodeId: node.id,
|
||||
nodeName: node.name,
|
||||
enabled: db.getNodeMeshEnabled(node.id),
|
||||
enabled: meshEnabled,
|
||||
localForwarderListening: node.id === localNodeId ? localListening : null,
|
||||
// `pilotConnected` stays at its original meaning: a pilot
|
||||
// tunnel is currently registered for this node. The
|
||||
@@ -2163,12 +2147,26 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
|
||||
pilotConnected: node.type !== 'remote' || ptm.hasActiveTunnel(node.id),
|
||||
reachableMode: reach.mode,
|
||||
reachableReason: reach.reason,
|
||||
reverseCallbackStatus: this.computeReverseCallbackStatus(node, meshEnabled, dialer),
|
||||
optedInStacks: db.listMeshStacks(node.id).map((s) => s.stack_name),
|
||||
activeStreamCount: this.activeStreams.size,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
private computeReverseCallbackStatus(
|
||||
node: { id: number; type: 'local' | 'remote'; mode: NodeMode },
|
||||
meshEnabled: boolean,
|
||||
dialer: MeshProxyTunnelDialer,
|
||||
): MeshReverseCallbackStatus {
|
||||
if (node.type !== 'remote' || node.mode !== 'proxy' || !meshEnabled) {
|
||||
return 'not_applicable';
|
||||
}
|
||||
if (dialer.hasBridge(node.id)) return 'connected';
|
||||
if (dialer.isDialing(node.id)) return 'connecting';
|
||||
return 'unavailable';
|
||||
}
|
||||
|
||||
/** True when the local Sencho's forwarder is started and bound to at least one alias port. */
|
||||
private isLocalForwarderActive(): boolean {
|
||||
return this.started && this.forwarder.getListenerPorts().length > 0;
|
||||
|
||||
@@ -1,276 +0,0 @@
|
||||
/**
|
||||
* PeerToCentralMeshSessionDialer: peer-side counterpart to the central
|
||||
* ingress at `/api/mesh/proxy-tunnel-from-peer`. Reads cached central
|
||||
* material from `MeshCentralRegistry`, opens a WebSocket to central
|
||||
* carrying the bootstrapped JWT in the Authorization header, and on a
|
||||
* successful upgrade installs a local `PilotTunnelBridge` whose loopback
|
||||
* URL the local proxy can target as if central had dialed us.
|
||||
*
|
||||
* Failure handling branches on the central's machine-readable reason
|
||||
* code returned in the 401 body: terminal codes (stale,
|
||||
* signature_invalid, fingerprint, node_deleted, instance_mismatch,
|
||||
* audience_mismatch) clear the cache so we stop dialing with a known-bad
|
||||
* credential. Transient codes (clock_skew, mode_mismatch) keep the cache
|
||||
* but record the rejection so an operator can see why we stalled.
|
||||
* HTTP 404 (older central without the peer ingress endpoint) keeps the
|
||||
* cache and arms a longer backoff window so we do not hammer the remote.
|
||||
*
|
||||
* Process-local rate limit: at most 5 dial attempts per 60s window.
|
||||
* Concurrency: a single in-flight promise is shared across callers so
|
||||
* parallel `ensureSession()` calls coalesce into one dial.
|
||||
*/
|
||||
import { EventEmitter } from 'events';
|
||||
import WebSocket from 'ws';
|
||||
import {
|
||||
MAX_FRAME_SIZE_BYTES,
|
||||
decodeBinaryFrame,
|
||||
decodeJsonFrame,
|
||||
wsDataToBuffer,
|
||||
wsDataToString,
|
||||
} from '../pilot/protocol';
|
||||
import { MeshCentralRegistry } from './MeshCentralRegistry';
|
||||
import {
|
||||
attachTcpStreamSwitchboard,
|
||||
resolveByComposeLabels,
|
||||
type TcpStreamSwitchboard,
|
||||
type ReverseTcpStreamHandle,
|
||||
} from '../mesh/tcpStreamSwitchboard';
|
||||
import { PilotMetrics } from './PilotMetrics';
|
||||
import { httpUrlToWs } from '../utils/wsUrl';
|
||||
import { sanitizeForLog } from '../utils/safeLog';
|
||||
|
||||
interface SwitchboardReverseDialer {
|
||||
openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null;
|
||||
}
|
||||
|
||||
const HANDSHAKE_TIMEOUT_MS = 15_000;
|
||||
const RATE_LIMIT_WINDOW_MS = 60_000;
|
||||
const RATE_LIMIT_MAX = 5;
|
||||
const ENDPOINT_UNAVAILABLE_BACKOFF_MS = 5 * 60_000;
|
||||
const CALLBACK_PATH = '/api/mesh/proxy-tunnel-from-peer';
|
||||
|
||||
/**
|
||||
* Reasons that indicate the cached central material is permanently bad
|
||||
* (wrong secret, wrong fingerprint, node removed on central, etc.) and
|
||||
* should be evicted so the next mesh-enable cycle remints fresh
|
||||
* material. `endpoint_not_found` is explicitly NOT in this set because
|
||||
* a 404 typically means "central is older than the peer ingress",
|
||||
* which is recoverable by waiting.
|
||||
*/
|
||||
const REJECT_REASONS_CLEAR = new Set([
|
||||
'stale',
|
||||
'token_fingerprint_mismatch',
|
||||
'node_deleted',
|
||||
'instance_mismatch',
|
||||
'signature_invalid',
|
||||
'audience_mismatch',
|
||||
'unknown',
|
||||
]);
|
||||
|
||||
interface DialError extends Error {
|
||||
reason?: string;
|
||||
httpStatus?: number;
|
||||
}
|
||||
|
||||
export class PeerToCentralMeshSessionDialer extends EventEmitter {
|
||||
private static instance: PeerToCentralMeshSessionDialer | null = null;
|
||||
private currentSession: TcpStreamSwitchboard | null = null;
|
||||
private currentWs: WebSocket | null = null;
|
||||
private inflight: Promise<TcpStreamSwitchboard | null> | null = null;
|
||||
private recentDials: number[] = [];
|
||||
private endpointUnavailableUntil = 0;
|
||||
|
||||
private constructor() { super(); }
|
||||
|
||||
public static getInstance(): PeerToCentralMeshSessionDialer {
|
||||
if (!this.instance) this.instance = new PeerToCentralMeshSessionDialer();
|
||||
return this.instance;
|
||||
}
|
||||
|
||||
public static resetForTest(): void {
|
||||
if (this.instance) {
|
||||
try { this.instance.currentSession?.cleanup('test reset'); } catch { /* ignore */ }
|
||||
try { this.instance.currentWs?.close(1000, 'test reset'); } catch { /* ignore */ }
|
||||
}
|
||||
this.instance = null;
|
||||
}
|
||||
|
||||
public hasSession(): boolean {
|
||||
return this.currentSession !== null;
|
||||
}
|
||||
|
||||
public async ensureSession(): Promise<TcpStreamSwitchboard | null> {
|
||||
if (this.currentSession) return this.currentSession;
|
||||
if (Date.now() < this.endpointUnavailableUntil) return null;
|
||||
if (this.isRateLimited()) return null;
|
||||
if (this.inflight) return this.inflight;
|
||||
this.inflight = this.dial().finally(() => { this.inflight = null; });
|
||||
return this.inflight;
|
||||
}
|
||||
|
||||
private isRateLimited(): boolean {
|
||||
const now = Date.now();
|
||||
this.recentDials = this.recentDials.filter((ts) => now - ts < RATE_LIMIT_WINDOW_MS);
|
||||
return this.recentDials.length >= RATE_LIMIT_MAX;
|
||||
}
|
||||
|
||||
private async dial(): Promise<TcpStreamSwitchboard | null> {
|
||||
const material = MeshCentralRegistry.getInstance().getActive();
|
||||
if (!material) return null;
|
||||
this.recentDials.push(Date.now());
|
||||
const wsUrl = httpUrlToWs(material.centralApiUrl) + CALLBACK_PATH;
|
||||
const ws = new WebSocket(wsUrl, {
|
||||
headers: { Authorization: `Bearer ${material.callbackJwt}` },
|
||||
handshakeTimeout: HANDSHAKE_TIMEOUT_MS,
|
||||
maxPayload: MAX_FRAME_SIZE_BYTES,
|
||||
});
|
||||
try {
|
||||
await this.awaitOpen(ws);
|
||||
} catch (err) {
|
||||
ws.on('error', () => { /* swallow tail */ });
|
||||
try { ws.close(); } catch { /* ignore */ }
|
||||
this.handleDialFailure(err, material.centralInstanceId);
|
||||
return null;
|
||||
}
|
||||
return this.attachSwitchboard(ws, material.centralInstanceId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Wire the peer-initiated callback WS into the local MeshService. The
|
||||
* R1-A2 design puts the peer end of the bridge in TcpStreamSwitchboard
|
||||
* mode (peer multiplexes streams; central side runs PilotTunnelBridge).
|
||||
* Without this wiring the WS opens cleanly but MeshService.reverseDialer
|
||||
* stays null, so MeshService.dialMeshTcpStream falls through to
|
||||
* PilotTunnelManager.ensureBridge(centralNodeId), which has no record
|
||||
* for central on a proxy-mode peer (peers do not enroll central) and
|
||||
* fails with proxy-tunnel.open.fail reason=no_target. End state matches
|
||||
* the v0.78.1 reverse-direction failure even though the callback bridge
|
||||
* is alive.
|
||||
*
|
||||
* Wiring symmetric to the central-initiated handler at
|
||||
* `meshProxyTunnel.ts:115-163`:
|
||||
* - attachTcpStreamSwitchboard with the same compose-label resolver
|
||||
* - SwitchboardReverseDialer that delegates to switchboard.openReverseStream
|
||||
* - setReverseDialer(localDialer, null) with CAS so a concurrent
|
||||
* central-initiated tunnel does not get silently overwritten
|
||||
* - ws.on('message') dispatches JSON/binary frames to the switchboard
|
||||
* - ws.on('close'/'error') tears down switchboard + clears reverseDialer
|
||||
*/
|
||||
private async attachSwitchboard(ws: WebSocket, instanceId: string): Promise<TcpStreamSwitchboard | null> {
|
||||
let switchboard: TcpStreamSwitchboard;
|
||||
try {
|
||||
switchboard = attachTcpStreamSwitchboard({
|
||||
ws,
|
||||
resolveTarget: resolveByComposeLabels,
|
||||
logLabel: 'MeshCallback',
|
||||
});
|
||||
} catch (err) {
|
||||
try { ws.close(1011, 'switchboard attach failed'); } catch { /* ignore */ }
|
||||
PilotMetrics.increment('mesh_callback_dials_failed_total');
|
||||
console.warn(`[PeerToCentralMeshSessionDialer] attach failed: ${sanitizeForLog((err as Error).message)}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
const { MeshService } = await import('./MeshService');
|
||||
const meshService = MeshService.getInstance();
|
||||
const localDialer: SwitchboardReverseDialer = {
|
||||
openMeshTcpStream(target) {
|
||||
return switchboard.openReverseStream(target);
|
||||
},
|
||||
};
|
||||
const installed = meshService.setReverseDialer(localDialer, null);
|
||||
if (!installed) {
|
||||
console.warn('[PeerToCentralMeshSessionDialer] reverse dialer already installed; rejecting concurrent callback bridge');
|
||||
switchboard.cleanup('reverse dialer already installed');
|
||||
try { ws.close(1013, 'reverse dialer already installed'); } catch { /* ignore */ }
|
||||
PilotMetrics.increment('mesh_callback_dials_failed_total');
|
||||
return null;
|
||||
}
|
||||
|
||||
const onMessage = (data: unknown, isBinary: boolean): void => {
|
||||
try {
|
||||
if (isBinary) {
|
||||
const buf = wsDataToBuffer(data);
|
||||
if (!buf) return;
|
||||
switchboard.handleBinaryFrame(decodeBinaryFrame(buf));
|
||||
return;
|
||||
}
|
||||
const text = wsDataToString(data);
|
||||
if (text == null) return;
|
||||
switchboard.handleJsonFrame(decodeJsonFrame(text));
|
||||
} catch (err) {
|
||||
console.warn(`[PeerToCentralMeshSessionDialer] malformed frame: ${sanitizeForLog((err as Error).message)}`);
|
||||
}
|
||||
};
|
||||
|
||||
let tornDown = false;
|
||||
const teardown = (): void => {
|
||||
if (tornDown) return;
|
||||
tornDown = true;
|
||||
ws.off('message', onMessage);
|
||||
try { switchboard.cleanup('mesh callback bridge closed'); } catch { /* ignore */ }
|
||||
meshService.setReverseDialer(null, localDialer);
|
||||
if (this.currentSession === switchboard) this.currentSession = null;
|
||||
if (this.currentWs === ws) this.currentWs = null;
|
||||
};
|
||||
|
||||
ws.on('message', onMessage);
|
||||
ws.once('close', teardown);
|
||||
ws.once('error', (err) => {
|
||||
console.warn(`[PeerToCentralMeshSessionDialer] ws error: ${sanitizeForLog(err.message)}`);
|
||||
teardown();
|
||||
});
|
||||
|
||||
this.currentSession = switchboard;
|
||||
this.currentWs = ws;
|
||||
PilotMetrics.increment('mesh_central_bootstraps_total');
|
||||
MeshCentralRegistry.getInstance().markUsed(instanceId);
|
||||
return switchboard;
|
||||
}
|
||||
|
||||
private awaitOpen(ws: WebSocket): Promise<void> {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const cleanup = (): void => {
|
||||
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();
|
||||
let body = '';
|
||||
res.on('data', (chunk: Buffer) => { body += chunk.toString('utf8'); });
|
||||
res.on('end', () => reject(this.buildUpgradeError(res.statusCode, body)));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private buildUpgradeError(statusCode: number | undefined, body: string): DialError {
|
||||
let reason = 'unknown';
|
||||
try {
|
||||
const parsed = JSON.parse(body) as { reason?: unknown };
|
||||
if (typeof parsed.reason === 'string') reason = parsed.reason;
|
||||
} catch { /* keep unknown */ }
|
||||
if (statusCode === 404) reason = 'endpoint_not_found';
|
||||
const err = new Error(`upgrade rejected: HTTP ${statusCode ?? 'unknown'} reason=${reason}`) as DialError;
|
||||
err.reason = reason;
|
||||
err.httpStatus = statusCode;
|
||||
return err;
|
||||
}
|
||||
|
||||
private handleDialFailure(err: unknown, instanceId: string): void {
|
||||
const reason = (err as DialError).reason ?? 'unknown';
|
||||
PilotMetrics.increment('mesh_callback_dials_failed_total');
|
||||
if (REJECT_REASONS_CLEAR.has(reason)) {
|
||||
MeshCentralRegistry.getInstance().clearForInstance(instanceId);
|
||||
PilotMetrics.increment('mesh_callback_auth_failures_total');
|
||||
} else {
|
||||
MeshCentralRegistry.getInstance().markRejected(instanceId, reason);
|
||||
if (reason === 'endpoint_not_found') {
|
||||
this.endpointUnavailableUntil = Date.now() + ENDPOINT_UNAVAILABLE_BACKOFF_MS;
|
||||
}
|
||||
}
|
||||
console.warn(`[PeerToCentralMeshSessionDialer] dial failed: ${sanitizeForLog(`reason=${reason}`)}`);
|
||||
}
|
||||
}
|
||||
@@ -28,16 +28,8 @@ export interface Counters {
|
||||
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-tunnel teardowns initiated by the dialer's idle sweep (zero active streams for the configured TTL). Effectively 0 by default since the dialer no longer idle-closes the bridge. */
|
||||
proxy_idle_closes: number;
|
||||
/** Proxy bridges registered via the peer-initiated dial-back path (`/api/mesh/proxy-tunnel-from-peer`). Disjoint from `proxy_bridges_total`, which counts central-initiated dials. */
|
||||
proxy_bridges_peer_initiated_total: number;
|
||||
/** Successful peer-to-central dial-back sessions (counted on WS open + bridge start). Disjoint from `proxy_bridges_peer_initiated_total`, which is incremented by the central-side ingress at register time. Useful for operators investigating asymmetric counts (peer thinks it dialed, central never registered). */
|
||||
mesh_central_bootstraps_total: number;
|
||||
/** Peer-to-central dial-back attempts that did not complete a WS open. Covers transport errors, upgrade rejections, and bridge.start failures. */
|
||||
mesh_callback_dials_failed_total: number;
|
||||
/** Subset of `mesh_callback_dials_failed_total` where central responded 401 with a terminal reason code (stale, signature_invalid, ...) that caused the peer to clear its cached central material. */
|
||||
mesh_callback_auth_failures_total: number;
|
||||
}
|
||||
|
||||
const ZERO_COUNTERS: Counters = {
|
||||
@@ -49,10 +41,6 @@ const ZERO_COUNTERS: Counters = {
|
||||
proxy_bridges_total: 0,
|
||||
proxy_dials_failed: 0,
|
||||
proxy_idle_closes: 0,
|
||||
proxy_bridges_peer_initiated_total: 0,
|
||||
mesh_central_bootstraps_total: 0,
|
||||
mesh_callback_dials_failed_total: 0,
|
||||
mesh_callback_auth_failures_total: 0,
|
||||
};
|
||||
|
||||
export const PILOT_METRICS_FLUSH_INTERVAL_MS = 1_000;
|
||||
|
||||
@@ -36,11 +36,10 @@ export class PilotTunnelCapacityError extends Error {
|
||||
/**
|
||||
* Discriminator for the two flavors of bridge that share the `bridges`
|
||||
* map: `'pilot'` is set by `registerTunnel` (agent-initiated long-lived
|
||||
* tunnel); `'proxy'` is set by `registerProxyBridge` /
|
||||
* `replaceOrRegisterProxyBridge` (central- or peer-initiated short-lived
|
||||
* proxy bridge). Used by the rejection-message formatter and by
|
||||
* `replaceOrRegisterProxyBridge` so a peer-initiated dial can supersede a
|
||||
* previous proxy bridge but never shadow a live pilot tunnel.
|
||||
* tunnel); `'proxy'` is set by `registerProxyBridge` (central-initiated
|
||||
* persistent bridge dialed by `MeshProxyTunnelDialer`). Used by the
|
||||
* rejection-message formatter so a pilot tunnel and a proxy bridge for the
|
||||
* same nodeId cannot coexist silently.
|
||||
*/
|
||||
export type BridgeKind = 'pilot' | 'proxy';
|
||||
|
||||
@@ -282,46 +281,12 @@ export class PilotTunnelManager extends EventEmitter {
|
||||
this.emit('proxy-bridge-up', nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a peer-initiated proxy bridge for an existing remote. If a proxy
|
||||
* bridge already exists for this nodeId, close it (the new dial is the source
|
||||
* of truth) and replace. If a pilot tunnel exists, refuse: pilot tunnels
|
||||
* always win over peer-initiated proxy bridges.
|
||||
*/
|
||||
public replaceOrRegisterProxyBridge(nodeId: number, bridge: PilotTunnelBridge): void {
|
||||
const existingKind = this.bridgeKinds.get(nodeId);
|
||||
if (existingKind === 'pilot') {
|
||||
throw new Error(this.formatBridgeConflict(nodeId, existingKind));
|
||||
}
|
||||
if (existingKind === 'proxy') {
|
||||
const old = this.bridges.get(nodeId);
|
||||
this.bridges.delete(nodeId);
|
||||
this.bridgeKinds.delete(nodeId);
|
||||
try { old?.close(1000, 'replaced-by-newer-proxy'); } catch { /* best-effort cleanup */ }
|
||||
}
|
||||
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.bridgeKinds.delete(nodeId);
|
||||
this.emit('proxy-bridge-down', nodeId, 'remote_closed');
|
||||
}
|
||||
});
|
||||
this.bridges.set(nodeId, bridge);
|
||||
this.bridgeKinds.set(nodeId, 'proxy');
|
||||
}
|
||||
|
||||
/**
|
||||
* Single source of truth for the "slot already held" rejection
|
||||
* message thrown by `registerProxyBridge` and
|
||||
* `replaceOrRegisterProxyBridge`. Picks the bridge-kind subject and
|
||||
* the rejection tail so a stale call site cannot drift from the
|
||||
* other. `undefined` collapses to the pilot branch defensively;
|
||||
* production code always sets `bridgeKinds` whenever `bridges` is
|
||||
* set, so this is unreachable in practice.
|
||||
* message thrown by `registerProxyBridge`. Picks the bridge-kind
|
||||
* subject and the rejection tail. `undefined` collapses to the pilot
|
||||
* branch defensively; production code always sets `bridgeKinds`
|
||||
* whenever `bridges` is set, so this is unreachable in practice.
|
||||
*/
|
||||
private formatBridgeConflict(nodeId: number, existingKind: BridgeKind | undefined): string {
|
||||
return existingKind === 'proxy'
|
||||
|
||||
Reference in New Issue
Block a user