Skip to content

fix(mobile): reconnect faster and stop resyncing more than needed after backgrounding #5154

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
t3dotgg wants to merge 4 commits into
base: main
Choose a base branch
from
Open

fix(mobile): reconnect faster and stop resyncing more than needed after backgrounding #5154

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4abafa4
fix(mobile): harden background resume and reconnection
t3dotgg Jul 31, 2026
4df84c9
fix(mobile): address review findings on resume hardening
t3dotgg Jul 31, 2026
2c8fe66
fix(mobile): dedup connected generations per supervisor, lock legacy …
t3dotgg Jul 31, 2026
b889ec4
fix(mobile): sol/macroscope review fixes for resume hardening
t3dotgg Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 39 additions & 1 deletion apps/mobile/src/connection/background-activity.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { EnvironmentRegistry } from "@t3tools/client-runtime/connection";
import { EnvironmentRegistry, EnvironmentSupervisor } from "@t3tools/client-runtime/connection";
import { EnvironmentRpcSubscriptionObserver, request } from "@t3tools/client-runtime/rpc";
import {
type BackgroundScope,
Expand Down Expand Up @@ -103,6 +103,44 @@ export const mobileBackgroundActivityReporterLayer = Layer.effectDiscard(
Stream.runForEach(() => Effect.sync(requestReport)),
Effect.forkScoped,
);
// Re-report on every newly connected session generation. The AppState
// report races the reconnect (the RPC fails while the socket is down and
// is ignored), so without this the server's lease can lag a resume by up
// to REPORT_INTERVAL_MS, keeping provider/VCS work paused.
const connectedGenerations = (environmentId: EnvironmentId) =>
registry.followStream(
environmentId,
Stream.unwrap(
Effect.map(EnvironmentSupervisor, (supervisor) =>
// The generation dedup must live inside the per-supervisor stream:
// a replacement supervisor restarts generations at 1, so deduping
// outside followStream would suppress its first connected event.
SubscriptionRef.changes(supervisor.state).pipe(
Stream.filter((state) => state.phase === "connected"),
Stream.map((state) => state.generation),
Stream.changes,
),
),
),
);
yield* Stream.concat(
Stream.fromEffect(SubscriptionRef.get(registry.entries)),
SubscriptionRef.changes(registry.entries),
).pipe(
Stream.map((entries) => [...entries.keys()].sort()),
Stream.changesWith((a, b) => a.length === b.length && a.every((id, i) => id === b[i])),
// Default switchMap concurrency (1) interrupts the previous inner
// stream on every environment-set change; unbounded here would leak a
// subscription set per change. mergeAll inside stays unbounded so all
// environments are observed concurrently.
Stream.switchMap((environmentIds) =>
Stream.mergeAll(environmentIds.map(connectedGenerations), {
concurrency: "unbounded",
}),
),
Stream.runForEach(() => Effect.sync(requestReport)),
Effect.forkScoped,
);
yield* Stream.fromQueue(reportRequests).pipe(
Stream.debounce("250 millis"),
Stream.runForEach(() => report),
Expand Down
39 changes: 37 additions & 2 deletions apps/mobile/src/connection/platform.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,11 +87,13 @@ const connectivityLayer = Connectivity.layer({

const wakeupsLayer = Wakeups.layer({
changes: Stream.merge(
Stream.callback<"application-active-probe" | "application-active-reconnect">((queue) =>
Stream.callback<
"application-active-probe" | "application-active-reconnect" | "network-path-changed"
>((queue) =>
Effect.acquireRelease(
Effect.sync(() => {
let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null;
return AppState.addEventListener("change", (state) => {
const appStateSubscription = AppState.addEventListener("change", (state) => {
if (state === "background") {
backgroundedAtMs = Date.now();
return;
Expand All @@ -101,6 +103,39 @@ const wakeupsLayer = Wakeups.layer({
backgroundedAtMs = null;
}
});
// WiFi <-> cellular keeps isConnected true while invalidating the
// socket's path, so the coarse online/offline signal never fires.
// Emit an advisory wakeup on interface-type changes while active;
// the supervisor probes the session rather than blindly replacing
// it, which keeps flapping paths cheap.
// Seed the current interface type so the first flip after startup
// is detected; the listener only reports changes.
let networkType: string | null = null;
void Network.getNetworkStateAsync()
.then((current) => {
networkType ??= current.type ?? null;
})
.catch(() => undefined);
const networkSubscription = Network.addNetworkStateListener((state) => {
const nextType = state.type ?? null;
const previousType = networkType;
networkType = nextType;
if (
previousType !== null &&
nextType !== null &&
nextType !== previousType &&
state.isConnected === true &&
AppState.currentState === "active"
) {
Queue.offerUnsafe(queue, "network-path-changed");
}
Comment thread
cursor[bot] marked this conversation as resolved.
});
return {
remove: () => {
appStateSubscription.remove();
networkSubscription.remove();
},
};
}),
(subscription) => Effect.sync(() => subscription.remove()),
).pipe(Effect.asVoid),
Expand Down
134 changes: 123 additions & 11 deletions packages/client-runtime/src/connection/supervisor.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import { EnvironmentId } from "@t3tools/contracts";
import { RelayClientTracer } from "@t3tools/shared/relayTracing";
import { describe, expect, it } from "@effect/vitest";
import * as Clock from "effect/Clock";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
Expand Down Expand Up @@ -358,7 +359,16 @@ describe("EnvironmentSupervisor", () => {
);
expect(yield* Ref.get(harness.prepareCount)).toBe(1);

for (const [index, delay] of [1_000, 2_000, 4_000, 8_000, 16_000, 16_000].entries()) {
// Jitter puts each delay in [rung/2, rung]. Read the scheduled retryAt
// per rung, assert it falls inside the jitter bounds (proving ladder
// progression and the 16s cap), then advance exactly to it.
for (const [index, rung] of [1_000, 2_000, 4_000, 8_000, 16_000, 16_000].entries()) {
const current = yield* SubscriptionRef.get(supervisor.state);
const now = yield* Clock.currentTimeMillis;
expect(current.retryAt).not.toBeNull();
const delay = (current.retryAt ?? now) - now;
expect(delay).toBeGreaterThanOrEqual(rung / 2);
expect(delay).toBeLessThanOrEqual(rung);
yield* TestClock.adjust(delay);
yield* eventuallyState(
supervisor.state,
Expand Down Expand Up @@ -539,9 +549,10 @@ describe("EnvironmentSupervisor", () => {
);
expect(yield* Ref.get(harness.prepareCount)).toBe(3);

yield* TestClock.adjust("999 millis");
// Jittered first-rung delay lands in [500ms, 1000ms].
yield* TestClock.adjust("499 millis");
expect(yield* Ref.get(harness.prepareCount)).toBe(3);
yield* TestClock.adjust("1 milli");
yield* TestClock.adjust("501 millis");
yield* eventuallyState(
supervisor.state,
(state) => state.phase === "backoff" && state.attempt === 2,
Expand Down Expand Up @@ -873,6 +884,82 @@ describe("EnvironmentSupervisor", () => {
}),
);

it.effect("probes the active session when the network path changes", () =>
Effect.gen(function* () {
const probeCount = yield* Ref.make(0);
const probeCalled = yield* Deferred.make<void>();
const harness = yield* makeHarness({
probe: () =>
Ref.update(probeCount, (count) => count + 1).pipe(
Effect.andThen(Deferred.succeed(probeCalled, undefined)),
),
});
const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
initiallyDesired: true,
}).pipe(Effect.provide(harness.dependencies));

yield* awaitState(supervisor.state, (state) => state.phase === "connected");
yield* harness.wake("network-path-changed");
yield* Deferred.await(probeCalled);

expect(yield* Ref.get(probeCount)).toBe(1);
expect(yield* Ref.get(harness.sessionCount)).toBe(1);
expect(yield* Ref.get(harness.releaseCount)).toBe(0);
expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connected");
}),
);

it.effect("does not cut backoff short when the network path flaps", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({
prepare: () => Effect.fail(transient()),
});
const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
initiallyDesired: true,
}).pipe(Effect.provide(harness.dependencies));

yield* awaitState(
supervisor.state,
(state) => state.phase === "backoff" && state.attempt === 1,
);
expect(yield* Ref.get(harness.prepareCount)).toBe(1);

// Advisory path-change wakeups have no session to probe during backoff
// and must not trigger an early retry.
yield* harness.wake("network-path-changed");
yield* harness.wake("network-path-changed");
for (let attempt = 0; attempt < 20; attempt += 1) {
yield* Effect.yieldNow;
}
expect(yield* Ref.get(harness.prepareCount)).toBe(1);
expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("backoff");
}),
);
Comment on lines +912 to +937

@coderabbitai coderabbitai Bot Aug 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Replace the fixed Effect.yieldNow loop with a typed wait.

This test waits for two queued "network-path-changed" wakeups to be processed by looping Effect.yieldNow a fixed 20 times before asserting prepareCount and phase. This does not wait for a typed receipt that the supervisor consumed both signals; it relies on an arbitrary iteration count matching the current fiber-scheduling depth. If the internal signal-processing chain gets deeper (e.g., another yield* is added upstream), this test can pass or fail non-deterministically without a related regression in production code.

Use a typed receipt instead, for example by extending the harness to expose an effect that completes once the signals queue is drained, or by using TestClock.adjust to a known point (since waitForRetrySignal races against Effect.sleep(delayMs), advancing time deterministically resolves the race without a magic iteration count).

As per coding guidelines, "Tests must wait for typed receipts and worker drains in event-sourced async flows; do not use sleeps, polling, or arbitrary timeouts to make tests pass."

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/client-runtime/src/connection/supervisor.test.ts` around lines 912 -
937, Replace the fixed Effect.yieldNow loop in the “does not cut backoff short
when the network path flaps” test with a deterministic typed wait. Extend the
visible harness or use TestClock so the test awaits confirmation that both
network-path-changed signals were consumed before asserting prepareCount and
supervisor.state.phase, while preserving the backoff and no-early-retry
assertions.

Source: Coding guidelines


it.effect("replaces the session when a probe after a network path change fails", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({
probe: (attempt) =>
attempt === 1
? Effect.fail(transient("The path changed under the socket."))
: Effect.void,
});
const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
initiallyDesired: true,
}).pipe(Effect.provide(harness.dependencies));

yield* awaitState(supervisor.state, (state) => state.phase === "connected");
yield* harness.wake("network-path-changed");
yield* awaitState(
supervisor.state,
(state) => state.phase === "connected" && state.generation === 2,
);

expect(yield* Ref.get(harness.sessionCount)).toBe(2);
expect(yield* Ref.get(harness.releaseCount)).toBe(1);
}),
);

it.effect("immediately replaces a mobile session after a long background resume", () =>
Effect.gen(function* () {
const probeCount = yield* Ref.make(0);
Expand Down Expand Up @@ -937,9 +1024,9 @@ describe("EnvironmentSupervisor", () => {

yield* awaitState(supervisor.state, (state) => state.phase === "connected");
yield* harness.wake("application-active");
yield* awaitState(supervisor.state, (state) => state.phase === "backoff");
yield* TestClock.adjust("1 second");
yield* eventuallyState(
// A failed probe already proves the session is dead: the supervisor
// replaces it immediately without entering backoff.
yield* awaitState(
supervisor.state,
(state) => state.phase === "connected" && state.generation === 2,
);
Expand All @@ -949,6 +1036,34 @@ describe("EnvironmentSupervisor", () => {
}).pipe(Effect.provide(TestClock.layer())),
);

it.effect("parks in blocked when the foreground probe fails with a blocked error", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({
probe: (attempt) => (attempt === 1 ? Effect.fail(blocked()) : Effect.void),
});
const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
initiallyDesired: true,
}).pipe(Effect.provide(harness.dependencies));

yield* awaitState(supervisor.state, (state) => state.phase === "connected");
yield* harness.wake("application-active");
// A blocked probe failure (auth revoked, permissions) must not churn
// immediate reconnects; the supervisor parks until an external signal.
yield* awaitState(
supervisor.state,
(state) => state.phase === "blocked" && state.lastFailure?.reason === "authentication",
);
expect(yield* Ref.get(harness.sessionCount)).toBe(1);
expect(yield* Ref.get(harness.releaseCount)).toBe(1);

yield* harness.wake("application-active");
yield* awaitState(
supervisor.state,
(state) => state.phase === "connected" && state.generation === 2,
);
}),
);

it.effect("quickly times out a stalled mobile foreground liveness probe", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({
Expand All @@ -961,11 +1076,8 @@ describe("EnvironmentSupervisor", () => {
yield* awaitState(supervisor.state, (state) => state.phase === "connected");
yield* harness.wake("application-active-probe");
yield* TestClock.adjust("3 seconds");
yield* awaitState(
supervisor.state,
(state) => state.phase === "backoff" && state.lastFailure?.reason === "timeout",
);
yield* TestClock.adjust("1 second");
// The 3s mobile probe timeout counts as a failed probe: the stale
// session is replaced immediately without a backoff delay.
yield* eventuallyState(
supervisor.state,
(state) => state.phase === "connected" && state.generation === 2,
Expand Down
57 changes: 44 additions & 13 deletions packages/client-runtime/src/connection/supervisor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import * as Fiber from "effect/Fiber";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Queue from "effect/Queue";
import * as Random from "effect/Random";
import * as Ref from "effect/Ref";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
Expand Down Expand Up @@ -101,8 +102,11 @@ export interface EnvironmentSupervisorOptions {
readonly initiallyDesired?: boolean;
}

function retryDelayMs(failureCount: number): number {
return RETRY_DELAYS_MS[Math.min(failureCount, RETRY_DELAYS_MS.length - 1)] ?? 16_000;
// Equal jitter: each retry waits between half and all of its ladder rung so
// clients sharing an environment don't reconnect in lockstep after a restart.
function retryDelayMs(failureCount: number, random: number): number {
const base = RETRY_DELAYS_MS[Math.min(failureCount, RETRY_DELAYS_MS.length - 1)] ?? 16_000;
return Math.round(base / 2 + random * (base / 2));
}

function annotateTarget(target: ConnectionTarget) {
Expand Down Expand Up @@ -414,13 +418,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
// replaces that lease and starts a fresh attempt without backoff.
return true;
}
if (next.reason === "application-active" || next.reason === "application-active-probe") {
if (
next.reason === "application-active" ||
next.reason === "application-active-probe" ||
next.reason === "network-path-changed"
) {
const probe = yield* lease.session.probe.pipe(
Effect.timeoutOrElse({
duration:
next.reason === "application-active-probe"
? MOBILE_CONNECTION_PROBE_TIMEOUT
: CONNECTION_PROBE_TIMEOUT,
next.reason === "application-active"
? CONNECTION_PROBE_TIMEOUT
: MOBILE_CONNECTION_PROBE_TIMEOUT,
orElse: () =>
Effect.fail(
new ConnectionTransientError({
Expand All @@ -441,7 +449,19 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
),
);
if (probeEvent._tag === "ProbeCompleted") {
yield* probeEvent.exit;
if (Exit.isFailure(probeEvent.exit)) {
const failure = Cause.findErrorOption(probeEvent.exit.cause);
if (Option.isSome(failure) && failure.value._tag === "ConnectionBlockedError") {
// Blocked failures (auth, permissions) must keep their
// classification so the run loop parks in the blocked
// state instead of churning immediate reconnects.
yield* probeEvent.exit;
}
// A transiently failed health check already proves the
// session is dead; replace it immediately instead of paying
// the retry ladder for a failure the probe just diagnosed.
return true;
}
break;
}
switch (probeEvent.signal._tag) {
Expand Down Expand Up @@ -615,6 +635,12 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
const next = yield* Queue.take(signals);
switch (next._tag) {
case "Wakeup":
// The path-change wakeup is advisory: it only prompts a probe
// of a connected session. With no session to probe, a flapping
// interface must not cut backoff delays short.
if (next.reason === "network-path-changed") {
break;
}
return ConnectionWakeups.isApplicationActiveWakeup(next.reason);
case "ConnectRequested":
case "DisconnectRequested":
Expand All @@ -627,11 +653,16 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
);
});

const waitForSignal = Queue.take(signals).pipe(
Effect.map(
(next) => next._tag === "Wakeup" && ConnectionWakeups.isApplicationActiveWakeup(next.reason),
),
);
const waitForSignal = Effect.gen(function* () {
for (;;) {
const next = yield* Queue.take(signals);
if (next._tag === "Wakeup" && next.reason === "network-path-changed") {
// Advisory only; see waitForRetrySignal.
continue;
}
return next._tag === "Wakeup" && ConnectionWakeups.isApplicationActiveWakeup(next.reason);
}
});

const run = Effect.fnUntraced(function* () {
let failureCount = 0;
Expand Down Expand Up @@ -710,7 +741,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
}

failureCount += 1;
const delayMs = retryDelayMs(failureCount - 1);
Comment thread
cursor[bot] marked this conversation as resolved.
const delayMs = retryDelayMs(failureCount - 1, yield* Random.next);
pendingRetry = Option.map(attemptSpan, (previousAttempt) => ({
previousAttempt,
failureCount,
Expand Down
Loading
Loading