diff --git a/README.md b/README.md index a09a23b..628e62b 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@

- Version + Version License Docker TypeScript @@ -339,7 +339,7 @@ Each network service supports multi-device management with conflict detection: ### Fleet operations -- **Firmware orchestration** — fleet version overview, live update checks, and MikroTik's official release notes in-app. Staged rollouts run in **waves** (wave 1 = canary) through a verified pipeline per device: pre-upgrade backup → install → ride out the reboot → confirm it returned healthy on the new version → next. **Halt on failure** stops the rollout if any device fails, and a device that comes back on the *old* version counts as a failure. Rollouts can be scheduled, and cancelling never interrupts an in-flight flash +- **Firmware orchestration** — fleet version overview, live update checks, and MikroTik's official release notes in-app. Staged rollouts run in **waves** (wave 1 = canary) through a verified pipeline per device: pre-upgrade backup → download and confirm the image landed → reboot → prove it restarted and came back healthy on the new version → next. **Halt on failure** stops the rollout if any device fails, and a device that comes back on the *old* version counts as a failure. Devices in a wave upgrade one at a time by default; raise **at once** to run them concurrently when a fleet of five-minute reboots would otherwise take all evening — waves stay sequential either way, so the canary still gates the rest. Rollouts can be scheduled, and cancelling never interrupts an in-flight flash - **Backups** — on demand or on a daily/weekly/monthly schedule for all online devices. Filter by device, type and date range; read one in place instead of downloading it; and compare any two as a diff — they are `/export` text, not opaque blobs. Config snapshots and their restorable `.rsc` are one artifact: delete either and both go, so they never drift apart, and a bulk delete says how many snapshots it will take with it - **Configuration templates** — reusable sets (DNS, NTP, syslog) pushed to many devices with per-device result reporting - **Bulk device add** — "Try All" on discovered devices runs as a server-side job that survives a closed browser tab, with live progress and cancel diff --git a/backend/package.json b/backend/package.json index acab926..cef2ccd 100644 --- a/backend/package.json +++ b/backend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-backend", - "version": "0.24.2-beta", + "version": "0.24.3-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 a7a834f..7bb0402 100644 --- a/backend/src/db/migrate.ts +++ b/backend/src/db/migrate.ts @@ -608,6 +608,10 @@ ALTER TABLE devices ADD COLUMN IF NOT EXISTS wifi_package VARCHAR(16); -- now meant visiting every device by hand — the exact task staged rollouts exist to -- remove. ALTER TABLE firmware_rollouts ADD COLUMN IF NOT EXISTS routerboot_after BOOLEAN NOT NULL DEFAULT FALSE; +-- How many devices in a wave may upgrade at once (issue #135). Default 1 keeps +-- the sequential behaviour every existing rollout was built around; raising it +-- trades canary strictness for wall-clock time, which is the operator's call. +ALTER TABLE firmware_rollouts ADD COLUMN IF NOT EXISTS wave_concurrency INTEGER NOT NULL DEFAULT 1; -- Temporarily dismissed "Things to handle" items (github issue #102). -- diff --git a/backend/src/routes/firmware.ts b/backend/src/routes/firmware.ts index 294ad08..aa9fa7f 100644 --- a/backend/src/routes/firmware.ts +++ b/backend/src/routes/firmware.ts @@ -5,6 +5,8 @@ import { siteScopeDevices, siteScopeByDevice } from '../utils/siteScope'; import { activeSite } from '../middleware/site'; import { DeviceCollector, DeviceRow } from '../services/mikrotik/DeviceCollector'; import { firmwareOrchestrator } from '../services/FirmwareOrchestrator'; +import { clampConcurrency, MAX_WAVE_CONCURRENCY } from '../utils/concurrency'; +import { findUpstreamWithinSelection } from '../utils/rolloutTopology'; const router = Router(); router.use(requireAuth); @@ -64,9 +66,13 @@ router.post('/check-all', requireWrite, async (req: Request, res: Response) => { // POST /api/firmware/rollouts — create a rollout (optionally scheduled) router.post('/rollouts', requireWrite, async (req: Request, res: Response) => { - const { name, halt_on_failure, pre_backup, routerboot_after, scheduled_at, devices, start } = req.body as { + const { + name, halt_on_failure, pre_backup, routerboot_after, scheduled_at, devices, start, + wave_concurrency, + } = req.body as { name?: string; halt_on_failure?: boolean; pre_backup?: boolean; routerboot_after?: boolean; scheduled_at?: string | null; devices?: { device_id: number; wave: number }[]; start?: boolean; + wave_concurrency?: number; }; if (!name || !name.trim()) return res.status(400).json({ error: 'name is required' }); if (!Array.isArray(devices) || devices.length === 0) return res.status(400).json({ error: 'devices array is required' }); @@ -113,9 +119,10 @@ router.post('/rollouts', requireWrite, async (req: Request, res: Response) => { } const rollout = await queryOne<{ id: number }>( - `INSERT INTO firmware_rollouts (name, halt_on_failure, pre_backup, routerboot_after, scheduled_at) - VALUES ($1,$2,$3,$4,$5) RETURNING id`, - [name.trim().slice(0, 100), halt_on_failure !== false, pre_backup !== false, routerboot_after === true, scheduled_at || null]); + `INSERT INTO firmware_rollouts (name, halt_on_failure, pre_backup, routerboot_after, scheduled_at, wave_concurrency) + VALUES ($1,$2,$3,$4,$5,$6) RETURNING id`, + [name.trim().slice(0, 100), halt_on_failure !== false, pre_backup !== false, routerboot_after === true, + scheduled_at || null, clampConcurrency(wave_concurrency ?? 1)]); for (const d of devices) { await query( `INSERT INTO firmware_rollout_devices (rollout_id, device_id, wave) VALUES ($1,$2,$3)`, @@ -134,6 +141,42 @@ router.post('/rollouts', requireWrite, async (req: Request, res: Response) => { res.status(201).json({ id: rollout!.id }); }); +/** + * GET /api/firmware/rollouts/upstream-check?ids=1,2,3 + * + * Which of the selected devices sit upstream of the others (#135). + * + * Running a wave concurrently means several devices reboot together, and a + * switch that carries the uplink to the rest takes them with it -- the manager + * then cannot reach the devices it is mid-upgrade on. Change Guard does not + * cover this: it protects configuration, not the path back to the device. + * + * Advisory only. It warns, it does not refuse: the operator may know the + * topology better than the discovered links do. + */ +router.get('/rollouts/upstream-check', async (req: Request, res: Response) => { + const ids = String(req.query.ids || '') + .split(',').map((v) => parseInt(v, 10)).filter(Number.isInteger); + if (ids.length < 2) return res.json({ upstream: [] }); + + const links = await query<{ + from_device_id: number; to_device_id: number | null; + from_interface: string | null; stp_role: string | null; + }>( + `SELECT from_device_id, to_device_id, from_interface, stp_role + FROM topology_links WHERE to_device_id IS NOT NULL` + ); + const names = await query<{ id: number; name: string }>( + `SELECT id, name FROM devices WHERE id = ANY($1::int[])`, [ids] + ); + const nameById = new Map(names.map((n) => [n.id, n.name])); + + const upstream = findUpstreamWithinSelection(ids, links).map((id) => ({ + id, name: nameById.get(id) ?? `Device ${id}`, + })); + return res.json({ upstream, maxConcurrency: MAX_WAVE_CONCURRENCY }); +}); + // GET /api/firmware/rollouts — recent rollouts with progress counts router.get('/rollouts', async (req: Request, res: Response) => { // A rollout is shown when it touched at least one device in the active site. diff --git a/backend/src/services/FirmwareOrchestrator.ts b/backend/src/services/FirmwareOrchestrator.ts index 86ba808..9050948 100644 --- a/backend/src/services/FirmwareOrchestrator.ts +++ b/backend/src/services/FirmwareOrchestrator.ts @@ -1,6 +1,7 @@ // Staged firmware rollout orchestrator. // -// Executes a rollout wave-by-wave (wave 1 = canary), each device sequentially: +// Executes a rollout wave-by-wave (wave 1 = canary). Devices within a wave run +// sequentially by default, or up to wave_concurrency at once (#135): // pre-upgrade backup → install RouterOS update → wait through the reboot → // verify it came back healthy on the new version → next device. // A failure marks the device failed and (with halt_on_failure) stops the whole @@ -11,6 +12,7 @@ import { query, queryOne } from '../config/database'; import { DeviceCollector, DeviceRow } from './mikrotik/DeviceCollector'; import { BackupService } from './BackupService'; +import { mapWithConcurrency, clampConcurrency } from '../utils/concurrency'; const REBOOT_GRACE_MS = 25_000; // let the device actually go down const DOWNLOAD_TIMEOUT_MS = 10 * 60_000; // large images on slow links @@ -28,6 +30,8 @@ interface RolloutRow { halt_on_failure: boolean; pre_backup: boolean; /** Follow the RouterOS upgrade with the pending RouterBOOT upgrade (issue #113). */ routerboot_after: boolean; + /** How many devices in a wave may upgrade at once; 1 is sequential (#135). */ + wave_concurrency: number; } interface RolloutDeviceRow { id: number; rollout_id: number; device_id: number; wave: number; status: string; @@ -87,18 +91,46 @@ export class FirmwareOrchestrator { `SELECT * FROM firmware_rollout_devices WHERE rollout_id=$1 ORDER BY wave ASC, id ASC`, [rollout.id]); + const concurrency = clampConcurrency(rollout.wave_concurrency); + const waves = [...new Set(items.map((i) => i.wave))].sort((a, b) => a - b); let halted = false; - for (const item of items) { - if (this.cancelRequested || halted) { - await query(`UPDATE firmware_rollout_devices SET status='skipped', - error=$2 WHERE id=$1 AND status='pending'`, - [item.id, this.cancelRequested ? 'Rollout cancelled' : 'Halted: earlier device failed']); - continue; + + // Waves stay strictly sequential -- that is the decision point, and it is + // what makes wave 1 a canary. Concurrency applies only *within* a wave. + // + // At the default of 1 this is the previous behaviour with one difference + // worth stating: halting is evaluated at the end of a wave rather than the + // instant a device fails. With a wave of one those are the same moment. With + // a larger wave it is the trade the operator asked for (#135) -- more devices + // carry a bad build before anything stops, in exchange for not waiting out + // ten five-minute reboots in series. + for (const wave of waves) { + if (this.cancelRequested || halted) break; + const inWave = items.filter((i) => i.wave === wave); + + if (concurrency > 1 && inWave.length > 1) { + console.log(`[Firmware] rollout #${rollout.id}: wave ${wave} — ` + + `${inWave.length} device(s), up to ${concurrency} at once`); + } + + const results = await mapWithConcurrency(inWave, concurrency, (item) => + this.cancelRequested + ? Promise.resolve(false) + : this.upgradeDevice(rollout, item) + ); + + if (rollout.halt_on_failure && results.some((ok) => !ok)) { + halted = true; + console.warn(`[Firmware] rollout #${rollout.id}: halting after failures in wave ${wave}`); } - const ok = await this.upgradeDevice(rollout, item); - if (!ok && rollout.halt_on_failure) halted = true; } + // Anything never reached is skipped rather than failed -- it did not run. + await query( + `UPDATE firmware_rollout_devices SET status='skipped', error=$2, finished_at=NOW() + WHERE rollout_id=$1 AND status='pending'`, + [rollout.id, this.cancelRequested ? 'Rollout cancelled' : 'Halted: earlier device failed']); + const finalStatus = this.cancelRequested ? 'cancelled' : halted ? 'failed' : 'completed'; await query(`UPDATE firmware_rollouts SET status=$2, finished_at=NOW() WHERE id=$1`, [rollout.id, finalStatus]); console.log(`[Firmware] rollout #${rollout.id} ${finalStatus}`); diff --git a/backend/src/utils/__tests__/concurrency.test.ts b/backend/src/utils/__tests__/concurrency.test.ts new file mode 100644 index 0000000..29632a6 --- /dev/null +++ b/backend/src/utils/__tests__/concurrency.test.ts @@ -0,0 +1,103 @@ +import { mapWithConcurrency, clampConcurrency, MAX_WAVE_CONCURRENCY } from '../concurrency'; + +const tick = (ms = 5) => new Promise((r) => setTimeout(r, ms)); + +describe('mapWithConcurrency', () => { + it('returns results in input order, not completion order', async () => { + // First item is slowest, so completion order is the reverse of input order. + const out = await mapWithConcurrency([30, 20, 1], 3, async (ms) => { + await tick(ms); + return ms; + }); + expect(out).toEqual([30, 20, 1]); + }); + + it('never exceeds the limit', async () => { + let inFlight = 0; + let peak = 0; + await mapWithConcurrency(Array.from({ length: 12 }, (_, i) => i), 3, async () => { + inFlight++; + peak = Math.max(peak, inFlight); + await tick(); + inFlight--; + }); + expect(peak).toBe(3); + }); + + // The behaviour every existing rollout depends on: unchanged unless asked for. + it('is strictly sequential at a limit of 1', async () => { + const order: number[] = []; + let inFlight = 0; + let peak = 0; + await mapWithConcurrency([1, 2, 3, 4], 1, async (n) => { + inFlight++; peak = Math.max(peak, inFlight); + await tick(); + order.push(n); + inFlight--; + }); + expect(peak).toBe(1); + expect(order).toEqual([1, 2, 3, 4]); + }); + + it('runs every item exactly once', async () => { + const seen: number[] = []; + await mapWithConcurrency(Array.from({ length: 25 }, (_, i) => i), 4, async (n) => { + await tick(1); + seen.push(n); + }); + expect(seen.sort((a, b) => a - b)).toEqual(Array.from({ length: 25 }, (_, i) => i)); + }); + + it('handles an empty list without starting anything', async () => { + let started = 0; + const out = await mapWithConcurrency([], 5, async () => { started++; return 1; }); + expect(out).toEqual([]); + expect(started).toBe(0); + }); + + it('does not start more runners than there is work', async () => { + let peak = 0; + let inFlight = 0; + await mapWithConcurrency([1, 2], 10, async () => { + inFlight++; peak = Math.max(peak, inFlight); + await tick(); + inFlight--; + }); + expect(peak).toBe(2); + }); + + // A rejecting worker would abandon the wave mid-flight; the orchestrator + // catches per device, but the helper must not be the thing that hides it. + it('propagates a worker rejection', async () => { + await expect( + mapWithConcurrency([1, 2, 3], 2, async (n) => { + if (n === 2) throw new Error('boom'); + return n; + }) + ).rejects.toThrow('boom'); + }); +}); + +describe('clampConcurrency', () => { + it('defaults to sequential for anything unusable', () => { + for (const bad of [undefined, null, '', 'abc', NaN, {}, []]) { + expect(clampConcurrency(bad)).toBe(1); + } + }); + + it('never returns less than 1', () => { + expect(clampConcurrency(0)).toBe(1); + expect(clampConcurrency(-8)).toBe(1); + }); + + // This number decides how many devices may be rebooting at once. + it('caps the upper end', () => { + expect(clampConcurrency(1000)).toBe(MAX_WAVE_CONCURRENCY); + expect(clampConcurrency(Infinity)).toBe(MAX_WAVE_CONCURRENCY); + }); + + it('floors fractional values', () => { + expect(clampConcurrency(3.9)).toBe(3); + expect(clampConcurrency('4')).toBe(4); + }); +}); diff --git a/backend/src/utils/__tests__/rolloutTopology.test.ts b/backend/src/utils/__tests__/rolloutTopology.test.ts new file mode 100644 index 0000000..4fc477e --- /dev/null +++ b/backend/src/utils/__tests__/rolloutTopology.test.ts @@ -0,0 +1,80 @@ +import { findUpstreamWithinSelection, type TopologyEdge } from '../rolloutTopology'; + +const link = ( + from: number, to: number | null, iface = 'ether1', role: string | null = 'root' +): TopologyEdge => ({ from_device_id: from, to_device_id: to, from_interface: iface, stp_role: role }); + +describe('findUpstreamWithinSelection', () => { + // The case the warning exists for: two devices reach the network through a + // switch, and all three are being upgraded together. + it('flags a device reached through, when the link is unambiguous', () => { + const links = [ + link(2, 1, 'ether1', 'root'), + link(3, 1, 'ether1', 'root'), + ]; + expect(findUpstreamWithinSelection([1, 2, 3], links)).toEqual([1]); + }); + + /** + * The failure that prompted the rewrite. Links are bidirectional and a trunk + * port often resolves to several neighbours, so counting adjacency flagged + * every device on a real four-device fleet. A warning that fires on + * everything trains people to ignore it. + */ + it('says nothing when a port resolves to several neighbours', () => { + const links = [ + link(7, 1, 'sfp28-2', 'root'), + link(7, 8, 'sfp28-2', 'root'), + link(7, 2, 'sfp28-2', 'root'), + ]; + expect(findUpstreamWithinSelection([1, 2, 7, 8], links)).toEqual([]); + }); + + it('is not fooled by the reverse direction of the same link', () => { + // Both ends report each other; only the root-port end means "upstream". + const links = [ + link(2, 1, 'ether1', 'root'), + link(1, 2, 'sfp1', 'designated'), + ]; + expect(findUpstreamWithinSelection([1, 2], links)).toEqual([1]); + }); + + it('ignores designated ports, which face away from the root', () => { + const links = [link(1, 2, 'sfp1', 'designated')]; + expect(findUpstreamWithinSelection([1, 2], links)).toEqual([]); + }); + + it('tolerates the -port suffix RouterOS sometimes reports', () => { + const links = [link(2, 1, 'ether1', 'root-port')]; + expect(findUpstreamWithinSelection([1, 2], links)).toEqual([1]); + }); + + it('ignores links to devices outside the selection', () => { + // A switch upstream of something not in this rollout is not a risk to it. + const links = [link(50, 1, 'ether1', 'root')]; + expect(findUpstreamWithinSelection([1, 2], links)).toEqual([]); + }); + + it('needs at least two selected devices', () => { + expect(findUpstreamWithinSelection([1], [link(2, 1)])).toEqual([]); + expect(findUpstreamWithinSelection([], [link(2, 1)])).toEqual([]); + }); + + it('tolerates unresolved neighbours, self links and missing roles', () => { + const links = [ + link(2, null, 'ether1', 'root'), + link(1, 1, 'ether1', 'root'), + link(3, 1, 'ether5', null), + link(2, 1, 'ether9', 'root'), + ]; + expect(findUpstreamWithinSelection([1, 2, 3], links)).toEqual([1]); + }); + + it('reports several upstream devices in a stable order', () => { + const links = [ + link(3, 2, 'ether1', 'root'), + link(4, 1, 'ether1', 'root'), + ]; + expect(findUpstreamWithinSelection([1, 2, 3, 4], links)).toEqual([1, 2]); + }); +}); diff --git a/backend/src/utils/concurrency.ts b/backend/src/utils/concurrency.ts new file mode 100644 index 0000000..b882c49 --- /dev/null +++ b/backend/src/utils/concurrency.ts @@ -0,0 +1,54 @@ +/** + * Run tasks with a ceiling on how many are in flight (issue #135). + * + * Firmware upgrades are not like bulk commands. A command is a short round + * trip, so CommandRunner can fire a whole wave at once with Promise.all. An + * upgrade holds an API connection for minutes across a download, a reboot and a + * verification — and connection pressure is precisely where this system has + * been shown to break. So concurrency is bounded rather than unlimited. + * + * Results are returned in input order regardless of completion order, because + * callers pair them with the devices they came from. + */ +export async function mapWithConcurrency( + items: T[], + limit: number, + worker: (item: T, index: number) => Promise +): Promise { + const max = Math.max(1, Math.floor(limit) || 1); + const results = new Array(items.length); + let next = 0; + + async function runner(): Promise { + while (true) { + const i = next++; + if (i >= items.length) return; + results[i] = await worker(items[i], i); + } + } + + // Never start more runners than there is work for them to do. + await Promise.all( + Array.from({ length: Math.min(max, items.length) }, () => runner()) + ); + return results; +} + +/** + * Clamp an operator-supplied concurrency to something sane. + * + * Rejecting silly values matters more here than elsewhere: this number decides + * how many devices can be rebooting simultaneously. + */ +export const MAX_WAVE_CONCURRENCY = 20; + +export function clampConcurrency(raw: unknown): number { + const n = Number(raw); + // NaN means "not a number at all" -> fall back to sequential. Infinity means + // "as many as possible", so it clamps to the cap: treating it as unusable + // would quietly do the opposite of what was asked. + if (Number.isNaN(n)) return 1; + if (n === Infinity) return MAX_WAVE_CONCURRENCY; + if (n === -Infinity) return 1; + return Math.min(MAX_WAVE_CONCURRENCY, Math.max(1, Math.floor(n))); +} diff --git a/backend/src/utils/rolloutTopology.ts b/backend/src/utils/rolloutTopology.ts new file mode 100644 index 0000000..a2e626f Binary files /dev/null and b/backend/src/utils/rolloutTopology.ts differ diff --git a/frontend/package.json b/frontend/package.json index 093f279..b98ba30 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-frontend", - "version": "0.24.2-beta", + "version": "0.24.3-beta", "private": true, "type": "module", "scripts": { diff --git a/frontend/src/pages/FirmwarePage.tsx b/frontend/src/pages/FirmwarePage.tsx index 134d469..184b30b 100644 --- a/frontend/src/pages/FirmwarePage.tsx +++ b/frontend/src/pages/FirmwarePage.tsx @@ -2,7 +2,7 @@ import { useMemo, useState } from 'react'; import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; import { ArrowUpCircle, RefreshCw, CheckCircle, XCircle, AlertTriangle, Clock, - HardDrive, ShieldAlert, Rocket, Ban, ChevronRight, Cpu, + HardDrive, ShieldAlert, Rocket, Ban, ChevronRight, Cpu, Zap, } from 'lucide-react'; import clsx from 'clsx'; import { firmwareApi } from '../services/api'; @@ -87,6 +87,11 @@ function RolloutPanel({ rolloutId, canWrite }: { rolloutId: number; canWrite: bo {rollout.pre_backup && pre-backup} {rollout.halt_on_failure && halt on failure} {rollout.routerboot_after && + RouterBOOT} + {(rollout.wave_concurrency ?? 1) > 1 && ( + + {rollout.wave_concurrency} at once + + )} {canWrite && active && (