refactor(session-tabs): one authority-repair lane shared by every session-tabs mirror (#27093)

Co-authored-by: m4air <m4air@m4airs-MacBook-Air.local>
2026-10-31 00:57:45 +00:00 · 2026-10-09 19:39:00 -07:00
parent 746e4b42a0
commit 33d90bb066
4 changed files with 165 additions and 139 deletions
@@ -1,131 +1,23 @@
-/**
- * Repairing a structured session snapshot the retired-epoch fence rejected.
- *
- * The fence is right to reject a subscription frame carrying a retired epoch — it cannot tell that
- * frame apart from a delayed one queued by a dead publisher generation, whose version can be
- * higher than the live cursor. What it cannot do is notice when the epoch's publisher is actually
- * still alive and has simply returned after another publisher briefly owned the worktree.
- *
- * So the drop is not treated as final: it schedules one authoritative `session.tabs.listAll`, whose
- * answer settles which epoch is current. If the epoch really is dead the census changes nothing; if
- * it is live, the census carries it and the tab lands. The fence itself is never relaxed for
- * subscription frames.
- *
- * The refresh is injected rather than imported so this module depends on nothing that in turn
- * depends on the snapshot apply — which is what lets the apply prune this module's state.
- */
-
+import { createSessionTabsAuthorityRepairLane } from '../web-session-tabs-sync/session-tabs-authority-repair'
 import {
  isCurrentLocalStructuredSessionGeneration,
  localStructuredSessionEpochHistoryByWorktree,
  localStructuredSessionGeneration
 } from './inventory-generation-fence'

-/** Bounded so a publisher that keeps re-sending a retired epoch cannot drive an endless refetch. */
-const MAX_REPAIR_ATTEMPTS = 3
-const BASE_REPAIR_DELAY_MS = 250
-const MAX_REPAIR_DELAY_MS = 5000
-/**
- * The cap decays rather than latching. A run of transient RPC failures must not hide a chat tab for
- * the renderer's lifetime; once a worktree has been quiet this long, a fresh drop is a fresh
- * problem and gets its full budget back.
- */
-const REPAIR_ATTEMPT_DECAY_MS = 60_000
+/** The local structured-session mirror's lane, keyed by worktree. */
+const localRepairLane = createSessionTabsAuthorityRepairLane({
+  logLabel: 'structured-session-tabs',
+  generation: localStructuredSessionGeneration,
+  isCurrent: isCurrentLocalStructuredSessionGeneration,
+  isStillRetired: (worktreeId, publicationEpoch) =>
+    localStructuredSessionEpochHistoryByWorktree
+      .get(worktreeId)
+      ?.retired.includes(publicationEpoch) ?? false
+})

-type RepairState = {
-  attempts: number
-  lastAttemptAt: number
-  timer: ReturnType<typeof setTimeout> | null
-}
-
-export type RetiredEpochRepairRunner = (expectedGeneration: number) => Promise<unknown>
-
-const repairsByWorktree = new Map<string, RepairState>()
-
-function repairState(worktreeId: string, now: number): RepairState {
-  const existing = repairsByWorktree.get(worktreeId)
-  if (!existing) {
-    const created: RepairState = { attempts: 0, lastAttemptAt: now, timer: null }
-    repairsByWorktree.set(worktreeId, created)
-    return created
-  }
-  if (now - existing.lastAttemptAt >= REPAIR_ATTEMPT_DECAY_MS) {
-    existing.attempts = 0
-  }
-  return existing
-}
-
-/**
- * Schedules the authoritative refetch for a dropped snapshot, coalescing repeat drops for the same
- * worktree into the one already pending.
- */
-export function scheduleRetiredEpochRepair(
-  worktreeId: string,
-  publicationEpoch: string,
-  runRepair: RetiredEpochRepairRunner
-): void {
-  const now = Date.now()
-  const state = repairState(worktreeId, now)
-  if (state.timer !== null) {
-    return
-  }
-  if (state.attempts >= MAX_REPAIR_ATTEMPTS) {
-    console.warn('[structured-session-tabs] retired publication epoch still unrepaired', {
-      worktree: worktreeId,
-      publicationEpoch,
-      attempts: state.attempts,
-      retryAfterMs: Math.max(0, REPAIR_ATTEMPT_DECAY_MS - (now - state.lastAttemptAt))
-    })
-    return
-  }
-  const generation = localStructuredSessionGeneration()
-  const delay = Math.min(BASE_REPAIR_DELAY_MS * 2 ** state.attempts, MAX_REPAIR_DELAY_MS)
-  state.attempts += 1
-  state.lastAttemptAt = now
-  state.timer = setTimeout(() => {
-    state.timer = null
-    if (!isCurrentLocalStructuredSessionGeneration(generation)) {
-      repairsByWorktree.delete(worktreeId)
-      return
-    }
-    void runRepair(generation)
-      .then(() => {
-        // Why re-check rather than trust the call: a refresh that succeeds without reviving the
-        // epoch has not repaired anything, and counting it as success would loop forever.
-        const stillRetired =
-          localStructuredSessionEpochHistoryByWorktree
-            .get(worktreeId)
-            ?.retired.includes(publicationEpoch) ?? false
-        if (!stillRetired) {
-          repairsByWorktree.delete(worktreeId)
-        }
-      })
-      .catch((error) => {
-        console.warn('[structured-session-tabs] retired-epoch repair refresh failed', error)
-      })
-  }, delay)
-}
-
-/**
- * Drops repair state for worktrees that no longer exist, alongside the publisher cursors it
- * shadows — without this every deleted worktree leaks an entry for the renderer's lifetime.
- */
+export const scheduleRetiredEpochRepair = localRepairLane.schedule
 export function forgetRetiredEpochRepairsOutside(knownWorktreeIds: ReadonlySet<string>): void {
-  for (const [worktreeId, state] of repairsByWorktree) {
-    if (!knownWorktreeIds.has(worktreeId)) {
-      if (state.timer !== null) {
-        clearTimeout(state.timer)
-      }
-      repairsByWorktree.delete(worktreeId)
-    }
-  }
-}
-
-export function resetRetiredEpochRepairsForTests(): void {
-  for (const state of repairsByWorktree.values()) {
-    if (state.timer !== null) {
-      clearTimeout(state.timer)
-    }
-  }
-  repairsByWorktree.clear()
+  localRepairLane.forget((worktreeId) => !knownWorktreeIds.has(worktreeId))
 }
+export const resetRetiredEpochRepairsForTests = localRepairLane.resetForTests
@@ -11,9 +11,8 @@ import {
 } from '../web-session-tabs-sync'
 import type { WebSessionTabsSyncState } from '../web-session-tabs-sync'
 import {
-  hasRetiredValue,
+  admitRetiredValueUnderAuthority,
  noteRetiredValue,
-  reviveRetiredValue,
  sameSessionTabsPublicationLineage
 } from '../web-session-tabs-sync/publisher-identity-fences'
 import { knownStructuredSessionWorktreeIds } from '../local-structured-session-tab-retirement'
@@ -130,12 +129,16 @@ export function applyLocalStructuredSessionTabSnapshots<
    // headless rebuild), and the structured publish inherits the worktree's existing epoch rather
    // than minting its own. So a retired epoch is not proof of a dead generation — only authority
    // can settle it, and the repair lane goes and asks.
-    if (hasRetiredValue(epochHistory, snapshot.publicationEpoch) && !sharesLineage) {
-      if (!options.authoritative) {
-        options.onRetiredEpochDrop?.(snapshot.worktree, snapshot.publicationEpoch)
-        continue
-      }
-      reviveRetiredValue(epochHistory, snapshot.publicationEpoch)
+    if (
+      !sharesLineage &&
+      !admitRetiredValueUnderAuthority(
+        epochHistory,
+        snapshot.publicationEpoch,
+        options.authoritative === true
+      )
+    ) {
+      options.onRetiredEpochDrop?.(snapshot.worktree, snapshot.publicationEpoch)
+      continue
    }
    if (prior && sharesLineage && snapshot.snapshotVersion <= prior.snapshotVersion) {
      continue
@@ -11,7 +11,7 @@ import {
 const SESSION_TABS_RETIRED_EPOCH_LIMIT = 8
 const SESSION_TABS_RETIRED_RUNTIME_ID_LIMIT = 8

-export function hasRetiredValue(history: RetiredValueHistory | undefined, value: string): boolean {
+function hasRetiredValue(history: RetiredValueHistory | undefined, value: string): boolean {
  return history?.retired.includes(value) ?? false
 }

@@ -37,16 +37,23 @@ export function noteRetiredValue(
 }

 /**
- * Un-retires one value, leaving every other retired generation fenced.
- *
- * Only an authority that names the value current may call this; reviving on a delayed frame's own
- * say-so is exactly the resurrection `retired` exists to prevent.
+ * Whether a snapshot under `value` may apply. A retired value applies only on an authoritative
+ * census that names it, which revives just that value (or its lineage); anything else stays fenced.
 */
-export function reviveRetiredValue(history: RetiredValueHistory | undefined, value: string): void {
-  const index = history?.retired.indexOf(value) ?? -1
-  if (history && index >= 0) {
-    history.retired.splice(index, 1)
+export function admitRetiredValueUnderAuthority(
+  history: RetiredValueHistory | undefined,
+  value: string,
+  authoritative: boolean,
+  sameValue: (retired: string, value: string) => boolean = (retired, candidate) =>
+    retired === candidate
+): boolean {
+  if (!history?.retired.some((retired) => sameValue(retired, value))) {
+    return true
  }
+  if (authoritative) {
+    history.retired = history.retired.filter((retired) => !sameValue(retired, value))
+  }
+  return authoritative
 }

 function normalizeSessionTabsRuntimeId(runtimeId: unknown): string | undefined {
@@ -0,0 +1,124 @@
+/**
+ * Repairing a session-tabs snapshot the retired-epoch fence rejected.
+ *
+ * The fence is right to reject a subscription frame carrying a retired epoch — it cannot tell that
+ * frame apart from a delayed one queued by a dead publisher generation, whose version can be
+ * higher than the live cursor. What it cannot do is notice when the epoch's publisher is actually
+ * still alive and has simply returned after another publisher briefly owned the worktree.
+ *
+ * So the drop is not treated as final: it schedules one authoritative census, whose answer settles
+ * which epoch is current. If the epoch really is dead the census changes nothing; if it is live,
+ * the census carries it and the tab lands. The fence itself is never relaxed for subscription
+ * frames.
+ *
+ * Each mirror (local structured sessions, each paired environment) owns one lane. The census and
+ * the generation fence are injected so this module depends on nothing that depends on the apply.
+ */
+
+/** Bounded so a publisher that keeps re-sending a retired epoch cannot drive an endless refetch. */
+const MAX_REPAIR_ATTEMPTS = 3
+const BASE_REPAIR_DELAY_MS = 250
+const MAX_REPAIR_DELAY_MS = 5000
+/**
+ * The cap decays rather than latching. A run of transient RPC failures must not hide a tab for the
+ * renderer's lifetime; once a worktree has been quiet this long, a fresh drop is a fresh problem
+ * and gets its full budget back.
+ */
+const REPAIR_ATTEMPT_DECAY_MS = 60_000
+
+type RepairState = {
+  attempts: number
+  lastAttemptAt: number
+  timer: ReturnType<typeof setTimeout> | null
+}
+
+export type RetiredEpochRepairRunner = (expectedGeneration: number) => Promise<unknown>
+
+export type SessionTabsAuthorityRepairLane = {
+  /** Schedules the census for a dropped snapshot, coalescing repeat drops for one key. */
+  schedule(key: string, publicationEpoch: string, runRepair: RetiredEpochRepairRunner): void
+  /** Drops state for keys that no longer exist, so deleted worktrees do not leak entries. */
+  forget(isGone: (key: string) => boolean): void
+  resetForTests(): void
+}
+
+export function createSessionTabsAuthorityRepairLane(fence: {
+  logLabel: string
+  /** Omitted when the runner fences itself (the paired mirror checks its own subscription). */
+  generation?: () => number
+  isCurrent?: (expectedGeneration: number) => boolean
+  /** Why re-check rather than trust the census: one that does not revive the epoch repaired nothing. */
+  isStillRetired: (key: string, publicationEpoch: string) => boolean
+}): SessionTabsAuthorityRepairLane {
+  const repairsByKey = new Map<string, RepairState>()
+
+  const repairState = (key: string, now: number): RepairState => {
+    const existing = repairsByKey.get(key)
+    if (!existing) {
+      const created: RepairState = { attempts: 0, lastAttemptAt: now, timer: null }
+      repairsByKey.set(key, created)
+      return created
+    }
+    if (now - existing.lastAttemptAt >= REPAIR_ATTEMPT_DECAY_MS) {
+      existing.attempts = 0
+    }
+    return existing
+  }
+
+  const clearAll = (keep: (key: string) => boolean): void => {
+    for (const [key, state] of repairsByKey) {
+      if (keep(key)) {
+        continue
+      }
+      if (state.timer !== null) {
+        clearTimeout(state.timer)
+      }
+      repairsByKey.delete(key)
+    }
+  }
+
+  return {
+    schedule(key, publicationEpoch, runRepair) {
+      const now = Date.now()
+      const state = repairState(key, now)
+      if (state.timer !== null) {
+        return
+      }
+      if (state.attempts >= MAX_REPAIR_ATTEMPTS) {
+        console.warn(`[${fence.logLabel}] retired publication epoch still unrepaired`, {
+          worktree: key,
+          publicationEpoch,
+          attempts: state.attempts,
+          retryAfterMs: Math.max(0, REPAIR_ATTEMPT_DECAY_MS - (now - state.lastAttemptAt))
+        })
+        return
+      }
+      const generation = fence.generation?.() ?? 0
+      const delay = Math.min(BASE_REPAIR_DELAY_MS * 2 ** state.attempts, MAX_REPAIR_DELAY_MS)
+      state.attempts += 1
+      state.lastAttemptAt = now
+      state.timer = setTimeout(() => {
+        state.timer = null
+        if (fence.isCurrent && !fence.isCurrent(generation)) {
+          repairsByKey.delete(key)
+          return
+        }
+        void runRepair(generation)
+          .then(() => {
+            if (!fence.isStillRetired(key, publicationEpoch)) {
+              repairsByKey.delete(key)
+            }
+          })
+          .catch((error) => {
+            console.warn(`[${fence.logLabel}] retired-epoch repair refresh failed`, error)
+          })
+      }, delay)
+    },
+    forget(isGone) {
+      clearAll((key) => !isGone(key))
+    },
+    resetForTests() {
+      clearAll(() => false)
+    }
+  }
+}