diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 46648e13b4b57..1f77ddad1b709 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -33,7 +33,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"; @@ -845,7 +845,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; diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index 559ea7677aac0..35c25a2d96855 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts @@ -28,7 +28,7 @@ import * as ProjectStore from "./ProjectStore.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { makeLinkedThreadsChangeDetector } from "../pullRequest/linkedThreads.ts"; import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; -import { forkParked } from "../serverActivation.ts"; +import { forkBackground, forkParked } from "../serverActivation.ts"; import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts"; import { makeNoPullRequestRechecks } from "./noPullRequestRechecks.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -488,7 +488,7 @@ export const make = Effect.gen(function* () { Effect.forever, ), ); - yield* forkParked( + yield* forkBackground( Effect.gen(function* () { yield* worker.enqueue({ threadId: null, refresh: false, backfill: true }); yield* worker.drain; diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 512648e6657d2..8ff61bdcf3e5c 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -30,7 +30,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 GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ProjectStore from "./ProjectStore.ts"; @@ -806,7 +806,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* enqueueFullSweep(); yield* fullSweeps.drain; diff --git a/apps/server/src/provider/ProviderHostLive.ts b/apps/server/src/provider/ProviderHostLive.ts index fed07efc035ee..94f2a1cd9f8b5 100644 --- a/apps/server/src/provider/ProviderHostLive.ts +++ b/apps/server/src/provider/ProviderHostLive.ts @@ -10,6 +10,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 ServerLifecycleEvents from "../serverLifecycleEvents.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as DocumentText from "./DocumentText.ts"; @@ -28,6 +29,7 @@ export const layer = Layer.effect( const lifecycleEvents = yield* Effect.serviceOption( ServerLifecycleEvents.ServerLifecycleEvents, ); + const background = yield* ServerActivation.ServerBackgroundStart; return ProviderHost.ProviderHost.of({ paths: { cwd: config.cwd, @@ -45,7 +47,12 @@ export const layer = Layer.effect( shouldRunBackgroundWork: backgroundPolicy.shouldRunScopeWork, ...Option.match(lifecycleEvents, { onNone: () => ({}), - onSome: (events) => ({ awaitServerReady: events.awaitReady }), + onSome: (events) => ({ + awaitServerReady: + background === undefined + ? events.awaitReady + : Effect.andThen(events.awaitReady, background.awaitTurn), + }), }), resolveAttachmentPath: (attachment) => resolveAttachmentPath({ attachmentsDir: config.attachmentsDir, attachment }), diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index 8a7ff0ac7fd37..99de89027dd36 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -122,6 +122,11 @@ export class ProviderRegistry extends Context.Service< }) => Effect.Effect>; readonly streamChanges: Stream.Stream>; + readonly subscribeChanges: Effect.Effect< + Stream.Stream>, + never, + Scope.Scope + >; } >()("t3/provider/ProviderRegistry") {} @@ -1221,6 +1226,9 @@ export const layer = Layer.effect( get streamChanges() { return Stream.fromPubSub(changesPubSub); }, + get subscribeChanges() { + return Effect.map(PubSub.subscribe(changesPubSub), Stream.fromSubscription); + }, } satisfies ProviderRegistry["Service"]; }), ); diff --git a/apps/server/src/provider/providerMaintenanceRunner.test.ts b/apps/server/src/provider/providerMaintenanceRunner.test.ts index 985347a66e96b..6eb1a31762528 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.test.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.test.ts @@ -227,6 +227,7 @@ function makeRegistry( Effect.succeed(lifecycleFor(provider)), setProviderMaintenanceActionState, streamChanges: Stream.empty, + subscribeChanges: Effect.succeed(Stream.empty), }; return { diff --git a/apps/server/src/provider/testUtils/providerRegistryMock.ts b/apps/server/src/provider/testUtils/providerRegistryMock.ts index ccf4979937b35..4a37e329379d6 100644 --- a/apps/server/src/provider/testUtils/providerRegistryMock.ts +++ b/apps/server/src/provider/testUtils/providerRegistryMock.ts @@ -21,6 +21,7 @@ const makeProviderRegistryMock = ( Effect.succeed(makeManualOnlyProviderMaintenanceCapabilities({ provider, packageName: null })), setProviderMaintenanceActionState: () => Effect.succeed(providers), streamChanges: Stream.empty, + subscribeChanges: Effect.succeed(Stream.empty), }); export const layer = (providers: ReadonlyArray = []) => diff --git a/apps/server/src/pullRequest/PullRequestReadCache.ts b/apps/server/src/pullRequest/PullRequestReadCache.ts index 7f03beef3a245..2ca4d4f056371 100644 --- a/apps/server/src/pullRequest/PullRequestReadCache.ts +++ b/apps/server/src/pullRequest/PullRequestReadCache.ts @@ -23,7 +23,7 @@ import * as PersistedCache from "effect/persistence/PersistedCache"; import * as Persistence from "effect/persistence/Persistence"; import * as ServerConfig from "../config.ts"; import { detachStackFrame } from "../git/detachStackFrame.ts"; -import { forkParked } from "../serverActivation.ts"; +import { forkBackground } from "../serverActivation.ts"; const CONCURRENT_READS = 512; /** Invalidation excludes only reads whose scopes share its stripe, not every PR read. */ @@ -243,7 +243,7 @@ export const layer = Layer.unwrap( Layer.provide( // Memoize cleanup acquisition with the store; Layer.tap runs per consumer. Layer.effectDiscard( - forkParked( + forkBackground( pruneExpiredEntryFiles(directory).pipe( Effect.catch((cause) => Effect.logWarning("Failed to prune PR cache files", { cause }), diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 30f003bf12891..f057ca21451ce 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -898,7 +898,13 @@ const layerMakeServer = Layer.unwrap( const ownership = yield* ServerOwnership.acquireServerOwnership(config.serverRuntimeStatePath); const activation = yield* Deferred.make(); 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(); const tailscaleParked = yield* Deferred.make(); const cloudLinkParked = yield* Deferred.make(); diff --git a/apps/server/src/serverActivation.test.ts b/apps/server/src/serverActivation.test.ts new file mode 100644 index 0000000000000..9abd1ea9e641d --- /dev/null +++ b/apps/server/src/serverActivation.test.ts @@ -0,0 +1,39 @@ +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"; + +it.effect("keeps maintenance parked until activation and client delivery", () => + Effect.gen(function* () { + const activation = yield* Deferred.make(); + const ran = yield* Deferred.make(); + 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 after the fallback when no client connects", () => + Effect.gen(function* () { + const background = yield* ServerActivation.makeBackgroundStart(Effect.void); + const waiter = yield* background.awaitTurn.pipe(Effect.forkScoped); + yield* TestClock.adjust("29 seconds"); + expect(waiter.pollUnsafe()).toBeUndefined(); + yield* TestClock.adjust("1 second"); + yield* Fiber.join(waiter); + }), +); diff --git a/apps/server/src/serverActivation.ts b/apps/server/src/serverActivation.ts index 2945951d847a0..3f4baa3e7eef5 100644 --- a/apps/server/src/serverActivation.ts +++ b/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 | undefined>( @@ -27,3 +27,41 @@ export const forkParkedFiber = Effect.fn("serverActivation.forkParkedFiber")(fun export const forkParked = ( effect: Effect.Effect, ): Effect.Effect => forkParkedFiber(effect).pipe(Effect.asVoid); + +interface BackgroundStart { + readonly awaitTurn: Effect.Effect; + readonly markClientServed: Effect.Effect; +} + +export class ServerBackgroundStart extends Context.Reference( + "t3/serverBackgroundStart", + { defaultValue: () => undefined }, +) {} + +export const makeBackgroundStart = Effect.fn("serverActivation.makeBackgroundStart")(function* ( + activation: Effect.Effect, +) { + const clientServed = yield* Deferred.make(); + // 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 = (effect: Effect.Effect) => + forkParked(Effect.andThen(awaitBackgroundTurn, effect)); diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 0d84e99169e1b..d21197a4ec766 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -45,7 +45,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 { getServerOwners } from "./telemetry/Identify.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; @@ -614,7 +614,7 @@ const make = Effect.fn("serverRuntimeStartup.make")(function* (options?: Startup : importPendingTranscripts ).pipe(forkParked); - yield* forkParked( + yield* forkBackground( Effect.gen(function* () { yield* Effect.logDebug("startup phase: recording startup heartbeat"); yield* recordStartupHeartbeat.pipe( @@ -623,6 +623,10 @@ const make = Effect.fn("serverRuntimeStartup.make")(function* (options?: Startup }), 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(); diff --git a/apps/server/src/storageCleanup.ts b/apps/server/src/storageCleanup.ts index dc8ccca041015..9296d8f96ae20 100644 --- a/apps/server/src/storageCleanup.ts +++ b/apps/server/src/storageCleanup.ts @@ -49,7 +49,7 @@ import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManag import { threadHasQueuedTurnStart } from "./orchestration-v2/ThreadSettlementService.ts"; import { MAX_OVERDUE_ARM_MS } from "./orchestration-v2/UsageLimitRecoveryWorker.ts"; import * as WorktreeBranchSync from "./orchestration-v2/WorktreeBranchSync.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"; @@ -1800,7 +1800,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")), diff --git a/apps/server/src/telemetry/AnalyticsService.ts b/apps/server/src/telemetry/AnalyticsService.ts index 384da92b3f312..ca0b83abc8e5e 100644 --- a/apps/server/src/telemetry/AnalyticsService.ts +++ b/apps/server/src/telemetry/AnalyticsService.ts @@ -20,6 +20,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 * as ManagedPolicy from "../policy/ManagedPolicy.ts"; import { getTelemetryIdentity, type TelemetryOwner } from "./Identify.ts"; @@ -352,7 +353,7 @@ export const make = Effect.gen(function* () { yield* Effect.forever(Effect.sleep(FLUSH_INTERVAL_MS).pipe(Effect.andThen(flushWhenDue)), { disableYield: true, - }).pipe(Effect.forkScoped); + }).pipe((effect) => Effect.andThen(awaitBackgroundTurn, effect), Effect.forkScoped); yield* Effect.addFinalizer(() => flush.pipe(Effect.timeout(SHUTDOWN_FLUSH_TIMEOUT), Effect.ignore), diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 660548c1dd68b..a923a4aa6e1b6 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -906,6 +906,7 @@ it.effect.each([ const settings = yield* ServerSettings.ServerSettingsService; const keybindingsSubscribed = yield* Deferred.make(); const settingsSubscribed = yield* Deferred.make(); + const providersSubscribed = yield* Deferred.make(); const keybindingsChanges = keybindings.subscribeChanges.pipe( Effect.tap(() => Deferred.succeed(keybindingsSubscribed, undefined)), ); @@ -940,7 +941,9 @@ it.effect.each([ }), Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]), - streamChanges: Stream.never, + subscribeChanges: Deferred.succeed(providersSubscribed, undefined).pipe( + Effect.as(Stream.never), + ), }), Layer.mock(UsageLimitSources.UsageLimitSources)({ streamChanges: Stream.make([]) }), Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({ @@ -1045,6 +1048,8 @@ it.effect.each([ }); yield* Deferred.await(discoveryStarted); assert.isTrue(yield* Deferred.isDone(remoteDiscoveryStarted)); + // Startup probes are released by this delivery, so their results must already be buffered. + assert.isTrue(yield* Deferred.isDone(providersSubscribed)); const duringDiscovery = yield* settings.updateSettings({ defaultAutoPull: true }); yield* Deferred.succeed(finishDiscovery, undefined); @@ -1317,8 +1322,8 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( }), Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.sync(() => registryProviders), - streamChanges: Stream.fromQueue(providerChanges).pipe( - Stream.tap(() => Queue.offer(pulled, undefined)), + subscribeChanges: Effect.succeed( + Stream.fromQueue(providerChanges).pipe(Stream.tap(() => Queue.offer(pulled, undefined))), ), refresh: () => Effect.sync(() => registryProviders), refreshWorkspaceSnapshot: () => Effect.sync(() => registryProviders), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 09c6d86c050fd..436e661f96ccc 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -183,6 +183,7 @@ import * as ProviderAuthService from "./provider/ProviderAuthService.ts"; import * as ProviderInstallation from "./provider/providerInstallation.ts"; import * as ServerSelfUpdate from "./cloud/selfUpdate.ts"; import * as NpmRelease from "./cloud/npmRelease.ts"; +import * as ServerActivation from "./serverActivation.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver"; @@ -3315,6 +3316,7 @@ export const layerWsRpc = ( // Buffer changes through discovery and the initial snapshot's send. const keybindingsChanges = yield* keybindings.subscribeChanges; const settingsChanges = yield* serverSettings.subscribeChanges; + const providerChanges = yield* providerRegistry.subscribeChanges; const remoteTargets = yield* Effect.forkScoped(remoteOpenTargets.resolveTargets(), { startImmediately: true, }); @@ -3343,10 +3345,7 @@ export const layerWsRpc = ( ); const providerStatuses = Stream.zipLatestWith( // Seed the changes-only stream so early source refreshes pair with current providers. - Stream.concat( - Stream.fromEffect(providerRegistry.getProviders), - providerRegistry.streamChanges, - ), + Stream.concat(Stream.fromEffect(providerRegistry.getProviders), providerChanges), usageLimitSources.streamChanges.pipe( Stream.changesWith( usageLimitsCommand ? sameUsageLimitCommandCoverage : () => true, @@ -3443,9 +3442,16 @@ export const layerWsRpc = ( rpcInitialItems([ { version: 1 as const, type: "snapshot" as const, config: initialConfig }, ]), - input.providerStatusDeltas === true - ? updates.pipe(withProviderStatusDeltas(initialConfig.providers)) - : updates, + // Deferred maintenance starts once a client has its first snapshot. + Stream.unwrap( + ServerActivation.markClientServed.pipe( + Effect.as( + input.providerStatusDeltas === true + ? updates.pipe(withProviderStatusDeltas(initialConfig.providers)) + : updates, + ), + ), + ), ); }), ), diff --git a/packages/provider-core/src/server/ProviderHost.ts b/packages/provider-core/src/server/ProviderHost.ts index 804e8ff4cdda7..6606819c71c44 100644 --- a/packages/provider-core/src/server/ProviderHost.ts +++ b/packages/provider-core/src/server/ProviderHost.ts @@ -44,7 +44,7 @@ export class ProviderHost extends Context.Service< readonly subscribe: Effect.Effect, never, Scope.Scope>; }; readonly shouldRunBackgroundWork: (scope: BackgroundScope) => Effect.Effect; - /** Completes once server startup finishes; first provider checks wait for it. */ + /** Completes once startup finishes and the first client is served; first provider checks wait for it. */ readonly awaitServerReady?: Effect.Effect; /** Absolute attachment path, or null if the ID does not resolve inside attachmentsDir. */ readonly resolveAttachmentPath: (attachment: ChatAttachment) => string | null;