diff --git a/BRANCH_DETAILS.md b/BRANCH_DETAILS.md new file mode 100644 index 00000000000..e00cf781368 --- /dev/null +++ b/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` diff --git a/apps/server/src/preview/PortScanner.test.ts b/apps/server/src/preview/PortScanner.test.ts index 69b5729164d..fb0ac2acbc2 100644 --- a/apps/server/src/preview/PortScanner.test.ts +++ b/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(); + const releaseReplay = yield* Deferred.make(); + const secondProbeCompleted = yield* Deferred.make(); + const secondDeliveryStarted = yield* Deferred.make(); + const deliveries: Array> = []; + 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)); + }), +); diff --git a/apps/server/src/preview/PortScanner.ts b/apps/server/src/preview/PortScanner.ts index c306fca2b33..a0c3bf3a1ec 100644 --- a/apps/server/src/preview/PortScanner.ts +++ b/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 = 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, right: ReadonlySet): boolean => + left.size === right.size && [...left].every((processId) => right.has(processId)); + const parseLsofOutput = ( raw: string, terminalByProcessId: ReadonlyMap = 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({ 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, + ) { + 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) => 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({ diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 08b1770a0a2..7cd876d8747 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1957,13 +1957,10 @@ const makeWsRpcLayer = ( WS_METHODS.subscribeDiscoveredLocalServers, Stream.callback((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); diff --git a/apps/web/src/components/DiffPanel.tsx b/apps/web/src/components/DiffPanel.tsx index 5cb2a1cfdea..7a9f283da73 100644 --- a/apps/web/src/components/DiffPanel.tsx +++ b/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; diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index 4ef94c2619f..9a286463a4e 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/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( 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}`), ); }),