diff --git a/AGENTS.md b/AGENTS.md index 573b995..e9a8d8a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,7 +12,7 @@ ## Verification 1. **Local checks** — `npm run build` (nest build) and `npm test` (vitest; no e2e suite yet). Both must pass before a PR is opened. -2. **Docker** — `docker compose build`, then after `docker compose up -d`: `curl -s -H "Host: bot.dev.kkhost.pl" http://localhost/api/status` returns JSON (`{ connected, myNodeNum, host, port }`). `"connected":false` is normal while the bridge/device is down — the service retries every 5 s. +2. **Docker** — `docker compose build`, then after `docker compose up -d`: `curl -s -H "Host: bot.dev.kkhost.pl" http://localhost/api/status` returns JSON (`{ connected, myNodeNum, host, port }`). `"connected":false` is normal while the bridge/device is down — the service retries every 5 s. An idle watchdog also force-reconnects an ESTABLISHED-but-silent connection (zombie bridge: outbound works, inbound dead) when no packet arrives from the bridge for `MESH_IDLE_TIMEOUT_MS` ms (default 600000 = 10 min; checked every 30 s), and the socket has TCP keepalive enabled (30 s). ## API diff --git a/CHANGELOG.md b/CHANGELOG.md index 8651679..5d58506 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## v0.21.1 — 2026-09-26 06:47 + +- Bridge zombie-connection fix (production incident: outbound sends kept working ~10 h while inbound packets stopped entirely and `/api/status` stayed `connected:true`). New **idle watchdog**: the service tracks `lastPacketAt`, updated on EVERY chunk received from the bridge in the lowest-level socket `'data'` path (any frame kind, decodable or not) and reset when a connection successfully starts. A 30 s interval checks: if connected AND no inbound traffic for `MESH_IDLE_TIMEOUT_MS` (new env, default **600000** = 10 min), it logs WARN ("no bridge traffic for X ms — forcing reconnect") and destroys the socket — the EXISTING connect/retry loop (`MESH_RECONNECT_MS`) then re-establishes with a fresh sync. No changes to connect-error retry/backoff behavior; this only adds detection for established-but-silent connections. +- TCP keepalive on the bridge socket: `socket.setKeepAlive(true, 30_000)` — cheap insurance against dead peers at the TCP level, complementing the application-level watchdog. +- Tests: 165 → 169 — silence past an overridden timeout forces disconnect + reconnect (fake bridge), inbound traffic keeps resetting the idle clock so no false trigger, default timeout does not fire in a test window, and `setKeepAlive(true, 30_000)` is asserted on the socket. + ## v0.21.0 — 2026-09-25 19:59 - `#rss get ` now replies with **3–5 SEPARATE mesh messages** instead of one small summary (owner request: the single-message summary was too small). Each item is its own packet, so no fragmentation — and each rides the trigger's channel threaded as a quote of the command packet. diff --git a/package.json b/package.json index feb67d9..4b0b281 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "mesh-bot", - "version": "0.21.0", + "version": "0.21.1", "description": "", "author": "", "private": true, diff --git a/src/mesh/mesh.service.spec.ts b/src/mesh/mesh.service.spec.ts index 7e78fdb..be0cffa 100644 --- a/src/mesh/mesh.service.spec.ts +++ b/src/mesh/mesh.service.spec.ts @@ -1,6 +1,7 @@ import { create, fromBinary, toBinary } from '@bufbuild/protobuf'; import * as P from '@meshtastic/protobufs'; import { BadRequestException, NotFoundException } from '@nestjs/common'; +import * as net from 'node:net'; import { DatabaseSync } from 'node:sqlite'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { BROADCAST_ADDR, extractFrames, frame } from './codec.js'; @@ -249,6 +250,107 @@ describe('MeshService connect', () => { }); }); +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe('MeshService idle watchdog', () => { + it('forces a reconnect when an established connection goes silent past MESH_IDLE_TIMEOUT_MS', async () => { + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable + process.env.MESH_IDLE_TIMEOUT_MS = '100'; // override — the default (600000) would never fire in a test + const svc = new MeshService(); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + // zombie case: connected but no inbound traffic at all — wait past the timeout, then drive a watchdog tick + await sleep(250); + svc.checkIdle(); + + // socket torn down → existing connect/retry loop re-establishes with a fresh session + await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected); + } finally { + delete process.env.MESH_RECONNECT_MS; + delete process.env.MESH_IDLE_TIMEOUT_MS; + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('inbound traffic keeps resetting the idle clock — no false trigger while packets arrive', async () => { + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + process.env.MESH_IDLE_TIMEOUT_MS = '100'; + const svc = new MeshService(); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + // steady drip of inbound frames for well over the timeout in wall time — each chunk resets lastPacketAt + for (let i = 0; i < 5; i++) { + await sleep(60); + bridge.sockets[0].write(nodeInfoFrame(42, kamilNodeInfo())); + } + + svc.checkIdle(); // wall time is past the timeout, but traffic kept resetting it + await sleep(100); // enough for a destroy to surface as 'close' (connected would flip false) + expect(svc.status().connected).toBe(true); + expect(bridge.sockets.length).toBe(1); + } finally { + delete process.env.MESH_IDLE_TIMEOUT_MS; + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('does not trigger under the default timeout (MESH_IDLE_TIMEOUT_MS unset)', async () => { + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + delete process.env.MESH_IDLE_TIMEOUT_MS; // default 600000 — a test can't wait that long, so silence must NOT fire + const svc = new MeshService(); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + await sleep(250); // silent, but far under the default budget + svc.checkIdle(); + + expect(svc.status().connected).toBe(true); + expect(bridge.sockets.length).toBe(1); + } finally { + delete process.env.MESH_IDLE_TIMEOUT_MS; + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('enables TCP keepalive on the bridge socket', async () => { + const setKeepAlive = vi.spyOn(net.Socket.prototype, 'setKeepAlive'); + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + const svc = new MeshService(); + try { + svc.onModuleInit(); + await waitFor(() => svc.status().connected); + expect(setKeepAlive).toHaveBeenCalledWith(true, 30_000); + } finally { + setKeepAlive.mockRestore(); + svc.onModuleDestroy(); + bridge.close(); + } + }); +}); + describe('MeshService send', () => { it('sends on the test channel (index 3) by default, never primary', async () => { const bridge = await startBridge(); diff --git a/src/mesh/mesh.service.ts b/src/mesh/mesh.service.ts index 2b76794..704f829 100644 --- a/src/mesh/mesh.service.ts +++ b/src/mesh/mesh.service.ts @@ -81,6 +81,8 @@ export interface TelemetryRecord { const HISTORY_LIMIT = 100; /** Rows kept in SQLite across restarts; older rows are pruned on every poller tick. */ const DB_RETENTION = 1000; +// how often the idle watchdog checks for a silent-but-connected bridge (ms) +const IDLE_CHECK_MS = 30_000; // one row per MessageRecord, keyed by packet id (tx ids and injected ids are random uint32, rx uses the wire id) export const MESSAGES_DDL = `CREATE TABLE IF NOT EXISTS messages( @@ -208,6 +210,11 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { ); // how often the poller re-upserts the history buffer into SQLite (ms) private readonly pollMs = parseInt(process.env.MESH_POLL_MS ?? '1000', 10); + // idle watchdog: force a reconnect when an ESTABLISHED connection goes silent — the zombie-bridge case where outbound sends keep working but inbound packets stop entirely. Read at construction like the other env config. + private readonly idleTimeoutMs = parseInt( + process.env.MESH_IDLE_TIMEOUT_MS ?? '600000', + 10, + ); /** SQLite database backing the message history; null = persistence disabled (tests construct without one). */ private db: DatabaseSync | null = null; @@ -255,6 +262,9 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { // telemetry history (newest kept, capped at HISTORY_LIMIT) private telemetryHistory: TelemetryRecord[] = []; private reconnectTimer: NodeJS.Timeout | null = null; + /** epoch ms of the last inbound chunk from the bridge (any frame kind); reset on connect — idle watchdog baseline */ + private lastPacketAt = Date.now(); + private idleTimer: NodeJS.Timeout | null = null; private shuttingDown = false; onModuleInit(): void { @@ -264,6 +274,8 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { // ponytail: full re-upsert of the ≤100-row buffer every tick — trivial load, zero bookkeeping this.pollTimer = setInterval(() => this.flushHistory(), this.pollMs); } + // idle watchdog: catches established-but-silent connections (zombie bridge) that the connect/retry loop never sees + this.idleTimer = setInterval(() => this.checkIdle(), IDLE_CHECK_MS); this.connect(); } @@ -271,6 +283,7 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { this.shuttingDown = true; if (this.reconnectTimer) clearTimeout(this.reconnectTimer); if (this.pollTimer) clearInterval(this.pollTimer); + if (this.idleTimer) clearInterval(this.idleTimer); this.socket?.destroy(); this.db?.close(); } @@ -641,9 +654,13 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { private connect(): void { if (this.shuttingDown) return; const socket = net.createConnection({ host: this.host, port: this.port }); + // TCP-level keepalive: cheap insurance against a dead peer that the idle watchdog (application level) complements + socket.setKeepAlive(true, 30_000); this.socket = socket; socket.on('connect', () => { this.connected = true; + // fresh connection — reset the idle clock so it can't self-trigger before any traffic arrives + this.lastPacketAt = Date.now(); // fresh sync on every (re)connect — drop stale channels/nodes/device info/telemetry from the old session; // message history is NOT reset: it's an append-only log backed by SQLite and survives reconnects this.channelMap = new Map(); @@ -663,6 +680,8 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { this.logger.log(`requested initial sync (want_config_id=${nonce})`); }); socket.on('data', (chunk) => { + // lowest-level receive path: every inbound chunk counts as bridge traffic, decodable or not — this is what the idle watchdog watches + this.lastPacketAt = Date.now(); const merged = new Uint8Array(this.buffer.length + chunk.length); merged.set(this.buffer, 0); merged.set(chunk, this.buffer.length); @@ -694,6 +713,22 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { this.reconnectTimer = setTimeout(() => this.connect(), this.reconnectMs); } + /** + * Idle watchdog tick: force a reconnect when an ESTABLISHED connection has seen no inbound bridge + * traffic for MESH_IDLE_TIMEOUT_MS — the zombie case where outbound sends keep working but inbound + * packets stop. Destroying the socket routes through 'close' → onDown → scheduleReconnect, so the + * existing connect/retry loop re-establishes with a fresh sync; no new retry machinery here. + * Public so tests can drive a tick without waiting for the real interval (like flushHistory()). + */ + checkIdle(): void { + if (!this.connected) return; // disconnected state is owned by the connect/retry loop + const idle = Date.now() - this.lastPacketAt; + if (idle > this.idleTimeoutMs) { + this.logger.warn(`no bridge traffic for ${idle} ms — forcing reconnect`); + this.socket?.destroy(); + } + } + private async handleFromRadio( fr: import('@meshtastic/protobufs').Mesh.FromRadio, ): Promise {