Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
8 changes: 4 additions & 4 deletions apps/mobile/src/connection/app-state-wakeups.ts
Original file line number Diff line number Diff line change
@@ -1,18 +1,18 @@
import type { Wakeups } from "@t3tools/client-runtime/connection";

export const MOBILE_BACKGROUND_RECONNECT_AFTER_MS = 10_000;
export const MOBILE_LONG_BACKGROUND_AFTER_MS = 10_000;

export type MobileApplicationActiveWakeup = Extract<
Wakeups.ConnectionWakeup,
"application-active-probe" | "application-active-reconnect"
"application-active-probe" | "application-active-long"
>;

export function mobileApplicationActiveWakeup(
backgroundedAtMs: number | null,
activeAtMs: number,
): MobileApplicationActiveWakeup {
return backgroundedAtMs !== null &&
activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS
? "application-active-reconnect"
activeAtMs - backgroundedAtMs >= MOBILE_LONG_BACKGROUND_AFTER_MS
? "application-active-long"
: "application-active-probe";
}
2 changes: 1 addition & 1 deletion apps/mobile/src/connection/platform.ts
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ const networkPathChanges = Stream.callback<"network-changed">((queue) =>
const layerWakeups = Wakeups.layer({
changes: Stream.mergeAll(
[
Stream.callback<"application-active-probe" | "application-active-reconnect">((queue) =>
Stream.callback<"application-active-probe" | "application-active-long">((queue) =>
Effect.acquireRelease(
Effect.sync(() => {
let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null;
Expand Down
1 change: 1 addition & 0 deletions apps/web/src/cloud/linkEnvironment.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@ function layerRegistry(options?: {
subscribeServerConfig: (input) => client.subscribeServerConfig(input),
ready: Effect.void,
probe: Effect.void,
inboundFrames: Effect.succeed(0),
closed: Effect.never,
};
const target = new PrimaryConnectionTarget({
Expand Down
2 changes: 2 additions & 0 deletions apps/web/test/fakeEnvironment.ts
Original file line number Diff line number Diff line change
Expand Up @@ -377,6 +377,7 @@ const session: EnvironmentConnectionLease["session"] = {
subscribe([{ version: 1 as const, type: "snapshot" as const, config: serverConfig }]),
ready: Effect.void,
probe: Effect.void,
inboundFrames: Effect.succeed(0),
closed: Effect.never,
};

Expand All @@ -398,6 +399,7 @@ export const fakeDriver = {
}),
checkRoute: () => Effect.succeed<RouteCheck>("answered"),
preflight: () => Effect.succeed(true),
checkHost: () => Effect.succeed<RouteCheck>("unchecked"),
};
export type FakeDriver = typeof fakeDriver;

Expand Down
25 changes: 17 additions & 8 deletions docs/internals/connection-runtime.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,23 @@ client of a restarted server reconnects in the same second; with a short cap, a
client that can never connect retries all day. Offline states and authentication
failures wait for a wakeup instead of spending attempts on unchanged conditions.

Foregrounding, an explicit retry, and an offline report probe the established
session, and only a failed probe reconnects. Offline reports are often wrong, for
example for a loopback server. A long mobile background suspension is the one
exception: it replaces the session at once, because the OS can kill a socket
without reporting closure, and a probe would hold a dead socket in "Resuming"
until it times out. That fresh attempt runs even while the network reports
offline. Foregrounding also wakes a pending retry immediately and
leaves an ordinary in-flight attempt alone.
Foregrounding, a network change, an explicit retry, and an offline report probe
the established session, and only a failed probe reconnects. Offline reports are
often wrong, for example for a loopback server. The OS can kill a suspended
socket without a close event, so a probe still unanswered after 500 ms asks the
host for its public descriptor over a fresh request, again each second until it
answers. If the host answers while the socket delivers no frame, the socket is
dead and is replaced, even while the network reports offline: at once on
loopback and LAN, and 2.5 seconds after the check starts elsewhere, where one
lost packet can stall a live socket. Without that evidence the probe deadline
decides: 3 seconds on loopback and LAN routes and after a long mobile
suspension, when the data on screen is stale; 15 seconds elsewhere, because a
tailnet over DERP or a VPN can be slower than that on a live socket. A long
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.

The [registry](../../packages/client-runtime/src/connection/registry.ts) scopes
connections by environment. An involuntary disconnect retains the registration
Expand Down
32 changes: 24 additions & 8 deletions packages/client-runtime/src/connection/driver.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Layer from "effect/Layer";
import * as Scope from "effect/Scope";
import * as FetchHttpClient from "effect/http/FetchHttpClient";
import * as HttpClient from "effect/http/HttpClient";

import type { ConnectionCatalogEntry, ConnectionRoute } from "./catalog.ts";
Expand Down Expand Up @@ -61,6 +62,8 @@ export class ConnectionDriver extends Context.Service<
entry: ConnectionCatalogEntry,
route: ConnectionRoute,
) => Effect.Effect<boolean>;
/** Asks the host behind a live session to answer a fresh request on the session's own origin. */
readonly checkHost: (prepared: PreparedConnection) => Effect.Effect<RouteCheck>;
}
>()("@t3tools/client-runtime/connection/driver/ConnectionDriver") {}

Expand Down Expand Up @@ -142,26 +145,39 @@ export const make = Effect.gen(function* () {
const sessions = yield* RpcSession.RpcSessionFactory;
const httpClient = yield* HttpClient.HttpClient;

const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => {
const httpBaseUrl = routeHttpBaseUrl(route);
if (httpBaseUrl === null) return Effect.succeed<RouteCheck>("unchecked");
// The descriptor is public, so this sends no credential to whatever
// answers at a saved LAN address on a different network.
return fetchRemoteEnvironmentDescriptor({
// The descriptor is public, so this sends no credential to whatever
// answers at a saved LAN address on a different network.
const checkDescriptor = (httpBaseUrl: string, environmentId: string) =>
fetchRemoteEnvironmentDescriptor({
httpBaseUrl,
timeoutMs: ROUTE_CHECK_TIMEOUT_MS,
}).pipe(
Effect.map((descriptor): RouteCheck =>
descriptor.environmentId === entry.target.environmentId ? "answered" : "silent",
descriptor.environmentId === environmentId ? "answered" : "silent",
),
Effect.orElseSucceed((): RouteCheck => "silent"),
// A cached descriptor would answer for a host that cannot be reached.
Effect.provideService(FetchHttpClient.RequestInit, { cache: "no-store" }),
Effect.provideService(HttpClient.HttpClient, httpClient),
);
Comment thread
coderabbitai[bot] marked this conversation as resolved.

const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => {
const httpBaseUrl = routeHttpBaseUrl(route);
if (httpBaseUrl === null) return Effect.succeed<RouteCheck>("unchecked");
return checkDescriptor(httpBaseUrl, entry.target.environmentId).pipe(
Effect.withSpan("ConnectionDriver.checkRoute", {
attributes: { "connection.target.kind": route.target._tag },
}),
);
};

const checkHost = (prepared: PreparedConnection) =>
checkDescriptor(prepared.httpBaseUrl, prepared.environmentId).pipe(
Effect.withSpan("ConnectionDriver.checkHost", {
attributes: { "connection.target.kind": prepared.target._tag },
}),
);

const connect = Effect.fn("ConnectionDriver.connect")(function* (
entry: ConnectionCatalogEntry,
reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect<void>,
Expand Down Expand Up @@ -203,7 +219,7 @@ export const make = Effect.gen(function* () {
Effect.withSpan("ConnectionDriver.preflight"),
);

return ConnectionDriver.of({ connect, checkRoute, preflight });
return ConnectionDriver.of({ connect, checkRoute, preflight, checkHost });
});

export const layer = Layer.effect(ConnectionDriver, make);
2 changes: 2 additions & 0 deletions packages/client-runtime/src/connection/registry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,7 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* (
Stream.die(new Error("Config is not used by registry tests.")),
ready: Effect.void,
probe: Effect.void,
inboundFrames: Effect.succeed(0),
closed: Deferred.await(closed),
} satisfies RpcSession.RpcSession),
() => Ref.update(releasedSessions, (count) => count + 1),
Expand All @@ -432,6 +433,7 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* (
checkRoute: (_entry, route) => checkRoute(route),
preflight: (_entry, route) =>
checkRoute(route).pipe(Effect.map((check) => check === "answered")),
checkHost: () => Effect.succeed<ConnectionDriver.RouteCheck>("unchecked"),
});

const layerCache = Layer.succeed(Persistence.EnvironmentCacheStore, cacheStore);
Expand Down
Loading
Loading