Skip to content

Commit bf5f04c

Browse files
juliusmarmingejuliusmarminge
committed
fix(clients): retain compatibility through discovery refreshes
1 parent 8e5c75a commit bf5f04c

2 files changed

Lines changed: 54 additions & 14 deletions

File tree

‎packages/client-runtime/src/connection/layer.ts‎

Lines changed: 17 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import type { ExecutionEnvironmentDescriptor } from "@t3tools/contracts";
1+
import type { RelayEnvironmentStatusResponse } from "@t3tools/contracts/relay";
22
import * as Effect from "effect/Effect";
33
import * as Layer from "effect/Layer";
44
import * as Stream from "effect/Stream";
@@ -19,26 +19,34 @@ export const watchDiscoveredCompatibility = Effect.fn("connection.watchDiscovere
1919
function* () {
2020
const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
2121
const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
22-
const seenDescriptors = new Map<string, ExecutionEnvironmentDescriptor>();
22+
const seenChecks = new Map<string, RelayEnvironmentStatusResponse>();
2323
yield* Stream.merge(
2424
SubscriptionRef.changes(discovery.state),
2525
SubscriptionRef.changes(registry.entries),
2626
).pipe(
2727
Stream.runForEach(() =>
2828
Effect.gen(function* () {
2929
const current = yield* SubscriptionRef.get(discovery.state);
30-
for (const environmentId of seenDescriptors.keys()) {
31-
if (!current.environments.has(environmentId)) seenDescriptors.delete(environmentId);
30+
if (!current.refreshing) {
31+
for (const environmentId of seenChecks.keys()) {
32+
if (!current.environments.has(environmentId)) seenChecks.delete(environmentId);
33+
}
3234
}
3335
for (const entry of current.environments.values()) {
34-
const descriptor = Option.getOrNull(entry.status)?.descriptor;
35-
if (descriptor === undefined) continue;
36+
const status = Option.getOrNull(entry.status);
37+
const descriptor = status?.descriptor;
38+
if (status === null || descriptor === undefined) continue;
3639
const environmentId = entry.environment.environmentId;
37-
const fresh = seenDescriptors.get(environmentId) !== descriptor;
40+
const previous = seenChecks.get(environmentId);
41+
const fresh =
42+
previous?.checkedAt !== status.checkedAt ||
43+
(previous.descriptor?.orchestrationProtocolVersion ?? 1) !==
44+
(descriptor.orchestrationProtocolVersion ?? 1) ||
45+
previous.descriptor?.serverVersion !== descriptor.serverVersion;
3846
const error = orchestrationProtocolCompatibilityError(descriptor);
39-
// Only a new descriptor for this environment can clear a socket rejection.
47+
// A replayed health result must not clear a newer socket rejection.
4048
if (error !== null || fresh) yield* registry.setCompatibility(environmentId, error);
41-
seenDescriptors.set(environmentId, descriptor);
49+
seenChecks.set(environmentId, status);
4250
}
4351
}).pipe(
4452
Effect.catch((error) =>

‎packages/client-runtime/src/connection/registry.test.ts‎

Lines changed: 37 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -681,7 +681,7 @@ describe("EnvironmentRegistry", () => {
681681
}),
682682
);
683683

684-
it.effect("only a fresh descriptor for the rejected environment unlocks it", () =>
684+
it.effect("only a fresh health check for the rejected environment unlocks it", () =>
685685
Effect.gen(function* () {
686686
const harness = yield* makeHarness([RELAY_TARGET], [], [], {
687687
initialDisabled: [RELAY_TARGET.environmentId],
@@ -694,7 +694,10 @@ describe("EnvironmentRegistry", () => {
694694
orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
695695
capabilities: { repositoryIdentity: true },
696696
});
697-
const discovered = (value: ExecutionEnvironmentDescriptor) => {
697+
const discovered = (
698+
value: ExecutionEnvironmentDescriptor,
699+
checkedAt = "2026-09-15T00:00:00Z",
700+
) => {
698701
const environment = {
699702
environmentId: value.environmentId,
700703
label: value.label,
@@ -709,7 +712,7 @@ describe("EnvironmentRegistry", () => {
709712
environmentId: value.environmentId,
710713
endpoint: environment.endpoint,
711714
status: "online",
712-
checkedAt: "2026-09-15T00:00:00Z",
715+
checkedAt,
713716
descriptor: value,
714717
};
715718
return {
@@ -727,8 +730,10 @@ describe("EnvironmentRegistry", () => {
727730
});
728731
const initial = yield* Deferred.make<void>();
729732
const unrelated = yield* Deferred.make<void>();
733+
const replayed = yield* Deferred.make<void>();
730734
const refreshed = yield* Deferred.make<void>();
731735
let firstEnvironmentCalls = 0;
736+
let secondEnvironmentCalls = 0;
732737
yield* Effect.gen(function* () {
733738
const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
734739
yield* watchDiscoveredCompatibility().pipe(
@@ -744,7 +749,13 @@ describe("EnvironmentRegistry", () => {
744749
firstEnvironmentCalls === 1 ? initial : refreshed,
745750
undefined,
746751
);
747-
} else yield* Deferred.succeed(unrelated, undefined);
752+
} else {
753+
secondEnvironmentCalls += 1;
754+
yield* Deferred.succeed(
755+
secondEnvironmentCalls === 1 ? unrelated : replayed,
756+
undefined,
757+
);
758+
}
748759
}),
749760
),
750761
),
@@ -776,11 +787,32 @@ describe("EnvironmentRegistry", () => {
776787
(yield* SubscriptionRef.get(registry.entries)).get(RELAY_TARGET.environmentId)
777788
?.unsupportedReason,
778789
).toBe(error.message);
790+
yield* SubscriptionRef.update(discoveryState, (state) => ({
791+
...state,
792+
refreshing: true,
793+
environments: new Map(),
794+
}));
795+
yield* SubscriptionRef.update(discoveryState, (state) => ({
796+
...state,
797+
refreshing: false,
798+
environments: new Map([
799+
[RELAY_TARGET.environmentId, discovered(descriptor(RELAY_TARGET.environmentId))],
800+
[
801+
SECOND_TARGET.environmentId,
802+
discovered(descriptor(SECOND_TARGET.environmentId), "2026-09-15T00:01:00Z"),
803+
],
804+
]),
805+
}));
806+
yield* Deferred.await(replayed);
807+
expect(
808+
(yield* SubscriptionRef.get(registry.entries)).get(RELAY_TARGET.environmentId)
809+
?.unsupportedReason,
810+
).toBe(error.message);
779811
yield* SubscriptionRef.update(discoveryState, (state) => ({
780812
...state,
781813
environments: new Map(state.environments).set(
782814
RELAY_TARGET.environmentId,
783-
discovered(descriptor(RELAY_TARGET.environmentId)),
815+
discovered(descriptor(RELAY_TARGET.environmentId), "2026-09-15T00:02:00Z"),
784816
),
785817
}));
786818
yield* Deferred.await(refreshed);

0 commit comments

Comments
 (0)