perf(server): defer maintenance until initial client delivery

apps/server/src/httpResponseErrorGuard.test.ts

@@ -3,7 +3,10 @@ import * as NodeHttp from "node:http";
 import * as NodeNet from "node:net";
 import { afterEach, describe, expect, it } from "vite-plus/test";
 
-import { guardHttpResponseWriteErrors } from "./httpResponseErrorGuard.ts";
+import {
+  guardHttpResponseWriteErrors,
+  holdRequestsUntilServing,
+} from "./httpResponseErrorGuard.ts";
 
 const servers: NodeHttp.Server[] = [];
 
@@ -101,3 +104,39 @@ describe("guardHttpResponseWriteErrors", () => {
     expect(writeErrors).toEqual([]);
   });
 });
+
+it("serves HTTP and websocket requests received before handlers attach", async () => {
+  const server = holdRequestsUntilServing(guardHttpResponseWriteErrors(NodeHttp.createServer()));
+  const port = await listen(server);
+  const parsed = Promise.withResolvers<void>();
+  let received = 0;
+  server.on("connection", (socket) =>
+    socket.once("data", () => {
+      if (++received === 2) parsed.resolve();
+    }),
+  );
+  const response = fetchStatus(port, "/early");
+  const client = NodeNet.connect(port, "127.0.0.1");
+  const upgrade = new Promise<string>((resolve, reject) => {
+    client.once("connect", () =>
+      client.write(
+        "GET /ws HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
+      ),
+    );
+    client.once("data", (data) => resolve(data.toString().split("\r\n")[0]!));
+    client.once("error", reject);
+  });
+  try {
+    await parsed.promise;
+    server.on("request", (_, outgoing) => {
+      outgoing.writeHead(204);
+      outgoing.end();
+    });
+    server.on("upgrade", (_, socket) => socket.end("HTTP/1.1 101 Switching Protocols\r\n\r\n"));
+    await expect(response).resolves.toBe(204);
+    await expect(upgrade).resolves.toBe("HTTP/1.1 101 Switching Protocols");
+  } finally {
+    client.destroy();
+    server.closeAllConnections();
+  }
+});

apps/server/src/httpResponseErrorGuard.ts

@@ -22,3 +22,29 @@ export function guardHttpResponseWriteErrors<T extends NodeHttp.Server>(
   });
   return server;
 }
+
+/** Node starts listening before HttpRouter.serve attaches its handlers. */
+export function holdRequestsUntilServing<T extends NodeHttp.Server>(server: T): T {
+  for (const event of ["request", "upgrade"] as const) {
+    const existing = new Set(server.listeners(event));
+    const parked: Array<ReadonlyArray<unknown>> = [];
+    const park = (...args: ReadonlyArray<unknown>) => {
+      parked.push(args);
+    };
+    const onNewListener = (name: string | symbol) => {
+      if (name !== event) return;
+      server.off("newListener", onNewListener);
+      server.off(event, park);
+      // newListener fires before attachment. Replay before subsequent connection I/O.
+      queueMicrotask(() => {
+        const handlers = server.listeners(event).filter((listener) => !existing.has(listener));
+        for (const args of parked.splice(0)) {
+          for (const handler of handlers) handler.call(server, ...args);
+        }
+      });
+    };
+    server.on(event, park);
+    server.on("newListener", onNewListener);
+  }
+  return server;
+}

apps/server/src/orchestration-v2/PullRequestSyncReactor.ts

@@ -31,7 +31,7 @@ import * as Stream from "effect/Stream";
 import * as GitManager from "../git/GitManager.ts";
 import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground, forkParked } from "../serverActivation.ts";
 import * as Orchestrator from "./Orchestrator.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 import { isTerminalRunStatus } from "./ThreadManagementService.ts";
@@ -463,7 +463,7 @@ export const make = Effect.gen(function* () {
         }
       }).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))),
     );
-    yield* forkParked(
+    yield* forkBackground(
       Effect.gen(function* () {
         yield* worker.enqueue("all");
         yield* worker.drain;

apps/server/src/orchestration-v2/ThreadPullRequestService.ts

@@ -27,7 +27,7 @@ import * as GitManager from "../git/GitManager.ts";
 import * as ProjectStore from "./ProjectStore.ts";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
 import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground, forkParked } from "../serverActivation.ts";
 import * as Orchestrator from "./Orchestrator.ts";
 
 class ThreadPullRequestServiceV2 extends Context.Service<
@@ -390,7 +390,7 @@ export const make = Effect.gen(function* () {
     "ThreadPullRequestServiceV2.start",
   )(function* () {
     yield* forkParked(Stream.runForEach(orchestrator.streamDomainEvents, processEvent));
-    yield* forkParked(
+    yield* forkBackground(
       Effect.gen(function* () {
         yield* worker.enqueue({ threadId: null, refresh: false, backfill: true });
         yield* worker.drain;

apps/server/src/orchestration-v2/ThreadSettlementService.ts

@@ -23,7 +23,7 @@ import * as GitManager from "../git/GitManager.ts";
 import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
 import * as ServerSettings from "../serverSettings.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground, forkParked } from "../serverActivation.ts";
 import * as TerminalManager from "../terminal/Manager.ts";
 import * as ProjectStore from "./ProjectStore.ts";
 import * as Orchestrator from "./Orchestrator.ts";
@@ -627,7 +627,7 @@ export const make = Effect.gen(function* () {
     const events = orchestrator.streamDomainEvents;
     const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
     let lastSettlementSettings = autoSettlementSettingsKey(initialSettings);
-    yield* forkParked(
+    yield* forkBackground(
       Effect.gen(function* () {
         yield* worker.enqueue(undefined);
         yield* worker.drain;

apps/server/src/provider/ProviderHostLive.ts

@@ -8,6 +8,7 @@ import { resolveAttachmentPath } from "../attachmentStore.ts";
 import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
 import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
 import * as ServerConfig from "../config.ts";
+import * as ServerActivation from "../serverActivation.ts";
 import * as ServerSettings from "../serverSettings.ts";
 import * as ProviderCredentialStore from "./ProviderCredentialStore.ts";
 
@@ -19,6 +20,7 @@ export const layer = Layer.effect(
     const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
     const secrets = yield* ServerSecretStore.ServerSecretStore;
     const crypto = yield* Crypto.Crypto;
+    const background = yield* ServerActivation.ServerBackgroundStart;
     return ProviderHost.ProviderHost.of({
       paths: {
         cwd: config.cwd,
@@ -34,6 +36,7 @@ export const layer = Layer.effect(
         subscribe: serverSettings.subscribeChanges,
       },
       shouldRunBackgroundWork: backgroundPolicy.shouldRunScopeWork,
+      ...(background === undefined ? {} : { awaitStartupProbe: background.awaitTurn }),
       resolveAttachmentPath: (attachment) =>
         resolveAttachmentPath({ attachmentsDir: config.attachmentsDir, attachment }),
       credentials: (namespace, bindingId) =>

apps/server/src/pullRequest/PullRequestReadCache.ts

@@ -21,7 +21,7 @@ import * as Persistable from "effect/persistence/Persistable";
 import * as PersistedCache from "effect/persistence/PersistedCache";
 import * as Persistence from "effect/persistence/Persistence";
 import * as ServerConfig from "../config.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground } from "../serverActivation.ts";
 
 const CONCURRENT_READS = 512;
 // Persistence prefixes every entry key with the store id, so each entry file
@@ -216,7 +216,7 @@ export const layer = Layer.unwrap(
       Layer.provide(
         KeyValueStore.layerFileSystem(directory).pipe(
           Layer.tap(() =>
-            forkParked(
+            forkBackground(
               pruneExpiredEntryFiles(directory).pipe(
                 Effect.catch((cause) =>
                   Effect.logWarning("Failed to prune PR cache files", { cause }),

apps/server/src/server.ts

@@ -30,7 +30,10 @@ import * as HostPowerMonitor from "./background/HostPowerMonitor.ts";
 import * as ServerConfig from "./config.ts";
 import { withUntracedRequests } from "./http.ts";
 import * as ServerHttp from "./http.ts";
-import { guardHttpResponseWriteErrors } from "./httpResponseErrorGuard.ts";
+import {
+  guardHttpResponseWriteErrors,
+  holdRequestsUntilServing,
+} from "./httpResponseErrorGuard.ts";
 import { fixPath } from "./os-jank.ts";
 import * as Ws from "./ws.ts";
 import * as AgentScopeLive from "./process/agentScope.ts";
@@ -287,16 +290,19 @@ const layerRelayClient = Layer.unwrap(
 const layerHttpServer = Layer.unwrap(
   Effect.gen(function* () {
     const config = yield* ServerConfig.ServerConfig;
-    return NodeHttpServer.layer(() => guardHttpResponseWriteErrors(NodeHttp.createServer()), {
-      host: config.host ?? "127.0.0.1",
-      port: config.port,
-      gracefulShutdownTimeout: HTTP_PREEMPTIVE_SHUTDOWN_GRACE_MS,
-      // Context takeover stays enabled (ws default) so the compression
-      // window is shared across frames — that also makes small frames cheap
-      // to compress, so no size threshold is set (ws only honors
-      // `threshold` when context takeover is disabled).
-      websocket: { perMessageDeflate: true },
-    });
+    return NodeHttpServer.layer(
+      () => holdRequestsUntilServing(guardHttpResponseWriteErrors(NodeHttp.createServer())),
+      {
+        host: config.host ?? "127.0.0.1",
+        port: config.port,
+        gracefulShutdownTimeout: HTTP_PREEMPTIVE_SHUTDOWN_GRACE_MS,
+        // Context takeover stays enabled (ws default) so the compression
+        // window is shared across frames — that also makes small frames cheap
+        // to compress, so no size threshold is set (ws only honors
+        // `threshold` when context takeover is disabled).
+        websocket: { perMessageDeflate: true },
+      },
+    );
   }),
 );
 
@@ -720,7 +726,13 @@ const layerMakeServer = Layer.unwrap(
     const ownership = yield* ServerOwnership.acquireServerOwnership(config.serverRuntimeStatePath);
     const activation = yield* Deferred.make<void>();
     const awaitActivation = Deferred.await(activation);
-    const layerActivation = Layer.succeed(ServerActivation.ServerActivation, awaitActivation);
+    const layerActivation = Layer.merge(
+      Layer.succeed(ServerActivation.ServerActivation, awaitActivation),
+      Layer.effect(
+        ServerActivation.ServerBackgroundStart,
+        ServerActivation.makeBackgroundStart(awaitActivation),
+      ),
+    );
     const runtimeStateParked = yield* Deferred.make<void>();
     const tailscaleParked = yield* Deferred.make<void>();
     const cloudLinkParked = yield* Deferred.make<void>();

apps/server/src/serverActivation.test.ts

@@ -1,6 +1,8 @@
 import { expect, it } from "@effect/vitest";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
+import * as TestClock from "effect/testing/TestClock";
 
 import * as ServerActivation from "./serverActivation.ts";
 
@@ -21,3 +23,37 @@ it.effect("proves a root is parked before returning and releases it with one gat
     }),
   ),
 );
+
+it.effect("keeps maintenance parked until activation and client delivery", () =>
+  Effect.gen(function* () {
+    const activation = yield* Deferred.make<void>();
+    const ran = yield* Deferred.make<void>();
+    const background = yield* ServerActivation.makeBackgroundStart(Deferred.await(activation));
+    yield* ServerActivation.forkBackground(Deferred.succeed(ran, undefined)).pipe(
+      Effect.provideService(ServerActivation.ServerActivation, Deferred.await(activation)),
+      Effect.provideService(ServerActivation.ServerBackgroundStart, background),
+    );
+
+    yield* TestClock.adjust("1 minute");
+    expect(yield* Deferred.isDone(ran)).toBe(false);
+    yield* Deferred.succeed(activation, undefined);
+    yield* TestClock.adjust("10 seconds");
+    expect(yield* Deferred.isDone(ran)).toBe(false);
+
+    yield* background.markClientServed;
+    yield* Deferred.await(ran);
+  }),
+);
+
+it.effect("releases background work within a bounded time without a client", () =>
+  Effect.gen(function* () {
+    const background = yield* ServerActivation.makeBackgroundStart(Effect.void);
+    const probes = yield* background.awaitTurn.pipe(Effect.forkScoped);
+    const maintenance = yield* background.awaitTurn.pipe(Effect.forkScoped);
+    yield* TestClock.adjust("30 seconds");
+    expect(probes.pollUnsafe()?._tag).toBe("Success");
+    expect(maintenance.pollUnsafe()?._tag).toBe("Success");
+    yield* Fiber.join(probes);
+    yield* Fiber.join(maintenance);
+  }),
+);

apps/server/src/serverActivation.ts

@@ -1,7 +1,7 @@
 import * as Context from "effect/Context";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
-import type * as Fiber from "effect/Fiber";
+import * as Fiber from "effect/Fiber";
 import type * as Scope from "effect/Scope";
 
 export class ServerActivation extends Context.Reference<Effect.Effect<void> | undefined>(
@@ -28,3 +28,39 @@ export const forkParkedFiber = <A, E, R>(
 export const forkParked = <A, E, R>(
   effect: Effect.Effect<A, E, R>,
 ): Effect.Effect<void, never, Scope.Scope | R> => forkParkedFiber(effect).pipe(Effect.asVoid);
+
+interface BackgroundStart {
+  readonly awaitTurn: Effect.Effect<void>;
+  readonly markClientServed: Effect.Effect<void>;
+}
+
+export class ServerBackgroundStart extends Context.Reference<BackgroundStart | undefined>(
+  "t3/serverBackgroundStart",
+  { defaultValue: () => undefined },
+) {}
+
+export const makeBackgroundStart = Effect.fnUntraced(function* (activation: Effect.Effect<void>) {
+  const clientServed = yield* Deferred.make<void>();
+  // Headless environments still run maintenance even if no client connects.
+  const ready = yield* activation.pipe(
+    Effect.andThen(Deferred.await(clientServed).pipe(Effect.timeoutOption("30 seconds"))),
+    Effect.forkScoped,
+  );
+  return {
+    awaitTurn: Fiber.join(ready).pipe(Effect.asVoid),
+    markClientServed: Deferred.succeed(clientServed, undefined).pipe(Effect.asVoid),
+  } satisfies BackgroundStart;
+});
+
+export const awaitBackgroundTurn = Effect.gen(function* () {
+  const background = yield* ServerBackgroundStart;
+  if (background !== undefined) yield* background.awaitTurn;
+});
+
+export const markClientServed = Effect.gen(function* () {
+  const background = yield* ServerBackgroundStart;
+  if (background !== undefined) yield* background.markClientServed;
+});
+
+export const forkBackground = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+  forkParked(Effect.andThen(awaitBackgroundTurn, effect));

apps/server/src/serverRuntimeStartup.ts

@@ -43,7 +43,7 @@ import * as ProjectStore from "./orchestration-v2/ProjectStore.ts";
 import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts";
 import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
 import * as ServerSettings from "./serverSettings.ts";
-import { forkParked, forkParkedFiber } from "./serverActivation.ts";
+import { forkBackground, forkParked, forkParkedFiber } from "./serverActivation.ts";
 import * as AnalyticsService from "./telemetry/AnalyticsService.ts";
 import * as ServerEnvironment from "./environment/ServerEnvironment.ts";
 import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
@@ -608,7 +608,7 @@ const make = (options?: StartupOptions) =>
           : importPendingTranscripts
       ).pipe(forkParked);
 
-      yield* forkParked(
+      yield* forkBackground(
         Effect.gen(function* () {
           yield* Effect.logDebug("startup phase: recording startup heartbeat");
           yield* recordStartupHeartbeat.pipe(
@@ -617,6 +617,10 @@ const make = (options?: StartupOptions) =>
             }),
             Effect.ignoreCause({ log: true }),
           );
+        }),
+      );
+      yield* forkParked(
+        Effect.gen(function* () {
           if (serverConfig.startupPresentation === "headless") {
             yield* Effect.logDebug("startup phase: headless access info");
             const accessInfo = yield* issueHeadlessServeAccessInfo();

apps/server/src/storageCleanup.ts

@@ -40,7 +40,7 @@ import * as Orchestrator from "./orchestration-v2/Orchestrator.ts";
 import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts";
 import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts";
 import { threadHasQueuedTurnStart } from "./orchestration-v2/ThreadSettlementService.ts";
-import { forkParked } from "./serverActivation.ts";
+import { forkBackground, forkParked } from "./serverActivation.ts";
 import * as Settings from "./serverSettings.ts";
 import * as TerminalManager from "./terminal/Manager.ts";
 import { isFilesystemRoot, managedWorktreesDirectories } from "./worktreesDirectory.ts";
@@ -1046,7 +1046,7 @@ const make = Effect.gen(function* () {
     const changes = yield* settingsService.subscribeChanges;
     const events = engine.streamDomainEvents;
     let lastSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
-    yield* forkParked(
+    yield* forkBackground(
       requestSweep.pipe(
         Effect.andThen(worker.drain),
         Effect.repeat(Schedule.spaced("1 hour")),

apps/server/src/telemetry/AnalyticsService.ts

@@ -23,6 +23,7 @@ import * as HttpClientRequest from "effect/http/HttpClientRequest";
 import * as HttpClientResponse from "effect/http/HttpClientResponse";
 
 import packageJson from "../../package.json" with { type: "json" };
+import { awaitBackgroundTurn } from "../serverActivation.ts";
 import * as ServerConfig from "../config.ts";
 import { getTelemetryIdentifier } from "./Identify.ts";
 
@@ -286,7 +287,7 @@ export const make = Effect.gen(function* () {
 
   yield* Effect.forever(Effect.sleep(FLUSH_INTERVAL_MS).pipe(Effect.flatMap(() => flushWhenDue)), {
     disableYield: true,
-  }).pipe(Effect.forkScoped);
+  }).pipe((effect) => Effect.andThen(awaitBackgroundTurn, effect), Effect.forkScoped);
 
   yield* Effect.addFinalizer(() => flush);