mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-07-26 11:49:16 +00:00
feat(pilot): add tcp tunnel frames + mesh sidecar package (#857)
Lays the dormant data-plane foundation for Sencho Mesh. The pilot tunnel gains TCP forwarding frames (tcp_open / tcp_open_ack / tcp_close JSON plus a 0x04 TcpData binary type) and a TcpStream surface on the bridge so a future MeshService can ride the existing WSS tunnel for cross-node container traffic. The agent rejects every tcp_open with mesh_not_enabled until a follow-up PR wires the Dockerode resolver gated by a mesh_stacks opt-in table; ships dormant. A new top-level mesh-sidecar/ package provides the per-node container that will host the L4 forwarder + control WS in production. Built as a small Node 22 alpine image and published in lockstep with the main sencho image via a parallel docker-publish workflow job. Tests cover protocol roundtrips on both packages and the sidecar forwarder end-to-end including resolve, splice, close, and stats.
This commit is contained in:
Anso
@@ -232,3 +232,75 @@ jobs:
|
||||
sbom.cdx.json
|
||||
sbom.spdx.json
|
||||
security/vex/sencho.openvex.json
|
||||
|
||||
push_mesh_sidecar:
|
||||
name: Push Sencho Mesh sidecar image to Docker Hub
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 20
|
||||
environment: production
|
||||
permissions:
|
||||
contents: read
|
||||
id-token: write
|
||||
steps:
|
||||
- name: Check out the repo
|
||||
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
|
||||
|
||||
- name: Set up QEMU
|
||||
uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4
|
||||
with:
|
||||
platforms: arm64
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4
|
||||
|
||||
- name: Log in to Docker Hub
|
||||
uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v4
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_TOKEN }}
|
||||
|
||||
- name: Install cosign
|
||||
uses: sigstore/cosign-installer@cad07c2e89fa2edd6e2d7bab4c1aa38e53f76003 # v4.1.1
|
||||
|
||||
- name: Extract metadata for Mesh sidecar image
|
||||
id: mesh_meta
|
||||
uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6
|
||||
with:
|
||||
# Sencho Mesh sidecar image. Versioned in lockstep with the main
|
||||
# sencho image so each Sencho release has a matched mesh sidecar.
|
||||
images: saelix/sencho-mesh
|
||||
tags: |
|
||||
type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/v') }}
|
||||
type=semver,pattern={{version}},enable=${{ startsWith(github.ref, 'refs/tags/v') }}
|
||||
type=semver,pattern={{major}}.{{minor}},enable=${{ startsWith(github.ref, 'refs/tags/v') }}
|
||||
|
||||
- name: Build and push Mesh sidecar image
|
||||
id: mesh_build
|
||||
uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7
|
||||
with:
|
||||
context: ./mesh-sidecar
|
||||
file: ./mesh-sidecar/Dockerfile
|
||||
push: true
|
||||
platforms: linux/amd64,linux/arm64
|
||||
tags: ${{ steps.mesh_meta.outputs.tags }}
|
||||
labels: ${{ steps.mesh_meta.outputs.labels }}
|
||||
cache-from: type=registry,ref=saelix/sencho-mesh:buildcache
|
||||
cache-to: type=registry,ref=saelix/sencho-mesh:buildcache,mode=max
|
||||
sbom: true
|
||||
provenance: mode=max
|
||||
|
||||
- name: Sign Mesh sidecar with cosign (keyless)
|
||||
env:
|
||||
TAGS: ${{ steps.mesh_meta.outputs.tags }}
|
||||
DIGEST: ${{ steps.mesh_build.outputs.digest }}
|
||||
run: |
|
||||
refs=()
|
||||
while IFS= read -r tag; do
|
||||
[ -n "$tag" ] || continue
|
||||
refs+=("${tag}@${DIGEST}")
|
||||
done <<< "$TAGS"
|
||||
if [ ${#refs[@]} -gt 0 ]; then
|
||||
cosign sign --yes "${refs[@]}"
|
||||
else
|
||||
echo "No tags to sign (likely a workflow_dispatch run on a non-tag ref)."
|
||||
fi
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
/**
|
||||
* Tests for the Sencho Mesh TCP frames added to the pilot tunnel protocol.
|
||||
*/
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import {
|
||||
BinaryFrameType,
|
||||
decodeBinaryFrame,
|
||||
decodeJsonFrame,
|
||||
encodeBinaryFrame,
|
||||
encodeJsonFrame,
|
||||
} from '../pilot/protocol';
|
||||
|
||||
describe('Mesh TCP JSON frames', () => {
|
||||
it('roundtrips a tcp_open frame', () => {
|
||||
const raw = encodeJsonFrame({
|
||||
t: 'tcp_open',
|
||||
s: 42,
|
||||
stack: 'api',
|
||||
service: 'db',
|
||||
port: 5432,
|
||||
});
|
||||
const decoded = decodeJsonFrame(raw);
|
||||
expect(decoded.t).toBe('tcp_open');
|
||||
if (decoded.t !== 'tcp_open') throw new Error('narrowing');
|
||||
expect(decoded.s).toBe(42);
|
||||
expect(decoded.stack).toBe('api');
|
||||
expect(decoded.service).toBe('db');
|
||||
expect(decoded.port).toBe(5432);
|
||||
});
|
||||
|
||||
it('roundtrips a tcp_open_ack success', () => {
|
||||
const raw = encodeJsonFrame({ t: 'tcp_open_ack', s: 42, ok: true });
|
||||
const decoded = decodeJsonFrame(raw);
|
||||
expect(decoded.t).toBe('tcp_open_ack');
|
||||
if (decoded.t !== 'tcp_open_ack') throw new Error('narrowing');
|
||||
expect(decoded.ok).toBe(true);
|
||||
expect(decoded.err).toBeUndefined();
|
||||
});
|
||||
|
||||
it('roundtrips a tcp_open_ack failure with error code', () => {
|
||||
const raw = encodeJsonFrame({
|
||||
t: 'tcp_open_ack',
|
||||
s: 42,
|
||||
ok: false,
|
||||
err: 'unreachable',
|
||||
});
|
||||
const decoded = decodeJsonFrame(raw);
|
||||
if (decoded.t !== 'tcp_open_ack') throw new Error('narrowing');
|
||||
expect(decoded.ok).toBe(false);
|
||||
expect(decoded.err).toBe('unreachable');
|
||||
});
|
||||
|
||||
it('roundtrips a tcp_close frame', () => {
|
||||
const raw = encodeJsonFrame({ t: 'tcp_close', s: 42 });
|
||||
const decoded = decodeJsonFrame(raw);
|
||||
expect(decoded.t).toBe('tcp_close');
|
||||
if (decoded.t !== 'tcp_close') throw new Error('narrowing');
|
||||
expect(decoded.s).toBe(42);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Mesh TcpData binary frames', () => {
|
||||
it('encodes the 0x04 type discriminator', () => {
|
||||
const payload = Buffer.from('hello');
|
||||
const encoded = encodeBinaryFrame(BinaryFrameType.TcpData, 1, payload);
|
||||
expect(encoded[0]).toBe(0x04);
|
||||
});
|
||||
|
||||
it('roundtrips streamId + payload', () => {
|
||||
const payload = Buffer.from('SELECT 1;');
|
||||
const encoded = encodeBinaryFrame(BinaryFrameType.TcpData, 0xdeadbeef, payload);
|
||||
const decoded = decodeBinaryFrame(encoded);
|
||||
expect(decoded.type).toBe(BinaryFrameType.TcpData);
|
||||
expect(decoded.streamId).toBe(0xdeadbeef);
|
||||
expect(decoded.payload.toString()).toBe('SELECT 1;');
|
||||
});
|
||||
|
||||
it('roundtrips an empty payload', () => {
|
||||
const encoded = encodeBinaryFrame(BinaryFrameType.TcpData, 7, Buffer.alloc(0));
|
||||
const decoded = decodeBinaryFrame(encoded);
|
||||
expect(decoded.type).toBe(BinaryFrameType.TcpData);
|
||||
expect(decoded.streamId).toBe(7);
|
||||
expect(decoded.payload.length).toBe(0);
|
||||
});
|
||||
|
||||
it('preserves binary payloads byte-for-byte', () => {
|
||||
const payload = Buffer.from([0x00, 0xff, 0x01, 0x80, 0x7f, 0x10]);
|
||||
const encoded = encodeBinaryFrame(BinaryFrameType.TcpData, 1, payload);
|
||||
const decoded = decodeBinaryFrame(encoded);
|
||||
expect(decoded.payload.equals(payload)).toBe(true);
|
||||
});
|
||||
|
||||
it('rejects an unknown binary frame type', () => {
|
||||
const buf = Buffer.alloc(5);
|
||||
buf.writeUInt8(0x99, 0);
|
||||
buf.writeUInt32BE(1, 1);
|
||||
expect(() => decodeBinaryFrame(buf)).toThrow(/unknown binary frame type/);
|
||||
});
|
||||
|
||||
it('continues to accept the existing http and ws binary types', () => {
|
||||
for (const t of [BinaryFrameType.HttpReqBody, BinaryFrameType.HttpResBody, BinaryFrameType.WsMessageBinary]) {
|
||||
const buf = encodeBinaryFrame(t, 1, Buffer.from('x'));
|
||||
expect(() => decodeBinaryFrame(buf)).not.toThrow();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -1,10 +1,12 @@
|
||||
import fs from 'fs';
|
||||
import net from 'net';
|
||||
import path from 'path';
|
||||
import http from 'http';
|
||||
import WebSocket from 'ws';
|
||||
import { getSenchoVersion } from '../services/CapabilityRegistry';
|
||||
import {
|
||||
BinaryFrameType,
|
||||
MeshErrCode,
|
||||
PROTOCOL_VERSION,
|
||||
decodeBinaryFrame,
|
||||
decodeJsonFrame,
|
||||
@@ -66,6 +68,7 @@ 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>();
|
||||
private shuttingDown = false;
|
||||
private readonly agentVersion: string;
|
||||
|
||||
@@ -143,6 +146,10 @@ 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 */ }
|
||||
}
|
||||
this.tcpStreams.clear();
|
||||
}
|
||||
|
||||
private scheduleReconnect(): void {
|
||||
@@ -199,6 +206,8 @@ 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_close': this.onTcpClose(frame.s); break;
|
||||
default:
|
||||
// Other frame types are primary-bound only; agent ignores.
|
||||
break;
|
||||
@@ -219,6 +228,12 @@ class PilotAgent {
|
||||
try { ws.send(frame.payload, { binary: true }); } catch { /* ignore */ }
|
||||
break;
|
||||
}
|
||||
case BinaryFrameType.TcpData: {
|
||||
const stream = this.tcpStreams.get(frame.streamId);
|
||||
if (!stream) return;
|
||||
try { stream.socket.write(frame.payload); } catch { /* ignore */ }
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
@@ -329,8 +344,101 @@ class PilotAgent {
|
||||
try { ws.close(code, reason); } catch { /* ignore */ }
|
||||
this.wsStreams.delete(streamId);
|
||||
}
|
||||
|
||||
// --- Sencho Mesh TCP dispatch (tunnel -> Compose service container) ---
|
||||
//
|
||||
// PR 1 rejects every tcp_open with mesh_not_enabled; the dial path is
|
||||
// exercised by tests via setMeshResolver but never lit in production until
|
||||
// PR 2 wires Dockerode resolution gated by the local mesh_stacks table.
|
||||
|
||||
private async onTcpOpen(frame: Extract<ReturnType<typeof decodeJsonFrame>, { t: 'tcp_open' }>): Promise<void> {
|
||||
const ws = this.ws;
|
||||
if (!ws) 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);
|
||||
|
||||
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);
|
||||
});
|
||||
socket.on('data', (chunk: Buffer) => {
|
||||
try {
|
||||
ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, frame.s, chunk), { binary: true });
|
||||
} catch { /* ignore */ }
|
||||
});
|
||||
socket.on('timeout', () => {
|
||||
if (entry.accepted) return;
|
||||
entry.accepted = true;
|
||||
sendAck(false, 'unreachable');
|
||||
this.tcpStreams.delete(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);
|
||||
return;
|
||||
}
|
||||
console.warn('[Pilot] tcp stream error:', sanitizeForLog(err.message));
|
||||
if (this.tcpStreams.delete(frame.s)) {
|
||||
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
|
||||
}
|
||||
});
|
||||
socket.on('close', () => {
|
||||
if (this.tcpStreams.delete(frame.s)) {
|
||||
try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private onTcpClose(streamId: number): void {
|
||||
const entry = this.tcpStreams.get(streamId);
|
||||
if (!entry) return;
|
||||
this.tcpStreams.delete(streamId);
|
||||
try { entry.socket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
/**
|
||||
* Always returns ok:false in PR 1. Tests inject a real resolver via
|
||||
* setMeshResolver; PR 2 will install the Dockerode-backed implementation.
|
||||
*/
|
||||
private async resolveMeshTarget(
|
||||
_stack: string,
|
||||
_service: string,
|
||||
_port: number,
|
||||
): Promise<MeshResolveResult> {
|
||||
return { ok: false, err: 'mesh_not_enabled' };
|
||||
}
|
||||
}
|
||||
|
||||
const MESH_CONNECT_TIMEOUT_MS = 10_000;
|
||||
|
||||
interface MeshTcpStream {
|
||||
socket: net.Socket;
|
||||
accepted: boolean;
|
||||
}
|
||||
|
||||
type MeshResolveResult =
|
||||
| { ok: true; host: string; port: number }
|
||||
| { ok: false; err: MeshErrCode };
|
||||
|
||||
function readPersistedToken(): string | null {
|
||||
try {
|
||||
if (fs.existsSync(TOKEN_PATH)) {
|
||||
|
||||
@@ -23,6 +23,7 @@ export enum BinaryFrameType {
|
||||
HttpReqBody = 0x01,
|
||||
HttpResBody = 0x02,
|
||||
WsMessageBinary = 0x03,
|
||||
TcpData = 0x04,
|
||||
}
|
||||
|
||||
// --- JSON envelope types ---
|
||||
@@ -39,7 +40,10 @@ export type JsonFrame =
|
||||
| WsRejectFrame
|
||||
| WsMessageTextFrame
|
||||
| WsCloseFrame
|
||||
| ControlFrame;
|
||||
| ControlFrame
|
||||
| TcpOpenFrame
|
||||
| TcpOpenAckFrame
|
||||
| TcpCloseFrame;
|
||||
|
||||
export interface HelloFrame {
|
||||
t: 'hello';
|
||||
@@ -119,6 +123,33 @@ export interface ControlFrame {
|
||||
payload?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sencho Mesh TCP frames. The primary asks the agent to open a TCP connection
|
||||
* to a Compose service on the agent's local Docker host. Bytes flow as
|
||||
* BinaryFrameType.TcpData. Mid-stream failures send tcp_close.
|
||||
*/
|
||||
export interface TcpOpenFrame {
|
||||
t: 'tcp_open';
|
||||
s: number;
|
||||
stack: string;
|
||||
service: string;
|
||||
port: number;
|
||||
}
|
||||
|
||||
export type MeshErrCode = 'mesh_not_enabled' | 'denied' | 'no_target' | 'unreachable' | 'agent_error';
|
||||
|
||||
export interface TcpOpenAckFrame {
|
||||
t: 'tcp_open_ack';
|
||||
s: number;
|
||||
ok: boolean;
|
||||
err?: MeshErrCode;
|
||||
}
|
||||
|
||||
export interface TcpCloseFrame {
|
||||
t: 'tcp_close';
|
||||
s: number;
|
||||
}
|
||||
|
||||
// --- Serialize / parse ---
|
||||
|
||||
export function encodeJsonFrame(frame: JsonFrame): string {
|
||||
@@ -161,7 +192,8 @@ export function decodeBinaryFrame(buf: Buffer): DecodedBinaryFrame {
|
||||
const type = buf.readUInt8(0) as BinaryFrameType;
|
||||
if (type !== BinaryFrameType.HttpReqBody &&
|
||||
type !== BinaryFrameType.HttpResBody &&
|
||||
type !== BinaryFrameType.WsMessageBinary) {
|
||||
type !== BinaryFrameType.WsMessageBinary &&
|
||||
type !== BinaryFrameType.TcpData) {
|
||||
throw new Error(`unknown binary frame type: ${type}`);
|
||||
}
|
||||
const streamId = buf.readUInt32BE(1);
|
||||
|
||||
@@ -31,7 +31,62 @@ interface WsStreamState {
|
||||
clientWs?: WebSocket;
|
||||
}
|
||||
|
||||
type StreamState = HttpStreamState | WsStreamState;
|
||||
interface TcpStreamState {
|
||||
kind: 'tcp';
|
||||
handle: TcpStream;
|
||||
bytesIn: number;
|
||||
bytesOut: number;
|
||||
openedAt: number;
|
||||
accepted: boolean;
|
||||
}
|
||||
|
||||
type StreamState = HttpStreamState | WsStreamState | TcpStreamState;
|
||||
|
||||
export interface TcpStreamSummary {
|
||||
streamId: number;
|
||||
bytesIn: number;
|
||||
bytesOut: number;
|
||||
openedAt: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sencho Mesh TCP stream handle. EventEmitter-based duplex-like surface that
|
||||
* MeshService consumes to bridge a local socket to a Compose service on the
|
||||
* remote node behind this pilot tunnel.
|
||||
*
|
||||
* Events:
|
||||
* 'open' tcp_open_ack ok received; safe to write/read
|
||||
* 'data' (Buffer) bytes from the remote socket
|
||||
* 'drain' send buffer below high-water mark
|
||||
* 'error' (err) open rejected or mid-stream tunnel error
|
||||
* 'close' stream closed (graceful or otherwise)
|
||||
*/
|
||||
export class TcpStream extends EventEmitter {
|
||||
public readonly streamId: number;
|
||||
private readonly bridge: PilotTunnelBridge;
|
||||
|
||||
constructor(streamId: number, bridge: PilotTunnelBridge) {
|
||||
super();
|
||||
this.streamId = streamId;
|
||||
this.bridge = bridge;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns false when the underlying tunnel buffer is above the high-water
|
||||
* mark; caller should pause its source until 'drain' fires.
|
||||
*/
|
||||
public write(chunk: Buffer): boolean {
|
||||
return this.bridge._writeTcpData(this.streamId, chunk);
|
||||
}
|
||||
|
||||
public end(): void {
|
||||
this.bridge._closeTcpStream(this.streamId);
|
||||
}
|
||||
|
||||
public destroy(): void {
|
||||
this.end();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-tunnel bridge: hosts a loopback HTTP server that demuxes requests into
|
||||
@@ -86,14 +141,85 @@ export class PilotTunnelBridge extends EventEmitter {
|
||||
});
|
||||
});
|
||||
this.pingTimer = setInterval(() => {
|
||||
if (this.tunnelWs.readyState === WebSocket.OPEN) {
|
||||
try { this.tunnelWs.ping(); } catch { /* surfaced via 'error' */ }
|
||||
if (this.tunnelWs.readyState !== WebSocket.OPEN) return;
|
||||
try { this.tunnelWs.ping(); } catch { /* surfaced via 'error' */ }
|
||||
// Coarse drain fan-out for TCP streams: ws does not expose a
|
||||
// socket-level 'drain' we can hook, so we let backpressure clear
|
||||
// by the next ping cycle.
|
||||
if (this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK) {
|
||||
for (const s of this.streams.values()) {
|
||||
if (s.kind === 'tcp' && s.accepted) s.handle.emit('drain');
|
||||
}
|
||||
}
|
||||
}, PING_INTERVAL_MS);
|
||||
}
|
||||
|
||||
public getLoopbackUrl(): string { return this.loopbackUrl; }
|
||||
public getConnectedAt(): number { return this.connectedAt; }
|
||||
public getBufferedAmount(): number { return this.tunnelWs.bufferedAmount; }
|
||||
public isOpen(): boolean { return !this.closed && this.tunnelWs.readyState === WebSocket.OPEN; }
|
||||
|
||||
/**
|
||||
* Open a TCP stream to a Compose service on the remote node. Caller listens
|
||||
* on the returned TcpStream for 'open' (when the remote agent has accepted),
|
||||
* 'data', 'error', and 'close'. Returns null if the tunnel is not open.
|
||||
*/
|
||||
public openTcpStream(target: { stack: string; service: string; port: number }): TcpStream | null {
|
||||
if (!this.isOpen()) return null;
|
||||
const streamId = this.streamIds.allocate();
|
||||
const handle = new TcpStream(streamId, this);
|
||||
this.streams.set(streamId, {
|
||||
kind: 'tcp',
|
||||
handle,
|
||||
bytesIn: 0,
|
||||
bytesOut: 0,
|
||||
openedAt: Date.now(),
|
||||
accepted: false,
|
||||
});
|
||||
this.sendJson({
|
||||
t: 'tcp_open',
|
||||
s: streamId,
|
||||
stack: target.stack,
|
||||
service: target.service,
|
||||
port: target.port,
|
||||
});
|
||||
return handle;
|
||||
}
|
||||
|
||||
/**
|
||||
* @internal Called only by TcpStream.write; applies the same 4 MB
|
||||
* backpressure rule used by HTTP request bodies. Public for cross-class
|
||||
* access only; not part of the bridge's outward API.
|
||||
*/
|
||||
public _writeTcpData(streamId: number, payload: Buffer): boolean {
|
||||
const s = this.streams.get(streamId);
|
||||
if (!s || s.kind !== 'tcp') return false;
|
||||
if (!this.isOpen()) return false;
|
||||
this.sendBinary(BinaryFrameType.TcpData, streamId, payload);
|
||||
s.bytesOut += payload.length;
|
||||
return this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK;
|
||||
}
|
||||
|
||||
/** @internal Called only by TcpStream.end / .destroy. */
|
||||
public _closeTcpStream(streamId: number): void {
|
||||
if (!this.streams.has(streamId)) return;
|
||||
this.streams.delete(streamId);
|
||||
this.sendJson({ t: 'tcp_close', s: streamId });
|
||||
}
|
||||
|
||||
/**
|
||||
* Snapshot of active TCP streams for the diagnostics sheet. Cheap; called
|
||||
* on demand by MeshService.
|
||||
*/
|
||||
public listTcpStreams(): TcpStreamSummary[] {
|
||||
const out: TcpStreamSummary[] = [];
|
||||
for (const [streamId, s] of this.streams) {
|
||||
if (s.kind === 'tcp') {
|
||||
out.push({ streamId, bytesIn: s.bytesIn, bytesOut: s.bytesOut, openedAt: s.openedAt });
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
public close(code = 1000, reason = 'closed by primary'): void {
|
||||
if (this.closed) return;
|
||||
@@ -314,6 +440,26 @@ export class PilotTunnelBridge extends EventEmitter {
|
||||
// called, and ping/pong are handled by the WS layer.
|
||||
break;
|
||||
}
|
||||
case 'tcp_open_ack': {
|
||||
const s = this.streams.get(frame.s);
|
||||
if (!s || s.kind !== 'tcp') return;
|
||||
if (frame.ok) {
|
||||
s.accepted = true;
|
||||
s.handle.emit('open');
|
||||
} else {
|
||||
this.streams.delete(frame.s);
|
||||
s.handle.emit('error', new Error(frame.err ?? 'tcp_open rejected'));
|
||||
s.handle.emit('close');
|
||||
}
|
||||
break;
|
||||
}
|
||||
case 'tcp_close': {
|
||||
const s = this.streams.get(frame.s);
|
||||
if (!s || s.kind !== 'tcp') return;
|
||||
this.streams.delete(frame.s);
|
||||
s.handle.emit('close');
|
||||
break;
|
||||
}
|
||||
default:
|
||||
// Ignore unknown JSON frame types for forward compatibility.
|
||||
break;
|
||||
@@ -342,6 +488,12 @@ export class PilotTunnelBridge extends EventEmitter {
|
||||
case BinaryFrameType.HttpReqBody:
|
||||
// Agent never originates request bodies; ignore for defense-in-depth.
|
||||
break;
|
||||
case BinaryFrameType.TcpData: {
|
||||
if (s.kind !== 'tcp') return;
|
||||
s.bytesIn += frame.payload.length;
|
||||
s.handle.emit('data', frame.payload);
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
@@ -374,12 +526,17 @@ export class PilotTunnelBridge extends EventEmitter {
|
||||
state.res.end();
|
||||
}
|
||||
} catch { /* ignore */ }
|
||||
} else {
|
||||
} else if (state.kind === 'ws') {
|
||||
if (state.clientWs) {
|
||||
try { state.clientWs.close(1011, 'tunnel closed'); } catch { /* ignore */ }
|
||||
} else if (state.rawSocket) {
|
||||
try { state.rawSocket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
} else {
|
||||
try {
|
||||
if (!state.accepted) state.handle.emit('error', new Error('tunnel closed before accept'));
|
||||
state.handle.emit('close');
|
||||
} catch { /* ignore */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
node_modules
|
||||
dist
|
||||
*.tsbuildinfo
|
||||
@@ -0,0 +1,17 @@
|
||||
# Multi-stage build for the Sencho Mesh sidecar.
|
||||
FROM node:22-alpine AS build
|
||||
WORKDIR /app
|
||||
COPY package.json tsconfig.json ./
|
||||
RUN npm install --no-audit --no-fund
|
||||
COPY src ./src
|
||||
RUN npm run build
|
||||
|
||||
FROM node:22-alpine AS runtime
|
||||
WORKDIR /app
|
||||
ENV NODE_ENV=production
|
||||
COPY package.json ./
|
||||
RUN npm install --omit=dev --no-audit --no-fund
|
||||
COPY --from=build /app/dist ./dist
|
||||
|
||||
USER node
|
||||
CMD ["node", "dist/index.js"]
|
||||
Generated
+1333
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"name": "@sencho/mesh-sidecar",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"description": "Sencho Mesh sidecar: per-node TCP forwarder + control WS bridge.",
|
||||
"main": "dist/index.js",
|
||||
"type": "commonjs",
|
||||
"scripts": {
|
||||
"build": "tsc",
|
||||
"start": "node dist/index.js",
|
||||
"test": "vitest run",
|
||||
"lint": "eslint src"
|
||||
},
|
||||
"dependencies": {
|
||||
"ws": "^8.19.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^25.3.0",
|
||||
"@types/ws": "^8.18.1",
|
||||
"typescript": "^6.0.2",
|
||||
"vitest": "^4.1.0"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
import net from 'net';
|
||||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
import { Forwarder, MeshController } from '../forwarder';
|
||||
|
||||
interface RecordedCalls {
|
||||
resolve: Array<{ connId: number; port: number; remoteAddr: string }>;
|
||||
sendData: Array<{ streamId: number; payload: Buffer }>;
|
||||
sendClose: number[];
|
||||
sendStats: Array<{ streamId: number; bytesIn: number; bytesOut: number }>;
|
||||
}
|
||||
|
||||
function makeRecordingController(): { controller: MeshController; calls: RecordedCalls } {
|
||||
const calls: RecordedCalls = { resolve: [], sendData: [], sendClose: [], sendStats: [] };
|
||||
const controller: MeshController = {
|
||||
resolve: (connId, port, remoteAddr) => calls.resolve.push({ connId, port, remoteAddr }),
|
||||
sendData: (streamId, payload) => calls.sendData.push({ streamId, payload: Buffer.from(payload) }),
|
||||
sendClose: (streamId) => calls.sendClose.push(streamId),
|
||||
sendStats: (streamId, bytesIn, bytesOut) => calls.sendStats.push({ streamId, bytesIn, bytesOut }),
|
||||
};
|
||||
return { controller, calls };
|
||||
}
|
||||
|
||||
async function getEphemeralPort(): Promise<number> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const server = net.createServer();
|
||||
server.unref();
|
||||
server.listen(0, '127.0.0.1', () => {
|
||||
const addr = server.address();
|
||||
if (!addr || typeof addr === 'string') {
|
||||
reject(new Error('no address'));
|
||||
return;
|
||||
}
|
||||
const port = addr.port;
|
||||
server.close(() => resolve(port));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function dial(port: number): Promise<net.Socket> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const socket = net.createConnection({ host: '127.0.0.1', port });
|
||||
socket.once('connect', () => resolve(socket));
|
||||
socket.once('error', reject);
|
||||
});
|
||||
}
|
||||
|
||||
async function waitFor<T>(check: () => T | undefined, timeoutMs = 1000): Promise<T> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
const value = check();
|
||||
if (value !== undefined && value !== null && (Array.isArray(value) ? value.length > 0 : true)) {
|
||||
return value as T;
|
||||
}
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
}
|
||||
throw new Error('timeout waiting for condition');
|
||||
}
|
||||
|
||||
describe('Forwarder', () => {
|
||||
let forwarder: Forwarder | null = null;
|
||||
|
||||
afterEach(async () => {
|
||||
if (forwarder) await forwarder.shutdown();
|
||||
forwarder = null;
|
||||
});
|
||||
|
||||
it('asks the controller to resolve when a client connects', async () => {
|
||||
const { controller, calls } = makeRecordingController();
|
||||
forwarder = new Forwarder(controller);
|
||||
forwarder.start();
|
||||
const port = await getEphemeralPort();
|
||||
await forwarder.listen(port);
|
||||
|
||||
const client = await dial(port);
|
||||
const resolved = await waitFor(() => calls.resolve.length ? calls.resolve : undefined);
|
||||
expect(resolved[0].port).toBe(port);
|
||||
expect(resolved[0].connId).toBeGreaterThan(0);
|
||||
|
||||
client.destroy();
|
||||
});
|
||||
|
||||
it('splices bytes both directions after resolve_ok', async () => {
|
||||
const { controller, calls } = makeRecordingController();
|
||||
forwarder = new Forwarder(controller);
|
||||
forwarder.start();
|
||||
const port = await getEphemeralPort();
|
||||
await forwarder.listen(port);
|
||||
|
||||
const received: Buffer[] = [];
|
||||
const client = await dial(port);
|
||||
client.on('data', (chunk: Buffer) => received.push(chunk));
|
||||
|
||||
const resolved = await waitFor(() => calls.resolve.length ? calls.resolve : undefined);
|
||||
const connId = resolved[0].connId;
|
||||
const streamId = 42;
|
||||
forwarder.handleResolveOk(connId, streamId, 'test.alias.sencho');
|
||||
|
||||
client.write('hello');
|
||||
const sent = await waitFor(() => calls.sendData.length ? calls.sendData : undefined);
|
||||
expect(sent[0].streamId).toBe(streamId);
|
||||
expect(sent[0].payload.toString()).toBe('hello');
|
||||
|
||||
forwarder.handleData(streamId, Buffer.from('hi back'));
|
||||
await waitFor(() => received.length ? received : undefined);
|
||||
expect(Buffer.concat(received).toString()).toBe('hi back');
|
||||
|
||||
client.destroy();
|
||||
});
|
||||
|
||||
it('sends a close frame when the client socket disconnects', async () => {
|
||||
const { controller, calls } = makeRecordingController();
|
||||
forwarder = new Forwarder(controller);
|
||||
forwarder.start();
|
||||
const port = await getEphemeralPort();
|
||||
await forwarder.listen(port);
|
||||
|
||||
const client = await dial(port);
|
||||
const resolved = await waitFor(() => calls.resolve.length ? calls.resolve : undefined);
|
||||
forwarder.handleResolveOk(resolved[0].connId, 7);
|
||||
|
||||
client.destroy();
|
||||
const closed = await waitFor(() => calls.sendClose.length ? calls.sendClose : undefined);
|
||||
expect(closed[0]).toBe(7);
|
||||
});
|
||||
|
||||
it('drops the local socket on resolve_err', async () => {
|
||||
const { controller, calls } = makeRecordingController();
|
||||
forwarder = new Forwarder(controller);
|
||||
forwarder.start();
|
||||
const port = await getEphemeralPort();
|
||||
await forwarder.listen(port);
|
||||
|
||||
const client = await dial(port);
|
||||
const closedPromise = new Promise<void>((resolve) => client.once('close', () => resolve()));
|
||||
const resolved = await waitFor(() => calls.resolve.length ? calls.resolve : undefined);
|
||||
|
||||
forwarder.handleResolveErr(resolved[0].connId, 'tunnel_down');
|
||||
await closedPromise;
|
||||
// No data was ever piped, so no close frame sent for this connId.
|
||||
expect(calls.sendClose.length).toBe(0);
|
||||
});
|
||||
|
||||
it('destroys the local socket on inbound close', async () => {
|
||||
const { controller, calls } = makeRecordingController();
|
||||
forwarder = new Forwarder(controller);
|
||||
forwarder.start();
|
||||
const port = await getEphemeralPort();
|
||||
await forwarder.listen(port);
|
||||
|
||||
const client = await dial(port);
|
||||
const closedPromise = new Promise<void>((resolve) => client.once('close', () => resolve()));
|
||||
const resolved = await waitFor(() => calls.resolve.length ? calls.resolve : undefined);
|
||||
forwarder.handleResolveOk(resolved[0].connId, 11);
|
||||
|
||||
forwarder.handleClose(11);
|
||||
await closedPromise;
|
||||
});
|
||||
|
||||
it('refuses new connections during shutdown', async () => {
|
||||
const { controller } = makeRecordingController();
|
||||
forwarder = new Forwarder(controller);
|
||||
forwarder.start();
|
||||
const port = await getEphemeralPort();
|
||||
await forwarder.listen(port);
|
||||
await forwarder.shutdown();
|
||||
|
||||
await expect(dial(port)).rejects.toThrow();
|
||||
forwarder = null;
|
||||
});
|
||||
|
||||
it('emits stats for active streams when the timer fires', async () => {
|
||||
const { controller, calls } = makeRecordingController();
|
||||
forwarder = new Forwarder(controller, { statsIntervalMs: 50 });
|
||||
forwarder.start();
|
||||
const port = await getEphemeralPort();
|
||||
await forwarder.listen(port);
|
||||
|
||||
const client = await dial(port);
|
||||
const resolved = await waitFor(() => calls.resolve.length ? calls.resolve : undefined);
|
||||
forwarder.handleResolveOk(resolved[0].connId, 99);
|
||||
|
||||
client.write('x');
|
||||
|
||||
const stats = await waitFor(() => calls.sendStats.length ? calls.sendStats : undefined, 1500);
|
||||
expect(stats[0].streamId).toBe(99);
|
||||
expect(stats[0].bytesOut).toBeGreaterThan(0);
|
||||
|
||||
client.destroy();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,73 @@
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import {
|
||||
BinaryFrameType,
|
||||
decodeControl,
|
||||
decodeData,
|
||||
encodeControl,
|
||||
encodeData,
|
||||
} from '../protocol';
|
||||
|
||||
describe('Mesh sidecar control frames', () => {
|
||||
it('roundtrips a hello frame', () => {
|
||||
const raw = encodeControl({ t: 'hello', version: 1, nodeId: 7, sidecarVersion: '0.0.1' });
|
||||
const decoded = decodeControl(raw);
|
||||
expect(decoded.t).toBe('hello');
|
||||
if (decoded.t !== 'hello') throw new Error('narrowing');
|
||||
expect(decoded.nodeId).toBe(7);
|
||||
});
|
||||
|
||||
it('roundtrips a listen / unlisten pair', () => {
|
||||
const listen = decodeControl(encodeControl({ t: 'listen', port: 5432 }));
|
||||
const unlisten = decodeControl(encodeControl({ t: 'unlisten', port: 5432 }));
|
||||
expect(listen.t).toBe('listen');
|
||||
expect(unlisten.t).toBe('unlisten');
|
||||
});
|
||||
|
||||
it('roundtrips resolve / resolve_ok / resolve_err', () => {
|
||||
const resolve = decodeControl(encodeControl({ t: 'resolve', connId: 1, port: 5432, remoteAddr: '10.0.0.5' }));
|
||||
const ok = decodeControl(encodeControl({ t: 'resolve_ok', connId: 1, streamId: 9, alias: 'db.api.opsix.sencho' }));
|
||||
const err = decodeControl(encodeControl({ t: 'resolve_err', connId: 1, code: 'tunnel_down' }));
|
||||
if (resolve.t !== 'resolve') throw new Error('narrowing');
|
||||
if (ok.t !== 'resolve_ok') throw new Error('narrowing');
|
||||
if (err.t !== 'resolve_err') throw new Error('narrowing');
|
||||
expect(resolve.port).toBe(5432);
|
||||
expect(ok.streamId).toBe(9);
|
||||
expect(err.code).toBe('tunnel_down');
|
||||
});
|
||||
|
||||
it('rejects malformed control frames', () => {
|
||||
expect(() => decodeControl('not json')).toThrow();
|
||||
expect(() => decodeControl('{}')).toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe('Mesh sidecar data frames', () => {
|
||||
it('encodes the 0x01 type discriminator', () => {
|
||||
const buf = encodeData(1, Buffer.from('hello'));
|
||||
expect(buf[0]).toBe(BinaryFrameType.Data);
|
||||
});
|
||||
|
||||
it('roundtrips streamId + payload', () => {
|
||||
const payload = Buffer.from('SELECT 1;');
|
||||
const buf = encodeData(0xdeadbeef, payload);
|
||||
const decoded = decodeData(buf);
|
||||
expect(decoded.streamId).toBe(0xdeadbeef);
|
||||
expect(decoded.payload.toString()).toBe('SELECT 1;');
|
||||
});
|
||||
|
||||
it('preserves binary payloads byte-for-byte', () => {
|
||||
const payload = Buffer.from([0x00, 0xff, 0x01, 0x80, 0x7f]);
|
||||
const decoded = decodeData(encodeData(1, payload));
|
||||
expect(decoded.payload.equals(payload)).toBe(true);
|
||||
});
|
||||
|
||||
it('rejects an unknown binary frame type', () => {
|
||||
const buf = Buffer.alloc(5);
|
||||
buf.writeUInt8(0x99, 0);
|
||||
expect(() => decodeData(buf)).toThrow(/unknown binary frame type/);
|
||||
});
|
||||
|
||||
it('rejects too-short frames', () => {
|
||||
expect(() => decodeData(Buffer.alloc(3))).toThrow(/too short/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,175 @@
|
||||
import WebSocket from 'ws';
|
||||
import { Forwarder, MeshController } from './forwarder';
|
||||
import {
|
||||
ControlFrame,
|
||||
PROTOCOL_VERSION,
|
||||
decodeControl,
|
||||
decodeData,
|
||||
encodeControl,
|
||||
encodeData,
|
||||
wsDataToBuffer,
|
||||
wsDataToString,
|
||||
} from './protocol';
|
||||
|
||||
const RECONNECT_MIN_MS = 1_000;
|
||||
const RECONNECT_MAX_MS = 30_000;
|
||||
const PING_INTERVAL_MS = 30_000;
|
||||
|
||||
export interface ControlClientOptions {
|
||||
controlUrl: string;
|
||||
token: string;
|
||||
nodeId: number;
|
||||
sidecarVersion: string;
|
||||
forwarder: Forwarder;
|
||||
}
|
||||
|
||||
/**
|
||||
* WS-backed implementation of MeshController. Keeps a single long-lived
|
||||
* connection to the local Sencho instance, surfaces inbound frames to the
|
||||
* Forwarder, and queues outbound frames when reconnecting.
|
||||
*/
|
||||
export class ControlClient implements MeshController {
|
||||
private readonly options: ControlClientOptions;
|
||||
private ws: WebSocket | null = null;
|
||||
private backoff = RECONNECT_MIN_MS;
|
||||
private pingTimer?: NodeJS.Timeout;
|
||||
private reconnectTimer?: NodeJS.Timeout;
|
||||
private shuttingDown = false;
|
||||
|
||||
constructor(options: ControlClientOptions) {
|
||||
this.options = options;
|
||||
}
|
||||
|
||||
public start(): void {
|
||||
this.connect();
|
||||
}
|
||||
|
||||
public async shutdown(): Promise<void> {
|
||||
this.shuttingDown = true;
|
||||
if (this.pingTimer) clearInterval(this.pingTimer);
|
||||
if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = undefined; }
|
||||
try { this.ws?.close(1000, 'sidecar shutdown'); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
// --- MeshController surface ---
|
||||
|
||||
public resolve(connId: number, port: number, remoteAddr: string): void {
|
||||
this.send({ t: 'resolve', connId, port, remoteAddr });
|
||||
}
|
||||
|
||||
public sendData(streamId: number, payload: Buffer): void {
|
||||
const ws = this.ws;
|
||||
if (!ws || ws.readyState !== WebSocket.OPEN) return;
|
||||
try { ws.send(encodeData(streamId, payload), { binary: true }); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
public sendClose(streamId: number): void {
|
||||
this.send({ t: 'close', streamId });
|
||||
}
|
||||
|
||||
public sendStats(streamId: number, bytesIn: number, bytesOut: number, lastActivity: number): void {
|
||||
this.send({ t: 'stream.stats', streamId, bytesIn, bytesOut, lastActivity });
|
||||
}
|
||||
|
||||
public sendLog(level: 'info' | 'warn' | 'error', message: string, details?: Record<string, unknown>): void {
|
||||
this.send({ t: 'log', level, message, details });
|
||||
}
|
||||
|
||||
// --- Connection lifecycle ---
|
||||
|
||||
private connect(): void {
|
||||
if (this.shuttingDown) return;
|
||||
|
||||
if (this.pingTimer) { clearInterval(this.pingTimer); this.pingTimer = undefined; }
|
||||
if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = undefined; }
|
||||
|
||||
const ws = new WebSocket(this.options.controlUrl, {
|
||||
headers: { Authorization: `Bearer ${this.options.token}` },
|
||||
handshakeTimeout: 15_000,
|
||||
});
|
||||
this.ws = ws;
|
||||
|
||||
ws.on('open', () => {
|
||||
this.backoff = RECONNECT_MIN_MS;
|
||||
this.send({
|
||||
t: 'hello',
|
||||
version: PROTOCOL_VERSION,
|
||||
nodeId: this.options.nodeId,
|
||||
sidecarVersion: this.options.sidecarVersion,
|
||||
});
|
||||
this.pingTimer = setInterval(() => {
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
try { ws.ping(); } catch { /* error event will handle */ }
|
||||
}
|
||||
}, PING_INTERVAL_MS);
|
||||
});
|
||||
|
||||
ws.on('message', (data, isBinary) => this.onMessage(data, isBinary));
|
||||
ws.on('close', () => {
|
||||
if (this.pingTimer) { clearInterval(this.pingTimer); this.pingTimer = undefined; }
|
||||
this.scheduleReconnect();
|
||||
});
|
||||
ws.on('error', () => {
|
||||
// 'close' will follow.
|
||||
});
|
||||
}
|
||||
|
||||
private scheduleReconnect(): void {
|
||||
if (this.shuttingDown) return;
|
||||
const jitter = Math.floor(Math.random() * 500);
|
||||
const delay = this.backoff + jitter;
|
||||
this.reconnectTimer = setTimeout(() => {
|
||||
this.reconnectTimer = undefined;
|
||||
this.connect();
|
||||
}, delay);
|
||||
this.backoff = Math.min(this.backoff * 2, RECONNECT_MAX_MS);
|
||||
}
|
||||
|
||||
private onMessage(data: WebSocket.RawData, isBinary: boolean): void {
|
||||
try {
|
||||
if (isBinary) {
|
||||
const buf = wsDataToBuffer(data);
|
||||
if (!buf) return;
|
||||
const decoded = decodeData(buf);
|
||||
this.options.forwarder.handleData(decoded.streamId, decoded.payload);
|
||||
} else {
|
||||
const text = wsDataToString(data);
|
||||
if (text == null) return;
|
||||
const frame = decodeControl(text);
|
||||
this.dispatchControl(frame);
|
||||
}
|
||||
} catch {
|
||||
// Malformed frames are dropped; tunnel stays up.
|
||||
}
|
||||
}
|
||||
|
||||
private dispatchControl(frame: ControlFrame): void {
|
||||
switch (frame.t) {
|
||||
case 'listen':
|
||||
void this.options.forwarder.listen(frame.port);
|
||||
break;
|
||||
case 'unlisten':
|
||||
void this.options.forwarder.unlisten(frame.port);
|
||||
break;
|
||||
case 'resolve_ok':
|
||||
this.options.forwarder.handleResolveOk(frame.connId, frame.streamId, frame.alias);
|
||||
break;
|
||||
case 'resolve_err':
|
||||
this.options.forwarder.handleResolveErr(frame.connId, frame.code);
|
||||
break;
|
||||
case 'close':
|
||||
this.options.forwarder.handleClose(frame.streamId);
|
||||
break;
|
||||
default:
|
||||
// hello / resolve / stream.stats / log are sidecar-originated;
|
||||
// ignore on the inbound path.
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private send(frame: ControlFrame): void {
|
||||
const ws = this.ws;
|
||||
if (!ws || ws.readyState !== WebSocket.OPEN) return;
|
||||
try { ws.send(encodeControl(frame)); } catch { /* ignore */ }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,180 @@
|
||||
import net from 'net';
|
||||
|
||||
const DEFAULT_STATS_INTERVAL_MS = 5_000;
|
||||
|
||||
export interface ForwarderOptions {
|
||||
statsIntervalMs?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Outbound channel to the Sencho instance hosting this sidecar. The
|
||||
* production implementation is the control WS client; tests inject a fake.
|
||||
*/
|
||||
export interface MeshController {
|
||||
resolve(connId: number, port: number, remoteAddr: string): void;
|
||||
sendData(streamId: number, payload: Buffer): void;
|
||||
sendClose(streamId: number): void;
|
||||
sendStats(streamId: number, bytesIn: number, bytesOut: number, lastActivity: number): void;
|
||||
}
|
||||
|
||||
interface PendingConn {
|
||||
connId: number;
|
||||
socket: net.Socket;
|
||||
port: number;
|
||||
}
|
||||
|
||||
interface ActiveStream {
|
||||
streamId: number;
|
||||
socket: net.Socket;
|
||||
bytesIn: number;
|
||||
bytesOut: number;
|
||||
lastActivity: number;
|
||||
alias?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* The Forwarder owns per-port TCP listeners and the live socket <-> stream
|
||||
* mapping. It is wire-agnostic: all outbound frames flow through MeshController
|
||||
* methods. Inbound frames are delivered via the handle* methods.
|
||||
*/
|
||||
export class Forwarder {
|
||||
private readonly controller: MeshController;
|
||||
private readonly statsIntervalMs: number;
|
||||
private readonly listeners = new Map<number, net.Server>();
|
||||
private readonly pendingConns = new Map<number, PendingConn>();
|
||||
private readonly activeStreams = new Map<number, ActiveStream>();
|
||||
private nextConnId = 1;
|
||||
private statsTimer?: NodeJS.Timeout;
|
||||
private shuttingDown = false;
|
||||
|
||||
constructor(controller: MeshController, options: ForwarderOptions = {}) {
|
||||
this.controller = controller;
|
||||
this.statsIntervalMs = options.statsIntervalMs ?? DEFAULT_STATS_INTERVAL_MS;
|
||||
}
|
||||
|
||||
public start(): void {
|
||||
this.statsTimer = setInterval(() => this.flushStats(), this.statsIntervalMs);
|
||||
}
|
||||
|
||||
public async listen(port: number): Promise<void> {
|
||||
if (this.listeners.has(port) || this.shuttingDown) return;
|
||||
const server = net.createServer((socket) => this.acceptConnection(port, socket));
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onError = (err: Error) => { server.removeListener('listening', onListening); reject(err); };
|
||||
const onListening = () => { server.removeListener('error', onError); resolve(); };
|
||||
server.once('error', onError);
|
||||
server.once('listening', onListening);
|
||||
server.listen(port);
|
||||
});
|
||||
this.listeners.set(port, server);
|
||||
}
|
||||
|
||||
public async unlisten(port: number): Promise<void> {
|
||||
const server = this.listeners.get(port);
|
||||
if (!server) return;
|
||||
this.listeners.delete(port);
|
||||
await new Promise<void>((resolve) => {
|
||||
server.close(() => resolve());
|
||||
});
|
||||
}
|
||||
|
||||
public handleResolveOk(connId: number, streamId: number, alias?: string): void {
|
||||
const pending = this.pendingConns.get(connId);
|
||||
if (!pending) {
|
||||
// Sencho thinks we have a conn but we don't (race with socket close).
|
||||
this.controller.sendClose(streamId);
|
||||
return;
|
||||
}
|
||||
this.pendingConns.delete(connId);
|
||||
|
||||
const stream: ActiveStream = {
|
||||
streamId,
|
||||
socket: pending.socket,
|
||||
bytesIn: 0,
|
||||
bytesOut: 0,
|
||||
lastActivity: Date.now(),
|
||||
alias,
|
||||
};
|
||||
this.activeStreams.set(streamId, stream);
|
||||
|
||||
pending.socket.on('data', (chunk: Buffer) => {
|
||||
stream.bytesOut += chunk.length;
|
||||
stream.lastActivity = Date.now();
|
||||
this.controller.sendData(streamId, chunk);
|
||||
});
|
||||
pending.socket.on('close', () => {
|
||||
if (this.activeStreams.delete(streamId)) {
|
||||
this.controller.sendClose(streamId);
|
||||
}
|
||||
});
|
||||
pending.socket.on('error', () => {
|
||||
if (this.activeStreams.delete(streamId)) {
|
||||
this.controller.sendClose(streamId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public handleResolveErr(connId: number, _code: string): void {
|
||||
const pending = this.pendingConns.get(connId);
|
||||
if (!pending) return;
|
||||
this.pendingConns.delete(connId);
|
||||
try { pending.socket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
public handleData(streamId: number, payload: Buffer): void {
|
||||
const stream = this.activeStreams.get(streamId);
|
||||
if (!stream) return;
|
||||
stream.bytesIn += payload.length;
|
||||
stream.lastActivity = Date.now();
|
||||
try { stream.socket.write(payload); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
public handleClose(streamId: number): void {
|
||||
const stream = this.activeStreams.get(streamId);
|
||||
if (!stream) return;
|
||||
this.activeStreams.delete(streamId);
|
||||
try { stream.socket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
public async shutdown(): Promise<void> {
|
||||
this.shuttingDown = true;
|
||||
if (this.statsTimer) { clearInterval(this.statsTimer); this.statsTimer = undefined; }
|
||||
for (const [, stream] of this.activeStreams) {
|
||||
try { stream.socket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
this.activeStreams.clear();
|
||||
for (const [, pending] of this.pendingConns) {
|
||||
try { pending.socket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
this.pendingConns.clear();
|
||||
const ports = Array.from(this.listeners.keys());
|
||||
await Promise.all(ports.map((p) => this.unlisten(p)));
|
||||
}
|
||||
|
||||
public getActiveStreamCount(): number { return this.activeStreams.size; }
|
||||
public getListenerPorts(): number[] { return Array.from(this.listeners.keys()); }
|
||||
|
||||
private acceptConnection(port: number, socket: net.Socket): void {
|
||||
if (this.shuttingDown) {
|
||||
try { socket.destroy(); } catch { /* ignore */ }
|
||||
return;
|
||||
}
|
||||
const connId = this.nextConnId++;
|
||||
this.pendingConns.set(connId, { connId, socket, port });
|
||||
const remoteAddr = socket.remoteAddress ?? '';
|
||||
socket.once('error', () => { this.pendingConns.delete(connId); });
|
||||
socket.once('close', () => { this.pendingConns.delete(connId); });
|
||||
this.controller.resolve(connId, port, remoteAddr);
|
||||
}
|
||||
|
||||
private flushStats(): void {
|
||||
const now = Date.now();
|
||||
for (const [streamId, stream] of this.activeStreams) {
|
||||
// Only emit for streams that saw activity in the last interval to
|
||||
// keep the activity log readable.
|
||||
if (now - stream.lastActivity <= this.statsIntervalMs * 2) {
|
||||
this.controller.sendStats(streamId, stream.bytesIn, stream.bytesOut, stream.lastActivity);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
import { ControlClient } from './control';
|
||||
import { Forwarder } from './forwarder';
|
||||
|
||||
const SIDECAR_VERSION = '0.0.1';
|
||||
|
||||
function requireEnv(name: string): string {
|
||||
const value = process.env[name];
|
||||
if (!value) {
|
||||
console.error(`[mesh-sidecar] missing required env: ${name}`);
|
||||
process.exit(1);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function main(): void {
|
||||
const controlUrl = requireEnv('SENCHO_CONTROL_URL');
|
||||
const token = requireEnv('SENCHO_MESH_TOKEN');
|
||||
const nodeIdRaw = requireEnv('MESH_NODE_ID');
|
||||
const nodeId = Number.parseInt(nodeIdRaw, 10);
|
||||
if (!Number.isFinite(nodeId)) {
|
||||
console.error('[mesh-sidecar] MESH_NODE_ID must be an integer');
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
let client: ControlClient | null = null;
|
||||
// Bridge with explicit named params so types stay tight; client is wired
|
||||
// immediately after construction so the null check is only racing against
|
||||
// the 5s stats timer first tick.
|
||||
const forwarder = new Forwarder({
|
||||
resolve: (connId, port, remoteAddr) => client?.resolve(connId, port, remoteAddr),
|
||||
sendData: (streamId, payload) => client?.sendData(streamId, payload),
|
||||
sendClose: (streamId) => client?.sendClose(streamId),
|
||||
sendStats: (streamId, bytesIn, bytesOut, lastActivity) =>
|
||||
client?.sendStats(streamId, bytesIn, bytesOut, lastActivity),
|
||||
});
|
||||
forwarder.start();
|
||||
|
||||
client = new ControlClient({
|
||||
controlUrl,
|
||||
token,
|
||||
nodeId,
|
||||
sidecarVersion: SIDECAR_VERSION,
|
||||
forwarder,
|
||||
});
|
||||
client.start();
|
||||
|
||||
const shutdown = async () => {
|
||||
await forwarder.shutdown();
|
||||
await client?.shutdown();
|
||||
process.exit(0);
|
||||
};
|
||||
process.on('SIGTERM', () => { void shutdown(); });
|
||||
process.on('SIGINT', () => { void shutdown(); });
|
||||
|
||||
console.log(`[mesh-sidecar] started for node=${nodeId} version=${SIDECAR_VERSION}`);
|
||||
}
|
||||
|
||||
main();
|
||||
@@ -0,0 +1,146 @@
|
||||
/**
|
||||
* Mesh sidecar control protocol. Wire-compatible shape with the pilot tunnel:
|
||||
* JSON text frames for control + a single binary frame type carrying the
|
||||
* tunneled bytes.
|
||||
*
|
||||
* [ 1 byte: BinaryFrameType ][ 4 bytes: streamId (BE) ][ payload bytes... ]
|
||||
*/
|
||||
|
||||
export const PROTOCOL_VERSION = 1;
|
||||
|
||||
export enum BinaryFrameType {
|
||||
Data = 0x01,
|
||||
}
|
||||
|
||||
export type ControlFrame =
|
||||
| ListenFrame
|
||||
| UnlistenFrame
|
||||
| HelloFrame
|
||||
| ResolveFrame
|
||||
| ResolveOkFrame
|
||||
| ResolveErrFrame
|
||||
| StreamStatsFrame
|
||||
| LogFrame
|
||||
| CloseFrame;
|
||||
|
||||
export interface HelloFrame {
|
||||
t: 'hello';
|
||||
version: number;
|
||||
nodeId: number;
|
||||
sidecarVersion: string;
|
||||
}
|
||||
|
||||
/** Sencho -> sidecar: start listening on a TCP port for inbound app traffic. */
|
||||
export interface ListenFrame {
|
||||
t: 'listen';
|
||||
port: number;
|
||||
}
|
||||
|
||||
/** Sencho -> sidecar: stop listening on the given port. */
|
||||
export interface UnlistenFrame {
|
||||
t: 'unlisten';
|
||||
port: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sidecar -> Sencho: a new local TCP connection arrived on this port; what
|
||||
* stream should I attach it to?
|
||||
*/
|
||||
export interface ResolveFrame {
|
||||
t: 'resolve';
|
||||
connId: number;
|
||||
port: number;
|
||||
remoteAddr?: string;
|
||||
}
|
||||
|
||||
/** Sencho -> sidecar: forward bytes for connId via streamId. */
|
||||
export interface ResolveOkFrame {
|
||||
t: 'resolve_ok';
|
||||
connId: number;
|
||||
streamId: number;
|
||||
alias?: string;
|
||||
}
|
||||
|
||||
/** Sencho -> sidecar: drop the connection with a reason. */
|
||||
export interface ResolveErrFrame {
|
||||
t: 'resolve_err';
|
||||
connId: number;
|
||||
code: 'no_route' | 'tunnel_down' | 'denied' | 'unreachable' | 'agent_error';
|
||||
message?: string;
|
||||
}
|
||||
|
||||
/** Sidecar -> Sencho: per-stream byte counters every ~5 seconds while open. */
|
||||
export interface StreamStatsFrame {
|
||||
t: 'stream.stats';
|
||||
streamId: number;
|
||||
bytesIn: number;
|
||||
bytesOut: number;
|
||||
lastActivity: number;
|
||||
}
|
||||
|
||||
/** Sidecar -> Sencho: structured log line surfaced into the activity log. */
|
||||
export interface LogFrame {
|
||||
t: 'log';
|
||||
level: 'info' | 'warn' | 'error';
|
||||
message: string;
|
||||
details?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** Either side -> close a stream. */
|
||||
export interface CloseFrame {
|
||||
t: 'close';
|
||||
streamId: number;
|
||||
}
|
||||
|
||||
export function encodeControl(frame: ControlFrame): string {
|
||||
return JSON.stringify(frame);
|
||||
}
|
||||
|
||||
export function decodeControl(raw: string): ControlFrame {
|
||||
const parsed = JSON.parse(raw);
|
||||
if (!parsed || typeof parsed !== 'object' || typeof parsed.t !== 'string') {
|
||||
throw new Error('invalid control frame: missing type discriminator');
|
||||
}
|
||||
return parsed as ControlFrame;
|
||||
}
|
||||
|
||||
export function encodeData(streamId: number, payload: Buffer): Buffer {
|
||||
if (!Number.isInteger(streamId) || streamId < 0 || streamId > 0xffffffff) {
|
||||
throw new Error(`invalid streamId: ${streamId}`);
|
||||
}
|
||||
const out = Buffer.allocUnsafe(5 + payload.length);
|
||||
out.writeUInt8(BinaryFrameType.Data, 0);
|
||||
out.writeUInt32BE(streamId, 1);
|
||||
payload.copy(out, 5);
|
||||
return out;
|
||||
}
|
||||
|
||||
export interface DecodedData {
|
||||
streamId: number;
|
||||
payload: Buffer;
|
||||
}
|
||||
|
||||
export function decodeData(buf: Buffer): DecodedData {
|
||||
if (buf.length < 5) throw new Error(`data frame too short: ${buf.length} bytes`);
|
||||
const type = buf.readUInt8(0);
|
||||
if (type !== BinaryFrameType.Data) throw new Error(`unknown binary frame type: ${type}`);
|
||||
return {
|
||||
streamId: buf.readUInt32BE(1),
|
||||
payload: buf.subarray(5),
|
||||
};
|
||||
}
|
||||
|
||||
export type WsRawData = Buffer | ArrayBuffer | Buffer[] | string;
|
||||
|
||||
export function wsDataToBuffer(data: WsRawData): Buffer | null {
|
||||
if (Buffer.isBuffer(data)) return data;
|
||||
if (data instanceof ArrayBuffer) return Buffer.from(data);
|
||||
if (Array.isArray(data)) return Buffer.concat(data.map((d) => Buffer.isBuffer(d) ? d : Buffer.from(d as ArrayBuffer)));
|
||||
return null;
|
||||
}
|
||||
|
||||
export function wsDataToString(data: WsRawData): string | null {
|
||||
if (typeof data === 'string') return data;
|
||||
const buf = wsDataToBuffer(data);
|
||||
return buf ? buf.toString('utf8') : null;
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "ES2022",
|
||||
"module": "commonjs",
|
||||
"lib": ["ES2022"],
|
||||
"outDir": "./dist",
|
||||
"rootDir": "./src",
|
||||
"strict": true,
|
||||
"esModuleInterop": true,
|
||||
"skipLibCheck": true,
|
||||
"forceConsistentCasingInFileNames": true,
|
||||
"resolveJsonModule": true
|
||||
},
|
||||
"include": ["src/**/*"],
|
||||
"exclude": ["node_modules", "dist"]
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
import { defineConfig } from 'vitest/config';
|
||||
|
||||
export default defineConfig({
|
||||
test: {
|
||||
environment: 'node',
|
||||
include: ['src/__tests__/**/*.test.ts'],
|
||||
exclude: ['dist/**', 'node_modules/**'],
|
||||
pool: 'forks',
|
||||
testTimeout: 15_000,
|
||||
},
|
||||
});
|
||||
Reference in New Issue
Block a user