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
4 changes: 2 additions & 2 deletions apps/web/src/connection/platform.ts
Original file line number Diff line number Diff line change
Expand Up @@ -812,11 +812,11 @@ const layerEnvironmentOwnedDataCleanup = Layer.succeed(
const layerRpcRequestObserver = Layer.succeed(
EnvironmentRpcRequestObserver,
EnvironmentRpcRequestObserver.of({
observe: ({ environmentId, method }) =>
observe: ({ environmentId, environmentLabel, method }) =>
Effect.sync(() => {
nextObservedRpcRequestId += 1;
const requestId = `${environmentId}:${nextObservedRpcRequestId}`;
trackRpcRequestSent(requestId, method, `${method} · ${environmentId}`);
trackRpcRequestSent(requestId, method, `${method} · ${environmentLabel}`);
return Effect.sync(() => {
acknowledgeRpcRequest(requestId);
});
Expand Down
13 changes: 8 additions & 5 deletions apps/web/src/rpc/requestLatencyState.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,15 @@ describe("requestLatencyState", () => {
expect(getSlowRpcAckRequests()).toEqual([]);
});

it("ignores usage summary requests", () => {
trackRpcRequestSent("1", WS_METHODS.serverGetUsageSummary);
vi.advanceTimersByTime(SLOW_RPC_ACK_THRESHOLD_MS * 2);
it.each([WS_METHODS.serverGetUsageSummary, WS_METHODS.serverReportClientActivity])(
"ignores background request %s",
(method) => {
trackRpcRequestSent("1", method);
vi.advanceTimersByTime(SLOW_RPC_ACK_THRESHOLD_MS * 2);

expect(getSlowRpcAckRequests()).toEqual([]);
});
expect(getSlowRpcAckRequests()).toEqual([]);
},
);

it.each(Object.values(WS_METHODS).filter((method) => method.startsWith("pullRequests.")))(
"ignores pull request workspace request %s",
Expand Down
7 changes: 6 additions & 1 deletion apps/web/src/rpc/requestLatencyState.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,12 @@ interface PendingRpcAckRequest {
}

const pendingRpcAckRequests = new Map<string, PendingRpcAckRequest>();
const untrackedRpcAckMethods = new Set<string>([WS_METHODS.serverGetUsageSummary]);
// Background reports nobody waits on. A busy host delays them first, and warning
// about a request the user never made does not say what is wrong.
const untrackedRpcAckMethods = new Set<string>([
WS_METHODS.serverGetUsageSummary,
WS_METHODS.serverReportClientActivity,
]);
const longRunningRpcAckMethods = new Set<string>([
WS_METHODS.serverUpdateProvider,
WS_METHODS.serverRefreshProviders,
Expand Down
3 changes: 3 additions & 0 deletions packages/client-runtime/src/errors/network.ts
Original file line number Diff line number Diff line change
@@ -1,2 +1,5 @@
export const NETWORK_BLOCKING_HINT =
"Your DNS or firewall may be blocking T3 Connect. Try another network, such as a different Wi-Fi or mobile data.";

/** For a host that stopped answering on an open connection, so the network path was already working. */
export const UNRESPONSIVE_HOST_HINT = "It may be overloaded, asleep, or off the network.";
5 changes: 5 additions & 0 deletions packages/client-runtime/src/errors/transport.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,11 @@ describe("isTransportConnectionErrorMessage", () => {
it("recognizes connection errors emitted by the Effect RPC session", () => {
expect(isTransportConnectionErrorMessage("Test environment disconnected.")).toBe(true);
expect(isTransportConnectionErrorMessage("Test environment stopped responding.")).toBe(true);
expect(
isTransportConnectionErrorMessage(
"Test environment stopped responding for 15s. It may be overloaded, asleep, or off the network.",
),
).toBe(true);
expect(
isTransportConnectionErrorMessage(
"Test environment could not establish a WebSocket connection.",
Expand Down
3 changes: 2 additions & 1 deletion packages/client-runtime/src/errors/transport.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { NETWORK_BLOCKING_HINT } from "./network.ts";
import { NETWORK_BLOCKING_HINT, UNRESPONSIVE_HOST_HINT } from "./network.ts";

const escapeRegExp = (text: string) => text.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");

Expand All @@ -13,6 +13,7 @@ const TRANSPORT_ERROR_PATTERNS = [
`\\b(?:is not connected|disconnected|stopped responding|could not establish a WebSocket connection)\\.(?: ${escapeRegExp(NETWORK_BLOCKING_HINT)})?$`,
"i",
),
new RegExp(`\\bstopped responding for \\d+s\\. ${escapeRegExp(UNRESPONSIVE_HOST_HINT)}$`, "i"),
/\bClientProtocolError\b/i,
/\bRpcClientError\b/i,
/\bping timeout\b/i,
Expand Down
6 changes: 3 additions & 3 deletions packages/client-runtime/src/rpc/client.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -137,9 +137,9 @@ describe("environment RPC", () => {
Effect.provideService(
EnvironmentRpcRequestObserver,
EnvironmentRpcRequestObserver.of({
observe: ({ environmentId, method }) =>
observe: ({ environmentId, environmentLabel, method }) =>
Effect.sync(() => {
observations.push(`start:${environmentId}:${method}`);
observations.push(`start:${environmentId}:${environmentLabel}:${method}`);
return Effect.sync(() => {
observations.push(`finish:${environmentId}:${method}`);
});
Expand All @@ -150,7 +150,7 @@ describe("environment RPC", () => {

expect(result).toEqual({ status: "available", version: "2026.6.0" });
expect(observations).toEqual([
`start:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`,
`start:${TARGET.environmentId}:${TARGET.label}:${WS_METHODS.cloudGetRelayClientStatus}`,
`finish:${TARGET.environmentId}:${WS_METHODS.cloudGetRelayClientStatus}`,
]);
}),
Expand Down
2 changes: 2 additions & 0 deletions packages/client-runtime/src/rpc/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ export class EnvironmentRpcUnavailableError extends Schema.TaggedError<Environme

export interface EnvironmentRpcRequestObservation {
readonly environmentId: string;
readonly environmentLabel: string;
readonly method: string;
}

Expand Down Expand Up @@ -203,6 +204,7 @@ export const requestGuarded = Effect.fn("EnvironmentRpc.request")(function* <
) => Effect.Effect<EnvironmentRpcSuccess<TTag>, EnvironmentRpcFailure<TTag>>;
const completeObservation = yield* observer.observe({
environmentId: supervisor.target.environmentId,
environmentLabel: supervisor.target.label,
method: tag,
});
return yield* method(input).pipe(Effect.ensuring(completeObservation));
Expand Down
19 changes: 16 additions & 3 deletions packages/client-runtime/src/rpc/session.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1084,10 +1084,22 @@ describe("RpcSessionFactory", () => {
}),
);

it.effect("closes the session after 15 seconds without a frame from the server", () =>
// Over T3 Connect too: the socket opened, so blaming DNS or a firewall misleads.
it.effect.each([
{ relay: false, label: "direct" },
{ relay: true, label: "relay" },
])("closes the $label session after 15 seconds without a frame from the server", ({ relay }) =>
Effect.gen(function* () {
const { factory, sockets } = yield* makeFactory();
const session = yield* factory.connect(PREPARED);
const session = yield* factory.connect({
...PREPARED,
target: relay
? new RelayConnectionTarget({
environmentId: TARGET.environmentId,
label: TARGET.label,
})
: TARGET,
});
const readyFiber = yield* Effect.forkChild(session.ready);
const closedFiber = yield* Effect.forkChild(Effect.flip(session.closed));
const socket = yield* awaitSocket(sockets);
Expand All @@ -1109,7 +1121,8 @@ describe("RpcSessionFactory", () => {
expect(error).toBeInstanceOf(ConnectionTransientError);
expect(error).toMatchObject({
reason: "transport",
detail: "Test environment stopped responding.",
detail:
"Test environment stopped responding for 15s. It may be overloaded, asleep, or off the network.",
});
}).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
);
Expand Down
19 changes: 11 additions & 8 deletions packages/client-runtime/src/rpc/session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import {
import * as Cause from "effect/Cause";
import * as Context from "effect/Context";
import * as Deferred from "effect/Deferred";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Equal from "effect/Equal";
import * as Exit from "effect/Exit";
Expand All @@ -25,7 +26,7 @@ import * as RpcSerialization from "effect/rpc/RpcSerialization";
import * as Socket from "effect/socket/Socket";

import { makeWsRpcProtocolClient, PING_TIMEOUT, type WsRpcProtocolClient } from "./protocol.ts";
import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
import { NETWORK_BLOCKING_HINT, UNRESPONSIVE_HOST_HINT } from "../errors/network.ts";
import type {
ConnectionAttemptError,
ConnectionTransientError,
Expand Down Expand Up @@ -122,6 +123,7 @@ function serverConfigReplayEvents(
}

const isSocketErrorReason = Schema.is(Socket.SocketErrorReason);
const PING_TIMEOUT_SECONDS = Duration.toSeconds(PING_TIMEOUT);

function mapSessionRpcError(
error: InitialConfigError | ProbeError | ServerConfigSubscriptionError,
Expand Down Expand Up @@ -189,13 +191,14 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
disconnected,
new ConnectionTransientErrorClass({
reason: "transport",
detail: `${
!wasConnected
? `${connection.label} could not establish a WebSocket connection.`
: timedOut
? `${connection.label} stopped responding.`
: `${connection.label} disconnected.`
}${networkHint}`,
// The network hint only fits a socket that never opened. Once it
// opened, the path worked, and a host that goes quiet is more
// often overloaded or asleep than blocked.
detail: !wasConnected
? `${connection.label} could not establish a WebSocket connection.${networkHint}`
: timedOut
? `${connection.label} stopped responding for ${PING_TIMEOUT_SECONDS}s. ${UNRESPONSIVE_HOST_HINT}`
: `${connection.label} disconnected.`,
...socketFailureLocation(via),
}),
),
Expand Down
Loading