fix(poller): stop the queue outrunning the workers, and cap job retention (v0.23.17-beta)

Addresses #114, where a ~60 device fleet saw devices stop reporting entirely
while others crawled, new devices never started polling, and a manual Sync
worked exactly once. All of it is one failure with two compounding causes.

**The queue grew faster than it drained.** Worker concurrency was hardcoded to 3
with no per-job timeout, and the scheduler enqueued a fast poll per device every
30 seconds regardless of whether the previous one had finished. Measured against
a live fleet a fast poll took ~1s at the median but 13s at p90, so three workers
sustained about 0.87 jobs/sec against an arrival rate of 2.0/sec at sixty
devices. The difference — roughly 98,000 jobs a day — accumulated in an unbounded
FIFO, and devices at the back of it were simply never polled. That is why Sync
appeared to work once and then stop: the Sync route runs inline and bypasses the
queue entirely, so it was the only thing still reaching those devices.

Three changes: concurrency is configurable and defaults to 12; every job runs
under a timeout so one unreachable device cannot hold a worker while the fleet
starves; and a device with a poll already queued or running is not enqueued
again. That last one is the important one — it makes a backlog impossible rather
than merely less likely. The claim is a Redis key with a TTL, so a worker that
dies frees its device on the next cycle instead of wedging it forever.

**Finished jobs were never removed.** BullMQ keeps them unless told otherwise.
A four-device development instance had accumulated 1,412,935 completed and
230,707 failed jobs across the three queues — 2,570,474 Redis keys and 1.16 GB,
against a Redis with no maxmemory configured. Left alone it grows until the host
dies. Retention is now bounded by age and count, and a background pass on
startup trims history from before the limits existed.

The two faults fed each other, which is measurable: after trimming 2.16M jobs
the mean fast poll on the same fleet fell from 3,448ms to 602ms and the slowest
from 16,095ms to 1,205ms. Multi-million-element sorted sets were slowing every
queue operation, which reduced throughput, which grew the backlog further.

**The reporter's follow-up is the real lesson** and is addressed too: there was
no way to tell whether a device had been missed, was unresponsive, or was merely
slow, and no view of queue depth or per-poll cost — so polling frequency was
being chosen by guesswork. device_poll_stats now records attempt and success
separately per device and kind, with duration and last error; their difference
is precisely the missing distinction. GET /api/system/poller reports queue
depths, worker settings, measured poll cost, and a headroom figure that states
whether the configured cadence is achievable at the current fleet size. Below
1.0 the backlog grows every cycle.

Verified on the live instance: 2,570,474 keys reduced to 2,336, memory 1.16 GB
to 32.25 MB, queues drained, per-device stats recording, no stranded in-flight
claims, and headroom reported at 149x for four devices — the same arithmetic
that would have read 0.43 for sixty devices before this change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

README.md

@@ -6,7 +6,7 @@
 </p>
 
 <p align="center">
-  <img src="https://img.shields.io/badge/version-0.23.16_Beta-blue" alt="Version" />
+  <img src="https://img.shields.io/badge/version-0.23.17_Beta-blue" alt="Version" />
   <img src="https://img.shields.io/badge/license-AGPLv3-blue" alt="License" />
   <img src="https://img.shields.io/badge/docker-compose-2496ED?logo=docker&logoColor=white" alt="Docker" />
   <img src="https://img.shields.io/badge/TypeScript-5.3-3178C6?logo=typescript&logoColor=white" alt="TypeScript" />

backend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "mikrotik-manager-backend",
-  "version": "0.23.16-beta",
+  "version": "0.23.17-beta",
   "description": "Mikrotik Network Management Platform - Backend API",
   "main": "dist/index.js",
   "scripts": {

backend/src/db/migrate.ts

@@ -792,6 +792,26 @@ CREATE TABLE IF NOT EXISTS lte_observed_bands (
   observations   INTEGER NOT NULL DEFAULT 1,
   PRIMARY KEY (device_id, interface_name, band)
 );
+
+-- Per-device polling telemetry (#114).
+--
+-- last_attempt_at and last_success_at are deliberately separate columns rather
+-- than one "last polled" field. Their difference is the only thing that
+-- distinguishes a device the poller never reached from one it reached and got no
+-- answer from — a distinction the product could not previously make, which left
+-- operators unable to tell a monitoring failure from a device failure.
+CREATE TABLE IF NOT EXISTS device_poll_stats (
+  device_id        INTEGER NOT NULL REFERENCES devices(id) ON DELETE CASCADE,
+  kind             VARCHAR(16) NOT NULL,
+  last_attempt_at  TIMESTAMPTZ,
+  last_success_at  TIMESTAMPTZ,
+  last_duration_ms INTEGER,
+  last_error       TEXT,
+  attempts         BIGINT NOT NULL DEFAULT 0,
+  failures         BIGINT NOT NULL DEFAULT 0,
+  PRIMARY KEY (device_id, kind)
+);
+CREATE INDEX IF NOT EXISTS idx_poll_stats_attempt ON device_poll_stats(last_attempt_at DESC);
 `;
 
 const DEFAULT_SETTINGS = [

backend/src/index.ts

@@ -61,7 +61,7 @@ import wirelessRoutes from './routes/wireless';
 import networkServicesRoutes from './routes/networkServices';
 import trafficAnalyticsRoutes from './routes/trafficAnalytics';
 import credentialPresetsRoutes from './routes/credentialPresets';
-import systemRoutes from './routes/system';
+import systemRoutes, { setPollerService as setSystemPoller } from './routes/system';
 import { auditMiddleware } from './middleware/auditMiddleware';
 
 // ─── Secret hygiene ───────────────────────────────────────────────────────────
@@ -373,10 +373,17 @@ async function start(): Promise<void> {
   const pollerService = new PollerService();
   pollerService.setSocketServer(io);
   setDevicesPoller(pollerService);
+  setSystemPoller(pollerService);
   setBulkAddPollerService(pollerService);
   setTopologyPoller(pollerService);
   setClientsPoller(pollerService);
   await pollerService.start();
+
+  // Trim job history left over from before retention limits existed. Runs in the
+  // background: an installation with millions of retained jobs would otherwise
+  // hold up startup, and nothing depends on it having finished.
+  void pollerService.cleanupJobHistory().catch((e) =>
+    console.error('[Poller] Job history cleanup failed:', e));
   await startBulkAddWorker();
 
   // NetFlow/IPFIX collector (binds its UDP socket only when netflow_enabled)

backend/src/routes/system.ts

@@ -4,10 +4,17 @@ import { readFileSync } from 'fs';
 import { join } from 'path';
 import { requireAuth } from '../middleware/auth';
 import { query } from '../config/database';
+import type { PollerService } from '../services/PollerService';
 
 const router = Router();
 router.use(requireAuth);
 
+// Injected from index.ts, matching how the devices routes receive it.
+let pollerService: PollerService | null = null;
+export function setPollerService(p: PollerService): void {
+  pollerService = p;
+}
+
 // The container runs `node dist/index.js` (not via npm), so npm_package_version
 // is unset there — read the bundled package.json instead, falling back to it.
 // Without this, current stays "0.0.0" and every release looks like an update.
@@ -46,6 +53,41 @@ function isNewer(latest: number[], current: number[]): boolean {
 
 // GET /api/system/version-check
 // Returns { current, latest, update_available } — cached 24 h in app_settings.
+/**
+ * GET /api/system/poller — is the poller keeping up?
+ *
+ * A fleet that outruns its workers looks, from the outside, exactly like a fleet
+ * of misbehaving devices: data goes stale and nothing says why. `headroom` below
+ * 1.0 means the backlog grows every cycle and some devices will stop reporting
+ * altogether (#114).
+ */
+router.get('/poller', async (_req: Request, res: Response) => {
+  try {
+    if (!pollerService) return res.status(503).json({ error: 'Poller not started' });
+    const health = await pollerService.getPollerHealth();
+
+    // Devices the poller has not managed to reach lately, worst first. This is
+    // the list that answers "which ones is it actually missing?".
+    const stale = await query(
+      `SELECT d.id, d.name, d.status, s.kind,
+              s.last_attempt_at, s.last_success_at, s.last_duration_ms, s.last_error,
+              s.attempts, s.failures,
+              EXTRACT(EPOCH FROM (NOW() - s.last_success_at))::int AS seconds_since_success
+         FROM devices d
+         JOIN device_poll_stats s ON s.device_id = d.id AND s.kind = 'fast'
+        WHERE d.status != 'disabled'
+          AND (s.last_success_at IS NULL OR s.last_success_at < NOW() - INTERVAL '3 minutes')
+        ORDER BY s.last_success_at ASC NULLS FIRST
+        LIMIT 50`
+    );
+
+    res.json({ ...health, stale_devices: stale });
+  } catch (error) {
+    console.error('Error building poller health:', error);
+    res.status(500).json({ error: 'Failed to read poller health' });
+  }
+});
+
 router.get('/version-check', async (_req: Request, res: Response) => {
   try {
     // Check cache

backend/src/services/PollerService.ts

@@ -9,6 +9,48 @@ import { alertService } from './AlertService';
 import { runConfigHealth } from './changeGuard/configHealth';
 import type { GuardDevice } from './changeGuard/ChangeGuard';
 
+// ─── Tuning ───────────────────────────────────────────────────────────────────
+
+/** How long finished jobs are kept, and how many, per queue. */
+const JOB_RETENTION_SEC = Number(process.env.POLLER_JOB_RETENTION_SEC || 3600);
+const JOB_RETENTION_COUNT = Number(process.env.POLLER_JOB_RETENTION_COUNT || 5000);
+const CLEANUP_BATCH = 5000;
+const CLEANUP_MAX_PASSES = 200;
+
+/**
+ * Worker concurrency, per queue.
+ *
+ * Was hardcoded at 3. Measured against a live fleet a fast poll takes ~1s at the
+ * median but 13s at p90, so three workers sustain roughly 0.9 jobs/sec — while
+ * sixty devices on a 30-second cadence arrive at 2.0/sec. Any fleet past about
+ * twenty-five devices outruns the workers permanently, and the backlog is what
+ * makes devices appear to stop reporting (#114).
+ */
+const POLL_CONCURRENCY = Math.max(1, Number(process.env.POLLER_CONCURRENCY || 12));
+
+/** Scheduler tick — one fast poll per device per interval. */
+const POLL_INTERVAL_MS = Math.max(10_000, Number(process.env.POLLER_INTERVAL_MS || 30_000));
+
+/**
+ * Hard ceiling on a single poll.
+ *
+ * Without one, an unreachable device holds a worker for as long as its sockets
+ * take to give up, and a handful of them starve every other device in the fleet.
+ */
+const JOB_TIMEOUT_MS = Math.max(5_000, Number(process.env.POLLER_JOB_TIMEOUT_MS || 45_000));
+
+/** Fails the job rather than letting it occupy a worker indefinitely. */
+function withTimeout<T>(work: Promise<T>, ms: number, label: string): Promise<T> {
+  let timer: ReturnType<typeof setTimeout>;
+  return Promise.race([
+    work.finally(() => clearTimeout(timer)),
+    new Promise<never>((_, reject) => {
+      timer = setTimeout(() => reject(new Error(`${label} exceeded ${ms}ms`)), ms);
+    }),
+  ]);
+}
+
+
 interface PollJob {
   deviceId: number;
   type: 'fast' | 'slow' | 'logs' | 'full' | 'macscan' | 'spectral' | 'apscan' | 'configsnap' | 'confighealth';
@@ -29,9 +71,55 @@ export class PollerService {
     const conn2 = createRedisConnection();
     const conn3 = createRedisConnection();
 
-    this.fastQueue = new Queue('poll-fast', { connection: conn1 });
+    // Retention is not optional. BullMQ keeps every finished job unless told
-    this.slowQueue = new Queue('poll-slow', { connection: conn2 });
+    // otherwise, and a poller enqueues forever — a four-device fleet had
-    this.logsQueue = new Queue('poll-logs', { connection: conn3 });
+    // accumulated 1.4M completed and 231k failed jobs, 2.5M Redis keys and
+    // 1.16 GB, against a Redis with no maxmemory. Left alone it grows until the
+    // host dies, and multi-million-element sorted sets slow every queue
+    // operation long before that, which starves the very polling this serves.
+    const defaultJobOptions = {
+      removeOnComplete: { age: JOB_RETENTION_SEC, count: JOB_RETENTION_COUNT },
+      removeOnFail:     { age: JOB_RETENTION_SEC, count: JOB_RETENTION_COUNT },
+    };
+
+    this.fastQueue = new Queue('poll-fast', { connection: conn1, defaultJobOptions });
+    this.slowQueue = new Queue('poll-slow', { connection: conn2, defaultJobOptions });
+    this.logsQueue = new Queue('poll-logs', { connection: conn3, defaultJobOptions });
+  }
+
+  /** Queues in one place, so health reporting and cleanup cannot drift apart. */
+  private get queues(): { name: string; queue: Queue }[] {
+    return [
+      { name: 'poll-fast', queue: this.fastQueue },
+      { name: 'poll-slow', queue: this.slowQueue },
+      { name: 'poll-logs', queue: this.logsQueue },
+    ];
+  }
+
+  /**
+   * Trim finished jobs left over from before retention existed.
+   *
+   * `defaultJobOptions` only governs jobs added from now on, so an installation
+   * that has been running for weeks keeps its entire backlog until something
+   * removes it. Bounded per pass and run in the background: cleaning millions of
+   * keys in one call would block Redis for everything else.
+   */
+  async cleanupJobHistory(): Promise<void> {
+    for (const { name, queue } of this.queues) {
+      for (const state of ['completed', 'failed'] as const) {
+        let removed = 0;
+        try {
+          for (let pass = 0; pass < CLEANUP_MAX_PASSES; pass++) {
+            const ids = await queue.clean(JOB_RETENTION_SEC * 1000, CLEANUP_BATCH, state);
+            removed += ids.length;
+            if (ids.length < CLEANUP_BATCH) break;
+          }
+        } catch (e) {
+          console.error(`[Poller] History cleanup failed for ${name}/${state}:`, (e as Error).message);
+        }
+        if (removed > 0) console.log(`[Poller] Trimmed ${removed} ${state} jobs from ${name}`);
+      }
+    }
   }
 
   setSocketServer(io: SocketServer): void {
@@ -58,6 +146,14 @@ export class PollerService {
 
   async scheduleDeviceSync(deviceId: number, type: PollJob['type'] = 'full'): Promise<void> {
     const jobData: PollJob = { deviceId, type };
+
+    // A poll already queued or running for this device makes another pointless.
+    // The TTL is generous relative to the job timeout so a crashed worker frees
+    // the device on the following cycle rather than stranding it.
+    if (type !== 'full' && !(await this.claimPoll(type, deviceId, Math.ceil(JOB_TIMEOUT_MS / 1000) * 3))) {
+      return;
+    }
+
     if (type === 'full') {
       await this.fastQueue.add('device-full-sync', jobData, {
         attempts: 3,
@@ -83,10 +179,9 @@ export class PollerService {
   }
 
   private startScheduler(): void {
-    // Schedule polls every 30 seconds
     this.schedulerInterval = setInterval(async () => {
       await this.schedulePollCycle();
-    }, 30_000);
+    }, POLL_INTERVAL_MS);
 
     // Also run immediately
     setTimeout(() => this.schedulePollCycle(), 5000);
@@ -474,6 +569,123 @@ export class PollerService {
     }
   }
 
+  /**
+   * Claim a device for one kind of poll, refusing if one is already outstanding.
+   *
+   * The scheduler runs on a fixed 30-second timer and used to enqueue
+   * unconditionally, so whenever the workers could not keep up the queue grew by
+   * the difference on every tick — forever. Devices at the back of it simply
+   * stopped being polled, which is what "last seen isn't refreshing at all"
+   * actually was (#114).
+   *
+   * A claim is a Redis SET NX with a TTL, so a worker that dies mid-poll cannot
+   * wedge a device permanently: the claim lapses and the next cycle re-enqueues.
+   * If Redis is unreachable we enqueue anyway — degrading to the old behaviour
+   * beats not polling at all.
+   */
+  /**
+   * Record what happened on a poll, per device and kind.
+   *
+   * The gap this closes is the one the reporter named: there was no way to tell
+   * whether a device was missed, unreachable, or simply slow. Attempt and
+   * success are stored separately so their difference is the answer, and the
+   * duration makes it possible to choose a polling interval from evidence
+   * instead of guesswork (#114).
+   */
+  private async recordPollOutcome(
+    deviceId: number, kind: string, durationMs: number, error: string | null,
+  ): Promise<void> {
+    try {
+      await query(
+        `INSERT INTO device_poll_stats (device_id, kind, last_attempt_at, last_success_at,
+                                        last_duration_ms, last_error, attempts, failures)
+         VALUES ($1, $2, NOW(), CASE WHEN $4::text IS NULL THEN NOW() END, $3, $4, 1,
+                 CASE WHEN $4::text IS NULL THEN 0 ELSE 1 END)
+         ON CONFLICT (device_id, kind) DO UPDATE SET
+           last_attempt_at  = NOW(),
+           last_success_at  = CASE WHEN $4::text IS NULL THEN NOW()
+                                   ELSE device_poll_stats.last_success_at END,
+           last_duration_ms = $3,
+           last_error       = $4,
+           attempts         = device_poll_stats.attempts + 1,
+           failures         = device_poll_stats.failures + CASE WHEN $4::text IS NULL THEN 0 ELSE 1 END`,
+        [deviceId, kind, durationMs, error]
+      );
+    } catch { /* never let bookkeeping break a poll */ }
+  }
+
+  /**
+   * Queue depth, worker settings and per-device freshness in one place.
+   *
+   * Exposed because a fleet outrunning its workers is invisible from the outside
+   * — the symptom is stale data, which looks like a device problem rather than a
+   * capacity problem. The headroom figure states plainly whether the configured
+   * cadence is achievable at the current fleet size.
+   */
+  async getPollerHealth(): Promise<Record<string, unknown>> {
+    const queues: Record<string, unknown>[] = [];
+    for (const { name, queue } of this.queues) {
+      try {
+        const c = await queue.getJobCounts('waiting', 'active', 'delayed', 'failed', 'completed');
+        queues.push({ name, ...c });
+      } catch (e) {
+        queues.push({ name, error: (e as Error).message });
+      }
+    }
+
+    const [{ count: deviceCount } = { count: 0 }] = await query<{ count: number }>(
+      `SELECT COUNT(*)::int AS count FROM devices WHERE status != 'disabled'`
+    );
+    const [timing] = await query<{ avg_ms: number | null; p90_ms: number | null }>(
+      `SELECT AVG(last_duration_ms)::int AS avg_ms,
+              (PERCENTILE_CONT(0.9) WITHIN GROUP (ORDER BY last_duration_ms))::int AS p90_ms
+         FROM device_poll_stats WHERE kind = 'fast' AND last_duration_ms IS NOT NULL`
+    );
+
+    // Can the workers keep up? Arrival is one fast poll per device per cycle.
+    const avgSec = (timing?.avg_ms || 0) / 1000;
+    const arrivalPerSec = deviceCount / (POLL_INTERVAL_MS / 1000);
+    const servicePerSec = avgSec > 0 ? POLL_CONCURRENCY / avgSec : null;
+
+    return {
+      queues,
+      workers: {
+        concurrency: POLL_CONCURRENCY,
+        job_timeout_ms: JOB_TIMEOUT_MS,
+        poll_interval_ms: POLL_INTERVAL_MS,
+      },
+      capacity: {
+        devices: deviceCount,
+        avg_fast_poll_ms: timing?.avg_ms ?? null,
+        p90_fast_poll_ms: timing?.p90_ms ?? null,
+        arrival_per_sec: Math.round(arrivalPerSec * 100) / 100,
+        service_per_sec: servicePerSec == null ? null : Math.round(servicePerSec * 100) / 100,
+        // Below 1.0 the backlog grows every cycle and devices go stale.
+        headroom: servicePerSec == null || arrivalPerSec === 0
+          ? null
+          : Math.round((servicePerSec / arrivalPerSec) * 100) / 100,
+      },
+      retention: { max_age_sec: JOB_RETENTION_SEC, max_count: JOB_RETENTION_COUNT },
+    };
+  }
+
+  private async claimPoll(kind: string, deviceId: number, ttlSec: number): Promise<boolean> {
+    try {
+      const { redis } = await import('../config/redis');
+      const res = await redis.set(`poll:inflight:${kind}:${deviceId}`, '1', 'EX', ttlSec, 'NX');
+      return res === 'OK';
+    } catch {
+      return true;
+    }
+  }
+
+  private async releasePoll(kind: string, deviceId: number): Promise<void> {
+    try {
+      const { redis } = await import('../config/redis');
+      await redis.del(`poll:inflight:${kind}:${deviceId}`);
+    } catch { /* claim will lapse on its own TTL */ }
+  }
+
   private async setTimestamp(key: string, ts: number, ttlSec = 600): Promise<void> {
     try {
       const { redis } = await import('../config/redis');
@@ -484,30 +696,40 @@ export class PollerService {
   private startWorkers(): void {
     const workerOptions = {
       connection: createRedisConnection(),
-      concurrency: 3,
+      concurrency: POLL_CONCURRENCY,
     };
 
-    this.fastWorker = new Worker(
+    /**
-      'poll-fast',
+     * Every job runs under a timeout, records how long it took, and releases its
+     * claim — whatever the outcome. Recording the attempt separately from the
+     * success is what makes "we never polled it" distinguishable from "we polled
+     * it and it did not answer", which was previously impossible to tell apart.
+     */
+    const run = (kind: string, fn: (data: PollJob) => Promise<void>) =>
       async (job: Job<PollJob>) => {
-        await this.processPollJob(job.data);
+        const started = Date.now();
-      },
+        try {
-      workerOptions
+          await withTimeout(fn(job.data), JOB_TIMEOUT_MS, `${kind} poll`);
-    );
+          await this.recordPollOutcome(job.data.deviceId, kind, Date.now() - started, null);
+        } catch (err) {
+          await this.recordPollOutcome(job.data.deviceId, kind, Date.now() - started, (err as Error).message);
+          throw err;
+        } finally {
+          await this.releasePoll(job.data.type, job.data.deviceId);
+        }
+      };
+
+    this.fastWorker = new Worker('poll-fast', run('fast', (d) => this.processPollJob(d)), workerOptions);
 
     this.slowWorker = new Worker(
       'poll-slow',
-      async (job: Job<PollJob>) => {
+      run('slow', (d) => this.processSlowJob(d)),
-        await this.processSlowJob(job.data);
-      },
       { ...workerOptions, connection: createRedisConnection() }
     );
 
     this.logsWorker = new Worker(
       'poll-logs',
-      async (job: Job<PollJob>) => {
+      run('logs', (d) => this.processLogsJob(d)),
-        await this.processLogsJob(job.data);
-      },
       { ...workerOptions, connection: createRedisConnection() }
     );
 

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "mikrotik-manager-frontend",
-  "version": "0.23.16-beta",
+  "version": "0.23.17-beta",
   "private": true,
   "type": "module",
   "scripts": {

frontend/src/version.ts

@@ -1 +1 @@
-export const APP_VERSION = 'v0.23.16 Beta';
+export const APP_VERSION = 'v0.23.17 Beta';