feat: C-015+P-006/008/009 Gateway连接生命周期
- 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优雅关闭
This commit is contained in:
231
packages/server/src/connection.ts
Normal file
231
packages/server/src/connection.ts
Normal file
@@ -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<boolean> {
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
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<typeof setInterval> | 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<void> {
|
||||
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<void> {
|
||||
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<ConnectionState>): void {
|
||||
this.state = { ...this.state, ...partial };
|
||||
this.onStateChange?.(this.currentState);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user