Reduce background Git ref and port polling

BRANCH_DETAILS.md

@@ -0,0 +1,26 @@
+# Background Git Ref And Port Polling
+
+Frequently mounted Git-ref and preview-discovery surfaces stay fresh without continuously repeating their most expensive subprocess work.
+
+Expected behavior:
+
+- Git ref lists revalidate their first page every 20 seconds instead of every five seconds. Loaded cursor pages are one-shot snapshots, and inactive ref atoms expire after 30 seconds.
+- Opening the composer branch selector or Diff panel comparison-ref menu explicitly refreshes local and remote refs, so user interaction does not wait for the background interval.
+- Preview port discovery performs one immediate scan when the first subscriber retains it. Subscriptions replay the latest snapshot instead of initiating a duplicate scan.
+- Subscription replay and concurrent snapshot broadcasts are serialized so a stale replay cannot arrive after a newer scan result.
+- Managed terminal process-set changes trigger an immediate port scan; unchanged registrations and redundant removals do not.
+- The broad `lsof` safety-net scan runs every 20 seconds when no server is known and every 10 seconds while a listener is present. This preserves discovery for servers started outside T3-managed terminals without a permanent three-second system-wide process sweep.
+
+Primary files:
+
+- `packages/client-runtime/src/state/vcs.ts`
+- `apps/web/src/components/DiffPanel.tsx`
+- `apps/server/src/preview/PortScanner.ts`
+- `apps/server/src/ws.ts`
+
+Focused regression coverage lives in `apps/server/src/preview/PortScanner.test.ts`. Keep coverage for snapshot replay without rescanning, ordered replay during concurrent broadcasts, and unchanged terminal registrations avoiding redundant probes.
+
+## Development Ports
+
+- Web: `5740`
+- Server/WebSocket: `13780`

apps/server/src/preview/PortScanner.test.ts

@@ -4,8 +4,10 @@ import { it as effectIt } from "@effect/vitest";
 import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
 import * as Net from "@t3tools/shared/Net";
 import * as Cause from "effect/Cause";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
+import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as PlatformError from "effect/PlatformError";
 import { expect } from "vite-plus/test";
@@ -155,3 +157,111 @@ effectIt("does not swallow process probe interruption", () =>
     }
   }),
 );
+
+effectIt("replays snapshots without rescanning unchanged terminal registrations", () => {
+  let probeCount = 0;
+  let replayCount = 0;
+  const layer = makeProbeFailureLayer(() =>
+    Effect.sync(() => {
+      probeCount += 1;
+      return {
+        stdout: "",
+        stderr: "",
+        code: null,
+        timedOut: false,
+        stdoutTruncated: false,
+        stderrTruncated: false,
+      };
+    }),
+  );
+
+  return Effect.gen(function* () {
+    const scanner = yield* PortScanner.PortDiscovery;
+    yield* scanner.subscribe(() => Effect.void);
+    yield* scanner.retain;
+    yield* scanner.registerTerminalProcesses({
+      threadId: "thread-1",
+      terminalId: "terminal-1",
+      processIds: [42],
+    });
+    yield* scanner.registerTerminalProcesses({
+      threadId: "thread-1",
+      terminalId: "terminal-1",
+      processIds: [42],
+    });
+    yield* scanner.subscribe(() =>
+      Effect.sync(() => {
+        replayCount += 1;
+      }),
+    );
+
+    expect(probeCount).toBe(2);
+    expect(replayCount).toBe(1);
+  }).pipe(Effect.provide(layer));
+});
+
+effectIt("serializes snapshot replay with concurrent broadcasts", () =>
+  Effect.gen(function* () {
+    const replayStarted = yield* Deferred.make<void>();
+    const releaseReplay = yield* Deferred.make<void>();
+    const secondProbeCompleted = yield* Deferred.make<void>();
+    const secondDeliveryStarted = yield* Deferred.make<void>();
+    const deliveries: Array<ReadonlyArray<number>> = [];
+    let probeCount = 0;
+    let deliveryCount = 0;
+    const layer = makeProbeFailureLayer(() =>
+      Effect.gen(function* () {
+        probeCount += 1;
+        if (probeCount === 2) {
+          yield* Deferred.succeed(secondProbeCompleted, undefined).pipe(Effect.ignore);
+        }
+        return {
+          stdout: probeCount === 1 ? "p100\ncnode\nn*:3000\n" : "p101\ncnode\nn*:3001\n",
+          stderr: "",
+          code: null,
+          timedOut: false,
+          stdoutTruncated: false,
+          stderrTruncated: false,
+        };
+      }),
+    );
+
+    yield* Effect.gen(function* () {
+      const scanner = yield* PortScanner.PortDiscovery;
+      yield* scanner.retain;
+
+      const subscription = yield* scanner
+        .subscribe((servers) =>
+          Effect.gen(function* () {
+            deliveryCount += 1;
+            if (deliveryCount === 1) {
+              yield* Deferred.succeed(replayStarted, undefined).pipe(Effect.ignore);
+              yield* Deferred.await(releaseReplay);
+            } else {
+              yield* Deferred.succeed(secondDeliveryStarted, undefined).pipe(Effect.ignore);
+            }
+            deliveries.push(servers.map((server) => server.port));
+          }),
+        )
+        .pipe(Effect.forkScoped);
+      yield* Deferred.await(replayStarted);
+
+      const registration = yield* scanner
+        .registerTerminalProcesses({
+          threadId: "thread-1",
+          terminalId: "terminal-1",
+          processIds: [101],
+        })
+        .pipe(Effect.forkScoped);
+      yield* Deferred.await(secondProbeCompleted);
+      yield* Effect.yieldNow;
+      expect(yield* Deferred.isDone(secondDeliveryStarted)).toBe(false);
+
+      yield* Deferred.succeed(releaseReplay, undefined);
+      yield* Fiber.join(subscription);
+      yield* Fiber.join(registration);
+
+      expect(deliveries).toEqual([[3000], [3001]]);
+    }).pipe(Effect.provide(layer));
+  }),
+);

apps/server/src/preview/PortScanner.ts

@@ -21,7 +21,7 @@ import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Ref from "effect/Ref";
-import * as Schedule from "effect/Schedule";
+import * as Semaphore from "effect/Semaphore";
 import * as Scope from "effect/Scope";
 
 import * as ProcessRunner from "../processRunner.ts";
@@ -50,7 +50,8 @@ export const COMMON_DEV_PORTS: ReadonlyArray<number> = Object.freeze([
   3000, 3001, 3333, 4173, 4200, 4321, 5000, 5173, 5174, 5175, 5500, 8000, 8080, 8081, 8888, 9000,
 ]);
 
-const POLL_INTERVAL = Duration.seconds(3);
+const ACTIVE_POLL_INTERVAL = Duration.seconds(10);
+const IDLE_POLL_INTERVAL = Duration.seconds(20);
 const LSOF_TIMEOUT_MS = 5_000;
 const WINDOWS_LISTENER_TIMEOUT_MS = 5_000;
 
@@ -79,6 +80,9 @@ const terminalOwnerKey = (owner: {
   readonly terminalId: string;
 }): string => `${owner.threadId}\u0000${owner.terminalId}`;
 
+const processIdsEqual = (left: ReadonlySet<number>, right: ReadonlySet<number>): boolean =>
+  left.size === right.size && [...left].every((processId) => right.has(processId));
+
 const parseLsofOutput = (
   raw: string,
   terminalByProcessId: ReadonlyMap<number, TerminalProcessOwner> = new Map(),
@@ -190,6 +194,7 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
   const net = yield* Net.NetService;
   const processRunner = yield* ProcessRunner.ProcessRunner;
   const hostPlatform = yield* HostProcessPlatform;
+  const notificationLock = yield* Semaphore.make(1);
   const stateRef = yield* Ref.make<ScannerState>({
     lastSnapshot: [],
     listeners: new Set(),
@@ -292,25 +297,49 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
     yield* Effect.forEach(listeners, (listener) => listener(servers), { discard: true });
   });
 
+  const publishSnapshot = Effect.fn("PortDiscovery.publishSnapshot")(function* (
+    next: ReadonlyArray<DiscoveredLocalServer>,
+  ) {
+    yield* notificationLock.withPermit(
+      Effect.gen(function* () {
+        const changed = yield* Ref.modify(stateRef, (state) =>
+          serversEqual(state.lastSnapshot, next)
+            ? [false, state]
+            : [true, { ...state, lastSnapshot: next }],
+        );
+        if (changed) yield* broadcast(next);
+      }),
+    );
+  });
+
   const pollTick = Effect.fn("PortDiscovery.pollTick")(
     function* () {
       if ((yield* Ref.get(stateRef)).retainCount <= 0) return;
       const next = yield* scanOnce();
-      const changed = yield* Ref.modify(stateRef, (state) =>
-        serversEqual(state.lastSnapshot, next)
-          ? [false, state]
-          : [true, { ...state, lastSnapshot: next }],
-      );
-      if (changed) yield* broadcast(next);
+      yield* publishSnapshot(next);
     },
     Effect.catchCause((cause: Cause.Cause<never>) =>
       Effect.logWarning("preview port scan failed", Cause.pretty(cause)),
     ),
   );
 
-  // Single layer-scoped polling fiber. Ticks are no-ops when no client is
-  // currently retained, so the cost is one Ref.get every POLL_INTERVAL.
-  yield* Effect.forkScoped(pollTick().pipe(Effect.repeat(Schedule.spaced(POLL_INTERVAL))));
+  // Keep broad listener discovery as a fallback, but avoid a system-wide lsof
+  // process every three seconds while the app is otherwise idle. Terminal PID
+  // changes trigger immediate scans below; the periodic loop is only the
+  // safety net for listeners started outside a managed terminal.
+  yield* Effect.forkScoped(
+    Effect.gen(function* () {
+      while (true) {
+        const state = yield* Ref.get(stateRef);
+        yield* Effect.sleep(
+          state.retainCount > 0 && state.lastSnapshot.length > 0
+            ? ACTIVE_POLL_INTERVAL
+            : IDLE_POLL_INTERVAL,
+        );
+        yield* pollTick();
+      }
+    }),
+  );
 
   const acquireRetention = Effect.fn("PortDiscovery.retain")(function* () {
     const wasIdle = yield* Ref.modify(stateRef, (state) => [
@@ -334,16 +363,23 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
   const subscribe: PortDiscovery["Service"]["subscribe"] = Effect.fn("PortDiscovery.subscribe")(
     (listener) =>
       Effect.acquireRelease(
-        Ref.update(stateRef, (state) => ({
-          ...state,
-          listeners: new Set([...state.listeners, listener]),
-        })),
+        notificationLock.withPermit(
+          Ref.modify(stateRef, (state) => [
+            state.lastSnapshot,
+            {
+              ...state,
+              listeners: new Set([...state.listeners, listener]),
+            },
+          ]).pipe(Effect.tap(listener)),
+        ),
         () =>
-          Ref.update(stateRef, (state) => {
-            const listeners = new Set(state.listeners);
-            listeners.delete(listener);
-            return { ...state, listeners };
-          }),
+          notificationLock.withPermit(
+            Ref.update(stateRef, (state) => {
+              const listeners = new Set(state.listeners);
+              listeners.delete(listener);
+              return { ...state, listeners };
+            }),
+          ),
       ),
   );
 
@@ -356,26 +392,33 @@ export const make = Effect.gen(function* PortDiscoveryMake() {
       const processIds = new Set(
         input.processIds.filter((processId) => Number.isInteger(processId) && processId > 0),
       );
-      yield* Ref.update(stateRef, (state) => {
+      const changed = yield* Ref.modify(stateRef, (state) => {
         const terminalProcesses = new Map(state.terminalProcesses);
         const key = terminalOwnerKey(owner);
+        const existing = terminalProcesses.get(key);
+        if (existing && processIdsEqual(existing.processIds, processIds)) {
+          return [false, state] as const;
+        }
         if (processIds.size === 0) {
+          if (!existing) return [false, state] as const;
           terminalProcesses.delete(key);
         } else {
           terminalProcesses.set(key, { owner, processIds });
         }
-        return { ...state, terminalProcesses };
+        return [true, { ...state, terminalProcesses }] as const;
       });
+      if (changed) yield* pollTick();
     });
 
   const unregisterTerminal: PortDiscovery["Service"]["unregisterTerminal"] = Effect.fn(
     "PortDiscovery.unregisterTerminal",
   )(function* (input) {
-    yield* Ref.update(stateRef, (state) => {
+    const changed = yield* Ref.modify(stateRef, (state) => {
       const terminalProcesses = new Map(state.terminalProcesses);
-      terminalProcesses.delete(terminalOwnerKey(input));
-      return { ...state, terminalProcesses };
+      const removed = terminalProcesses.delete(terminalOwnerKey(input));
+      return [removed, removed ? { ...state, terminalProcesses } : state] as const;
     });
+    if (changed) yield* pollTick();
   });
 
   return PortDiscovery.of({

apps/server/src/ws.ts

@@ -1957,13 +1957,10 @@ const makeWsRpcLayer = (
             WS_METHODS.subscribeDiscoveredLocalServers,
             Stream.callback<DiscoveredLocalServerList>((queue) =>
               Effect.gen(function* () {
+                // Retention performs one immediate scan when discovery was
+                // idle. Subscribe replays that snapshot to every connection,
+                // including connections that join an already-retained scanner.
                 yield* portDiscovery.retain;
-                const initial = yield* portDiscovery.scan();
-                const initialScannedAt = DateTime.formatIso(yield* DateTime.now);
-                yield* Queue.offer(queue, {
-                  servers: initial,
-                  scannedAt: initialScannedAt,
-                });
                 yield* portDiscovery.subscribe((servers) =>
                   Effect.gen(function* () {
                     const scannedAt = DateTime.formatIso(yield* DateTime.now);

apps/web/src/components/DiffPanel.tsx

@@ -581,7 +581,12 @@ export default function DiffPanel({
               filteredItems={filteredBaseRefItems}
               value={selectedBaseRef ?? AUTOMATIC_BASE_REF}
               onOpenChange={(open) => {
-                if (!open) setBaseRefQuery("");
+                if (!open) {
+                  setBaseRefQuery("");
+                  return;
+                }
+                localBranchRefs.refresh();
+                remoteBranchRefs.refresh();
               }}
               onValueChange={(value) => {
                 if (!value) return;

packages/client-runtime/src/state/vcs.ts

@@ -23,7 +23,8 @@ import { followStreamInEnvironment } from "./runtime.ts";
 import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts";
 
 const OFFLINE_BRANCH_LIST_LIMIT = 100;
-const VCS_REFS_REVALIDATE_INTERVAL = "5 seconds";
+const VCS_REFS_REVALIDATE_INTERVAL = "20 seconds";
+const VCS_REFS_IDLE_TTL_MS = 30_000;
 
 function canUseVcsRefsCache(input: VcsListRefsInput): boolean {
   return (
@@ -107,7 +108,10 @@ export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChange
     Stream.switchMap((generation) =>
       generation === null
         ? Stream.empty
-        : Stream.tick(VCS_REFS_REVALIDATE_INTERVAL).pipe(
+        : (input.cursor === undefined
+            ? Stream.tick(VCS_REFS_REVALIDATE_INTERVAL)
+            : Stream.succeed(undefined)
+          ).pipe(
             Stream.mapEffect(
               () =>
                 refresh().pipe(
@@ -151,7 +155,7 @@ export function createVcsEnvironmentAtoms<R, E>(
       return runtime
         .atom(cachedVcsRefsChanges(environmentId, input))
         .pipe(
-          Atom.setIdleTTL(5 * 60_000),
+          Atom.setIdleTTL(VCS_REFS_IDLE_TTL_MS),
           Atom.withLabel(`environment-data:vcs:list-refs:${environmentId}:${inputKey}`),
         );
     }),