From 38d23cde02ff5028ea3908dd04900dbb901db6c7 Mon Sep 17 00:00:00 2001 From: 2GT-Rich <107219997+2GT-Rich@users.noreply.github.com> Date: Mon, 31 Aug 2026 18:28:51 -0700 Subject: [PATCH] fix(poller): stop the queue outrunning the workers, and cap job retention (v0.23.17-beta) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- README.md | 2 +- backend/package.json | 2 +- backend/src/db/migrate.ts | 20 ++ backend/src/index.ts | 9 +- backend/src/routes/system.ts | 42 +++++ backend/src/services/PollerService.ts | 258 ++++++++++++++++++++++++-- frontend/package.json | 2 +- frontend/src/version.ts | 2 +- 8 files changed, 314 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 2c48636..f131f26 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@

- Version + Version License Docker TypeScript diff --git a/backend/package.json b/backend/package.json index df6281e..fcb9841 100644 --- a/backend/package.json +++ b/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": { diff --git a/backend/src/db/migrate.ts b/backend/src/db/migrate.ts index aca2014..b3244a1 100644 --- a/backend/src/db/migrate.ts +++ b/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 = [ diff --git a/backend/src/index.ts b/backend/src/index.ts index 5eaa18a..472da7c 100644 --- a/backend/src/index.ts +++ b/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 { 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) diff --git a/backend/src/routes/system.ts b/backend/src/routes/system.ts index 5431e2b..8c28ea9 100644 --- a/backend/src/routes/system.ts +++ b/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 diff --git a/backend/src/services/PollerService.ts b/backend/src/services/PollerService.ts index 7b95f52..e36f86f 100644 --- a/backend/src/services/PollerService.ts +++ b/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(work: Promise, ms: number, label: string): Promise { + let timer: ReturnType; + return Promise.race([ + work.finally(() => clearTimeout(timer)), + new Promise((_, 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 }); - this.slowQueue = new Queue('poll-slow', { connection: conn2 }); - this.logsQueue = new Queue('poll-logs', { connection: conn3 }); + // Retention is not optional. BullMQ keeps every finished job unless told + // otherwise, and a poller enqueues forever — a four-device fleet had + // 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 { + 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 { 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 { + 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> { + const queues: Record[] = []; + 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 { + 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 { + 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 { 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) => async (job: Job) => { - await this.processPollJob(job.data); - }, - workerOptions - ); + const started = Date.now(); + try { + 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) => { - await this.processSlowJob(job.data); - }, + run('slow', (d) => this.processSlowJob(d)), { ...workerOptions, connection: createRedisConnection() } ); this.logsWorker = new Worker( 'poll-logs', - async (job: Job) => { - await this.processLogsJob(job.data); - }, + run('logs', (d) => this.processLogsJob(d)), { ...workerOptions, connection: createRedisConnection() } ); diff --git a/frontend/package.json b/frontend/package.json index 9e9c70f..2d064da 100644 --- a/frontend/package.json +++ b/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": { diff --git a/frontend/src/version.ts b/frontend/src/version.ts index 74b21ac..2252b75 100644 --- a/frontend/src/version.ts +++ b/frontend/src/version.ts @@ -1 +1 @@ -export const APP_VERSION = 'v0.23.16 Beta'; +export const APP_VERSION = 'v0.23.17 Beta';