Skip to content
5 changes: 4 additions & 1 deletion docs/internals/connection-runtime.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,10 @@ suspension's probe that times out also replaces the socket regardless of the
network report. The web and desktop foreground probe skips the fresh request,
because a busy host can answer it before a queued probe. Foregrounding also
wakes a pending retry immediately. A long suspension restarts an in-flight
attempt; a short one leaves it alone.
attempt, and so does a web or desktop tab returning to view during a reconnect;
a short mobile suspension and a first attempt leave it alone. An interrupted
attempt stops sharing its pending T3 Connect credential renewal, so the next
attempt starts its own instead of joining one that may be stalled too.

The [registry](../../packages/client-runtime/src/connection/registry.ts) scopes
connections by environment. An involuntary disconnect retains the registration
Expand Down
46 changes: 46 additions & 0 deletions packages/client-runtime/src/authorization/layer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -921,6 +921,52 @@ describe("RemoteEnvironmentAuthorization", () => {
}),
);

it.effect("starts a fresh renewal for a connection attempt that replaces a stalled one", () =>
Effect.gen(function* () {
const stalled = yield* Deferred.make<void>();
const release = yield* Deferred.make<void>();
const bootstraps = yield* Ref.make(0);
const harness = yield* makeHarness({
beforeBootstrap: Ref.getAndUpdate(bootstraps, (count) => count + 1).pipe(
Effect.flatMap((count) =>
count === 0
? Deferred.succeed(stalled, undefined).pipe(Effect.andThen(Deferred.await(release)))
: Effect.void,
),
),
responses: [
Response.json(DESCRIPTOR),
accessToken("fresh-token"),
websocketTicket("fresh-ticket"),
Response.json(DESCRIPTOR),
accessToken("late-token"),
],
});
yield* Effect.gen(function* () {
const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
const stalledAttempt = yield* remote
.authorizeDpop({ expectedEnvironmentId: ENVIRONMENT_ID })
.pipe(Effect.forkChild);
yield* Deferred.await(stalled);
const request = yield* remote
.authorizeDpopHttp({ expectedEnvironmentId: ENVIRONMENT_ID })
.pipe(Effect.forkChild);
yield* Queue.takeN(harness.tokenReads, 2);
yield* Fiber.interrupt(stalledAttempt);

const authorized = yield* remote.authorizeDpop({ expectedEnvironmentId: ENVIRONMENT_ID });
expect(authorized.socketUrl).toContain("wsTicket=fresh-ticket");
expect(yield* Ref.get(harness.bootstrapCalls)).toBe(2);

yield* Deferred.succeed(release, undefined);
const late = yield* Fiber.join(request);
expect(late.httpAuthorization.accessToken, "a request that joined it keeps waiting").toBe(
"late-token",
);
}).pipe(Effect.provide(harness.layer));
}),
);

it.effect("does not block another environment behind a pending renewal", () =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>();
Expand Down
46 changes: 29 additions & 17 deletions packages/client-runtime/src/authorization/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -364,8 +364,15 @@ export const make = Effect.gen(function* () {
},
);

const forgetPendingToken = (environmentId: EnvironmentId, id: object) => {
if (pendingTokens.get(environmentId)?.id === id) {
pendingTokens.delete(environmentId);
}
};

const getDpopToken = Effect.fn("clientRuntime.connection.remote.getDpopToken")(function* (
input: Parameters<RemoteEnvironmentAuthorization["Service"]["authorizeDpopHttp"]>[0],
connectionAttempt = false,
) {
const session = yield* cloudSession.identity;
if (Option.isNone(session)) {
Expand Down Expand Up @@ -403,14 +410,14 @@ export const make = Effect.gen(function* () {
accessToken: cached.value.accessToken,
identity,
});
return { token: Effect.succeed(cached.value), fromCache: true };
return { token: Effect.succeed(cached.value), pendingId: undefined, fromCache: true };
}
if (input.rejectedAccessToken !== undefined) {
yield* removeRejectedToken(input.expectedEnvironmentId, input.rejectedAccessToken);
}
const pending = pendingTokens.get(input.expectedEnvironmentId);
if (pending?.identity === identity && pending.thumbprint === thumbprint) {
return { token: pending.token, fromCache: false };
return { token: pending.token, pendingId: pending.id, fromCache: false };
}

const id = {};
Expand All @@ -435,9 +442,7 @@ export const make = Effect.gen(function* () {
accessToken: token.accessToken,
identity,
});
if (pendingTokens.get(input.expectedEnvironmentId)?.id === id) {
pendingTokens.delete(input.expectedEnvironmentId);
}
forgetPendingToken(input.expectedEnvironmentId, id);
return token;
}),
),
Expand All @@ -454,25 +459,32 @@ export const make = Effect.gen(function* () {
}),
Effect.ensuring(
tokenLock.withPermits(1)(
Effect.sync(() => {
if (pendingTokens.get(input.expectedEnvironmentId)?.id === id) {
pendingTokens.delete(input.expectedEnvironmentId);
}
}),
Effect.sync(() => forgetPendingToken(input.expectedEnvironmentId, id)),
),
),
// One caller's timeout must not cancel renewal for other HTTP requests.
Effect.forkIn(serviceScope),
);
const token = Fiber.join(fiber);
pendingTokens.set(input.expectedEnvironmentId, { id, identity, thumbprint, token });
return { token, fromCache: false };
return { token, pendingId: id, fromCache: false };
}),
);
yield* Effect.annotateCurrentSpan({
"connection.remote_token_cache": selected.fromCache ? "hit" : "miss",
});
const token = yield* selected.token;
const { pendingId } = selected;
const token = yield* connectionAttempt && pendingId !== undefined
? // Attempts are interrupted to replace a stalled one, so the replacement
// starts its own renewal while other waiters keep this one.
selected.token.pipe(
Effect.onInterrupt(() =>
tokenLock.withPermits(1)(
Effect.sync(() => forgetPendingToken(input.expectedEnvironmentId, pendingId)),
),
),
)
: selected.token;
yield* assertSession(identity);
return { token, fromCache: selected.fromCache, identity };
});
Expand Down Expand Up @@ -508,7 +520,7 @@ export const make = Effect.gen(function* () {
socketUrl,
...(descriptor === undefined ? {} : { descriptor }),
});
let selected = yield* getDpopToken(input);
let selected = yield* getDpopToken(input, true);
if (selected.fromCache) {
const cachedSocket = yield* createDpopSocketUrl(
selected.token,
Expand All @@ -528,10 +540,10 @@ export const make = Effect.gen(function* () {
return yield* mapDpopSocketError(cachedSocket.failure);
}
yield* assertSession(selected.identity);
selected = yield* getDpopToken({
...input,
rejectedAccessToken: selected.token.accessToken,
});
selected = yield* getDpopToken(
{ ...input, rejectedAccessToken: selected.token.accessToken },
true,
);
}
const socket = yield* createDpopSocketUrl(selected.token, undefined, endpoint).pipe(
Effect.result,
Expand Down
79 changes: 79 additions & 0 deletions packages/client-runtime/src/connection/supervisor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -683,6 +683,85 @@ describe("EnvironmentSupervisor", () => {
}).pipe(Effect.provide(TestClock.layer())),
);

it.effect("restarts connection setup that stalled while a web tab was hidden", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({
prepare: (attempt) =>
attempt === 2 || attempt === 3 ? Effect.never : Effect.succeed(PREPARED_CONNECTION),
});
const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
initiallyDesired: true,
}).pipe(Effect.provide(harness.dependencies));

yield* awaitState(supervisor.state, (state) => state.phase === "connected");
yield* harness.closeLatestSession();
yield* awaitState(
supervisor.state,
(state) => state.phase === "backoff" && state.attempt === 1,
);
yield* TestClock.adjust("3 seconds");
yield* awaitState(
supervisor.state,
(state) => state.phase === "connecting" && state.attempt === 2,
);

yield* harness.wake("application-active");
yield* TestClock.adjust("1 second");

expect(
yield* Ref.get(harness.prepareCount),
"a visible tab should start a fresh attempt instead of waiting out the stalled one",
).toBe(3);

yield* harness.wake("application-active");
yield* TestClock.adjust("1 second");

expect(
yield* Ref.get(harness.prepareCount),
"returning again should also restart the replacement attempt",
).toBe(4);
yield* awaitState(
supervisor.state,
(state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1,
);
}).pipe(Effect.provide(TestClock.layer())),
);

it.effect("restarts a reconnect that a tab return started from backoff", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({
prepare: (attempt) => (attempt === 2 ? Effect.never : Effect.succeed(PREPARED_CONNECTION)),
});
const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
initiallyDesired: true,
}).pipe(Effect.provide(harness.dependencies));

yield* awaitState(supervisor.state, (state) => state.phase === "connected");
yield* harness.closeLatestSession();
yield* awaitState(
supervisor.state,
(state) => state.phase === "backoff" && state.attempt === 1,
);
yield* harness.wake("application-active");
yield* awaitState(
supervisor.state,
(state) => state.phase === "connecting" && state.attempt === 1,
);

yield* harness.wake("application-active");
yield* TestClock.adjust("1 second");

expect(
yield* Ref.get(harness.prepareCount),
"a reconnect whose retry count was reset is still restarted by a tab return",
).toBe(3);
yield* awaitState(
supervisor.state,
(state) => state.phase === "connected" && state.generation === 2,
);
}).pipe(Effect.provide(TestClock.layer())),
);

it.effect("probes the active session without reconnecting on application activation", () =>
Effect.gen(function* () {
const probeCount = yield* Ref.make(0);
Expand Down
11 changes: 9 additions & 2 deletions packages/client-runtime/src/connection/supervisor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -579,7 +579,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
);
});

const waitForEstablishmentInterrupt = Effect.fnUntraced(function* () {
const waitForEstablishmentInterrupt = Effect.fnUntraced(function* (reconnecting: boolean) {
// A primary attempt keeps running through an offline report, so the
// network returning restarts it in case it stalled meanwhile.
let deviceOffline = (yield* connectivity.status) === "offline";
Expand Down Expand Up @@ -608,6 +608,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
if (next.reason === "application-active-long") {
return true;
}
// A reconnect may have stalled while the tab was hidden, so returning restarts it.
if (next.reason === "application-active" && reconnecting) {
return true;
}
if (next.reason === "credentials-changed" && usesRelay) {
yield* logManagedRelayAccountChange;
return false;
Expand Down Expand Up @@ -899,7 +903,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
exit,
})),
),
waitForEstablishmentInterrupt().pipe(
// Only the first attempt of the first connection is not a reconnect.
waitForEstablishmentInterrupt(
generation > 1 || attempt > 1 || lastFailure !== null || ignoreOffline,
).pipe(
Effect.map((resetRetry): EstablishmentEvent => ({
_tag: "Interrupted",
resetRetry,
Expand Down
Loading