Add failover channel events to activity journal

src/server/features/failover/failoverService.ts

@@ -91,7 +91,42 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
   let timer: NodeJS.Timeout | null = null;
   let collectorEnabled: boolean | null = null;
   let roundPromise: Promise<void> | null = null;
+  let roundGeneration: number | null = null;
   let decisionMemory: FailoverDecisionMemory | undefined;
+  const healthMemory: Record<FailoverRole, 'healthy' | 'unhealthy' | undefined> = {
+    primary: undefined,
+    reserve: undefined,
+  };
+  function clearHealthMemory() {
+    healthMemory.primary = undefined;
+    healthMemory.reserve = undefined;
+  }
+  function recordHealthTransition(
+    role: FailoverRole,
+    health: FailoverHealth,
+    capturedGeneration?: number,
+  ) {
+    if (capturedGeneration !== undefined && capturedGeneration !== generation) return;
+    if (health !== 'healthy' && health !== 'unhealthy') return;
+    const previous = healthMemory[role];
+    healthMemory[role] = health;
+    if (previous === health) return;
+    if (previous === undefined && health === 'healthy') return;
+    const type = health === 'unhealthy'
+      ? role === 'primary' ? 'failover.primary_unavailable' : 'failover.reserve_unavailable'
+      : role === 'primary' ? 'failover.primary_recovered' : 'failover.reserve_recovered';
+    dependencies.onEvent?.({
+      type,
+      severity: health === 'unhealthy' ? 'warning' : 'info',
+      source: 'failover',
+      dedupeKey: null,
+      data: {
+        role,
+        reason: health === 'unhealthy' ? 'probe-failed' : 'probe-recovered',
+      },
+    });
+  }
+
   let snapshot = createIdleFailoverSnapshot(dependencies.state.read().failoverPolicy, epoch, sequence);
 
   function appliedMatchesDesired(state: StoredState) {
@@ -133,6 +168,7 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
     generation += 1;
     clearTimer();
     decisionMemory = undefined;
+    clearHealthMemory();
     await disableCollector();
     const idle = createIdleFailoverSnapshot(policy, epoch, sequence);
     if (passiveRole) {
@@ -181,6 +217,7 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
     const running = await dependencies.runtime.isRunning();
     if (!running || !state.appliedFailoverPolicy || !currentRole(state)) {
       generation += 1;
+      clearHealthMemory();
       clearTimer();
       await disableCollector();
       const pending = activeSnapshot(state, 'idle');
@@ -355,6 +392,8 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
       reserve = { health: 'unknown' as const, failingServiceIds: [] };
     }
     if (capturedGeneration !== generation || !dependencies.state.read().failoverPolicy.enabled) return;
+    recordHealthTransition('primary', primary.health, capturedGeneration);
+    recordHealthTransition('reserve', reserve.health, capturedGeneration);
     const activityResponse = record(await dependencies.dataplane.readFailoverActivity(
       policy.trafficGuard.thresholdBytesPerSecond,
     ));
@@ -437,15 +476,6 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
         data: { reason: decision.reason },
       });
     }
-    if (primary.health === 'healthy' && previousSnapshot.primary.health === 'unhealthy') {
-      dependencies.onEvent?.({
-        type: 'failover.recovered',
-        severity: 'info',
-        source: 'failover',
-        dedupeKey: `failover.recovered:${state.revision}:primary`,
-        data: { role: 'primary', reason: decision.reason },
-      });
-    }
     if (decision.switchTo) {
       try {
         const switched = await dependencies.serialize(async () => {
@@ -522,11 +552,25 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
     }
   }
 
-  async function runRound() {
-    if (roundPromise) return roundPromise;
+  async function runRound(): Promise<void> {
+    if (roundPromise) {
+      const pending = roundPromise;
+      if (roundGeneration === generation) return pending;
+      try {
+        await pending;
+      } catch {
+        // The original caller owns the stale round error; continue with current-generation work.
+      }
+      if (roundPromise && roundPromise !== pending) return roundPromise;
+      return runRound();
+    }
     const capturedGeneration = generation;
-    roundPromise = performRound(capturedGeneration).finally(() => {
-      roundPromise = null;
+    let trackedPromise: Promise<void>;
+    trackedPromise = performRound(capturedGeneration).finally(() => {
+      if (roundPromise === trackedPromise) {
+        roundPromise = null;
+        roundGeneration = null;
+      }
       if (capturedGeneration === generation && snapshot.reason === 'checking-channels') {
         const failed = activeSnapshot(dependencies.state.read(), 'error');
         failed.reason = 'health-unknown';
@@ -541,7 +585,9 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
         schedule(Math.min(policy.intervalMs, decisionDelay));
       }
     });
-    return roundPromise;
+    roundPromise = trackedPromise;
+    roundGeneration = capturedGeneration;
+    return trackedPromise;
   }
 
   function save(value: unknown) {
@@ -574,6 +620,7 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
       };
     });
     decisionMemory = undefined;
+    if (before.failoverPolicy.enabled !== policy.enabled) clearHealthMemory();
     if (before.failoverPolicy.enabled !== policy.enabled) {
       dependencies.onEvent?.({
         type: policy.enabled ? 'failover.enabled' : 'failover.disabled',
@@ -637,7 +684,7 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
       if (!state.failoverPolicy.enabled || !state.appliedFailoverPolicy || !currentRole(state)) {
         throw new HarborError('REQUEST_INVALID');
       }
-      generation += 1;
+      const capturedGeneration = ++generation;
       clearTimer();
       const checkedAt = now().toISOString();
       publish({ ...snapshot, reason: 'checking-channels' });
@@ -652,6 +699,9 @@ export function createFailoverService(dependencies: FailoverServiceDependencies)
         primary = { health: 'unknown' as const, failingServiceIds: [] };
         reserve = { health: 'unknown' as const, failingServiceIds: [] };
       }
+      if (capturedGeneration !== generation || !dependencies.state.read().failoverPolicy.enabled) return;
+      recordHealthTransition('primary', primary.health, capturedGeneration);
+      recordHealthTransition('reserve', reserve.health, capturedGeneration);
       const next = activeSnapshot(dependencies.state.read(), 'observing');
       next.primary = { ...next.primary, ...primary, checkedAt, stateSince: checkedAt };
       next.reserve = { ...next.reserve, ...reserve, checkedAt, stateSince: checkedAt };