Fix native provisioning scheduling and progress reporting

src/lib/server/pipeline/worker.ts

@@ -16,6 +16,7 @@ import { STATUS_REFRESH_MS } from '../../server-freshness.js';
 import { recordMetrics } from '../fleet/metrics.js';
 import { selectedZombieMap } from '../../map-aliases.js';
 import { runMaintenance } from '../maintenance/index.js';
+import { trackActivity } from './activity.js';
 
 const TICK_MS = 15_000;
 // Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the
@@ -27,26 +28,38 @@ let tickTimer: NodeJS.Timeout | null = null;
 let statusTimer: NodeJS.Timeout | null = null;
 
 let tickInFlight: Promise<void> | null = null;
+let tickRequested = false;
+const requesterFlights = new Map<string, Promise<void>>();
 
-function tick(): Promise<void> {
+function tick(requestAnother = false): Promise<void> {
+    if (requestAnother) tickRequested = true;
     if (tickInFlight) return tickInFlight;
-    tickInFlight = runTick().finally(() => { tickInFlight = null; });
+    tickInFlight = (async () => {
+        do {
+            tickRequested = false;
+            try { await trackActivity('provisioning', 'Processing provisioning requests', () => processProvisioning({ wait: false })); }
+            catch (err) { console.error('[worker] provisioning error:', err); }
+        } while (tickRequested);
+    })().finally(() => { tickInFlight = null; });
     return tickInFlight;
 }
 
-async function reconcileRequesters(): Promise<void> {
-    try { await sweep(); } catch (err) { console.error('[worker] campaign error:', err); }
-    try { await betaTick(); } catch (err) { console.error('[worker] beta error:', err); }
-    try { await runMaintenance(); } catch (err) { console.error('[worker] maintenance error:', err); }
-}
-
-async function runTick(): Promise<void> {
-    // Callers submit durable requests first. Exactly one owner admits and runs
-    // lifecycle work, then callers observe the resulting jobs by ID.
-    await reconcileRequesters();
-    try { await processProvisioning(); }
-    catch (err) { console.error('[worker] provisioning error:', err); }
-    await reconcileRequesters();
+function reconcileRequesters(): void {
+    // Observers submit durable intent independently. Provisioning still owns
+    // admission and rechecks source eligibility immediately before mutation.
+    const requesters: [string, string, () => Promise<unknown>][] = [
+        ['campaign', 'Reconciling campaigns', sweep],
+        ['beta', 'Reconciling beta tests', betaTick],
+        ['maintenance', 'Checking maintenance and recovery', runMaintenance]
+    ];
+    for (const [key, label, work] of requesters) {
+        if (requesterFlights.has(key)) continue;
+        const flight = trackActivity(key, label, work)
+            .catch(err => { console.error(`[worker] ${key} error:`, err); })
+            .then(async () => { await tick(true); })
+            .finally(() => { requesterFlights.delete(key); });
+        requesterFlights.set(key, flight);
+    }
 }
 
 /**
@@ -185,8 +198,9 @@ export function startWorker(): void {
     started = true;
 
     void tick();
+    reconcileRequesters();
     void fanoutStatus();
-    tickTimer = setInterval(tick, TICK_MS);
+    tickTimer = setInterval(() => { void tick(); reconcileRequesters(); }, TICK_MS);
     statusTimer = setInterval(fanoutStatus, STATUS_REFRESH_MS);
 
     console.log('[worker] started — sweep every', TICK_MS, 'ms, status every', STATUS_REFRESH_MS, 'ms');
@@ -205,5 +219,9 @@ export function stopWorker(): void {
  * fires immediately after a state change instead of waiting for the next tick.
  */
 export async function kick(): Promise<void> {
-    await tick();
+    reconcileRequesters();
+    await tick(true);
 }
+
+/** Agent receipts need observation promptly, without launching another fleet scan. */
+export async function kickProvisioning(): Promise<void> { await tick(true); }