[pull] main from liveblocks:main

CHANGELOG.md

@@ -1,5 +1,13 @@
 ## vNEXT (not yet released)
 
+## v3.19.3
+
+### `@liveblocks/client`
+
+- Fix unexpected disconnects that could happen while receiving large or
+  long-running streaming responses from the server (e.g. when loading a large
+  initial storage state).
+
 ## v3.19.2
 
 ### `@liveblocks/client`

docs/pages/get-started/nextjs-comments-ai.mdx

@@ -23,7 +23,7 @@ your Next.js `/app` directory application.
     <StepTitle>Have a Comments app ready</StepTitle>
     <StepContent>
 
-      To add AI replies to comment thread, you first need to have a Liveblocks
+      To add AI replies to a comment thread, you first need to have a Liveblocks
       Comments app set up with secret key authentication and resolved users.
       Open up your app, or set up comments if you haven’t already.
 
@@ -278,7 +278,7 @@ export async function handleAiCommentReply(data: {
     system: `You are a helpful assistant replying inside a Liveblocks comment thread.
 
   - Reply concisely and to the point.
-  - Reply in plain text. Do not use markdown.
+  - You can use inline markdown.
   - Your user ID is ${AI_USER_INFO.id}.`,
     messages,
   });
@@ -336,7 +336,7 @@ export async function handleAiCommentReply(data: {
   <StepTitle>Complete!</StepTitle>
   <StepContent>
     You now have an AI agent capable of replying to mentions in comment threads.
-    When it’s mentioned in a acomment, it’ll leave a placeholder comment, and
+    When it’s mentioned in a comment, it’ll leave a placeholder comment, and
     edit it after generating a response.
   </StepContent>
 </Step>

packages/liveblocks-chat-sdk-adapter/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@liveblocks/chat-sdk-adapter",
-  "version": "3.19.2",
+  "version": "3.19.3",
   "description": "Liveblocks adapter for the Chat SDK.",
   "license": "Apache-2.0",
   "author": "Liveblocks Inc.",

packages/liveblocks-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@liveblocks/client",
-  "version": "3.19.2",
+  "version": "3.19.3",
   "description": "A client that lets you interact with Liveblocks servers. Liveblocks is the all-in-one toolkit to build collaborative products like Figma, Notion, and more.",
   "license": "Apache-2.0",
   "author": "Liveblocks Inc.",

packages/liveblocks-core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@liveblocks/core",
-  "version": "3.19.2",
+  "version": "3.19.3",
   "description": "Private internals for Liveblocks. DO NOT import directly from this package!",
   "type": "module",
   "main": "./dist/index.cjs",

packages/liveblocks-core/src/connection.ts

@@ -4,7 +4,7 @@ import type { Observable } from "./lib/EventSource";
 import { makeBufferableEventSource, makeEventSource } from "./lib/EventSource";
 import * as console from "./lib/fancy-console";
 import type { BuiltinEvent, Patchable, Target } from "./lib/fsm";
-import { FSM } from "./lib/fsm";
+import { FSM, IGNORE } from "./lib/fsm";
 import type { Json } from "./lib/Json";
 import { tryParseJson, withTimeout } from "./lib/utils";
 import { ServerMsgCode } from "./protocol/ServerMsg";
@@ -98,7 +98,7 @@ type Event =
   | { type: "NAVIGATOR_OFFLINE" } // e.g. browser goes offline
 
   // Events that the connection manager will internally deal with
-  | { type: "PONG" }
+  | { type: "ALIVE" } // Previously called "PONG", but widened to include any socket activity
   | { type: "EXPLICIT_SOCKET_ERROR"; event: IWebSocketEvent }
   | { type: "EXPLICIT_SOCKET_CLOSE"; event: IWebSocketCloseEvent }
 
@@ -171,8 +171,9 @@ const BACKOFF_DELAYS_SLOW = [2_000, 30_000, 60_000, 300_000] as const;
 
 /**
  * The client will send a PING to the server every 30 seconds, after which it
- * must receive a PONG back within the next 2 seconds. If that doesn't happen,
- * this is interpreted as an implicit connection loss event.
+ * must receive a PONG (or any other sign of activity) back within the next
+ * 2 seconds. If nothing arrives in that window, the connection is treated as
+ * implicitly lost.
  */
 const HEARTBEAT_INTERVAL = 30_000;
 const PONG_TIMEOUT = 2_000;
@@ -302,8 +303,8 @@ function enableTracing(machine: FSM<Context, Event, State>) {
     machine.events.didExitState.subscribe(({ state, durationMs }) =>
       log(`Exited ${state} after ${durationMs.toFixed(0)}ms`)
     ),
-    machine.events.didIgnoreEvent.subscribe((e) =>
-      log("Ignored event", e.type, e, "(current state won't handle it)")
+    machine.events.didIgnoreUnexpectedEvent.subscribe((e) =>
+      log("Ignored unexpected event", e.type, e, "(no transition declared)")
     ),
   ];
   return () => {
@@ -503,10 +504,14 @@ function createConnectionStateMachine<T extends BaseAuthResult>(
   const onSocketClose = (event: IWebSocketCloseEvent) =>
     machine.send({ type: "EXPLICIT_SOCKET_CLOSE", event });
 
-  const onSocketMessage = (event: IWebSocketMessageEvent) =>
-    event.data === "pong"
-      ? machine.send({ type: "PONG" })
-      : onMessage.notify(event);
+  const onSocketMessage = (event: IWebSocketMessageEvent) => {
+    // Every inbound message counts as activity, not just explicit PONGs
+    machine.send({ type: "ALIVE" });
+
+    if (event.data !== "pong") {
+      onMessage.notify(event);
+    }
+  };
 
   function teardownSocket(socket: IWebSocketInstance | null) {
     if (socket) {
@@ -764,18 +769,25 @@ function createConnectionStateMachine<T extends BaseAuthResult>(
           effect: [increaseBackoffDelay, logPrematureErrorOrCloseEvent(err)],
         };
       }
-    );
+    )
+    .addTransitions("@connecting.busy", {
+      // The socket message listener is attached during @connecting.busy (see
+      // onEnterAsync above), so server frames (most notably the actor-id
+      // handshake) can fire onSocketMessage and emit a ALIVE before we
+      // reach @ok.*. That's fine. Heartbeat only matters in @ok.*.
+      ALIVE: IGNORE,
+    });
 
   //
   // Configure the @ok.* states
   //
   // Keeps a heartbeat alive with the server whenever in the @ok.* state group.
   // 30 seconds after entering the "@ok.connected" state, it will emit
-  // a heartbeat, and awaits a PONG back that should arrive within 2 seconds.
-  // If this happens, then it transitions back to normal "connected" state, and
-  // the cycle repeats. If the PONG is not received timely, then we interpret
-  // it as an implicit connection loss, and transition to reconnect (throw away
-  // this socket, and open a new one).
+  // a heartbeat, and awaits a PONG (or any other sign of activity) back that
+  // should arrive within 2 seconds. If this happens, it transitions back to
+  // "@ok.connected" and the cycle repeats. If nothing arrives in time, we
+  // interpret it as an implicit connection loss and transition to reconnect
+  // (throw away this socket, and open a new one).
   //
 
   const sendHeartbeat: Target<Context, Event | BuiltinEvent, State> = {
@@ -799,6 +811,7 @@ function createConnectionStateMachine<T extends BaseAuthResult>(
     .addTransitions("@ok.connected", {
       NAVIGATOR_OFFLINE: maybeHeartbeat, // Don't take the browser's word for it when it says it's offline. Do a ping/pong to make sure.
       WINDOW_GOT_FOCUS: sendHeartbeat,
+      ALIVE: IGNORE,
     });
 
   machine.addTransitions("@idle.zombie", {
@@ -827,7 +840,7 @@ function createConnectionStateMachine<T extends BaseAuthResult>(
       };
     })
 
-    .addTransitions("@ok.awaiting-pong", { PONG: "@ok.connected" })
+    .addTransitions("@ok.awaiting-pong", { ALIVE: "@ok.connected" })
     .addTimedTransition("@ok.awaiting-pong", PONG_TIMEOUT, {
       target: "@connecting.busy",
       // Log implicit connection loss and drop the current open socket
@@ -844,7 +857,7 @@ function createConnectionStateMachine<T extends BaseAuthResult>(
       EXPLICIT_SOCKET_ERROR: (_, context) => {
         if (context.socket?.readyState === 1 /* WebSocket.OPEN */) {
           // TODO Do we need to forward this error to the client?
-          return null; /* Do not leave OK state, socket is still usable */
+          return IGNORE; /* Do not leave OK state, socket is still usable */
         }
 
         return {

packages/liveblocks-core/src/lib/__tests__/fsm.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, test, vi } from "vitest";
 
-import { distance, FSM, patterns } from "../fsm";
+import { distance, FSM, IGNORE, patterns } from "../fsm";
 import { wait } from "../utils";
 
 async function failAfter(ms: number): Promise<void> {
@@ -555,7 +555,7 @@ describe("finite state machine", () => {
         GO: () =>
           n++ % 2 === 0
             ? "one" // Transition if n is even
-            : null, // Otherwise, do nothing
+            : IGNORE, // Otherwise, do nothing
       })
       .start();
 
@@ -572,6 +572,136 @@ describe("finite state machine", () => {
     expect(fsm.currentState).toEqual("one");
   });
 
+  describe("IGNORE sentinel", () => {
+    test("static IGNORE keeps state unchanged", () => {
+      const fsm = new FSM({})
+        .addState("foo")
+        .addTransitions("foo", { GO: IGNORE })
+        .start();
+
+      expect(fsm.currentState).toEqual("foo");
+      fsm.send({ type: "GO" });
+      expect(fsm.currentState).toEqual("foo");
+    });
+
+    test("static IGNORE fires no observable notifications", () => {
+      const didReceive = vi.fn();
+      const willTransition = vi.fn();
+      const didIgnoreUnexpected = vi.fn();
+
+      const fsm = new FSM({})
+        .addState("foo")
+        .addTransitions("foo", { GO: IGNORE })
+        .start();
+
+      fsm.events.didReceiveEvent.subscribe(didReceive);
+      fsm.events.willTransition.subscribe(willTransition);
+      fsm.events.didIgnoreUnexpectedEvent.subscribe(didIgnoreUnexpected);
+
+      fsm.send({ type: "GO" });
+
+      expect(didReceive).not.toHaveBeenCalled();
+      expect(willTransition).not.toHaveBeenCalled();
+      expect(didIgnoreUnexpected).not.toHaveBeenCalled();
+    });
+
+    test("missing transition still fires didIgnoreUnexpectedEvent (control)", () => {
+      const didIgnoreUnexpected = vi.fn();
+
+      // Declare GO somewhere so the FSM accepts it as a known event type,
+      // but not in state "foo" — so sending GO in "foo" goes unhandled.
+      const fsm = new FSM({})
+        .addState("foo")
+        .addState("bar")
+        .addTransitions("bar", { GO: "foo" })
+        .start();
+
+      fsm.events.didIgnoreUnexpectedEvent.subscribe(didIgnoreUnexpected);
+
+      expect(fsm.currentState).toEqual("foo");
+      fsm.send({ type: "GO" });
+      expect(didIgnoreUnexpected).toHaveBeenCalledTimes(1);
+    });
+
+    test("dynamic IGNORE (function returning IGNORE) is silent except for didReceiveEvent", () => {
+      const didReceive = vi.fn();
+      const willTransition = vi.fn();
+      const didIgnoreUnexpected = vi.fn();
+
+      const fsm = new FSM({})
+        .addState("foo")
+        .addTransitions("foo", { GO: () => IGNORE })
+        .start();
+
+      fsm.events.didReceiveEvent.subscribe(didReceive);
+      fsm.events.willTransition.subscribe(willTransition);
+      fsm.events.didIgnoreUnexpectedEvent.subscribe(didIgnoreUnexpected);
+
+      fsm.send({ type: "GO" });
+
+      // We had to invoke the function to learn it returns IGNORE.
+      expect(didReceive).toHaveBeenCalledTimes(1);
+      // But no transition happened, and the event is not "unexpected".
+      expect(willTransition).not.toHaveBeenCalled();
+      expect(didIgnoreUnexpected).not.toHaveBeenCalled();
+      expect(fsm.currentState).toEqual("foo");
+    });
+
+    test("mixing IGNORE with real transitions in the same mapping", () => {
+      const fsm = new FSM({})
+        .addState("foo")
+        .addState("bar")
+        .addTransitions("foo", { GO: "bar", PING: IGNORE })
+        .start();
+
+      fsm.send({ type: "PING" });
+      expect(fsm.currentState).toEqual("foo"); // silent no-op
+
+      fsm.send({ type: "GO" });
+      expect(fsm.currentState).toEqual("bar"); // normal transition
+    });
+
+    test("IGNORE works via wildcard pattern for a state group", () => {
+      const didIgnoreUnexpected = vi.fn();
+
+      const fsm = new FSM({})
+        .addState("group.one")
+        .addState("group.two")
+        .addState("other")
+        .addTransitions("group.*", { PING: IGNORE, GO: "other" })
+        .start();
+
+      fsm.events.didIgnoreUnexpectedEvent.subscribe(didIgnoreUnexpected);
+
+      expect(fsm.currentState).toEqual("group.one");
+      fsm.send({ type: "PING" });
+      expect(fsm.currentState).toEqual("group.one");
+      expect(didIgnoreUnexpected).not.toHaveBeenCalled();
+
+      // Switch to group.two by adding a transition path; reuse "GO".
+      // To exercise PING in group.two we need to be there first.
+      // Build a separate FSM for that leg to keep this test focused.
+      const fsm2 = new FSM({})
+        .addState("group.two")
+        .addTransitions("group.*", { PING: IGNORE })
+        .start();
+      const didIgnoreUnexpected2 = vi.fn();
+      fsm2.events.didIgnoreUnexpectedEvent.subscribe(didIgnoreUnexpected2);
+      fsm2.send({ type: "PING" });
+      expect(fsm2.currentState).toEqual("group.two");
+      expect(didIgnoreUnexpected2).not.toHaveBeenCalled();
+    });
+
+    test("declaring IGNORE for an already-declared event still throws", () => {
+      expect(() =>
+        new FSM({})
+          .addState("foo")
+          .addTransitions("foo", { GO: "foo" })
+          .addTransitions("foo", { GO: IGNORE })
+      ).toThrow(/transition already exists/);
+    });
+  });
+
   describe("time-based transitions", () => {
     test("time-based transitions", () => {
       vi.useFakeTimers();