diff --git a/CHANGELOG.md b/CHANGELOG.md index 022a38078e..b0606147f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,12 @@ ## vNEXT (not yet released) -- Export internal utility +## v3.19.5 + +### `@liveblocks/client` + +- Fix a `LiveList` divergence after reconnects: a pending `push` could under + specific timing conditions during a reconnect still cause a divergence between + clients, despite the fix from 3.19.4. ## v3.19.4 diff --git a/packages/liveblocks-chat-sdk-adapter/package.json b/packages/liveblocks-chat-sdk-adapter/package.json index 35ddbbda8e..c318196445 100644 --- a/packages/liveblocks-chat-sdk-adapter/package.json +++ b/packages/liveblocks-chat-sdk-adapter/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/chat-sdk-adapter", - "version": "3.19.4", + "version": "3.19.5", "description": "Liveblocks adapter for the Chat SDK.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-client/package.json b/packages/liveblocks-client/package.json index e1d451c38c..0a5a2c839f 100644 --- a/packages/liveblocks-client/package.json +++ b/packages/liveblocks-client/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/client", - "version": "3.19.4", + "version": "3.19.5", "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.", diff --git a/packages/liveblocks-core/e2e/README.md b/packages/liveblocks-core/e2e/README.md index fc7cfb6948..9e010158de 100644 --- a/packages/liveblocks-core/e2e/README.md +++ b/packages/liveblocks-core/e2e/README.md @@ -44,6 +44,6 @@ Run a specific test file: npx turbo test:e2e -- e2e/list-insert.test.ts ``` -**Note**: Since these tests run against an actual production deployment, they -require a `LIVEBLOCKS_PUBLIC_KEY` environment variable to connect to the -Liveblocks service. +**Note**: These tests run against a local Liveblocks dev server, which the +`test:e2e` script starts automatically (`liveblocks dev`). No API key is +needed; set `LIVEBLOCKS_DEV_SERVER_PORT` to override the default port (1154). diff --git a/packages/liveblocks-core/e2e/list-consistency.test.ts b/packages/liveblocks-core/e2e/list-consistency.test.ts index 7f3c38bb2f..d1f8ddd3d5 100644 --- a/packages/liveblocks-core/e2e/list-consistency.test.ts +++ b/packages/liveblocks-core/e2e/list-consistency.test.ts @@ -91,7 +91,7 @@ test( // This test verifies that undo/redo operations maintain consistency across clients // when operations are performed on different clients in a distributed environment. async ({ root1, root2, room1, room2, control, assert }) => { - // Client A does a move operation: move C (index 2) to position 0 + // Client A moves 🟒 (index 2) to position 0; Client B deletes it root1.get("list").move(2, 0); root2.get("list").delete(2); assert( diff --git a/packages/liveblocks-core/e2e/list-push-reconnect-divergence.test.ts b/packages/liveblocks-core/e2e/list-push-reconnect-divergence.test.ts index c9ac62011b..b97ae32ceb 100644 --- a/packages/liveblocks-core/e2e/list-push-reconnect-divergence.test.ts +++ b/packages/liveblocks-core/e2e/list-push-reconnect-divergence.test.ts @@ -1,7 +1,8 @@ import { expect, test } from "vitest"; import { LiveList } from "../src/crdts/LiveList"; -import { prepareTestsConflicts } from "./utils"; +import { withTimeout } from "../src/lib/utils"; +import { prepareTestsConflicts, waitUntilStatus } from "./utils"; const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); @@ -21,16 +22,15 @@ async function waitUntil( } /** - * Deterministic reproduction of the offline.test.ts "client synchronizes - * offline changes" divergence. + * Deterministic version of the offline.test.ts "client synchronizes offline + * changes" scenario. * - * The trigger is an item that the server has already stored but that the + * The tricky case is an item that the server has already stored but that the * pushing client still considers pending (unacknowledged), because the client - * never received the ack. On reconnect, the client's optimistic tail-bump - * moves that pending push past a sibling the other client added, re-sends it at - * its original key, and the server bare-acks it (already stored, no - * reposition), so the bump is never undone. The two clients then disagree on - * the order. + * never received the ack. On reconnect, the snapshot also carries a sibling + * the other client appended in the meantime. The still-pending item must keep + * its server position (before the sibling) on both clients, rather than being + * optimistically bumped past it by its own re-sent push. */ test( "a pending push the server already stored keeps its server position after reconnect", @@ -40,11 +40,12 @@ test( const list1 = root1.get("list"); const list2 = root2.get("list"); - // 1. Client A pushes P and flushes it to the server, but drops every - // incoming message first, so the server's ack/echo never reaches A: P - // is stored server-side (so B sees it) yet stays *pending* on A. + // 1. Client A pushes P and flushes it to the server, but stalls its + // downlink first, so the server's ack/echo never reaches A before the + // connection drops: P is stored server-side (so B sees it) yet stays + // *pending* on A. list1.push("P"); - control.dropIncomingA(); + control.pauseIncomingA(); control.flushSyncA(); await waitUntil( () => [...list2].includes("P"), @@ -62,9 +63,9 @@ test( "Client B sees [P, Q]" ); - // 4. A reconnects. The snapshot carries both P and Q; A's tail-bump moves - // its still-pending P past Q, then re-sends P at its original key. The - // server bare-acks (P already there), so A's bump is never undone. + // 4. A reconnects. The snapshot carries both P and Q, and A re-sends its + // still-pending P. P is already stored server-side, so it must keep + // its server position (before Q) on both clients. room1.reconnect(); await waitUntil( @@ -76,8 +77,76 @@ test( await sleep(500); // Both clients must agree on the server's order, [P, Q]. - expect([...list1]).toEqual([...list2]); - expect([...list2]).toEqual(["P", "Q"]); + expect(list1.toJSON()).toEqual(list2.toJSON()); + expect(list2.toJSON()).toEqual(["P", "Q"]); + } + ) +); + +/** + * Same divergence, but reached *after* the snapshot reconcile, via a live op. + * + * After a reconnect, the snapshot reconcile itself adopts the server's + * positions, but the re-sent pending push stays unacknowledged until the + * server's ack lands. A remote sibling push arriving as a live op inside that + * window triggers the optimistic tail-bump, which moves the pending push past + * the sibling. The server already stored the push, so the re-send is acked + * without a repositioning op, and the bump is never undone. + * + * In the wild this window is widened by large list items (their re-send is + * slow to reach the server), which is why the bug shows up intermittently and + * mostly with big payloads. The test simulates that slowness by stalling A's + * uplink while the re-send sits on it. + */ +test( + "a sibling pushed while a re-sent pending push awaits its ack keeps its server position", + prepareTestsConflicts( + { list: new LiveList([]) }, + async ({ root1, root2, room1, control }) => { + const list1 = root1.get("list"); + const list2 = root2.get("list"); + + // 1. Client A pushes P and flushes it to the server, but stalls its + // downlink first, so the server's ack/echo never reaches A before the + // connection drops: P is stored server-side (so B sees it) yet stays + // *pending* on A. + list1.push("P"); + control.pauseIncomingA(); + await control.flushA(); // Ensure client B sees P + + // 2. A reconnects, and we stall the fresh socket's uplink right after it + // connects: at that point its FETCH_STORAGE request is already out + // (sent synchronously on connect), but the snapshot needs a server + // round trip, so the reconcile hasn't run yet. The reconcile then + // puts the re-send of P on the stalled uplink instead of on the wire, + // keeping P pending. The reconcile signals completion through the + // storageDidLoad event. + const reconciled$ = room1.events.storageDidLoad.waitUntil(); + room1.reconnect(); + await waitUntilStatus(room1, "connecting"); + await waitUntilStatus(room1, "connected"); + control.pauseA(); + await withTimeout( + reconciled$, + 10_000, + "Client A did not reconcile after reconnect within 10s" + ); + + // 3. B pushes Q. It reaches A as a live op while P is still pending. + list2.push("Q"); + await control.flushB(); + + // 4. Only now release P's re-send. P is already stored server-side, so + // it must keep its server position (before Q) on both clients. + await control.flushA(); + + // Let any acks settle. (flushA's beacon is confirmed by Client B, so it + // says nothing about A having received its ack yet.) + await sleep(500); + + // Both clients must agree on the server's order, [P, Q]. + expect(list2.toJSON()).toEqual(["P", "Q"]); + expect(list1.toJSON()).toEqual(list2.toJSON()); } ) ); diff --git a/packages/liveblocks-core/e2e/list-push.test.ts b/packages/liveblocks-core/e2e/list-push.test.ts index 33b7295173..af383286ab 100644 --- a/packages/liveblocks-core/e2e/list-push.test.ts +++ b/packages/liveblocks-core/e2e/list-push.test.ts @@ -4,11 +4,11 @@ import { LiveList } from "../src/crdts/LiveList"; import { prepareTestsConflicts } from "./utils"; // Two actors append to the same LiveList near-simultaneously: client A appends -// a1 then a2; client B appends b1 without yet having seen a1/a2, so b1 guesses -// the head position. By the time b1 reaches the server, a1 and a2 are already -// stored, and the position conflict is resolved *between* them β€” so the list -// settles as [a1, b1, a2] instead of append order [a1, a2, b1]. -// A server-authoritative append must place b1 at the true end. +// a1 then a2; client B appends b1 without yet having seen a1/a2, so b1's +// client-computed position is stale by the time it reaches the server (a1 and +// a2 are already stored there). Because the op is tagged with intent: "push", +// the server ignores that stale position and appends b1 at the true end, so +// both clients settle in append order: [a1, a2, b1]. test( "concurrent pushes settle in append order, never wedged", prepareTestsConflicts( diff --git a/packages/liveblocks-core/e2e/list-set.test.ts b/packages/liveblocks-core/e2e/list-set.test.ts index 54a96bd67d..67ac742bed 100644 --- a/packages/liveblocks-core/e2e/list-set.test.ts +++ b/packages/liveblocks-core/e2e/list-set.test.ts @@ -215,7 +215,7 @@ test( list: new LiveList(["a"]), }, async ({ root1, root2, control, assert }) => { - // Client A replaces "a" with "X" + // Client A replaces "a" with "🟒" root1.get("list").set(0, "🟒"); // Client B simultaneously deletes "a" @@ -245,8 +245,6 @@ test( }, async ({ root1, root2, control, assert }) => { // Client A changes "a" to "🟒" and moves it after "b" - // This is done in a batch to ensure the default throttling won't - // send the second operation in the message queue root1.get("list").set(0, "🟒"); root1.get("list").move(0, 1); assert( diff --git a/packages/liveblocks-core/e2e/storage-notification-reconnect.test.ts b/packages/liveblocks-core/e2e/storage-notification-reconnect.test.ts index de7488d3c8..e5f3eaa9b9 100644 --- a/packages/liveblocks-core/e2e/storage-notification-reconnect.test.ts +++ b/packages/liveblocks-core/e2e/storage-notification-reconnect.test.ts @@ -17,9 +17,10 @@ * * NOTE ON CONTROL KEYS: several LiveObject tests carry an unchanged scalar key * (e.g. `keep`) that the mutation never touches. The reconnect path routes a - * snapshot through `getTreesDiffOperations`, which re-sends the *full* - * UPDATE_OBJECT data β€” so an unchanged key can be spuriously re-notified. The - * control key is what makes that bug observable; do not remove it. + * snapshot through `getTreesDiffOperations`, whose UPDATE_OBJECT ops must + * carry only the keys that actually changed β€” a full-data re-send would + * spuriously re-notify the unchanged key. The control key is what makes that + * observable; do not remove it. */ import { expect, onTestFinished, test } from "vitest"; import WebSocket from "ws"; @@ -483,11 +484,10 @@ test("LiveObject: nested-object deletes fire equivalent notifications online and ).toEqual({ x: 1 }); }); -// Baseline (passes today): when the transitioned key is the object's *only* -// scalar, moving it into a child node empties the object's `data`, so the -// snapshot diff produces an UPDATE_OBJECT with empty data β€” nothing left for -// the full-data re-send to spuriously re-notify. Contrast with the next test, -// which adds a surviving scalar sibling and exposes that exact leak. +// Baseline: when the transitioned key is the object's *only* scalar, moving +// it into a child node empties the object's `data`, so the snapshot diff has +// no other scalar keys to consider. The next test adds a surviving scalar +// sibling, which the diff must not spuriously re-notify. test("LiveObject: scalarβ†’nested-object transition (sole key) fires equivalent notifications online and on reconnect", async () => { const { online, reconnect } = await bothPhases( () => ({ @@ -573,9 +573,8 @@ test("LiveObject: nested-objectβ†’scalar transition fires equivalent notificatio // - the online path saw the intermediate churn while the reconnect path saw // only the collapsed net result. // -// Unlike the bug-spec tests above, these are expected to PASS on the current -// path β€” they lock in the collapse semantics so the reconcile refactor can't -// regress them. +// These lock in the collapse semantics so the reconcile refactor can't regress +// them. // ───────────────────────────────────────────────────────────────────────────── const insertedItems = (deltas: ListUpdate["updates"]): unknown[] => diff --git a/packages/liveblocks-core/e2e/utils.ts b/packages/liveblocks-core/e2e/utils.ts index c7b71bcccb..05d8d195db 100644 --- a/packages/liveblocks-core/e2e/utils.ts +++ b/packages/liveblocks-core/e2e/utils.ts @@ -24,12 +24,23 @@ async function initializeRoomForTest< TM extends BaseMetadata = BaseMetadata, CM extends BaseMetadata = BaseMetadata, >(roomId: string, initialPresence: NoInfr

, initialStorage: NoInfr) { - let ws: PausableWebSocket | null = null; - - class PausableWebSocket extends WebSocket { - sendBuffer: string[] = []; - _isSendPaused = false; - _dropIncoming = false; + let ws: ControlledWebSocket | null = null; + + /** + * A WebSocket whose two directions can each be stalled, like a real + * network pipe. A stalled direction buffers frames FIFO; un-stalling + * delivers them in their original order, so the stream order can never be + * changed by these controls, only delayed. + * + * All state is per-socket. A reconnect creates a fresh, unstalled socket, + * and whatever was still buffered on the old socket is simply never + * delivered: those frames were in flight when the connection died. + */ + class ControlledWebSocket extends WebSocket { + // When non-null, the direction is stalled and the array buffers its + // frames, in order. When null, frames flow through. + sendBuffer: string[] | null = null; + recvBuffer: unknown[][] | null = null; constructor(address: string | URL) { super(address); @@ -38,50 +49,65 @@ async function initializeRoomForTest< } /** - * Stops sending messages through to the server. Effectively starts - * buffering messages in-memory until .resume() is called. + * Stalls the uplink: outgoing messages are buffered in-memory instead of + * sent, until .resume() is called. Models a slow upload or backpressure. */ pause() { - this._isSendPaused = true; + this.sendBuffer ??= []; + } + + /** + * Stalls the downlink: messages the server sends are buffered instead of + * delivered, until .resumeIncoming() is called. Models slow delivery. + * + * Stalling the downlink and then reconnecting simulates "the server + * received and processed our op, but its ack/echo never reached us": the + * buffered ack dies with the socket, and the op stays pending + * (unacknowledged) on this client across the reconnect. + */ + pauseIncoming() { + this.recvBuffer ??= []; } /** - * Immediately sends all buffered messages to the server and stops - * buffering any new messages. + * Immediately sends all buffered messages to the server, in order, and + * stops buffering any new messages. */ resume() { - this._isSendPaused = false; - for (const item of this.sendBuffer) { + const sendBuffer = this.sendBuffer; + this.sendBuffer = null; + for (const item of sendBuffer ?? []) { super.send(item); } - this.sendBuffer = []; + } + + /** + * Immediately delivers all buffered incoming messages, in order, and + * stops buffering any new ones. + */ + resumeIncoming() { + const recvBuffer = this.recvBuffer; + this.recvBuffer = null; + for (const args of recvBuffer ?? []) { + super.emit("message", ...args); + } } send(data: string) { - if (this._isSendPaused) { + if (this.sendBuffer !== null) { this.sendBuffer.push(data); } else { super.send(data); } } - /** - * Silently drops every message the server sends from now on, as if the - * network ate them. Used to keep an op "pending" on this client even - * though the server already received and processed it: the server's - * ack/echo never reaches the client, so it never clears from - * unacknowledgedOps. - */ - dropIncoming() { - this._dropIncoming = true; - } - // `ws` delivers incoming frames by emitting a "message" event (both - // addEventListener and .on() listeners run through this). Swallow those - // emissions while dropping, leaving every other event untouched. - // eslint-disable-next-line @typescript-eslint/no-explicit-any -- matches EventEmitter.emit's own signature + // addEventListener and .on() listeners run through this). Buffer those + // emissions while the downlink is stalled, leaving every other event + // untouched. emit(eventName: string | symbol, ...args: any[]): boolean { - if (this._dropIncoming && eventName === "message") { + if (this.recvBuffer !== null && eventName === "message") { + this.recvBuffer.push(args); return false; } return super.emit(eventName, ...args); @@ -92,7 +118,7 @@ async function initializeRoomForTest< __DANGEROUSLY_disableThrottling: true, publicApiKey: "pk_localdev", polyfills: { - WebSocket: PausableWebSocket, + WebSocket: ControlledWebSocket, }, baseUrl: BASE_URL, }); @@ -140,39 +166,55 @@ export function prepareTestsConflicts( /** Test utilities to exactly control message passing */ control: { /** - * Sends all buffered messages from Client A to Client B, and waits - * until Client B has processed them. + * Flushes all messages from Client A and waits until Client B has + * received them. + * Note that this will hang if client B's downlink has been stalled, as + * it's unable to receive any messages until resumed. */ + // TODO: Rename to flushAtoB() later flushA: () => Promise; /** - * Sends all buffered messages from Client B to Client A, and waits - * until Client A has processed them. + * Flushes all messages from Client B and waits until Client A has + * received them. + * Note that this will hang if client A's downlink has been stalled, as + * it's unable to receive any messages until resumed. */ + // TODO: Rename to flushBtoA() later flushB: () => Promise; /** - * Flushes Client A's buffered sends to the server without waiting for a - * beacon round-trip. Use when Client A is dropping incoming messages (so - * a beacon would never return). + * Flushes Client A's buffered sends to the server (unlike flushA, + * without waiting for any confirmation). */ + // TODO: Rename to flushToServerA() later flushSyncA: () => void; /** - * Flushes Client B's buffered sends to the server without waiting for a - * beacon round-trip. Use when Client B is dropping incoming messages (so - * a beacon would never return). + * Flushes Client B's buffered sends to the server (unlike flushB, + * without waiting for any confirmation). */ + // TODO: Rename to flushToServerB() later flushSyncB: () => void; /** - * Makes client A silently drop every message the server sends from now - * on, keeping its in-flight ops "pending" even after the server has - * processed them. + * Stalls client A's uplink: outgoing messages buffer in order instead + * of being sent, until the next flush. Note that this stalls the + * *current* socket; a socket freshly created by a reconnect starts + * unstalled. */ - dropIncomingA: () => void; + pauseA: () => void; + /** Same as pauseA, for client B. */ + pauseB: () => void; /** - * Makes client B silently drop every message the server sends from now - * on, keeping its in-flight ops "pending" even after the server has - * processed them. + * Stalls client A's downlink: messages from the server buffer in order + * instead of being delivered, until resumeIncomingA(). Stalled messages + * die with the socket, so following this up with a reconnect simulates + * "the server processed our op, but its ack never reached us". */ - dropIncomingB: () => void; + pauseIncomingA: () => void; + /** Same as pauseIncomingA, for client B. */ + pauseIncomingB: () => void; + /** Delivers client A's stalled incoming messages, and stops stalling. */ + resumeIncomingA: () => void; + /** Delivers client B's stalled incoming messages, and stops stalling. */ + resumeIncomingB: () => void; }; }) => Promise ): () => Promise { @@ -287,15 +329,18 @@ export function prepareTestsConflicts( actor2.ws.pause(); }, - dropIncomingA: () => { - actor1.ws.dropIncoming(); - }, - - dropIncomingB: () => { - actor2.ws.dropIncoming(); - }, + pauseA: () => actor1.ws.pause(), + pauseB: () => actor2.ws.pause(), + pauseIncomingA: () => actor1.ws.pauseIncoming(), + pauseIncomingB: () => actor2.ws.pauseIncoming(), + resumeIncomingA: () => actor1.ws.resumeIncoming(), + resumeIncomingB: () => actor2.ws.resumeIncoming(), }; + // TODO Maybe make this the default behavior of the ControlledWebSocket + // class, and clearly document this. _Send_ is paused by default, but + // _recv_ is not. I think that'd be a nice default? + // TODO Not super sure though how it related to the one-time sync below. actor1.ws.pause(); actor2.ws.pause(); @@ -355,7 +400,6 @@ export function prepareTestsConflicts( // Surface the full storage pool of both clients (every node, its parent, // its position key, and its value) so convergence failures are debuggable // from the test output alone. - // eslint-disable-next-line no-console console.error( `\n=== Storage pool dump on failure ===\n${actor1.room._dump()}\n\n${actor2.room._dump()}\n` ); @@ -422,7 +466,7 @@ export function prepareSingleClientTest( * asynchronously reached a particular status. Status must be reached within * a limited time window, or else this will fail, to avoid hanging. */ -async function waitUntilStatus( +export async function waitUntilStatus( room: Room, targetStatus: Status ): Promise { diff --git a/packages/liveblocks-core/package.json b/packages/liveblocks-core/package.json index 450f167d81..d8ec5de895 100644 --- a/packages/liveblocks-core/package.json +++ b/packages/liveblocks-core/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/core", - "version": "3.19.4", + "version": "3.19.5", "description": "Private internals for Liveblocks. DO NOT import directly from this package!", "type": "module", "main": "./dist/index.cjs", diff --git a/packages/liveblocks-core/src/crdts/AbstractCrdt.ts b/packages/liveblocks-core/src/crdts/AbstractCrdt.ts index 06fb988489..b372123188 100644 --- a/packages/liveblocks-core/src/crdts/AbstractCrdt.ts +++ b/packages/liveblocks-core/src/crdts/AbstractCrdt.ts @@ -373,19 +373,8 @@ export abstract class AbstractCrdt { this.#pool = pool; } - /** - * @internal - * `fromSnapshot` is set when the op is part of a full-state snapshot - * reconstruction (the reconnect reconcile) rather than a live incremental op. - * Only LiveList uses it, to suppress its optimistic push tail-bump: the bump - * predicts where the server will place pending pushes, but a snapshot already - * holds the final positions, so there's nothing to predict. - */ - abstract _attachChild( - op: CreateOp, - source: OpSource, - fromSnapshot?: boolean - ): ApplyResult; + /** @internal */ + abstract _attachChild(op: CreateOp, source: OpSource): ApplyResult; /** @internal */ _detach(): void { diff --git a/packages/liveblocks-core/src/crdts/LiveList.ts b/packages/liveblocks-core/src/crdts/LiveList.ts index 9d1d260e62..064f25be88 100644 --- a/packages/liveblocks-core/src/crdts/LiveList.ts +++ b/packages/liveblocks-core/src/crdts/LiveList.ts @@ -429,7 +429,7 @@ export class LiveList extends AbstractCrdt { return result.modified.updates[0]; } - #applyRemoteInsert(op: CreateOp, fromSnapshot: boolean): ApplyResult { + #applyRemoteInsert(op: CreateOp): ApplyResult { if (this._pool === undefined) { throw new Error("Can't attach child if managed pool is not present"); } @@ -452,12 +452,11 @@ export class LiveList extends AbstractCrdt { // a view that excludes our still-unacked pushes, so they can't address a // position inside our pending tail block. // - // The bump is a purely-local, live-only anti-flicker prediction of where - // the server will place things. While reconstructing from a server snapshot - // (reconnect reconcile) we already have the answer, so we don't predict: - // bumping there would override the snapshot's positions with a guess, and - // the diff carries no corrective op to undo it. - const bumpDeltas = fromSnapshot ? [] : this.#bumpUnackedPushesAbove(key); + // The bump is a purely-local anti-flicker prediction of where the server + // will place things. It's sound because #unackedPushNodes only yields ops + // the server has provably not processed yet (ops whose fate became + // unknown in a disconnect are excluded at that source). + const bumpDeltas = this.#bumpUnackedPushesAbove(key); return { modified: makeUpdate(this, [ @@ -474,6 +473,13 @@ export class LiveList extends AbstractCrdt { * the single source of truth, so an item drops out the instant its op is * acked, with no per-instance membership to leak. Yielded in push order. * + * Excludes ops that may already be stored on the server (they were in + * flight when a connection died, so their fate is unknown): the bump + * prediction assumes the server has not processed the op yet, which is only + * guaranteed for ops sent on the current connection. For these excluded + * ops, the server's (re-)ack states the authoritative position; predicting + * locally could produce a wrong position that no ack would correct. + * * Restricted to items currently in `#items`: a pushed node whose op is still * pending may have been pulled out of the list (e.g. implicitly deleted by a * remote set, or removed by an undo) while still living in the pool, and such @@ -488,6 +494,9 @@ export class LiveList extends AbstractCrdt { if (op.intent !== "push") { continue; } + if (this._pool.unacknowledgedOps.isPossiblyStored(op.opId)) { + continue; + } const node = this._pool.getNode(op.id); if (node !== undefined && this.#items.includes(node)) { yield node; @@ -701,11 +710,7 @@ export class LiveList extends AbstractCrdt { } /** @internal */ - _attachChild( - op: CreateOp, - source: OpSource, - fromSnapshot: boolean = false - ): ApplyResult { + _attachChild(op: CreateOp, source: OpSource): ApplyResult { if (this._pool === undefined) { throw new Error("Can't attach child if managed pool is not present"); } @@ -722,7 +727,7 @@ export class LiveList extends AbstractCrdt { } } else { if (source === OpSource.THEIRS) { - result = this.#applyRemoteInsert(op, fromSnapshot); + result = this.#applyRemoteInsert(op); } else if (source === OpSource.OURS) { result = this.#applyInsertAck(op); } else { diff --git a/packages/liveblocks-core/src/crdts/UnacknowledgedOps.ts b/packages/liveblocks-core/src/crdts/UnacknowledgedOps.ts index 9036ceee71..1346e21847 100644 --- a/packages/liveblocks-core/src/crdts/UnacknowledgedOps.ts +++ b/packages/liveblocks-core/src/crdts/UnacknowledgedOps.ts @@ -25,6 +25,16 @@ export interface ReadonlyUnacknowledgedOps { parentId: string, parentKey: string ): Iterable; + + /** + * Whether the given pending op may already have been processed by the + * server. True for ops that were in flight when a connection died: the + * server may have stored them with the ack getting lost in the disconnect, + * or may never have received them. Until the (re-sent) op's ack arrives, + * the client cannot know which, so optimistic predictions that assume the + * op has not been processed yet are unsound for these ops. + */ + isPossiblyStored(opId: string): boolean; } /** @@ -54,6 +64,9 @@ export class UnacknowledgedOps implements ReadonlyUnacknowledgedOps { new Map(); // parentId -> (opId -> Create op) #createOpsByParent: Map> = new Map(); + // opIds of pending ops that were in flight when a connection died, so the + // server may already have processed them. See isPossiblyStored(). + #possiblyStoredOpIds: Set = new Set(); #posKey(parentId: string, parentKey: string): PositionKey { return `${parentId}\n${parentKey}`; @@ -99,6 +112,7 @@ export class UnacknowledgedOps implements ReadonlyUnacknowledgedOps { } this.#byOpId.delete(opId); + this.#possiblyStoredOpIds.delete(opId); if (isCreateOp(op)) { const posKey = this.#posKey(op.parentId, op.parentKey); @@ -144,4 +158,19 @@ export class UnacknowledgedOps implements ReadonlyUnacknowledgedOps { values(): IterableIterator { return this.#byOpId.values(); } + + isPossiblyStored(opId: string): boolean { + return this.#possiblyStoredOpIds.has(opId); + } + + /** + * Mark every currently pending op as possibly stored on the server. Called + * when the connection dies: all of these ops were in flight, and their + * (possibly lost) acks would have been the only way to know their fate. + */ + markAllAsPossiblyStored(): void { + for (const opId of this.#byOpId.keys()) { + this.#possiblyStoredOpIds.add(opId); + } + } } diff --git a/packages/liveblocks-core/src/room.ts b/packages/liveblocks-core/src/room.ts index 8d96aa934d..c74579b929 100644 --- a/packages/liveblocks-core/src/room.ts +++ b/packages/liveblocks-core/src/room.ts @@ -1715,6 +1715,14 @@ export function createRoom< function onDidDisconnect() { clearTimeout(context.buffer.flushTimerID); + + // Every op still pending at this point was in flight on the connection + // that just died: the server may have processed it (with its ack lost in + // the disconnect), or never received it. Mark them, so that optimistic + // position predictions (like the LiveList tail-bump) stop applying to + // them: such predictions are only sound for ops the server has provably + // not processed yet. + context.unacknowledgedOps.markAllAsPossiblyStored(); } // Register events handlers for events coming from the socket @@ -1894,17 +1902,17 @@ export function createRoom< // XXX_vincent Smell, needs a deeper refactor soon! A reconnect // snapshot is a stream of *nodes* (the full authoritative state), but // here we fabricate a diff of *ops* and replay it through the live - // op-apply path. That path carries live-only optimistic semantics (the - // LiveList push tail-bump, "temporary position until the backend sends - // a fix" shifts, pending-conflict resolution) that are nonsensical - // when the stream we are applying already IS the fix. The - // `fromSnapshot` flag below patches only the one leak that bit us (the - // bump); it does not address the others. The proper fix is - // a node-stream reconcile that updates the tree in place, unified with - // the `_fromItems` path used on initial load, so a node stream never - // enters the op path at all. Until then `fromSnapshot` is a stopgap. + // op-apply path. That path carries live-only optimistic semantics + // ("temporary position until the backend sends a fix" shifts, + // pending-conflict resolution) that are nonsensical when the stream we + // are applying already IS the fix. (The LiveList push tail-bump is + // fine, though: it skips every op whose position the snapshot may + // already own, so replaying the diff cannot mispredict.) The proper + // fix is a node-stream reconcile that updates the tree in place, + // unified with the `_fromItems` path used on initial load, so a node + // stream never enters the op path at all. const ops = getTreesDiffOperations(currentItems, nodes); - const result = applyRemoteOps(ops, /* fromSnapshot */ true); + const result = applyRemoteOps(ops); notify(result.updates); } else { context.root = LiveObject._fromItems( @@ -2019,27 +2027,20 @@ export function createRoom< return { opsToEmit: opsWithOpIds, reverse, updates }; } - function applyRemoteOps( - ops: readonly ServerWireOp[], - // True when `ops` reconstruct state from a server snapshot (the reconnect - // reconcile) rather than being live ops. Disables the live-only LiveList - // push tail-bump. - fromSnapshot: boolean = false - ): { + function applyRemoteOps(ops: readonly ServerWireOp[]): { // Updates to notify about afterwards updates: { storageUpdates: Map; presence: boolean; }; } { - return applyOps([], ops, /* isLocal */ false, fromSnapshot); + return applyOps([], ops, /* isLocal */ false); } function applyOps( pframes: readonly PresenceStackframe

[], ops: readonly Op[], - isLocal: boolean, - fromSnapshot: boolean = false + isLocal: boolean ): { reverse: Stackframe

[]; updates: { @@ -2094,7 +2095,7 @@ export function createRoom< source = OpSource.THEIRS; } - const applyOpResult = applyOp(op, source, fromSnapshot); + const applyOpResult = applyOp(op, source); if (applyOpResult.modified) { const nodeId = applyOpResult.modified.node._id; @@ -2131,11 +2132,7 @@ export function createRoom< }; } - function applyOp( - op: Op, - source: OpSource, - fromSnapshot: boolean = false - ): ApplyResult { + function applyOp(op: Op, source: OpSource): ApplyResult { // Explicit case to handle ignored Ops if (isIgnoredOp(op)) { return { modified: false }; @@ -2181,7 +2178,7 @@ export function createRoom< return { modified: false }; } - return parentNode._attachChild(op, source, fromSnapshot); + return parentNode._attachChild(op, source); } } } diff --git a/packages/liveblocks-emails/package.json b/packages/liveblocks-emails/package.json index 86f5f3e7c4..28b8eb248e 100644 --- a/packages/liveblocks-emails/package.json +++ b/packages/liveblocks-emails/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/emails", - "version": "3.19.4", + "version": "3.19.5", "description": "A set of functions and utilities to make sending emails based on Liveblocks notification events easy. Liveblocks is the all-in-one toolkit to build collaborative products like Figma, Notion, and more.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-node-lexical/package.json b/packages/liveblocks-node-lexical/package.json index 81c263d639..f1e0c0928a 100644 --- a/packages/liveblocks-node-lexical/package.json +++ b/packages/liveblocks-node-lexical/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/node-lexical", - "version": "3.19.4", + "version": "3.19.5", "description": "A server-side utility that lets you modify lexical documents hosted in Liveblocks.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-node-prosemirror/package.json b/packages/liveblocks-node-prosemirror/package.json index bfaed89539..9ab95dd073 100644 --- a/packages/liveblocks-node-prosemirror/package.json +++ b/packages/liveblocks-node-prosemirror/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/node-prosemirror", - "version": "3.19.4", + "version": "3.19.5", "description": "A server-side utility that lets you modify prosemirror and tiptap documents hosted in Liveblocks.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-node/package.json b/packages/liveblocks-node/package.json index 3d9ea01655..0111d07a34 100644 --- a/packages/liveblocks-node/package.json +++ b/packages/liveblocks-node/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/node", - "version": "3.19.4", + "version": "3.19.5", "description": "A server-side utility that lets you set up a Liveblocks authentication endpoint. Liveblocks is the all-in-one toolkit to build collaborative products like Figma, Notion, and more.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-react-blocknote/package.json b/packages/liveblocks-react-blocknote/package.json index 44c24b5655..16c0b943ec 100644 --- a/packages/liveblocks-react-blocknote/package.json +++ b/packages/liveblocks-react-blocknote/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/react-blocknote", - "version": "3.19.4", + "version": "3.19.5", "description": "An integration of BlockNote + React to enable collaboration, comments, live cursors, and more with Liveblocks.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-react-flow/package.json b/packages/liveblocks-react-flow/package.json index ac61950232..c49f655432 100644 --- a/packages/liveblocks-react-flow/package.json +++ b/packages/liveblocks-react-flow/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/react-flow", - "version": "3.19.4", + "version": "3.19.5", "description": "An integration of React Flow to enable collaboration and realtime cursors with Liveblocks.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-react-lexical/package.json b/packages/liveblocks-react-lexical/package.json index 71f0545fdd..657e592094 100644 --- a/packages/liveblocks-react-lexical/package.json +++ b/packages/liveblocks-react-lexical/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/react-lexical", - "version": "3.19.4", + "version": "3.19.5", "description": "An integration of Lexical + React to enable collaboration, comments, live cursors, and more with Liveblocks.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-react-tiptap/package.json b/packages/liveblocks-react-tiptap/package.json index 109dff9247..640ec8402b 100644 --- a/packages/liveblocks-react-tiptap/package.json +++ b/packages/liveblocks-react-tiptap/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/react-tiptap", - "version": "3.19.4", + "version": "3.19.5", "description": "An integration of TipTap + React to enable collaboration, comments, live cursors, and more with Liveblocks.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-react-ui/package.json b/packages/liveblocks-react-ui/package.json index a476bc3cb9..a9734254b5 100644 --- a/packages/liveblocks-react-ui/package.json +++ b/packages/liveblocks-react-ui/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/react-ui", - "version": "3.19.4", + "version": "3.19.5", "description": "A set of React pre-built components for the Liveblocks products. Liveblocks is the all-in-one toolkit to build collaborative products like Figma, Notion, and more.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-react/package.json b/packages/liveblocks-react/package.json index 8446fc1870..346b85c51b 100644 --- a/packages/liveblocks-react/package.json +++ b/packages/liveblocks-react/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/react", - "version": "3.19.4", + "version": "3.19.5", "description": "A set of React hooks and providers to use Liveblocks declaratively. Liveblocks is the all-in-one toolkit to build collaborative products like Figma, Notion, and more.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-redux/package.json b/packages/liveblocks-redux/package.json index dac1fc537c..790b81566f 100644 --- a/packages/liveblocks-redux/package.json +++ b/packages/liveblocks-redux/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/redux", - "version": "3.19.4", + "version": "3.19.5", "description": "A store enhancer to integrate Liveblocks into Redux stores. Liveblocks is the all-in-one toolkit to build collaborative products like Figma, Notion, and more.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-server/CHANGELOG.md b/packages/liveblocks-server/CHANGELOG.md index 5417f97cd9..99a9b6d09b 100644 --- a/packages/liveblocks-server/CHANGELOG.md +++ b/packages/liveblocks-server/CHANGELOG.md @@ -1,5 +1,11 @@ ## vNEXT (not yet released) +## v1.6.1 + +- Fix a `LiveList` divergence after reconnects: when a client re-sends a pending + `push` op whose node the server had already stored (the original ack got lost + in the disconnect) + ## v1.6.0 - Update internal storage format of dev server. Note that your local dev rooms diff --git a/packages/liveblocks-server/package.json b/packages/liveblocks-server/package.json index 7708f0aaf1..45a427b03c 100644 --- a/packages/liveblocks-server/package.json +++ b/packages/liveblocks-server/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/server", - "version": "1.6.0", + "version": "1.6.1", "description": "Liveblocks backend server foundation.", "type": "module", "main": "./dist/index.js", diff --git a/packages/liveblocks-server/src/Room.ts b/packages/liveblocks-server/src/Room.ts index 1322458fe0..b272989d1b 100644 --- a/packages/liveblocks-server/src/Room.ts +++ b/packages/liveblocks-server/src/Room.ts @@ -1617,24 +1617,31 @@ export class Room { r.action === "accepted" ? [r.op] : [] ); - const opsToSendBack: ServerWireOp[] = result.flatMap((r) => { - switch (r.action) { - case "ignored": - // HACK! We send a cleverly composed message, that will act - // as an acknowledgement to all old clients out there in - // the wild. - return r.ignoredOpId !== undefined - ? [ackIgnoredOp(r.ignoredOpId)] - : []; - - case "accepted": - return r.fix !== undefined ? [r.fix] : []; - - // istanbul ignore next - default: - return assertNever(r, "Unhandled case"); + const opsToSendBack: ServerWireOp[] = result.flatMap( + (r): ServerWireOp[] => { + switch (r.action) { + case "ignored": + // HACK! We send a cleverly composed message, that will act + // as an acknowledgement to all old clients out there in + // the wild. + return r.ignoredOpId !== undefined + ? [ackIgnoredOp(r.ignoredOpId)] + : []; + + case "rectified": + // The op was already applied earlier; re-acknowledge it with + // its stored, authoritative position. + return [r.ackOp, r.fix]; + + case "accepted": + return r.fix !== undefined ? [r.fix] : []; + + // istanbul ignore next + default: + return assertNever(r, "Unhandled case"); + } } - }); + ); if (opsToForward.length > 0) { scheduleFanOut({ diff --git a/packages/liveblocks-server/src/Storage.ts b/packages/liveblocks-server/src/Storage.ts index 2d57658594..1cbac87c59 100644 --- a/packages/liveblocks-server/src/Storage.ts +++ b/packages/liveblocks-server/src/Storage.ts @@ -37,7 +37,18 @@ import type { } from "~/protocol"; import type { Pos } from "~/types"; -type ApplyOpResult = OpAccepted | OpIgnored; +/** + * The three possible outcomes of applying a client op. They differ along + * when the op (first) changed storage state, who hears about it, and what + * gets sent back to the originating client: + * + * | | state change | fan out to others | reply to sender | + * |-------------|--------------|-------------------|------------------| + * | OpAccepted | now | yes | ack echo (+ fix) | + * | OpRectified | in the past | no | ack echo + fix | + * | OpIgnored | never | no | bare (H)Ack | + */ +type ApplyOpResult = OpAccepted | OpIgnored | OpRectified; export type OpAccepted = { action: "accepted"; @@ -50,6 +61,25 @@ export type OpIgnored = { ignoredOpId?: string; }; +export type OpRectified = { + action: "rectified"; + /** + * Echo of the client's op, with the stored, authoritative parentKey. Sent + * back to the originating client as the acknowledgement, instead of the + * bare (H)Ack. Used for re-sent CREATE ops whose node the server already + * stored: the echo carries the authoritative position, so the client can + * correct any optimistic local position it may have predicted while the op + * was pending. Never fanned out to others: they already received the op + * when it was originally accepted. + */ + ackOp: CreateOp & HasOpId; + /** + * A corrective op to send back to the originating client, stating that + * same authoritative position (see ackOp). + */ + fix: FixOp; +}; + function accept(op: ClientWireOp, fix?: FixOp): OpAccepted { return { action: "accepted", op, fix }; } @@ -58,6 +88,14 @@ function ignore(ignoredOp: ClientWireOp): OpIgnored { return { action: "ignored", ignoredOpId: ignoredOp.opId }; } +function rectify(op: CreateOp & HasOpId, parentKey: string): OpRectified { + return { + action: "rectified", + ackOp: { ...op, parentKey }, + fix: { type: OpCode.SET_PARENT_KEY, id: op.id, parentKey }, + }; +} + function nodeFromCreateChildOp(op: CreateOp): SerializedChild { switch (op.type) { case OpCode.CREATE_LIST: @@ -164,7 +202,25 @@ export class Storage { private applyCreateOp(op: CreateOp & HasOpId): ApplyOpResult { if (this.driver.has_node(op.id)) { - // Node already exists, the operation is ignored + // Node already exists, meaning this op was already applied earlier + // (e.g. it was re-sent after a reconnect because its original ack + // never arrived), so it won't get applied again. For pushed list + // items, rectify: send the stored, authoritative position back to the + // originating client, because a bare ack would leave any optimistic + // local position prediction on that client uncorrected. Only pushes + // need this: they're the only ops the client locally repositions while + // pending. Note that unlike acceptAndFix, rectifying happens even when + // the stored key equals the op's key: the client's *local* key may + // have drifted from both, and the server cannot see that. + if (op.intent === "push") { + const stored = this.driver.get_node(op.id); + if ( + stored?.parentId !== undefined && + this.driver.get_node(stored.parentId)?.type === CrdtType.LIST + ) { + return rectify(op, stored.parentKey); + } + } return ignore(op); } diff --git a/packages/liveblocks-yjs/package.json b/packages/liveblocks-yjs/package.json index 85d01c0ff5..44ee5b37dd 100644 --- a/packages/liveblocks-yjs/package.json +++ b/packages/liveblocks-yjs/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/yjs", - "version": "3.19.4", + "version": "3.19.5", "description": "Integrate your existing or new Yjs documents with Liveblocks.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/packages/liveblocks-zustand/package.json b/packages/liveblocks-zustand/package.json index 839d868bbc..a3cd0244c7 100644 --- a/packages/liveblocks-zustand/package.json +++ b/packages/liveblocks-zustand/package.json @@ -1,6 +1,6 @@ { "name": "@liveblocks/zustand", - "version": "3.19.4", + "version": "3.19.5", "description": "A middleware for Zustand to automatically synchronize your stores with Liveblocks. Liveblocks is the all-in-one toolkit to build collaborative products like Figma, Notion, and more.", "license": "Apache-2.0", "author": "Liveblocks Inc.", diff --git a/tools/liveblocks-cli/CHANGELOG.md b/tools/liveblocks-cli/CHANGELOG.md deleted file mode 100644 index 5417f97cd9..0000000000 --- a/tools/liveblocks-cli/CHANGELOG.md +++ /dev/null @@ -1,69 +0,0 @@ -## vNEXT (not yet released) - -## v1.6.0 - -- Update internal storage format of dev server. Note that your local dev rooms - are not automatically migrated and will appear as empty rooms after the - upgrade. - -## v1.5.0 - -- Add `--random-port` (`-P`) flag to `liveblocks dev`: bind a random free port - instead of an explicit port number. With `--cmd` (`-c`), the chosen port is - exposed to the command via `LIVEBLOCKS_DEV_SERVER_PORT`. Ideal for CI (no port - collisions ever). -- Fix `LiveList.push()` so concurrent pushes from multiple clients no longer - settle out of order. - -## v1.4.1 - -- Fix: `client.getOrCreateRoom()` no longer errors when the room already exists, - matching production behavior. -- Fix: Yjs document updates made via `PUT /v2/rooms//ydoc` now get - broadcast to connected WebSocket clients, matching production behavior. - -## v1.4.0 - -- Add support for `client.mutateStorage()` (from `@liveblocks/node`) - -## v1.3.0 - -- Add feeds support (`feeds:write` permission) -- Add verbose logging toggle -- Fix permission validation to accept all valid permission combinations -- Support passing extra arguments to `--cmd` (`-c`), appended to the command or - replacing `{}` if present - -## v1.2.0 - -- Add live socket inspector view -- Add maintenance mode toggle (to reject new WebSocket connections) - -## v1.1.0 - -### Added - -- ID token authentication support -- Read-only rooms support -- Room permissions and room metadata -- Room filtering support - -### Changed - -- Room Node.js methods and REST APIs are now fully supported - -See https://liveblocks.io/docs/tools/dev-server for the updated feature matrix. - -## v1.0.17 - -Initial release. Dev server supports: - -- Storage (all CRDTs) -- Presence -- Broadcast -- Text editors (Tiptap, BlockNote, Lexical) -- Public key authentication -- Access token authentication -- Room Node.js methods and REST APIs (partial) - -See https://liveblocks.io/docs/tools/dev-server for all details. diff --git a/tools/liveblocks-cli/CHANGELOG.md b/tools/liveblocks-cli/CHANGELOG.md new file mode 120000 index 0000000000..5cd6da9ba6 --- /dev/null +++ b/tools/liveblocks-cli/CHANGELOG.md @@ -0,0 +1 @@ +../../shared/liveblocks-server/CHANGELOG.md \ No newline at end of file diff --git a/tools/liveblocks-cli/package.json b/tools/liveblocks-cli/package.json index 386271ec89..2eeb79f6ad 100644 --- a/tools/liveblocks-cli/package.json +++ b/tools/liveblocks-cli/package.json @@ -1,6 +1,6 @@ { "name": "liveblocks", - "version": "1.6.0", + "version": "1.6.1", "description": "Liveblocks command line interface", "type": "module", "bin": {