fix(state): back off when the connection closes before connecting

packages/reflex-base/news/+reconnect-backoff.bugfix.md

@@ -0,0 +1 @@
+Retry with backoff and show the connection error when the server or a proxy closes the connection before it is established, for example when browser storage makes the hydrate message too large, instead of reconnecting in a tight loop.

packages/reflex-base/src/reflex_base/.templates/web/utils/state.js

@@ -1001,11 +1001,18 @@ export const connect = async (
 
   // Unknown substates already reported on this connection.
   const reported_missing_substates = new Set();
+  // Failed connection attempts since the last connect, for the retry backoff.
+  let failed_attempts = 0;
+  // The scheduled retry of the last failed attempt.
+  let retry_timer;
 
   // Once the socket is open, hydrate the page.
   socket.current.on("connect", async () => {
     socket.current.wait_connect = false;
     reported_missing_substates.clear();
+    failed_attempts = 0;
+    window.clearTimeout(retry_timer);
+    retry_timer = undefined;
     setConnectErrors([]);
     window.__reflex_otel?.onSocketConnect?.();
     window.addEventListener("pagehide", pagehideHandler);
@@ -1016,23 +1023,35 @@ export const connect = async (
     }
   });
 
-  socket.current.on("connect_error", (error) => {
+  // Record a failed connection attempt and retry it with incremental backoff.
+  const retryConnect = (error) => {
     socket.current.wait_connect = false;
-    let n_connect_errors = 0;
-    setConnectErrors((connectErrors) => {
-      const new_errors = [...connectErrors.slice(-9), error];
-      n_connect_errors = new_errors.length;
-      return new_errors;
-    });
-    window.setTimeout(() => {
+    failed_attempts = Math.min(failed_attempts + 1, 10);
+    setConnectErrors((connectErrors) => [...connectErrors.slice(-9), error]);
+    window.clearTimeout(retry_timer);
+    retry_timer = window.setTimeout(() => {
+      retry_timer = undefined;
       if (socket.current && !socket.current.connected) {
         socket.current.reconnect();
       }
-    }, 200 * n_connect_errors); // Incremental backoff
-  });
+    }, 200 * failed_attempts); // Incremental backoff
+  };
+
+  socket.current.on("connect_error", retryConnect);
 
   socket.current.on("disconnect", (reason, details) => {
-    socket.current.wait_connect = false;
+    // A transport closed before the namespace connected is a failed attempt,
+    // e.g. a CONNECT packet over the server or a proxy message size limit.
+    if (socket.current.wait_connect) {
+      retryConnect(
+        new Error(`connection closed before connecting (${reason})`),
+      );
+      return;
+    }
+    // A transport error reports connect_error first, which scheduled the retry.
+    if (retry_timer !== undefined) {
+      return;
+    }
     window.__reflex_otel?.onSocketDisconnect?.(reason);
     const try_reconnect =
       reason !== "io server disconnect" && reason !== "io client disconnect";

tests/integration/tests_playwright/test_hydration_storage.py

@@ -1,7 +1,9 @@
 """Browser regressions for client-storage writes during hydration and assigned defaults."""
 
 import re
+import time
 from collections.abc import Callable, Generator
+from itertools import pairwise
 
 import pytest
 from playwright.sync_api import Page, WebSocketRoute, expect
@@ -651,3 +653,59 @@ def test_synced_storage_event_while_disconnected_sends_current_value(
     expect(page.locator("#sync-value")).to_have_text("fresh")
     assert not any("update_vars_internal" in m and '"stale"' in m for m in socket.sent)
     assert page.evaluate("localStorage.getItem('hydrate-sync')") == "fresh"
+
+
+@pytest.mark.parametrize("limit", ["server", "proxy"])
+def test_refused_hydration_retries_with_backoff(
+    hydration_storage_app: AppHarness, page: Page, limit: str
+):
+    """A CONNECT packet dropped for its size is retried with backoff and reported.
+
+    Args:
+        hydration_storage_app: The running app.
+        page: A fresh browser page.
+        limit: Whether the server limit or a smaller proxy limit drops the packet.
+    """
+    assert hydration_storage_app.frontend_url is not None
+    opens: list[float] = []
+    page.on(
+        "websocket",
+        lambda ws: opens.append(time.monotonic()) if "/_event" in ws.url else None,
+    )
+    if limit == "proxy":
+
+        def proxy(route: WebSocketRoute):
+            """Close the transport on a large frame, like a proxy with its own limit.
+
+            Args:
+                route: The intercepted websocket.
+            """
+            server = route.connect_to_server()
+            route.on_message(
+                lambda m: route.close() if len(m) > 100_000 else server.send(m)
+            )
+            server.on_message(route.send)
+
+        page.route_web_socket(re.compile(r".*/_event"), proxy)
+    page.goto(hydration_storage_app.frontend_url)
+    expect(page.locator("#hydrated")).to_have_text("true")
+
+    size = 1_100_000 if limit == "server" else 200_000
+    page.evaluate("n => localStorage.setItem('hydrate-local', 'x'.repeat(n))", size)
+    opens.clear()
+    page.reload()
+    expect(page.get_by_text("Cannot connect to server")).to_be_visible()
+    deadline = time.monotonic() + 10
+    while len(opens) < 4 and time.monotonic() < deadline:
+        page.wait_for_timeout(100)
+    # Each retry waits 200 ms longer than the one before; an immediate
+    # reconnect retries within a few ms.
+    gaps = [b - a for a, b in pairwise(opens[:4])]
+    assert len(gaps) == 3, gaps
+    assert all(gap > 0.15 for gap in gaps), gaps
+    assert gaps[0] < gaps[2], gaps
+    expect(page.locator("#hydrated")).to_have_text("false")
+
+    page.evaluate("localStorage.removeItem('hydrate-local')")
+    page.reload()
+    expect(page.locator("#hydrated")).to_have_text("true")

tests/units/compiler/state_js.test.mjs

@@ -19,13 +19,17 @@ async function setup({
   hidden = false,
   storedToken = "original-token",
   clientStorage = {},
+  dropConnects = 0,
+  dropReason = "transport close",
   react = { useCallback() {}, useEffect() {}, useRef() {}, useState() {} },
 } = {}) {
   const sockets = [];
   const microtasks = [];
   const timers = new Map();
   const firstHydrates = [];
   const updates = [];
+  let connectErrors = [];
+  const errorUpdates = [];
   const window = new EventTarget();
   window.location = new URL("http://localhost:3000/page?query=value");
   const storage = new Map(storedToken ? [["token", storedToken]] : []);
@@ -58,10 +62,12 @@ async function setup({
     clearTimeout: (fn) => timers.delete(fn),
   });
   window.setTimeout = context.setTimeout;
+  window.clearTimeout = context.clearTimeout;
 
   function io(url, opts) {
     const handlers = new Map();
     const socket = {
+      handlers,
       connected: false,
       namespaceConnects: 0,
       disconnects: 0,
@@ -90,6 +96,15 @@ async function setup({
           assert.ok(handlers.has(name), `missing handler: ${name}`);
         }
         this.namespaceConnects++;
+        if (this.namespaceConnects <= dropConnects) {
+          // The transport closes before the CONNECT is acknowledged. A
+          // transport error reports connect_error before the disconnect.
+          if (dropReason === "transport error") {
+            handlers.get("connect_error")(new Error("xhr post error"));
+          }
+          handlers.get("disconnect")(dropReason);
+          return;
+        }
         this.connected = true;
         handlers.get("connect")();
         handlers.get("new_token")("assigned-token");
@@ -176,6 +191,14 @@ async function setup({
     localStorage,
     socket,
     updates,
+    get connectErrors() {
+      // Like React with pending work, apply queued updaters only on render.
+      for (const update of errorUpdates.splice(0)) {
+        connectErrors =
+          typeof update === "function" ? update(connectErrors) : update;
+      }
+      return connectErrors;
+    },
     module,
     dispose,
     flush: () => microtasks.splice(0).forEach((fn) => fn()),
@@ -184,7 +207,7 @@ async function setup({
         socket,
         { child: (delta) => updates.push(delta.count) },
         transports,
-        () => {},
+        (update) => errorUpdates.push(update),
         {},
         () => {},
         { current: {} },
@@ -252,6 +275,69 @@ test("reconnect uses the assigned token and requests a full hydrate", async () =
   assert.equal(app.socket.current.namespaceConnects, 2);
 });
 
+test("a transport closed before the namespace connects backs off with an error", async () => {
+  const app = await setup({ dropConnects: Infinity });
+  app.flush();
+  await app.connect();
+  const socket = app.socket.current;
+  assert.equal(socket.namespaceConnects, 1);
+  assert.equal(app.connectErrors.length, 1);
+  assert.deepEqual([...app.timers.values()], [200]);
+
+  const [retry] = app.timers.keys();
+  app.timers.clear();
+  retry();
+  assert.equal(socket.namespaceConnects, 2);
+  assert.equal(app.connectErrors.length, 2);
+  assert.deepEqual([...app.timers.values()], [400]);
+});
+
+test("a transport error before the namespace connects retries once with backoff", async () => {
+  const app = await setup({ dropConnects: 2, dropReason: "transport error" });
+  app.flush();
+  await app.connect();
+  const socket = app.socket.current;
+  assert.equal(socket.namespaceConnects, 1);
+  assert.equal(app.connectErrors.length, 1);
+  assert.deepEqual([...app.timers.values()], [200]);
+
+  const [retry] = app.timers.keys();
+  app.timers.clear();
+  retry();
+  assert.equal(socket.namespaceConnects, 2);
+  assert.deepEqual([...app.timers.values()], [400]);
+});
+
+test("a connect cancels the pending retry", async () => {
+  const app = await setup({ dropConnects: 1 });
+  app.flush();
+  await app.connect();
+  const socket = app.socket.current;
+  assert.equal(app.timers.size, 1);
+
+  // Another path, such as the page becoming visible, reconnects first.
+  socket.reconnect();
+  assert.equal(socket.connected, true);
+  assert.equal(app.timers.size, 0);
+
+  socket.connected = false;
+  socket.handlers.get("disconnect")("io server disconnect");
+  assert.equal(socket.namespaceConnects, 2);
+  assert.equal(app.timers.size, 0);
+});
+
+test("a dropped connection reconnects at once", async () => {
+  const app = await setup();
+  app.flush();
+  await app.connect();
+  const socket = app.socket.current;
+  socket.connected = false;
+  socket.handlers.get("disconnect")("transport close");
+  assert.equal(socket.namespaceConnects, 2);
+  assert.equal(app.connectErrors.length, 0);
+  assert.equal(app.timers.size, 0);
+});
+
 for (const reason of ["error", "timeout", "pagehide", "hot reload"]) {
   test(`discard an unclaimed warm transport on ${reason}`, async () => {
     const app = await setup();

tests/units/reflex_base/client_state_updates.mjs

@@ -65,6 +65,7 @@ async function createConnection(
     dependencies: {
       "test:browser": {
         window: {
+          clearTimeout,
           location: {
             pathname: "/",
             search: "",