From 34ead6e471956a2d05b7a0ff1b928cfb37b0b6e1 Mon Sep 17 00:00:00 2001 From: LukeMackin Date: Wed, 22 Jul 2026 17:29:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20C-015+P-006/008/009=20Gateway=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - connection.ts: ConnectionManager(注册/心跳/重连) - checkConnectivity+P-006: /health HEAD探测 - registerWithGateway: POST /devices/register - sendHeartbeat+P-008: 15s心跳 POST /devices/:id/heartbeat - backoffDelay+P-009: 指数退避+30%抖动 - main.ts: bootstrap后启动ConnectionManager - SIGINT/SIGTERM优雅关闭 --- packages/server/src/connection.ts | 231 ++++++++++++++++++++++++++++++ packages/server/src/main.ts | 11 ++ 2 files changed, 242 insertions(+) create mode 100644 packages/server/src/connection.ts diff --git a/packages/server/src/connection.ts b/packages/server/src/connection.ts new file mode 100644 index 0000000..b416676 --- /dev/null +++ b/packages/server/src/connection.ts @@ -0,0 +1,231 @@ +/** + * C-015 + P-006/008/009: Gateway Connection Lifecycle + * + * - Register device with Gateway on startup + * - Heartbeat keepalive every N seconds + * - Exponential backoff reconnection on disconnect + * - Connectivity check before registration + */ + +import { type DeviceConfig, type DeviceIdentity } from "@yuzu-gca/shared"; + +export interface ConnectionState { + status: "disconnected" | "connecting" | "connected" | "reconnecting"; + lastHeartbeat: number; + reconnectAttempt: number; + registeredAt: number | null; +} + +/** + * Check if the Gateway is reachable. + * Returns true if a HEAD request to gateway.url succeeds. + */ +export async function checkConnectivity(gatewayUrl: string): Promise { + try { + const resp = await fetch(`${gatewayUrl}/health`, { + method: "HEAD", + signal: AbortSignal.timeout(5000), + }); + return resp.ok; + } catch { + return false; + } +} + +/** + * Register this device with the Gateway. + * POST /devices/register with device identity and capabilities. + * Returns the assigned device token on success. + */ +export async function registerWithGateway( + gatewayUrl: string, + config: DeviceConfig, +): Promise<{ token: string } | null> { + try { + const resp = await fetch(`${gatewayUrl}/devices/register`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + id: config.identity.id, + name: config.identity.name, + platform: config.identity.platform, + arch: config.identity.arch, + hostname: config.identity.hostname, + capabilities: config.capabilities, + transport: config.gateway.transport, + }), + signal: AbortSignal.timeout(10000), + }); + if (!resp.ok) { + console.error(`[Gateway] Registration failed: HTTP ${resp.status}`); + return null; + } + const data = await resp.json() as { token: string }; + console.log(`[Gateway] Registered as ${config.identity.name} (${config.identity.id})`); + return { token: data.token }; + } catch (err) { + console.error(`[Gateway] Registration error:`, (err as Error).message); + return null; + } +} + +/** + * Send a heartbeat to the Gateway. + * POST /devices/:id/heartbeat + * Returns true if Gateway acknowledged. + */ +export async function sendHeartbeat( + gatewayUrl: string, + deviceId: string, + token: string, +): Promise { + try { + const resp = await fetch(`${gatewayUrl}/devices/${deviceId}/heartbeat`, { + method: "POST", + headers: { + "Content-Type": "application/json", + "Authorization": `Bearer ${token}`, + }, + body: JSON.stringify({ timestamp: Date.now() }), + signal: AbortSignal.timeout(5000), + }); + return resp.ok; + } catch { + return false; + } +} + +/** + * Unregister from Gateway. + * DELETE /devices/:id + */ +export async function unregisterFromGateway( + gatewayUrl: string, + deviceId: string, + token: string, +): Promise { + try { + await fetch(`${gatewayUrl}/devices/${deviceId}`, { + method: "DELETE", + headers: { "Authorization": `Bearer ${token}` }, + signal: AbortSignal.timeout(5000), + }); + } catch { + // Best effort + } +} + +/** + * Compute exponential backoff delay with jitter. + */ +export function backoffDelay(attempt: number, maxMs: number = 30000): number { + const base = Math.min(1000 * Math.pow(2, attempt), maxMs); + const jitter = Math.random() * base * 0.3; // 0-30% jitter + return base + jitter; +} + +/** + * Connection manager — handles the full lifecycle: + * check → register → heartbeat loop → disconnect → reconnect with backoff + */ +export class ConnectionManager { + private state: ConnectionState = { + status: "disconnected", + lastHeartbeat: 0, + reconnectAttempt: 0, + registeredAt: null, + }; + private heartbeatTimer: ReturnType | null = null; + private token: string = ""; + private stopped: boolean = false; + + constructor( + private config: DeviceConfig, + private onStateChange?: (state: ConnectionState) => void, + ) {} + + get currentState(): ConnectionState { return { ...this.state }; } + + /** Start the connection lifecycle (check → register → heartbeat) */ + async start(): Promise { + this.stopped = false; + await this.connect(); + } + + /** Stop everything */ + stop(): void { + this.stopped = true; + this.clearHeartbeat(); + if (this.token) { + unregisterFromGateway(this.config.gateway.url, this.config.identity.id, this.token) + .catch(() => {}); + } + this.setState({ status: "disconnected", registeredAt: null }); + } + + private async connect(): Promise { + if (this.stopped) return; + this.setState({ status: "connecting" }); + + const reachable = await checkConnectivity(this.config.gateway.url); + if (!reachable) { + console.warn("[Gateway] Unreachable, retrying..."); + this.setState({ status: "reconnecting" }); + this.scheduleReconnect(); + return; + } + + const result = await registerWithGateway(this.config.gateway.url, this.config); + if (!result) { + this.setState({ status: "reconnecting" }); + this.scheduleReconnect(); + return; + } + + this.token = result.token; + this.state.reconnectAttempt = 0; + this.setState({ status: "connected", registeredAt: Date.now() }); + this.startHeartbeat(); + } + + private startHeartbeat(): void { + this.clearHeartbeat(); + this.heartbeatTimer = setInterval(async () => { + if (this.stopped) return; + const ok = await sendHeartbeat( + this.config.gateway.url, + this.config.identity.id, + this.token, + ); + this.state.lastHeartbeat = Date.now(); + if (!ok && !this.stopped) { + console.warn("[Gateway] Heartbeat failed, reconnecting..."); + this.clearHeartbeat(); + this.setState({ status: "reconnecting" }); + this.scheduleReconnect(); + } + }, this.config.gateway.heartbeatInterval); + } + + private clearHeartbeat(): void { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + } + + private scheduleReconnect(): void { + if (this.stopped) return; + const delay = backoffDelay( + this.state.reconnectAttempt++, + this.config.gateway.maxReconnectDelay, + ); + console.log(`[Gateway] Reconnecting in ${Math.round(delay)}ms (attempt ${this.state.reconnectAttempt})`); + setTimeout(() => this.connect(), delay); + } + + private setState(partial: Partial): void { + this.state = { ...this.state, ...partial }; + this.onStateChange?.(this.currentState); + } +} diff --git a/packages/server/src/main.ts b/packages/server/src/main.ts index 894bddb..26109fd 100644 --- a/packages/server/src/main.ts +++ b/packages/server/src/main.ts @@ -8,6 +8,7 @@ import { bootstrap, type ServerOptions } from "./index.js"; import * as tools from "./tools.js"; +import { ConnectionManager } from "./connection.js"; import type { DeviceConfig } from "@yuzu-gca/shared"; import * as os from "node:os"; import { randomUUID } from "node:crypto"; @@ -59,6 +60,16 @@ console.log(` Port: ${port}`); bootstrap(opts, port).then(() => { console.log(`[Yuzu-GCA] Ready.`); + + // Start Gateway connection lifecycle + const connMgr = new ConnectionManager(config, (state) => { + console.log(`[Gateway] ${state.status} (attempt ${state.reconnectAttempt})`); + }); + connMgr.start(); + + // Graceful shutdown + process.on("SIGINT", () => { connMgr.stop(); process.exit(0); }); + process.on("SIGTERM", () => { connMgr.stop(); process.exit(0); }); }).catch(err => { console.error(`[Yuzu-GCA] Fatal:`, err); process.exit(1);