fix(server): subscribe before provider settings watcher (#9271)

t3-code[bot] committed 2 months ago

Commit f90e2f21 parent b57726c2 files · +19 −6Browse files

Unified · Split

2 files

apps/server/src/provider/Layers/ProviderInstanceRegistryHydration.ts +5 −4
apps/server/src/provider/Layers/ProviderRegistry.test.ts +14 −2

apps/server/src/provider/Layers/ProviderInstanceRegistryHydration.ts+5−4

@@ -31,7 +31,7 @@ · 30 unmodified lines
31	31	* 1. Read the current `ServerSettings` once and use it to seed the
32	32	* registry's initial state via `ProviderInstanceRegistryMutableLayer`.
33	33	* 2. Fork a daemon fiber (lifetime tied to the layer's scope) that
34		− * subscribes to `ServerSettingsService.streamChanges` and calls
	34	+ * acquires `ServerSettingsService.subscribeChanges` and calls
35	35	* `ProviderInstanceRegistryMutator.reconcile` on every emission.
36	36	*
37	37	* Failures inside the watcher are logged and swallowed so a single bad
@@ -118,7 +118,8 @@ const SettingsWatcherLive = Layer.effectDiscard( · 80 unmodified lines
118	118	Effect.gen(function* () {
119	119	const mutator = yield* ProviderInstanceRegistryMutator;
120	120	const serverSettings = yield* ServerSettingsService;
121		− yield* serverSettings.streamChanges.pipe(
	121	+ const settingsChanges = yield* serverSettings.subscribeChanges;
	122	+ yield* settingsChanges.pipe(
122	123	Stream.runForEach((next) =>
123	124	mutator
124	125	.reconcile(deriveProviderInstanceConfigMap(next))
@@ -141,8 +142,8 @@ const SettingsWatcherLive = Layer.effectDiscard( · 16 unmodified lines
141	142	* - `ProviderInstanceRegistryMutableLayer` produces the registry +
142	143	* mutator from the initial config map. Its scope owns every
143	144	* per-instance child scope created during reconcile.
144		− * - `SettingsWatcherLive` consumes the mutator and runs a daemon fiber
145		− * in the same scope.
	145	+ * - `SettingsWatcherLive` consumes the mutator, acquires its settings
	146	+ * subscription before forking, and runs a daemon fiber in the same scope.
146	147	*
147	148	* Composing via `Layer.provideMerge` makes the watcher's deps available
148	149	* from the mutable layer while still surfacing the registry as an output.

apps/server/src/provider/Layers/ProviderRegistry.test.ts+14−2

@@ -1944,7 +1944,8 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te · 1943 unmodified lines
1944	1944	const firstMissing = `t3code_codex_first_`;
1945	1945	const secondMissing = `t3code_codex_second_`;
1946	1946	const spawnedCommands: Array<string> = [];
1947		− const serverSettings = yield* makeMutableServerSettingsService(
	1947	+ const allowLazySettingsStream = yield* Deferred.make<void>();
	1948	+ const mutableServerSettings = yield* makeMutableServerSettingsService(
1948	1949	decodeServerSettings(
1949	1950	deepMerge(encodedDefaultServerSettings, {
1950	1951	providers: {
@@ -1957,6 +1958,14 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te · 6 unmodified lines
1957	1958	}),
1958	1959	),
1959	1960	);
	1961	+ const serverSettings = {
	1962	+ ...mutableServerSettings,
	1963	+ streamChanges: Stream.unwrap(
	1964	+ Deferred.await(allowLazySettingsStream).pipe(
	1965	+ Effect.as(mutableServerSettings.streamChanges),
	1966	+ ),
	1967	+ ),
	1968	+ } satisfies ServerSettingsModule.ServerSettingsService["Service"];
1960	1969	const scope = yield* Scope.make();
1961	1970	yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void));
1962	1971	const providerRegistryLayer = ProviderRegistryLive.pipe(
@@ -2016,7 +2025,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te · 53 unmodified lines
2016	2025	assert.deepStrictEqual(spawnedCommands, [firstMissing]);
2017	2026
2018	2027	// Drive a settings change. The Hydration layer's
2019		− // `SettingsWatcherLive` consumes this via `streamChanges`,
	2028	+ // `SettingsWatcherLive` consumes this via `subscribeChanges`,
2020	2029	// calls `reconcile`, which rebuilds the codex instance (the
2021	2030	// envelope changed because `binaryPath` differs → `entryEqual`
2022	2031	// is false). The registry's `Stream.runForEach(
@@ -2028,6 +2037,9 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te · 5 unmodified lines
2028	2037	codex: { enabled: true, binaryPath: secondMissing },
2029	2038	},
2030	2039	});
	2040	+ // Start the lazy stream only after publishing. A watcher that did
	2041	+ // not subscribe before forking has already lost this update.
	2042	+ yield* Deferred.succeed(allowLazySettingsStream, undefined);
2031	2043
2032	2044	// Poll until the injected process boundary observes the new
2033	2045	// executable. This verifies the public settings-to-probe behavior