Bridge idle watchdog + forced reconnect (v0.21.1) #44

Merged
kamilkrzywda merged 1 commits from fix/bridge-idle-reconnect into main 2026-09-26 08:53:06 +02:00
5 changed files with 145 additions and 2 deletions
Showing only changes of commit af1c77fd54 - Show all commits
+1 -1
View File
@@ -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
+6
View File
@@ -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 <name|#tag>` 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.
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "mesh-bot",
"version": "0.21.0",
"version": "0.21.1",
"description": "",
"author": "",
"private": true,
+102
View File
@@ -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();
+35
View File
@@ -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<void> {