Skip to content

Commit 8aefde8

Browse files
levelorbitlevelorbit
committed
packages/client-runtime/connection/supervisor: keep a socket whose frame beats its replacement
The silent-socket check reads the frame count every 100 ms. A replacement that connected between two reads took over even though the socket had just delivered a frame, which could fail a request in flight on a healthy session. A replacement that is not an upgrade now records the live socket's frame count when it starts and is closed instead of adopted if the count moved.
1 parent 1dc841f commit 8aefde8

2 files changed

Lines changed: 42 additions & 2 deletions

File tree

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

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1218,6 +1218,33 @@ describe("EnvironmentSupervisor", () => {
12181218
}).pipe(Effect.provide(TestClock.layer())),
12191219
);
12201220

1221+
it.effect("keeps a socket whose frame lands just before its replacement connects", () =>
1222+
Effect.gen(function* () {
1223+
const frames = yield* Ref.make(0);
1224+
const harness = yield* makeSilentSocketHarness({
1225+
prepare: (attempt, target) =>
1226+
Effect.sleep(attempt === 1 ? "0 millis" : "950 millis").pipe(
1227+
Effect.as(preparedFor(target)),
1228+
),
1229+
probe: (attempt) => (attempt === 1 ? Effect.sleep("3 seconds") : Effect.void),
1230+
inboundFrames: Ref.get(frames),
1231+
});
1232+
const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, {
1233+
initiallyDesired: true,
1234+
}).pipe(Effect.provide(harness.dependencies));
1235+
1236+
yield* awaitState(supervisor.state, (state) => state.phase === "connected");
1237+
yield* harness.wake("application-active-probe");
1238+
// The frame lands between two 100 ms frame checks, and the replacement
1239+
// connects at 1450 ms, before the next one.
1240+
yield* TestClock.adjust("1420 millis");
1241+
yield* Ref.update(frames, (count) => count + 1);
1242+
yield* TestClock.adjust("5 seconds");
1243+
1244+
expect((yield* SubscriptionRef.get(supervisor.state)).generation).toBe(1);
1245+
}).pipe(Effect.provide(TestClock.layer())),
1246+
);
1247+
12211248
it.effect("waits out a LAN resume probe when the host does not answer a fresh request", () =>
12221249
Effect.gen(function* () {
12231250
const harness = yield* makeHarness({

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

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,8 @@ interface Replacement {
118118
readonly startedAt: bigint;
119119
readonly synchronizedRoutes: Ref.Ref<number>;
120120
readonly upgradeTo: Option.Option<string>;
121+
// The live socket's frame count when a suspect replacement started.
122+
readonly liveFrames: Option.Option<number>;
121123
readonly reveal: Effect.Effect<void>;
122124
}
123125

@@ -710,7 +712,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
710712
// `generation` is the live session's; the replacement reports progress as
711713
// the first attempt of the next one, which it becomes if the session ends.
712714
const startReplacement = Effect.fn("EnvironmentSupervisor.startReplacement")(
713-
function* (generation: number, upgradeTo: Option.Option<string>) {
715+
function* (generation: number, upgradeTo: Option.Option<string>, live?: ConnectedLease) {
714716
const pending = yield* Ref.get(pendingReplacement);
715717
if (Option.isSome(pending)) {
716718
if (!(yield* replacementExpired(pending.value))) return;
@@ -757,6 +759,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
757759
startedAt,
758760
synchronizedRoutes,
759761
upgradeTo,
762+
liveFrames:
763+
live === undefined || Option.isSome(upgradeTo)
764+
? Option.none()
765+
: Option.some(yield* live.session.inboundFrames),
760766
reveal,
761767
};
762768
yield* Ref.set(pendingReplacement, Option.some(replacement));
@@ -835,6 +841,12 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
835841
if (next._tag === "ReplacementReady") {
836842
const pending = yield* Ref.get(pendingReplacement);
837843
if (Option.isNone(pending) || pending.value !== next.replacement) return undefined;
844+
// A frame since it started shows the socket alive, even between frame checks.
845+
const liveFrames = next.replacement.liveFrames;
846+
if (Option.isSome(liveFrames) && (yield* lease.session.inboundFrames) !== liveFrames.value) {
847+
yield* discardSuspectReplacement;
848+
return undefined;
849+
}
838850
// One on a worse route waits as the fallback in case the socket is dead.
839851
const exit = next.replacement.fiber.pollUnsafe();
840852
if (Option.isNone(next.replacement.upgradeTo) && exit !== undefined && Exit.isSuccess(exit)) {
@@ -1015,7 +1027,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
10151027
lease,
10161028
routeKind,
10171029
longResume,
1018-
startReplacement(generation, Option.none()),
1030+
startReplacement(generation, Option.none(), lease),
10191031
),
10201032
);
10211033
recheck = false;
@@ -1036,6 +1048,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
10361048
? startReplacement(
10371049
generation,
10381050
Option.flatMap(taken, (replacement) => replacement.upgradeTo),
1051+
lease,
10391052
)
10401053
: Effect.void,
10411054
),

0 commit comments

Comments
 (0)