From 745a7cda93a265dfdf34373f1f549b3939bd6792 Mon Sep 17 00:00:00 2001 From: 2GT-Rich <107219997+2GT-Rich@users.noreply.github.com> Date: Tue, 1 Sep 2026 08:07:49 -0700 Subject: [PATCH] fix(poller): drain stale backlogs, and stop reporting health that isn't (v0.23.19-beta) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v0.23.17 stopped the poller creating an unbounded backlog but did nothing about the one already there. A reporter with sixty devices upgraded and still had 752,318 jobs queued, nine stale devices, and an endpoint cheerfully reporting 1.59x headroom. Fair to call that not fixed. Three faults, all mine. **Retention never touched pending work.** cleanupJobHistory cleaned 'completed' and 'failed' only. The queue of outstanding polls — the part the operator actually feels — was out of scope, so an upgrade left it draining for days. Stale pending jobs are now swept, at startup and every five minutes. The rule is that a periodic poll which has waited longer than its own cadence is worthless: running it yields a reading the next cycle would have taken anyway, while occupying a worker the live fleet is waiting on. Discarding queued work is safe precisely because it is periodic — the schedule re-enqueues within one interval. **Deduplication failed in the case it existed for.** The in-flight claim was a Redis key with a 135-second TTL. Under a large backlog a job waits hours, so the claim lapsed long before it ran and the scheduler re-enqueued every cycle: the guard worked only on a queue that was already short. Replaced with a deterministic job id, which cannot lapse. While a job for a device and kind exists in any state, adding it again is a no-op. That requires finished jobs to be removed immediately or the id would stay taken and the device would never be polled again — which costs nothing now that device_poll_stats records every attempt, duration and error. **The headroom metric lied.** It describes flow and is blind to accumulated work, so it read "healthy" beside three-quarters of a million queued jobs. It now reports backlog and drain ETA alongside, plus a single status verdict that cannot say ok while a backlog exists. Verified by injecting 15,125 jobs: status went to 'draining' with headroom still at 85x, which is exactly the case the old metric got wrong. Also added, because the reporter asked and was right to: an operator-triggered drain (POST /api/system/poller/drain), and a Polling tab in Settings showing queue depth, headroom, backlog, per-device staleness and the drain control. It was all buried in an API response before. A dashboard insight fires when the poller falls behind, so a starved poller announces itself rather than presenting as a fleet of broken devices. Two corrections found by testing rather than reasoning: * BullMQ rejects ':' in a custom job id. `fast:2` threw on every enqueue and killed the scheduler outright — three of four devices went unpolled for eight minutes before it was caught. Hyphens now, and the dedup is exercised directly: 500 enqueues of one device yield one queued job. * Redis maxmemory-policy moves from volatile-lru to noeviction. I had reasoned that only TTL-bearing keys are evicted and BullMQ job hashes carry none — true, but its *control* keys do. Job locks and stalled-check markers both expire, and evicting one causes duplicate processing or a stalled-job storm. BullMQ warns about this on connect and was right. Co-Authored-By: Claude Opus 5 --- .env.example | 5 +- README.md | 2 +- backend/package.json | 2 +- backend/src/index.ts | 8 + backend/src/routes/operations.ts | 25 ++ backend/src/routes/system.ts | 22 +- backend/src/services/PollerService.ts | 185 ++++++++++---- backend/src/services/pollerRef.ts | 16 ++ docker-compose.yml | 19 +- frontend/package.json | 2 +- .../components/system/PollerHealthCard.tsx | 226 ++++++++++++++++++ frontend/src/pages/SettingsPage.tsx | 11 +- frontend/src/services/api.ts | 39 +++ frontend/src/version.ts | 2 +- 14 files changed, 501 insertions(+), 63 deletions(-) create mode 100644 backend/src/services/pollerRef.ts create mode 100644 frontend/src/components/system/PollerHealthCard.tsx diff --git a/.env.example b/.env.example index e6ffdd6..c126ff2 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,7 @@ INFLUXDB_ADMIN_PASSWORD=admin_password_123 # POLLER_JOB_RETENTION_SEC=3600 # POLLER_JOB_RETENTION_COUNT=5000 -# Redis memory ceiling. Only TTL-bearing keys are evicted (volatile-lru), so -# queued jobs are never discarded; the cap exists to protect the host. +# Redis memory ceiling. Eviction is disabled (noeviction) because BullMQ's locks +# and stalled-check keys carry TTLs and must never be evicted; the cap exists to +# protect the host, and retention keeps usage well below it. # REDIS_MAXMEMORY=512mb diff --git a/README.md b/README.md index 3da4552..bd2c849 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 d2636fa..81a29d0 100644 --- a/backend/package.json +++ b/backend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-backend", - "version": "0.23.18-beta", + "version": "0.23.19-beta", "description": "Mikrotik Network Management Platform - Backend API", "main": "dist/index.js", "scripts": { diff --git a/backend/src/index.ts b/backend/src/index.ts index 472da7c..be9acbc 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -18,6 +18,7 @@ import { redis } from './config/redis'; import { runMigrations } from './db/migrate'; import { errorHandler } from './middleware/errorHandler'; import { PollerService } from './services/PollerService'; +import { setSharedPollerService } from './services/pollerRef'; import { setBulkAddPollerService, startBulkAddWorker, @@ -374,6 +375,7 @@ async function start(): Promise { pollerService.setSocketServer(io); setDevicesPoller(pollerService); setSystemPoller(pollerService); + setSharedPollerService(pollerService); setBulkAddPollerService(pollerService); setTopologyPoller(pollerService); setClientsPoller(pollerService); @@ -384,6 +386,12 @@ async function start(): Promise { // hold up startup, and nothing depends on it having finished. void pollerService.cleanupJobHistory().catch((e) => console.error('[Poller] Job history cleanup failed:', e)); + + // Pending work accumulated before dedup existed is dropped too, not just + // finished-job history. An upgrade that leaves hundreds of thousands of stale + // polls queued has not actually fixed anything the operator can see (#114). + void pollerService.dropStalePending().catch((e) => + console.error('[Poller] Stale-pending sweep failed:', e)); await startBulkAddWorker(); // NetFlow/IPFIX collector (binds its UDP socket only when netflow_enabled) diff --git a/backend/src/routes/operations.ts b/backend/src/routes/operations.ts index a4ca577..901d401 100644 --- a/backend/src/routes/operations.ts +++ b/backend/src/routes/operations.ts @@ -152,6 +152,31 @@ router.get('/insights', async (_req: Request, res: Response) => { }); } } + // The poller's own health belongs here. A fleet outrunning its workers looks + // exactly like a fleet of broken devices — stale data everywhere and no + // explanation — and the operator who most needs to know is the one least + // likely to go reading an API endpoint (#114). + try { + const { pollerService } = await import('../services/pollerRef'); + const health = pollerService ? await pollerService.getPollerHealth() : null; + if (health && health.status !== 'ok') { + const cap = health.capacity as Record; + const backlog = cap.backlog ?? 0; + const saturated = health.status === 'saturated'; + attention.push({ + sev: saturated ? 'error' : 'warn', + category: 'poller', + title: saturated + ? 'Polling cannot keep up with this fleet' + : `Polling is behind — ${backlog.toLocaleString()} jobs queued`, + body: saturated + ? `Each cycle enqueues ${cap.arrival_per_sec}/s but the workers clear only ${cap.service_per_sec}/s, so devices will go stale. Raise POLLER_CONCURRENCY or lengthen the interval.` + : `Clearing at the current rate takes ${cap.drain_eta_sec == null ? 'longer than it is accumulating' : `about ${Math.round((cap.drain_eta_sec || 0) / 60)} min`}. Devices poll late until it clears.`, + action: 'Open poller health', path: '/settings?tab=poller', + }); + } + } catch { /* health is advisory — never let it break the dashboard */ } + for (const d of devices) { if (d.firmware_update_available) { attention.push({ diff --git a/backend/src/routes/system.ts b/backend/src/routes/system.ts index 8c28ea9..fd2c84e 100644 --- a/backend/src/routes/system.ts +++ b/backend/src/routes/system.ts @@ -2,7 +2,7 @@ import { Router } from 'express'; import type { Request, Response } from 'express'; import { readFileSync } from 'fs'; import { join } from 'path'; -import { requireAuth } from '../middleware/auth'; +import { requireAuth, requireWrite } from '../middleware/auth'; import { query } from '../config/database'; import type { PollerService } from '../services/PollerService'; @@ -88,6 +88,26 @@ router.get('/poller', async (_req: Request, res: Response) => { } }); +/** + * POST /api/system/poller/drain — clear every queued poll now. + * + * The automatic sweep only discards jobs past their cadence, which still leaves + * a large backlog draining for hours. Nothing is lost: the scheduler re-enqueues + * whatever is still due on its next tick, so the cost is at most one interval of + * freshness. + */ +router.post('/poller/drain', requireWrite, async (_req: Request, res: Response) => { + try { + if (!pollerService) return res.status(503).json({ error: 'Poller not started' }); + const drained = await pollerService.drainQueues(); + const total = Object.values(drained).reduce((a, b) => a + b, 0); + res.json({ drained, total, message: `Cleared ${total} queued job(s). Polling resumes on the next cycle.` }); + } catch (error) { + console.error('Error draining poller queues:', error); + res.status(500).json({ error: 'Failed to drain queues' }); + } +}); + 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 76ac9ba..9fb54dc 100644 --- a/backend/src/services/PollerService.ts +++ b/backend/src/services/PollerService.ts @@ -15,7 +15,23 @@ import type { GuardDevice } from './changeGuard/ChangeGuard'; 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; +const CLEANUP_MAX_PASSES = 400; + +/** + * How long a *pending* poll may wait before it is worthless. + * + * A periodic job that has waited longer than its own cadence has already been + * superseded: running it produces a reading the next cycle would have taken + * anyway, while occupying a worker the fleet needs. Retention limits do not + * touch pending work, so a backlog accumulated before those limits existed sat + * untouched — one reporter upgraded and still had 752,318 jobs queued, roughly + * nine days of draining (#114). + */ +const STALE_WAIT_MS: Record = { + 'poll-fast': 120_000, + 'poll-logs': 240_000, + 'poll-slow': 900_000, +}; /** * Worker concurrency, per queue. @@ -147,12 +163,30 @@ 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; - } + /** + * Deduplication is delegated to BullMQ rather than guarded by a lock. + * + * The previous attempt used a Redis key with a TTL, which fails in exactly + * the situation it exists for: with a large backlog a job waits hours, the + * TTL lapses long before it runs, and the scheduler enqueues another every + * cycle. A deterministic job id cannot lapse — while a job for this device + * and kind exists in any state, adding it again is a no-op. + * + * That requires finished jobs to be removed immediately, or the id would + * stay taken and the device would never be polled again. Losing that history + * costs nothing now: device_poll_stats records every attempt, its duration + * and its error, which is more useful than the job payload ever was. + */ + const periodic = { + // Hyphens, not colons: BullMQ reserves ':' for its own key structure and + // rejects a custom id containing one. Getting this wrong throws on every + // enqueue, which stops the scheduler dead — caught only by exercising the + // id directly rather than trusting the format. + jobId: `poll-${type}-${deviceId}`, + removeOnComplete: true, + removeOnFail: true, + attempts: 1, + }; if (type === 'full') { await this.fastQueue.add('device-full-sync', jobData, { @@ -167,21 +201,21 @@ export class PollerService { // also doubled the recorded failure count: 231k failed jobs on a four-device // fleet turned out to be ~1,000 hours of genuine downtime, counted twice. } else if (type === 'fast') { - await this.fastQueue.add('device-fast-poll', jobData, { attempts: 1 }); + await this.fastQueue.add('device-fast-poll', jobData, periodic); } else if (type === 'slow') { - await this.slowQueue.add('device-slow-poll', jobData, { attempts: 1 }); + await this.slowQueue.add('device-slow-poll', jobData, periodic); } else if (type === 'logs') { - await this.logsQueue.add('device-logs-poll', jobData, { attempts: 1 }); + await this.logsQueue.add('device-logs-poll', jobData, periodic); } else if (type === 'macscan') { - await this.fastQueue.add('device-macscan', jobData, { attempts: 1 }); + await this.fastQueue.add('device-macscan', jobData, periodic); } else if (type === 'spectral') { - await this.slowQueue.add('device-spectral', jobData, { attempts: 1 }); + await this.slowQueue.add('device-spectral', jobData, periodic); } else if (type === 'apscan') { - await this.slowQueue.add('device-apscan', jobData, { attempts: 1 }); + await this.slowQueue.add('device-apscan', jobData, periodic); } else if (type === 'configsnap') { - await this.slowQueue.add('device-configsnap', jobData, { attempts: 1 }); + await this.slowQueue.add('device-configsnap', jobData, periodic); } else if (type === 'confighealth') { - await this.slowQueue.add('device-confighealth', jobData, { attempts: 1 }); + await this.slowQueue.add('device-confighealth', jobData, periodic); } } @@ -316,6 +350,17 @@ export class PollerService { .catch((e) => console.error('[Poller] Stale topology-link cleanup error:', e)); } + // Stale pending sweep — every 5 minutes. Cheap when there is nothing to do, + // and it means a backlog from any cause self-heals rather than requiring + // an operator to notice it. + const staleKey = 'task:stale_pending'; + const lastStale = await this.getTimestamp(staleKey); + if (now - lastStale > 300_000) { + await this.setTimestamp(staleKey, now); + this.dropStalePending().catch((e) => + console.error('[Poller] Stale-pending sweep error:', e)); + } + // Firmware update check — runs once per day const firmwareKey = 'task:firmware_check'; const lastFirmware = await this.getTimestamp(firmwareKey); @@ -577,19 +622,55 @@ export class PollerService { } /** - * Claim a device for one kind of poll, refusing if one is already outstanding. + * Discard pending polls that have waited longer than their own cadence. * - * 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). + * Retention limits govern finished jobs only, so an installation that built a + * backlog before those limits existed keeps every queued job — and each one, + * when it finally runs, produces a reading that the next cycle would have + * taken anyway. Executing them is worse than dropping them: they occupy the + * workers the live fleet is waiting on. * - * 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. + * Dropping queued work is safe precisely because it is periodic. The schedule + * re-enqueues within one cycle, so the cost of discarding a stale poll is at + * most one interval of freshness, against hours of backlog otherwise. */ + async dropStalePending(): Promise { + let total = 0; + for (const { name, queue } of this.queues) { + const maxAge = STALE_WAIT_MS[name] ?? 300_000; + try { + for (let pass = 0; pass < CLEANUP_MAX_PASSES; pass++) { + const ids = await queue.clean(maxAge, CLEANUP_BATCH, 'wait'); + total += ids.length; + if (ids.length < CLEANUP_BATCH) break; + } + } catch (e) { + console.error(`[Poller] Stale-pending sweep failed for ${name}:`, (e as Error).message); + } + } + if (total > 0) console.log(`[Poller] Dropped ${total} stale pending job(s)`); + return total; + } + + /** + * Remove every queued poll immediately, on request. + * + * The automatic sweep only drops jobs past their cadence, which still leaves an + * operator staring at a backlog that takes hours to clear. This is the "clear + * it now" they actually want; nothing is lost because the next scheduler tick + * re-enqueues whatever is still due. + */ + async drainQueues(): Promise> { + const drained: Record = {}; + for (const { name, queue } of this.queues) { + const before = await queue.getJobCounts('waiting', 'delayed'); + await queue.drain(true); + drained[name] = (before.waiting || 0) + (before.delayed || 0); + } + console.log('[Poller] Queues drained on request:', drained); + return drained; + } + /** * Record what happened on a poll, per device and kind. * @@ -654,7 +735,29 @@ export class PollerService { const arrivalPerSec = deviceCount / (POLL_INTERVAL_MS / 1000); const servicePerSec = avgSec > 0 ? POLL_CONCURRENCY / avgSec : null; + const headroom = servicePerSec == null || arrivalPerSec === 0 + ? null + : Math.round((servicePerSec / arrivalPerSec) * 100) / 100; + + // Backlog is reported alongside headroom because headroom alone lies. + // It describes flow — whether the workers can keep up from here — and says + // nothing about work already queued. An earlier version of this endpoint + // reported a comfortable 1.59 to an operator with 752,318 jobs waiting and + // nine stale devices, which is worse than no number at all (#114). + const backlog = queues.reduce((n, q) => n + (Number(q.waiting) || 0), 0); + const netDrainPerSec = servicePerSec == null ? null : servicePerSec - arrivalPerSec; + const drainEtaSec = backlog === 0 ? 0 + : netDrainPerSec == null || netDrainPerSec <= 0 ? null + : Math.round(backlog / netDrainPerSec); + + // A single verdict, so nobody has to interpret three numbers correctly. + const status = + headroom != null && headroom < 1 ? 'saturated' + : backlog > deviceCount * 3 ? 'draining' + : 'ok'; + return { + status, queues, workers: { concurrency: POLL_CONCURRENCY, @@ -668,31 +771,19 @@ export class PollerService { 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, + headroom, + backlog, + // null means the backlog is not shrinking at the current rate. + drain_eta_sec: drainEtaSec, + }, + retention: { + max_age_sec: JOB_RETENTION_SEC, + max_count: JOB_RETENTION_COUNT, + stale_pending_ms: STALE_WAIT_MS, }, - 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'); @@ -721,8 +812,6 @@ export class PollerService { } 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); } }; diff --git a/backend/src/services/pollerRef.ts b/backend/src/services/pollerRef.ts new file mode 100644 index 0000000..1c42ffa --- /dev/null +++ b/backend/src/services/pollerRef.ts @@ -0,0 +1,16 @@ +/** + * A single place to reach the running PollerService. + * + * Routes receive it by injection from index.ts, which works well for a route + * module but not for code reached indirectly — the dashboard insights build in + * a helper that has no injection point of its own. A tiny module-level + * reference avoids threading the dependency through every caller, and stays + * null until the poller starts so nothing can accidentally depend on it early. + */ +import type { PollerService } from './PollerService'; + +export let pollerService: PollerService | null = null; + +export function setSharedPollerService(p: PollerService): void { + pollerService = p; +} diff --git a/docker-compose.yml b/docker-compose.yml index c48f867..a99bea3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -85,16 +85,21 @@ services: redis: image: redis:7-alpine - # A memory ceiling so a runaway producer can never take the host down. The - # eviction policy is deliberately volatile-lru rather than allkeys-*: BullMQ - # job data carries no TTL and must never be evicted (that would silently - # destroy queued work), while the poller's caches and in-flight claims all - # set one and are safe to drop. If pressure ever exhausts the volatile keys - # Redis refuses writes instead of evicting queue state — loud, and survivable. + # A memory ceiling so a runaway producer can never take the host down, with + # eviction disabled. + # + # volatile-lru looks tempting — it would evict only keys with an expiry, and + # BullMQ's job hashes have none. But its *control* keys do: job locks and + # per-queue stalled-check markers both carry TTLs, and losing one causes + # duplicate processing or a stalled-job storm. BullMQ warns about this on + # connect, and it is right. noeviction means Redis refuses writes when full + # rather than quietly corrupting queue state: the host stays up, and the + # failure is loud. Retention limits and the stale-pending sweep are what keep + # us away from the ceiling in the first place. command: > redis-server --appendonly yes --maxmemory ${REDIS_MAXMEMORY:-512mb} - --maxmemory-policy volatile-lru + --maxmemory-policy noeviction volumes: - redis_data:/data restart: unless-stopped diff --git a/frontend/package.json b/frontend/package.json index 822ff00..c79c270 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-frontend", - "version": "0.23.18-beta", + "version": "0.23.19-beta", "private": true, "type": "module", "scripts": { diff --git a/frontend/src/components/system/PollerHealthCard.tsx b/frontend/src/components/system/PollerHealthCard.tsx new file mode 100644 index 0000000..81fe532 --- /dev/null +++ b/frontend/src/components/system/PollerHealthCard.tsx @@ -0,0 +1,226 @@ +import { useState } from 'react'; +import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; +import { Activity, AlertTriangle, CheckCircle2, Clock, RefreshCw, Trash2 } from 'lucide-react'; +import { pollerApi, type PollerHealth } from '../../services/api'; + +/** + * The poller's own health, in the UI rather than buried in an API response. + * + * A fleet that outruns its workers presents as stale data everywhere, which + * reads as a device problem rather than a capacity one. Making queue depth and + * headroom visible is what turns "polling frequency is set by guesswork" into an + * arithmetic question (#114). + */ + +const STATUS: Record = { + ok: { + label: 'Keeping up', tone: 'text-green-700 bg-green-100 dark:text-green-300 dark:bg-green-900/40', + icon: CheckCircle2, + blurb: 'Workers are clearing polls faster than the schedule creates them.', + }, + draining: { + label: 'Working through a backlog', tone: 'text-amber-800 bg-amber-100 dark:text-amber-300 dark:bg-amber-900/40', + icon: Clock, + blurb: 'Capacity is sufficient, but queued work from earlier is still clearing. Devices poll late until it does.', + }, + saturated: { + label: 'Cannot keep up', tone: 'text-red-700 bg-red-100 dark:text-red-300 dark:bg-red-900/40', + icon: AlertTriangle, + blurb: 'Polls are being created faster than they can run, so the backlog grows every cycle and devices will go stale.', + }, +}; + +const fmtDuration = (sec: number | null): string => { + if (sec == null) return 'not shrinking'; + if (sec < 60) return `${sec}s`; + if (sec < 3600) return `${Math.round(sec / 60)} min`; + return `${(sec / 3600).toFixed(1)} h`; +}; + +function Stat({ label, value, hint, alarm }: { + label: string; value: string; hint?: string; alarm?: boolean; +}) { + return ( +

+
{label}
+
{value}
+ {hint &&
{hint}
} +
+ ); +} + +export default function PollerHealthCard() { + const queryClient = useQueryClient(); + const [drainResult, setDrainResult] = useState(''); + const [confirmDrain, setConfirmDrain] = useState(false); + + const { data, isLoading } = useQuery({ + queryKey: ['poller-health'], + queryFn: () => pollerApi.health().then(r => r.data), + refetchInterval: 15_000, + }); + + const drain = useMutation({ + mutationFn: () => pollerApi.drain(), + onSuccess: (res) => { + setDrainResult(res.data.message); + setConfirmDrain(false); + queryClient.invalidateQueries({ queryKey: ['poller-health'] }); + }, + }); + + if (isLoading || !data) { + return
Loading poller health…
; + } + + const s = STATUS[data.status] ?? STATUS.ok; + const Icon = s.icon; + const cap = data.capacity; + const backlog = cap.backlog; + + return ( +
+
+
+ +
+
+

Polling

+ {s.label} +
+

{s.blurb}

+
+
+ +
+ + {/* Capacity. headroom and backlog are shown together deliberately: headroom + alone once reported "healthy" to an operator with 752k jobs queued. */} +
+ + + 0 ? `clears in ${fmtDuration(cap.drain_eta_sec)}` : 'nothing waiting'} + alarm={backlog > cap.devices * 3} + /> + + + +
+ + {cap.headroom != null && cap.headroom < 1.5 && ( +
+
Little margin left
+

+ Demand is {cap.arrival_per_sec}/s against {cap.service_per_sec ?? '—'}/s of capacity. Slow + devices push individual polls well past the average, so bursts will fall behind even while + the average looks fine. Raise POLLER_CONCURRENCY or lengthen the interval. +

+
+ )} + + {/* Queues */} +
+ + + + + + + + + + + {data.queues.map(q => ( + + + + + + + ))} + +
QueueWaitingActiveFailed
{q.name} 0 ? 'text-amber-700 dark:text-amber-400 font-medium' + : 'text-gray-600 dark:text-slate-300'}`}> + {(q.waiting ?? 0).toLocaleString()} + {q.active ?? 0}{q.failed ?? 0}
+
+ + {/* Clearing the queue */} +
+
+
+ Clear the queue.{' '} + Discards everything currently queued. Nothing is lost — the scheduler re-enqueues whatever + is still due on its next cycle, so the cost is at most one interval of freshness. Stale + polls are also dropped automatically once they are older than their own cadence. +
+ {confirmDrain ? ( +
+ + +
+ ) : ( + + )} +
+ {drainResult &&

{drainResult}

} +
+ + {/* Which devices are actually being missed */} +
+

+ Devices not polled recently {data.stale_devices.length > 0 && `(${data.stale_devices.length})`} +

+ {data.stale_devices.length === 0 ? ( +

+ Every device has been polled successfully in the last few minutes. +

+ ) : ( +
    + {data.stale_devices.map(d => ( +
  • + {d.name} + + {d.last_success_at + ? `last success ${fmtDuration(d.seconds_since_success)} ago` + : 'never polled successfully'} + + {/* Attempt vs success is the distinction that says whether we + missed it or it did not answer. */} + {d.last_error + ? responded with: {d.last_error} + : not attempted recently} +
    • + ))} +
+ )} +
+
+ ); +} diff --git a/frontend/src/pages/SettingsPage.tsx b/frontend/src/pages/SettingsPage.tsx index aea0edb..0deb84e 100644 --- a/frontend/src/pages/SettingsPage.tsx +++ b/frontend/src/pages/SettingsPage.tsx @@ -3,10 +3,12 @@ import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; import { Settings, Users, Key, Plus, Trash2, CheckCircle, AlertCircle, Pencil, X, ShieldCheck, ShieldAlert, RefreshCw, Upload, Lock, Bell, Send, KeyRound, ClipboardList, FileText, Zap, LogIn, + Activity, } from 'lucide-react'; import { settingsApi, authApi, certApi, alertsApi, auditLogApi, tagsApi, maintenanceApi, configTemplatesApi } from '../services/api'; import type { MaintenanceWindow, ConfigTemplate } from '../services/api'; import type { CertInfo, AlertRule, AlertChannel } from '../services/api'; +import PollerHealthCard from '../components/system/PollerHealthCard'; import { useAuthStore } from '../store/authStore'; import { useThemeStore } from '../store/themeStore'; import type { User, UserRole } from '../types'; @@ -49,7 +51,7 @@ export default function SettingsPage() { const { theme, setTheme } = useThemeStore(); const isAdmin = user?.role === 'admin'; const canWrite = user?.role !== 'viewer'; - const [activeTab, setActiveTab] = useState<'general' | 'users' | 'sso' | 'credentials' | 'security' | 'certificate' | 'alerting' | 'audit' | 'tags' | 'maintenance' | 'templates' | 'automation'>('general'); + const [activeTab, setActiveTab] = useState<'general' | 'users' | 'sso' | 'credentials' | 'security' | 'certificate' | 'alerting' | 'audit' | 'tags' | 'maintenance' | 'templates' | 'automation' | 'poller'>('general'); const [auditSearch, setAuditSearch] = useState(''); const [auditPage, setAuditPage] = useState(1); const [newTagName, setNewTagName] = useState(''); @@ -477,6 +479,7 @@ export default function SettingsPage() { ...(isAdmin ? [{ key: 'sso' as const, label: 'SSO / OIDC', icon: LogIn }] : []), { key: 'certificate' as const, label: 'Certificate', icon: Lock }, ...(isAdmin ? [{ key: 'audit' as const, label: 'Audit Log', icon: ClipboardList }] : []), + ...(isAdmin ? [{ key: 'poller' as const, label: 'Polling', icon: Activity }] : []), ], }, { @@ -925,6 +928,12 @@ export default function SettingsPage() { )} {/* ── Users ── */} + {activeTab === 'poller' && ( +
+ +
+ )} + {activeTab === 'users' && (
{/* Role legend */} diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index 6215f49..d3ae180 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -301,6 +301,45 @@ export const lteApi = { }), }; + +// ─── Poller health (#114) ───────────────────────────────────────────────────── + +export interface PollerQueue { + name: string; waiting?: number; active?: number; delayed?: number; + failed?: number; completed?: number; paused?: number; error?: string; +} + +export interface PollerHealth { + status: 'ok' | 'draining' | 'saturated'; + queues: PollerQueue[]; + workers: { concurrency: number; job_timeout_ms: number; poll_interval_ms: number }; + capacity: { + devices: number; + avg_fast_poll_ms: number | null; + p90_fast_poll_ms: number | null; + arrival_per_sec: number; + service_per_sec: number | null; + /** Below 1.0 the backlog grows every cycle. */ + headroom: number | null; + backlog: number; + /** null when the backlog is not shrinking at the current rate. */ + drain_eta_sec: number | null; + }; + retention: { max_age_sec: number; max_count: number; stale_pending_ms: Record }; + stale_devices: { + id: number; name: string; status: string; kind: string; + last_attempt_at: string | null; last_success_at: string | null; + last_duration_ms: number | null; last_error: string | null; + attempts: number; failures: number; seconds_since_success: number | null; + }[]; +} + +export const pollerApi = { + health: () => api.get('/system/poller'), + drain: () => api.post<{ drained: Record; total: number; message: string }>( + '/system/poller/drain'), +}; + export const devicesApi = { list: () => api.get('/devices'), discovered: () => api.get('/devices/discovered'), diff --git a/frontend/src/version.ts b/frontend/src/version.ts index a457959..c2e492a 100644 --- a/frontend/src/version.ts +++ b/frontend/src/version.ts @@ -1 +1 @@ -export const APP_VERSION = 'v0.23.18 Beta'; +export const APP_VERSION = 'v0.23.19 Beta';