perf(mobile): connect, reconnect and thread open do less work on bad wifi

apps/server/src/socketFrameDeflate.ts

@@ -0,0 +1,5 @@
+// @effect-diagnostics nodeBuiltinImport:off -- zlib deflates 6-8x faster than a JS deflate on the event loop.
+import * as NodeZlib from "node:zlib";
+
+export const deflateSocketFrame = (frame: string): Uint8Array =>
+  NodeZlib.deflateRawSync(frame, { level: 1 });

apps/server/src/ws.ts

@@ -265,6 +265,7 @@ import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http
 import { isNetworkPath } from "@t3tools/shared/path";
 import * as RelayClient from "@t3tools/shared/relayClient";
 import {
+  SOCKET_FRAME_DEFLATE_QUERY_PARAM,
   SOCKET_FRAME_PARTS_QUERY_PARAM,
   splitLargeSocketFrames,
 } from "@t3tools/shared/socketFrameParts";
@@ -274,6 +275,7 @@ import {
   withUsageLimitsCommands,
 } from "@t3tools/shared/usageLimits";
 import * as AgentSessionScanner from "./project/AgentSessionScanner.ts";
+import { deflateSocketFrame } from "./socketFrameDeflate.ts";
 import * as AgentSessionImporter from "./project/AgentSessionImporter.ts";
 import * as UsageLimitSources from "./usage/UsageLimitSources.ts";
 import { requestUrl } from "./requestUrl.ts";
@@ -3632,6 +3634,12 @@ export const layer = Layer.unwrap(
             ),
           ),
         );
+        // A socket with permessage-deflate is already compressed on the wire.
+        const deflateFrames =
+          url.value.searchParams.has(SOCKET_FRAME_DEFLATE_QUERY_PARAM) &&
+          !request.headers["sec-websocket-extensions"]?.includes("permessage-deflate")
+            ? deflateSocketFrame
+            : undefined;
         const framedRpcWebSocketHttpEffect = url.value.searchParams.has(
           SOCKET_FRAME_PARTS_QUERY_PARAM,
         )
@@ -3641,7 +3649,11 @@ export const layer = Layer.unwrap(
                 // The RPC server upgrades the request itself, so its socket is
                 // wrapped through the request it reads.
                 Object.create(request, {
-                  upgrade: { value: Effect.map(request.upgrade, splitLargeSocketFrames) },
+                  upgrade: {
+                    value: Effect.map(request.upgrade, (socket) =>
+                      splitLargeSocketFrames(socket, deflateFrames),
+                    ),
+                  },
                 }),
               ),
             )

packages/client-runtime/src/authorization/layer.test.ts

@@ -317,6 +317,49 @@ describe("RemoteEnvironmentAuthorization", () => {
     }),
   );
 
+  it.effect("reuses a descriptor a host check just read, once and only while fresh", () =>
+    Effect.gen(function* () {
+      const descriptor = {
+        ...DESCRIPTOR,
+        orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
+      };
+      const harness = yield* makeHarness({
+        responses: [
+          websocketTicket("first-ticket"),
+          Response.json(descriptor),
+          websocketTicket("second-ticket"),
+          Response.json(descriptor),
+          websocketTicket("third-ticket"),
+        ],
+      });
+
+      yield* Effect.gen(function* () {
+        const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
+        const authorize = remote.authorizeBearer({
+          expectedEnvironmentId: ENVIRONMENT_ID,
+          httpBaseUrl: ENDPOINT.httpBaseUrl,
+          wsBaseUrl: ENDPOINT.wsBaseUrl,
+          bearerToken: "bearer-token",
+          connectionMethod: "direct",
+        });
+        yield* remote.rememberCheckedDescriptor(ENDPOINT.httpBaseUrl, descriptor as never);
+        yield* authorize;
+        yield* authorize;
+        yield* remote.rememberCheckedDescriptor(ENDPOINT.httpBaseUrl, descriptor as never);
+        yield* TestClock.adjust("6 seconds");
+        yield* authorize;
+      }).pipe(Effect.provide(harness.layer));
+
+      expect(harness.fetch.calls.map(([url]) => new URL(String(url)).pathname)).toEqual([
+        "/api/auth/websocket-ticket/rpc",
+        "/.well-known/t3/environment",
+        "/api/auth/websocket-ticket/rpc",
+        "/.well-known/t3/environment",
+        "/api/auth/websocket-ticket/rpc",
+      ]);
+    }),
+  );
+
   it.effect("blocks a warm direct bearer prepare when the environment changes identity", () =>
     Effect.gen(function* () {
       const harness = yield* makeHarness({

packages/client-runtime/src/authorization/service.ts

@@ -82,6 +82,11 @@ export class RemoteEnvironmentAuthorization extends Context.Service<
       readonly expectedEnvironmentId: EnvironmentId;
       readonly rejectedAccessToken?: string;
     }) => Effect.Effect<AuthorizedRemoteHttpEnvironment, ConnectionAttemptError>;
+    /** Lets the next direct authorization of this address skip its own descriptor read. */
+    readonly rememberCheckedDescriptor: (
+      httpBaseUrl: string,
+      descriptor: ExecutionEnvironmentDescriptor,
+    ) => Effect.Effect<void>;
     /**
      * Renews a live socket's token before it expires, so the server moves the
      * socket to the new session instead of closing it. Runs until interrupted.
@@ -95,6 +100,7 @@ export class RemoteEnvironmentAuthorization extends Context.Service<
 
 const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000;
 const DPOP_AUTHORIZATION_TIMEOUT_MS = 30_000;
+const CHECKED_DESCRIPTOR_TTL_MS = 5_000;
 
 function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) {
   return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError"
@@ -135,6 +141,13 @@ export const make = Effect.gen(function* () {
     }
   >();
 
+  // A host check that just answered hands its descriptor to the reconnect it
+  // triggers, once, so the reconnect skips a round trip.
+  const checkedDescriptors = new Map<
+    string,
+    { readonly descriptor: ExecutionEnvironmentDescriptor; readonly at: number }
+  >();
+
   /**
    * A saved LAN address can belong to another machine on a different network.
    */
@@ -144,9 +157,15 @@ export const make = Effect.gen(function* () {
       readonly httpBaseUrl: string;
       readonly connectionMethod: ClientConnectionMethod;
     }) {
-      const descriptor = yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe(
-        Effect.provideService(HttpClient.HttpClient, httpClient),
-      );
+      const checked = checkedDescriptors.get(input.httpBaseUrl);
+      checkedDescriptors.delete(input.httpBaseUrl);
+      const descriptor =
+        checked !== undefined &&
+        (yield* Clock.currentTimeMillis) - checked.at <= CHECKED_DESCRIPTOR_TTL_MS
+          ? checked.descriptor
+          : yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe(
+              Effect.provideService(HttpClient.HttpClient, httpClient),
+            );
       if (descriptor.environmentId !== input.expectedEnvironmentId) {
         return yield* environmentMismatchError({
           expected: input.expectedEnvironmentId,
@@ -527,6 +546,10 @@ export const make = Effect.gen(function* () {
 
   return RemoteEnvironmentAuthorization.of({
     authorizeBearer,
+    // Untraced: it runs on every live session's host check.
+    rememberCheckedDescriptor: Effect.fnUntraced(function* (httpBaseUrl, descriptor) {
+      checkedDescriptors.set(httpBaseUrl, { descriptor, at: yield* Clock.currentTimeMillis });
+    }),
     authorizeDpop: Effect.fn("environment.authorization")(function (input) {
       return authorizeDpop(input);
     }),

packages/client-runtime/src/connection/driver.ts

@@ -147,13 +147,18 @@ export const make = Effect.gen(function* () {
 
   // 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) =>
+  const checkDescriptor = (httpBaseUrl: string, environmentId: string, remember = false) =>
     fetchRemoteEnvironmentDescriptor({
       httpBaseUrl,
       timeoutMs: ROUTE_CHECK_TIMEOUT_MS,
     }).pipe(
-      Effect.map((descriptor): RouteCheck =>
-        descriptor.environmentId === environmentId ? "answered" : "silent",
+      Effect.flatMap((descriptor) =>
+        descriptor.environmentId !== environmentId
+          ? Effect.succeed<RouteCheck>("silent")
+          : (remember
+              ? remote.rememberCheckedDescriptor(httpBaseUrl, descriptor)
+              : Effect.void
+            ).pipe(Effect.as<RouteCheck>("answered")),
       ),
       Effect.orElseSucceed((): RouteCheck => "silent"),
       // A cached descriptor would answer for a host that cannot be reached.
@@ -171,8 +176,9 @@ export const make = Effect.gen(function* () {
     );
   };
 
+  // Only a live session's host check feeds the reconnect it may trigger.
   const checkHost = (prepared: PreparedConnection) =>
-    checkDescriptor(prepared.httpBaseUrl, prepared.environmentId).pipe(
+    checkDescriptor(prepared.httpBaseUrl, prepared.environmentId, true).pipe(
       Effect.withSpan("ConnectionDriver.checkHost", {
         attributes: { "connection.target.kind": prepared.target._tag },
       }),

packages/client-runtime/src/connection/resolver.test.ts

@@ -108,6 +108,7 @@ const makeDependencies = Effect.fn("TestConnectionResolver.makeDependencies")((o
           },
         })),
     authorizeDpopHttp: () => Effect.die("unused"),
+    rememberCheckedDescriptor: () => Effect.void,
     renewDpopBeforeExpiry: () => Effect.die("unused"),
   });
   const ssh = ClientCapabilities.SshEnvironmentGateway.of({

packages/client-runtime/src/rpc/session.test.ts

@@ -8,7 +8,7 @@ import {
   WS_METHODS,
   UsageLimitSourceId,
 } from "@t3tools/contracts";
-import { splitSocketFrame } from "@t3tools/shared/socketFrameParts";
+import { splitDeflatedSocketFrame, splitSocketFrame } from "@t3tools/shared/socketFrameParts";
 import { describe, expect, it } from "@effect/vitest";
 import * as Cause from "effect/Cause";
 import * as Deferred from "effect/Deferred";
@@ -91,7 +91,7 @@ class TestWebSocket {
     this.emit("open", { type: "open" });
   }
 
-  serverMessage(data: string) {
+  serverMessage(data: string | ArrayBuffer) {
     this.emit("message", { data, type: "message" });
   }
 
@@ -277,34 +277,52 @@ const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents
 });
 
 describe("RpcSessionFactory", () => {
-  it.effect("connects while a large config snapshot downloads at 4000 bytes per second", () =>
-    Effect.gen(function* () {
-      const { factory, sockets } = yield* makeFactory();
-      const session = yield* factory.connect(PREPARED);
-      const readyFiber = yield* Effect.forkChild(session.ready);
-      const socket = yield* awaitSocket(sockets);
-      socket.open();
-      const request = yield* awaitRequest(socket);
-      // Multi-byte text, so parts must be sized in UTF-8 bytes and never split a surrogate pair.
-      const config = { ...SERVER_CONFIG, cwd: "/" + "é€😀x".repeat(32 * 1024) };
-      const frame = encodeJson({
-        _tag: "Chunk",
-        requestId: request.id,
-        values: [{ version: 1, type: "snapshot", config: encodeServerConfig(config) }],
-      });
-      const parts = new URL(socket.url).searchParams.has("frameParts")
-        ? splitSocketFrame(frame)
-        : [frame];
-
-      for (const part of parts) {
-        const bytes = new TextEncoder().encode(part);
-        yield* TestClock.adjust((bytes.length / 4000) * 1000);
-        socket.serverMessage(new TextDecoder().decode(bytes));
-      }
-      yield* Fiber.join(readyFiber);
+  it.effect.each([
+    { label: "text parts", deflated: false },
+    { label: "deflated binary parts", deflated: true },
+  ])(
+    "connects while a large config snapshot downloads at 4000 bytes per second as $label",
+    ({ deflated }) =>
+      Effect.gen(function* () {
+        const { factory, sockets } = yield* makeFactory();
+        const session = yield* factory.connect(PREPARED);
+        const readyFiber = yield* Effect.forkChild(session.ready);
+        const socket = yield* awaitSocket(sockets);
+        socket.open();
+        const request = yield* awaitRequest(socket);
+        // Multi-byte text, so parts must be sized in UTF-8 bytes and never split a surrogate pair.
+        const config = { ...SERVER_CONFIG, cwd: "/" + "é€😀x".repeat(32 * 1024) };
+        const frame = encodeJson({
+          _tag: "Chunk",
+          requestId: request.id,
+          values: [{ version: 1, type: "snapshot", config: encodeServerConfig(config) }],
+        });
+        const url = new URL(socket.url).searchParams;
+        expect(url.has("frameParts") && url.has("frameDeflate")).toBe(true);
+        const parts = deflated
+          ? splitDeflatedSocketFrame(
+              yield* Effect.promise(() =>
+                new Response(
+                  new Blob([frame]).stream().pipeThrough(new CompressionStream("deflate-raw")),
+                )
+                  .arrayBuffer()
+                  .then((buffer) => new Uint8Array(buffer)),
+              ),
+              // Repeated text deflates small, so a small part size forces several parts.
+              1024,
+            )
+          : splitSocketFrame(frame);
+        expect(parts.length).toBeGreaterThan(1);
+
+        for (const part of parts) {
+          const bytes = typeof part === "string" ? new TextEncoder().encode(part) : part;
+          yield* TestClock.adjust((bytes.length / 4000) * 1000);
+          socket.serverMessage(typeof part === "string" ? part : bytes.slice().buffer);
+        }
+        yield* Fiber.join(readyFiber);
 
-      expect(yield* session.initialConfig).toEqual(config);
-    }),
+        expect(yield* session.initialConfig).toEqual(config);
+      }),
   );
 
   it.effect("owns one scoped websocket attempt and exposes readiness and closure", () =>
@@ -314,7 +332,7 @@ describe("RpcSessionFactory", () => {
       const readyFiber = yield* Effect.forkChild(session.ready);
       const socket = yield* awaitSocket(sockets);
 
-      expect(socket.url).toBe(`${PREPARED.socketUrl}&frameParts=1`);
+      expect(socket.url).toBe(`${PREPARED.socketUrl}&frameParts=1&frameDeflate=1`);
       socket.open();
       yield* completeInitialConfig(socket);
       yield* Fiber.join(readyFiber);

packages/client-runtime/src/rpc/session.ts

@@ -6,6 +6,7 @@ import {
 } from "@t3tools/contracts";
 import {
   joinSocketFrameParts,
+  SOCKET_FRAME_DEFLATE_QUERY_PARAM,
   SOCKET_FRAME_PARTS_QUERY_PARAM,
 } from "@t3tools/shared/socketFrameParts";
 import * as Cause from "effect/Cause";
@@ -231,6 +232,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
     };
     const socketUrl = new URL(connection.socketUrl);
     socketUrl.searchParams.set(SOCKET_FRAME_PARTS_QUERY_PARAM, "1");
+    socketUrl.searchParams.set(SOCKET_FRAME_DEFLATE_QUERY_PARAM, "1");
     const layerSocket = Layer.effect(
       Socket.Socket,
       Effect.map(Effect.service(Socket.Socket), joinSocketFrameParts),

packages/client-runtime/src/state/environmentHttpAuth.test.ts

@@ -146,6 +146,7 @@ function makeHarness(reply: (requestNumber: number) => Response | Promise<Respon
           },
         };
       }),
+    rememberCheckedDescriptor: () => Effect.void,
     renewDpopBeforeExpiry: () => Effect.die("HTTP requests must not renew a socket's token."),
   });
   const signer = ManagedRelay.ManagedRelayDpopSigner.of({