feat(firmware): optional parallel upgrades within a wave (v0.24.3-beta)

Closes #135. The reporter's numbers: CRS switches take ~5 minutes each to
reboot, so ten devices is an hour of mostly waiting.

Devices in a wave can now upgrade concurrently, up to a per-rollout limit.
Waves themselves stay strictly sequential — that is the decision point, and
what makes wave 1 a canary. The default is 1, which is the behaviour every
existing rollout was built around, and existing rows backfill to it.

CommandRunner already runs a wave with Promise.all, but an upgrade is not a
bulk command: it holds an API connection for minutes across a download, a
reboot and a verification, and connection pressure is exactly where this
system has been shown to break. So concurrency is bounded rather than
unlimited, via mapWithConcurrency, capped at 20 and clamped server-side.

The trade is stated rather than hidden. Halting is now evaluated at the end
of a wave instead of the instant a device fails, so a wave of five can put a
bad build on five devices before stopping. At the default of 1 those are the
same moment. The UI says so plainly whenever the setting is above 1.

An upstream warning covers the risk Change Guard cannot: several devices
rebooting together is fine for independent routers and dangerous for a switch
the others are reached through, and a device unreachable because its uplink
rebooted has no config change to revert.

The first implementation of that warning was wrong in an instructive way. It
flagged any device another selected device linked to — and on the real
four-device fleet it flagged all four, because discovered links are
bidirectional and a trunk port often resolves to several neighbours at once.
A warning that fires on everything is worse than none: it teaches people to
dismiss it. It now requires the link to be the reporting device's STP root
port *and* that port to resolve to exactly one neighbour, so it stays quiet
when it cannot tell — which, on this fleet's ambiguous topology, it correctly
does.

Verified on hardware with devices already up to date, so the wave machinery
ran end to end without flashing anything:

  - sequential: 04:20:21→:26, then :26→:31. No overlap, 12s.
  - concurrency 2: both started within 10ms of each other, 7s.
  - concurrency 5 across two waves: wave 2 started 3ms *after* wave 1
    finished, never during it.

Clamping checked at the API: 999 stored as 20, 0 and "abc" as 1.

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

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).
 --

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.

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}`);

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);
+  });
+});

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]);
+  });
+});

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<T, R>(
+  items: T[],
+  limit: number,
+  worker: (item: T, index: number) => Promise<R>
+): Promise<R[]> {
+  const max = Math.max(1, Math.floor(limit) || 1);
+  const results = new Array<R>(items.length);
+  let next = 0;
+
+  async function runner(): Promise<void> {
+    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)));
+}