refactor(heartbeat): extract protocol-agnostic HeartbeatScheduler; MeshCore adopts (#3962)

docs/internal/dev-notes/REMEDIATION_EPIC.md

@@ -31,8 +31,8 @@ Each unchecked phase = one worktree → architect spec → implementation → re
 
 Phase 2 started 2026-07-07. Decisions carried forward from the Phase 0+1 interview: behavior-preserving; each numbered sub-task is one PR; full suite green before merge.
 
-- [ ] **2.1** — `MeshCoreManager implements ISourceManager`: one unified `sourceManagerRegistry` for all source types; `meshcoreRegistry.ts` reduced to a `@deprecated` shim (delete after one release). WP1 (interface+guards+config move) → WP2+WP3a (lifecycle+loop hardening) → WP3b (read-site migration) → WP4 (shim+tests+docs). See `task21_spec.md` and PR on branch `feature/3962-p2-meshcore-isourcemanager`.
-- [ ] **2.2a** — Heartbeat/status-probing: extract shared service parameterized by `ISourceManager` (plan §2.2 ¶1).
+- [x] **2.1** — `MeshCoreManager implements ISourceManager`: one unified `sourceManagerRegistry` for all source types; `meshcoreRegistry.ts` reduced to a `@deprecated` shim (delete after one release). WP1 (interface+guards+config move) → WP2+WP3a (lifecycle+loop hardening) → WP3b (read-site migration) → WP4 (shim+tests+docs). See `task21_spec.md` and PR on branch `feature/3962-p2-meshcore-isourcemanager`.
+- [x] **2.2a** — Heartbeat/status-probing: extract shared service parameterized by `ISourceManager` (plan §2.2 ¶1).
 - [ ] **2.2b** — Auto-announce: MeshCore + Meshtastic cycles → one `autoAnnounceService` with per-protocol adapters (plan §2.2 ¶2).
 - [ ] **2.2c** — Auto-responder: `checkAutoResponder` in both managers → shared service (plan §2.2 ¶3).
 - [ ] **2.2d** — Distance-delete scheduling: MeshCore's `DistanceDeleteScheduler` construction → unified ownership in `services/` (plan §2.2 ¶4).
@@ -61,3 +61,5 @@ Record per-phase: PR link, deviations from plan, follow-ups.
 - **1.4** — PR #3989 merged (count-based lint ratchet, eslint-baseline.json 410 files/2,515 violations; no-explicit-any/exhaustive-deps/prefer-const → error; raw-fetch ban in components/pages; CI lint now BLOCKING — closes the 0.5 deviation). Census correction: 727 pre-existing errors (468 config false-positives fixed at config level, not baselined).
 - **1.5** — PR #3990 merged (ok()/fail() helpers + 2 exemplar conversions + CLAUDE.md convention; ApiService does-not-unwrap constraint documented). No deviations.
 - **1.6** — PR TBD (schemaDrift.test.ts + schemaDrift.allowlist.ts; 15-entry allowlist; census confirmed 15 divergences — 10 onlyInBootstrap, 3 onlyInReplay, 2 sqlMismatch — matching the architect spec exactly). Finding flagged for Phase 3.3: 7 createIndexes()-only indexes (idx_messages_createdAt/fromNodeId/toNodeId, idx_nodes_updatedAt, idx_route_segments_distance/recordholder/timestamp) are silently lost on a replay-only fresh install; Phase 3.3 must add migrations creating each before deleting createIndexes(). No deviations.
+- **2.1** — PR #4004 merged (unified registry; ISourceManager on both managers; meshcoreRegistry → deprecated shim; ensureMeshCoreManagerStarted helper). Adversarial review (8 finders/11 verifiers) caught a confirmed by-id crash class (meshcore ids reaching Meshtastic-only routes via resolveSourceManager/security routes/purge-node) — fixed at shared-resolver altitude with regression tests. Backlog noted: disconnect() idempotency guard, notifyNewNodeDiscovered name fetch, start() contract hardening.
+- **2.2a** — PR TBD. Premise correction: no cross-protocol heartbeat duplication exists (Meshtastic = transport-level keepalive push; MQTT = none) — reduced scope to a HeartbeatScheduler leaf utility adopted by MeshCore only; reconnect machinery untouched (Phase 4.2). New files: `src/server/services/heartbeatScheduler.ts` (protocol-agnostic scheduler: interval/start/stop/in-flight-guard/pre+post-await connected-gate) + `heartbeatScheduler.test.ts` (8 fake-timer unit tests covering every gate). MeshCore adoption: `startHeartbeat`/`stopHeartbeat` delegate to per-instance HeartbeatScheduler; `runHeartbeatProbe` deleted; `heartbeatTimer`/`heartbeatProbeInFlight` fields removed; `heartbeatProbe`/`onHeartbeatOk` added as manager callbacks; `recordHeartbeatFailure`/`beginReconnect` chain unchanged. New integration test in meshcoreNativeBackend.test.ts: N consecutive probe failures → heartbeat_failed×N + state='reconnecting'.

src/server/meshcoreManager.ts

@@ -27,6 +27,7 @@ import {
 import meshcorePacketLogService from './services/meshcorePacketLogService.js';
 import { notificationService } from './services/notificationService.js';
 import { DistanceDeleteScheduler } from './services/distanceDeleteScheduler.js';
+import { HeartbeatScheduler } from './services/heartbeatScheduler.js';
 import type { DbMeshCorePacket } from '../db/repositories/meshcore.js';
 import type { ISourceManager, SourceStatus } from './sourceManagerRegistry.js';
 import { decodeMeshCorePacket } from '../utils/meshcorePacketDecode.js';
@@ -583,7 +584,7 @@ class MeshCoreManager extends EventEmitter implements ISourceManager {
 
   // Heartbeat / auto-reconnect state (native-backend only).
   private connectionState: MeshCoreConnectionState = 'disconnected';
-  private heartbeatTimer: NodeJS.Timeout | null = null;
+  private heartbeatScheduler: HeartbeatScheduler | null = null;
   private reconnectTimer: NodeJS.Timeout | null = null;
   private heartbeatConsecutiveFailures: number = 0;
 
@@ -604,7 +605,6 @@ class MeshCoreManager extends EventEmitter implements ISourceManager {
   private deviceTimeSyncTimer: NodeJS.Timeout | null = null;
   private static readonly DEVICE_TIME_SYNC_INTERVAL_MS = 6 * 60 * 60 * 1000; // 6h
   private heartbeatLastSuccessAt: number | null = null;
-  private heartbeatProbeInFlight: boolean = false;
   private reconnectAttempts: number = 0;
   private nextReconnectAt: number | null = null;
   private shouldReconnect: boolean = false;
@@ -5632,7 +5632,7 @@ class MeshCoreManager extends EventEmitter implements ISourceManager {
 
   /**
    * Start the heartbeat probe loop. Called automatically from connect() when
-   * the native backend is in use. Idempotent: if interval is 0 or a timer
+   * the native backend is in use. Idempotent: if interval is 0 or a scheduler
    * is already running, it's a no-op.
    */
   private startHeartbeat(): void {
@@ -5641,51 +5641,46 @@ class MeshCoreManager extends EventEmitter implements ISourceManager {
       // Heartbeat disabled — preserves prior behaviour.
       return;
     }
-    if (this.heartbeatTimer) return;
+    if (this.heartbeatScheduler?.running) return;
     this.shouldReconnect = true;
-    this.heartbeatTimer = setInterval(() => {
-      this.runHeartbeatProbe().catch((err) => {
-        logger.warn(`[MeshCore:${this.sourceId}] heartbeat probe threw: ${(err as Error).message}`);
-      });
-    }, intervalSecs * 1000);
+    this.heartbeatScheduler = new HeartbeatScheduler({
+      label: `MeshCore:${this.sourceId}`,
+      intervalMs: intervalSecs * 1000,
+      timeoutMs: this.config?.heartbeatTimeoutMs ?? 5000,
+      probe: (t) => this.heartbeatProbe(t),
+      isConnected: () => this.connectionState === 'connected' && !!this.nativeBackend,
+      onSuccess: (ms) => this.onHeartbeatOk(ms),
+      onFailure: (e) => this.recordHeartbeatFailure(e),
+    });
+    this.heartbeatScheduler.start();
     logger.info(`[MeshCore:${this.sourceId}] Heartbeat started (every ${intervalSecs}s)`);
   }
 
   private stopHeartbeat(): void {
-    if (this.heartbeatTimer) {
-      clearInterval(this.heartbeatTimer);
-      this.heartbeatTimer = null;
-    }
-    this.heartbeatProbeInFlight = false;
+    this.heartbeatScheduler?.stop();
+    this.heartbeatScheduler = null;
   }
 
-  private async runHeartbeatProbe(): Promise<void> {
-    if (this.heartbeatProbeInFlight) return;
-    if (this.connectionState !== 'connected') return;
-    if (!this.nativeBackend) return;
-
-    this.heartbeatProbeInFlight = true;
-    const timeoutMs = this.config?.heartbeatTimeoutMs ?? 5000;
-    const probeStartedAt = Date.now();
-    try {
-      const response = await this.nativeBackend.sendCommand('get_device_time', {}, timeoutMs);
-      // Probe arrived after a teardown — drop the result.
-      if (this.connectionState !== 'connected') return;
-
-      if (response.success) {
-        const latencyMs = Date.now() - probeStartedAt;
-        this.heartbeatConsecutiveFailures = 0;
-        this.heartbeatLastSuccessAt = Date.now();
-        this.emit('heartbeat_ok', { sourceId: this.sourceId, latencyMs });
-      } else {
-        this.recordHeartbeatFailure(new Error(response.error ?? 'probe failed'));
-      }
-    } catch (err) {
-      if (this.connectionState !== 'connected') return;
-      this.recordHeartbeatFailure(err as Error);
-    } finally {
-      this.heartbeatProbeInFlight = false;
+  /**
+   * Wire-level probe operation: sends a cheap `get_device_time` RTC read to
+   * the native backend and returns `true` on success or throws on failure, so
+   * the {@link HeartbeatScheduler} can route the result to the appropriate
+   * callback without knowing about MeshCore command vocabulary.
+   */
+  private async heartbeatProbe(timeoutMs: number): Promise<boolean> {
+    if (!this.nativeBackend) {
+      throw new Error('no native backend');
     }
+    const response = await this.nativeBackend.sendCommand('get_device_time', {}, timeoutMs);
+    if (response.success) return true;
+    throw new Error(response.error ?? 'probe failed');
+  }
+
+  /** Called by the scheduler on a successful probe. */
+  private onHeartbeatOk(latencyMs: number): void {
+    this.heartbeatConsecutiveFailures = 0;
+    this.heartbeatLastSuccessAt = Date.now();
+    this.emit('heartbeat_ok', { sourceId: this.sourceId, latencyMs });
   }
 
   private recordHeartbeatFailure(err: Error): void {

src/server/meshcoreNativeBackend.test.ts

@@ -1321,4 +1321,43 @@ describe('MeshCoreManager heartbeat (native backend)', () => {
     await mgr.disconnect();
     expect(mgr.getHeartbeatStatus().state).toBe('disconnected');
   });
+
+  it('N consecutive probe failures → heartbeat_failed emitted N times and beginReconnect reached', async () => {
+    // Install a mock where getDeviceTime() always throws so every heartbeat
+    // probe returns { success: false } via sendCommand's catch wrapper.
+    class FailingConnection extends MockConnection {
+      async getDeviceTime(): Promise<{ epochSecs: number } | null> {
+        throw new Error('simulated device timeout');
+      }
+    }
+    const lastRef = installMockModule(FailingConnection as unknown as typeof MockConnection) as any;
+
+    const mgr = new MeshCoreManager('hb-fail-src');
+    const ok = await mgr.connect({
+      connectionType: ConnectionType.SERIAL,
+      serialPort: '/dev/ttyUSB0',
+      firmwareType: 'companion',
+      heartbeatIntervalSeconds: 1,    // 1s interval
+      heartbeatTimeoutMs: 200,
+      heartbeatMaxFailures: 3,
+    });
+    expect(ok).toBe(true);
+    // Verify we actually got a FailingConnection instance
+    expect(lastRef.current).toBeInstanceOf(FailingConnection);
+
+    const failedEvents: any[] = [];
+    mgr.on('heartbeat_failed', (e) => failedEvents.push(e));
+
+    // Wait for 3 probes to fire and fail (3× 1s interval + a little slack).
+    await new Promise((r) => setTimeout(r, 3500));
+
+    // All 3 failures should have been emitted.
+    expect(failedEvents.length).toBeGreaterThanOrEqual(3);
+    expect(failedEvents[0]).toMatchObject({ sourceId: 'hb-fail-src', consecutiveFailures: 1 });
+    expect(failedEvents[1]).toMatchObject({ sourceId: 'hb-fail-src', consecutiveFailures: 2 });
+    expect(failedEvents[2]).toMatchObject({ sourceId: 'hb-fail-src', consecutiveFailures: 3 });
+
+    // After heartbeatMaxFailures, beginReconnect() fires and transitions to 'reconnecting'.
+    expect(mgr.getHeartbeatStatus().state).toBe('reconnecting');
+  });
 });