diff --git a/README.md b/README.md index 4df4195..e96b87b 100644 --- a/README.md +++ b/README.md @@ -157,7 +157,7 @@ The Rust workspace contains a minimal async WebSocket server (`loro-websocket-se ## Protocol Highlights -- Magic bytes per CRDT: "%LOR" (Loro doc), "%EPH" (Loro ephemeral), "%YJS", "%YAW", … +- Magic bytes per CRDT: "%LOR" (Loro doc), "%EPH" (Loro ephemeral), "%EPS" (persisted Loro ephemeral – tells the server to keep the latest state so new peers can load it immediately), "%YJS", "%YAW", … - Messages: JoinRequest/JoinResponseOk/JoinError, DocUpdate, DocUpdateFragmentHeader/Fragment, UpdateError, Leave - Limits: 256 KiB per message; large updates must be fragmented; default reassembly timeout 10s - Multi‑room: room ID is part of every message; one connection can join multiple rooms diff --git a/packages/loro-adaptors/README.md b/packages/loro-adaptors/README.md index 82f446b..f465c0f 100644 --- a/packages/loro-adaptors/README.md +++ b/packages/loro-adaptors/README.md @@ -14,6 +14,7 @@ The websocket client (`loro-websocket`) speaks the binary wire protocol. These a - `LoroAdaptor`: wraps a `LoroDoc` and streams local updates to the connection; applies remote updates on receipt - `LoroEphemeralAdaptor`: wraps an `EphemeralStore` for transient presence/cursor data +- `LoroPersistentStoreAdaptor`: wraps an `EphemeralStore` but marks updates as persisted so the server stores them for new peers - `EloLoroAdaptor`: wraps a `LoroDoc` and packages updates into %ELO containers with AES‑GCM; decrypts inbound containers and imports plaintext. ## Usage @@ -23,6 +24,7 @@ import { LoroWebsocketClient } from "loro-websocket"; import { LoroAdaptor, LoroEphemeralAdaptor, + LoroPersistentStoreAdaptor, EloLoroAdaptor, } from "loro-adaptors"; import { LoroDoc, EphemeralStore } from "loro-crdt"; @@ -41,6 +43,14 @@ const eph = new EphemeralStore(30_000); const ephAdaptor = new LoroEphemeralAdaptor(eph); const roomEph = await client.join({ roomId: "demo", crdtAdaptor: ephAdaptor }); +// Persisted presence that should be available to late joiners +const persistedStore = new EphemeralStore(30_000); +const persistedAdaptor = new LoroPersistentStoreAdaptor(persistedStore); +const roomPersisted = await client.join({ + roomId: "demo-persisted", + crdtAdaptor: persistedAdaptor, +}); + // %ELO (end‑to‑end encrypted Loro) const key = new Uint8Array(32); const elo = new EloLoroAdaptor({ @@ -55,6 +65,7 @@ doc.commit(); // Cleanup await roomEph.destroy(); await roomDoc.destroy(); +await roomPersisted.destroy(); await secure.destroy(); ``` @@ -62,6 +73,7 @@ await secure.destroy(); - `new LoroAdaptor(doc?: LoroDoc, config?: { onImportError?, onUpdateError? })` - `new LoroEphemeralAdaptor(store?: EphemeralStore)` +- `new LoroPersistentStoreAdaptor(store?: EphemeralStore)` - `new EloLoroAdaptor(docOrConfig: LoroDoc | { getPrivateKey, ivFactory?, onDecryptError?, onUpdateError? })` - `getPrivateKey: (keyId?) => Promise<{ keyId: string, key: CryptoKey | Uint8Array }>` - Optional `ivFactory()` for testing (12‑byte IV) diff --git a/packages/loro-adaptors/src/adaptors.ts b/packages/loro-adaptors/src/adaptors.ts index d9674e2..ea68ccd 100644 --- a/packages/loro-adaptors/src/adaptors.ts +++ b/packages/loro-adaptors/src/adaptors.ts @@ -1,4 +1,5 @@ export * from "./loro-adaptor"; export * from "./loro-ephemeral-adaptor"; +export * from "./loro-persistent-store-adaptor"; export * from "./elo-adaptor"; export * from "./flock-adaptor"; diff --git a/packages/loro-adaptors/src/loro-persistent-store-adaptor.ts b/packages/loro-adaptors/src/loro-persistent-store-adaptor.ts new file mode 100644 index 0000000..2c0d92b --- /dev/null +++ b/packages/loro-adaptors/src/loro-persistent-store-adaptor.ts @@ -0,0 +1,86 @@ +import { EphemeralStore } from "loro-crdt"; +import { CrdtType, JoinResponseOk } from "loro-protocol"; +import type { CrdtAdaptorContext, CrdtDocAdaptor } from "./types"; + +export class LoroPersistentStoreAdaptor implements CrdtDocAdaptor { + readonly crdtType = CrdtType.LoroEphemeralStorePersisted; + + private store: EphemeralStore; + private ctx?: CrdtAdaptorContext; + private localUpdateUnsubscribe?: () => void; + private destroyed = false; + + constructor(store?: EphemeralStore) { + this.store = store || new EphemeralStore(); + } + + waitForReachingServerVersion(): Promise { + return Promise.resolve(); + } + + cmpVersion(_v: Uint8Array) { + return 0 as const; + } + + handleJoinErr?: undefined; + + getStore(): EphemeralStore { + return this.store; + } + + setCtx(ctx: CrdtAdaptorContext): void { + this.ctx = ctx; + this.localUpdateUnsubscribe = this.store.subscribeLocalUpdates(updates => { + if (!this.destroyed && this.ctx) { + this.ctx.send([updates]); + } + }); + } + + getVersion(): Uint8Array { + return new Uint8Array(); + } + + getAlternativeVersion(_currentVersion: Uint8Array): Uint8Array | undefined { + return undefined; + } + + async handleJoinOk(_res: JoinResponseOk): Promise { + if (this.destroyed) return; + if (this.ctx) { + const allState = this.store.encodeAll(); + if (allState.length > 0) { + this.ctx.send([allState]); + } + } + } + + applyUpdate(updates: Uint8Array[]): void { + if (this.destroyed || !updates?.length) return; + for (const update of updates) { + if (update?.length > 0) { + try { + this.store.apply(update); + } catch (error) { + this.ctx?.onImportError( + error instanceof Error ? error : new Error(String(error)), + [update] + ); + } + } + } + } + + destroy(): void { + if (this.destroyed) return; + this.destroyed = true; + + if (this.localUpdateUnsubscribe) { + this.localUpdateUnsubscribe(); + this.localUpdateUnsubscribe = undefined; + } + + this.store.destroy(); + this.ctx = undefined; + } +} diff --git a/packages/loro-adaptors/src/server/index.ts b/packages/loro-adaptors/src/server/index.ts index a73e842..f8abc92 100644 --- a/packages/loro-adaptors/src/server/index.ts +++ b/packages/loro-adaptors/src/server/index.ts @@ -1,5 +1,6 @@ export * from "./server-registry"; export * from "./server-loro-adaptor"; export * from "./server-loro-ephemeral-adaptor"; +export * from "./server-loro-persistent-store-adaptor"; export * from "./server-yjs-awareness-adaptor"; export * from "./server-flock-adaptor"; diff --git a/packages/loro-adaptors/src/server/server-loro-persistent-store-adaptor.ts b/packages/loro-adaptors/src/server/server-loro-persistent-store-adaptor.ts new file mode 100644 index 0000000..a92bd69 --- /dev/null +++ b/packages/loro-adaptors/src/server/server-loro-persistent-store-adaptor.ts @@ -0,0 +1,136 @@ +import { EphemeralStore } from "loro-crdt"; +import { + CrdtType, + Permission, + MessageType, + JoinResponseOk, + UpdateError, + UpdateErrorCode, +} from "loro-protocol"; +import type { CrdtServerAdaptor } from "../types"; + +export interface LoroPersistentStoreServerAdaptorConfig { + timeout?: number; +} + +export class LoroPersistentStoreServerAdaptor implements CrdtServerAdaptor { + readonly crdtType = CrdtType.LoroEphemeralStorePersisted; + private readonly timeout: number; + + constructor(config: LoroPersistentStoreServerAdaptorConfig = {}) { + this.timeout = config.timeout ?? 10_000; + } + + createEmpty(): Uint8Array { + const store = new EphemeralStore(this.timeout); + try { + return store.encodeAll(); + } finally { + store.inner.free(); + } + } + + handleJoinRequest( + documentData: Uint8Array, + _clientVersion: Uint8Array, + permission: Permission + ): { + response: JoinResponseOk; + updates?: Uint8Array[]; + } { + const response: JoinResponseOk = { + type: MessageType.JoinResponseOk, + crdt: this.crdtType, + roomId: "", + permission, + version: new Uint8Array(), + }; + + const updates = documentData.length > 0 ? [documentData] : undefined; + return { response, updates }; + } + + applyUpdates( + documentData: Uint8Array, + updates: Uint8Array[], + permission: Permission + ): { + success: boolean; + newDocumentData?: Uint8Array; + error?: UpdateError; + broadcastUpdates?: Uint8Array[]; + } { + if (permission === "read") { + return { + success: false, + error: { + type: MessageType.UpdateError, + crdt: this.crdtType, + roomId: "", + code: UpdateErrorCode.PermissionDenied, + message: "Read-only permission, cannot apply updates", + }, + }; + } + + const store = new EphemeralStore(this.timeout); + const broadcastUpdates: Uint8Array[] = []; + + try { + if (documentData.length > 0) { + store.apply(documentData); + } + for (const update of updates) { + if (update.length > 0) { + store.apply(update); + broadcastUpdates.push(update); + } + } + + const newDocumentData = store.encodeAll(); + + return { + success: true, + newDocumentData, + broadcastUpdates: + broadcastUpdates.length > 0 ? broadcastUpdates : undefined, + }; + } catch (error) { + return { + success: false, + error: { + type: MessageType.UpdateError, + crdt: this.crdtType, + roomId: "", + code: UpdateErrorCode.InvalidUpdate, + message: error instanceof Error ? error.message : "Invalid update", + }, + }; + } finally { + store.destroy(); + store.inner.free(); + } + } + + getVersion(_documentData: Uint8Array): Uint8Array { + return new Uint8Array(); + } + + merge(documents: Uint8Array[]): Uint8Array { + const store = new EphemeralStore(this.timeout); + for (const data of documents) { + if (data.length > 0) { + store.apply(data); + } + } + try { + return store.encodeAll(); + } finally { + store.destroy(); + store.inner.free(); + } + } +} + +export const loroPersistentStoreServerAdaptor = + new LoroPersistentStoreServerAdaptor(); diff --git a/packages/loro-adaptors/tests/adaptors.test.ts b/packages/loro-adaptors/tests/adaptors.test.ts index f468655..b9d47a5 100644 --- a/packages/loro-adaptors/tests/adaptors.test.ts +++ b/packages/loro-adaptors/tests/adaptors.test.ts @@ -1,6 +1,10 @@ import { describe, it, expect, beforeEach, vi } from "vitest"; import { LoroDoc, EphemeralStore, VersionVector } from "loro-crdt"; -import { LoroAdaptor, LoroEphemeralAdaptor } from "../src/adaptors"; +import { + LoroAdaptor, + LoroEphemeralAdaptor, + LoroPersistentStoreAdaptor, +} from "../src/adaptors"; import { CrdtType, JoinResponseOk, MessageType } from "loro-protocol"; describe("LoroAdaptor", () => { @@ -125,6 +129,46 @@ describe("LoroEphemeralAdaptor", () => { }); }); +describe("LoroPersistentStoreAdaptor", () => { + let adaptor: LoroPersistentStoreAdaptor; + let store: EphemeralStore; + + beforeEach(() => { + store = new EphemeralStore(); + adaptor = new LoroPersistentStoreAdaptor(store); + }); + + it("should have correct CRDT type", () => { + expect(adaptor.crdtType).toBe(CrdtType.LoroEphemeralStorePersisted); + }); + + it("should return the underlying EphemeralStore", () => { + expect(adaptor.getStore()).toBe(store); + }); + + it("should send full state on join", async () => { + const mockSend = vi.fn(); + adaptor.setCtx({ + send: mockSend, + onJoinFailed: vi.fn(), + onImportError: vi.fn(), + }); + + store.set("cursor", { x: 1 }); + const joinOk: JoinResponseOk = { + type: MessageType.JoinResponseOk, + crdt: CrdtType.LoroEphemeralStorePersisted, + roomId: "room", + permission: "write", + version: new Uint8Array(), + }; + + await adaptor.handleJoinOk(joinOk); + + expect(mockSend).toHaveBeenCalledWith([expect.any(Uint8Array)]); + }); +}); + describe("Utility functions", () => { it("should create LoroAdaptor with new LoroDoc", () => { const adaptor = new LoroAdaptor(); diff --git a/packages/loro-protocol/src/encoding.ts b/packages/loro-protocol/src/encoding.ts index 09bdb90..07c9346 100644 --- a/packages/loro-protocol/src/encoding.ts +++ b/packages/loro-protocol/src/encoding.ts @@ -22,6 +22,7 @@ import { const validCrdtTypes = [ CrdtType.Loro, CrdtType.LoroEphemeralStore, + CrdtType.LoroEphemeralStorePersisted, CrdtType.Yjs, CrdtType.YjsAwareness, CrdtType.Elo, @@ -35,6 +36,7 @@ function isCrdtType(x: unknown): x is CrdtType { typeof x === "string" && (x === CrdtType.Loro || x === CrdtType.LoroEphemeralStore || + x === CrdtType.LoroEphemeralStorePersisted || x === CrdtType.Yjs || x === CrdtType.YjsAwareness || x === CrdtType.Flock || diff --git a/packages/loro-protocol/src/protocol.ts b/packages/loro-protocol/src/protocol.ts index 2c4b131..1bb79c2 100644 --- a/packages/loro-protocol/src/protocol.ts +++ b/packages/loro-protocol/src/protocol.ts @@ -5,6 +5,7 @@ export const MAX_MESSAGE_SIZE = 256 * 1024; export const CrdtType = { Loro: "%LOR", LoroEphemeralStore: "%EPH", + LoroEphemeralStorePersisted: "%EPS", Yjs: "%YJS", YjsAwareness: "%YAW", Elo: "%ELO", @@ -13,6 +14,7 @@ export const CrdtType = { export const CrdtTypeId = { Loro: "loro", LoroEphemeralStore: "eph", + LoroEphemeralStorePersisted: "eps", Yjs: "yjs", YjsAwareness: "yaw", Elo: "elo", @@ -23,6 +25,7 @@ export type CrdtId = (typeof CrdtTypeId)[keyof typeof CrdtTypeId]; export const MagicBytesToCrdtId = { "%LOR": "loro", "%EPH": "eph", + "%EPS": "eps", "%YJS": "yjs", "%YAW": "yaw", "%ELO": "elo", diff --git a/packages/loro-websocket/src/server/bin.ts b/packages/loro-websocket/src/server/bin.ts index e5b0cf9..718ca07 100644 --- a/packages/loro-websocket/src/server/bin.ts +++ b/packages/loro-websocket/src/server/bin.ts @@ -15,7 +15,14 @@ if (!Number.isSafeInteger(port) || port <= 0 || port > 65_535) { } const server = new SimpleServer({ port }); -server.start().then(() => { - // eslint-disable-next-line no-console - console.log(`SimpleServer listening on ws://localhost:${port}`); -}); +server + .start() + .then(() => { + // eslint-disable-next-line no-console + console.log(`SimpleServer listening on ws://localhost:${port}`); + }) + .catch((error) => { + // eslint-disable-next-line no-console + console.error("Failed to start SimpleServer:", error); + process.exit(1); + }); diff --git a/packages/loro-websocket/src/server/crdt-doc.ts b/packages/loro-websocket/src/server/crdt-doc.ts index ddb06c4..d176e5a 100644 --- a/packages/loro-websocket/src/server/crdt-doc.ts +++ b/packages/loro-websocket/src/server/crdt-doc.ts @@ -125,6 +125,44 @@ class LoroEphemeralCrdtDoc implements CrdtDoc { } } +class LoroPersistentStoreCrdtDoc implements CrdtDoc { + private store: EphemeralStore; + constructor() { + this.store = new EphemeralStore(); + } + getVersion(): Uint8Array { + return new Uint8Array(); + } + computeBackfill(_clientVersion: Uint8Array): Uint8Array[] | null { + const data = this.store.encodeAll(); + return data && data.length ? [data] : null; + } + applyUpdates(updates: Uint8Array[]) { + try { + for (const u of updates) this.store.apply(u); + return { ok: true } as const; + } catch (e) { + return { ok: false, error: e instanceof Error ? e.message : String(e) } as const; + } + } + shouldPersist(): boolean { + return true; + } + exportSnapshot(): Uint8Array | null { + return this.store.encodeAll(); + } + importSnapshot(data: Uint8Array): void { + if (data.length) { + this.store.apply(data); + } + } + allowBackfillWhenNoOtherClients(): boolean { + // Allow a single reconnecting client to hydrate from the persisted snapshot + // even if no other peers are currently connected to the room. + return true; + } +} + class EloCrdtDoc implements CrdtDoc { private elo: EloDoc; constructor() { @@ -159,4 +197,8 @@ class EloCrdtDoc implements CrdtDoc { // Register default CRDT doc implementations registerCrdtDoc(CrdtType.Loro, () => new LoroCrdtDoc()); registerCrdtDoc(CrdtType.LoroEphemeralStore, () => new LoroEphemeralCrdtDoc()); +registerCrdtDoc( + CrdtType.LoroEphemeralStorePersisted, + () => new LoroPersistentStoreCrdtDoc() +); registerCrdtDoc(CrdtType.Elo, () => new EloCrdtDoc()); diff --git a/protocol.md b/protocol.md index 8cb0e36..327c6be 100644 --- a/protocol.md +++ b/protocol.md @@ -21,6 +21,7 @@ Exception: the protocol also defines two out‑of‑band keepalive frames, "ping - The first 4 bytes are magic bytes that indicate the CRDT type: - "%LOR": Loro Document - "%EPH": [Loro Ephemeral Store](https://loro.dev/docs/api/js#ephemeralstore) + - "%EPS": Persisted Loro Ephemeral Store (marks data that should be stored server-side so new peers can hydrate immediately) - "%YJS": Yjs - "%YAW": Yjs Awareness - "%FLO": Flock Document @@ -31,6 +32,10 @@ Exception: the protocol also defines two out‑of‑band keepalive frames, "ping - Payload, which depends on the message type. - Message size should not exceed 256 KB. +Implementations use `%EPS` when the ephemeral payloads must survive beyond a single client session. Tagging frames with this CRDT +type tells the server to persist the latest store so that future peers can immediately download the full state instead of waiting +for another client to resend their presence data. + Note: Keepalive frames are special and bypass this envelope entirely. When the entire frame payload is exactly the text string "ping" or "pong" (WebSocket text frames), it MUST be treated as a keepalive and NOT parsed using the fields above. See Keepalive: Ping/Pong. ## Terminology diff --git a/rust/loro-protocol/src/protocol.rs b/rust/loro-protocol/src/protocol.rs index 8fd1382..41c9fac 100644 --- a/rust/loro-protocol/src/protocol.rs +++ b/rust/loro-protocol/src/protocol.rs @@ -14,6 +14,8 @@ pub enum CrdtType { Loro, /// "%EPH" LoroEphemeralStore, + /// "%EPS" + LoroEphemeralStorePersisted, /// "%YJS" Yjs, /// "%YAW" @@ -27,6 +29,7 @@ impl CrdtType { match self { CrdtType::Loro => *b"%LOR", CrdtType::LoroEphemeralStore => *b"%EPH", + CrdtType::LoroEphemeralStorePersisted => *b"%EPS", CrdtType::Yjs => *b"%YJS", CrdtType::YjsAwareness => *b"%YAW", CrdtType::Elo => *b"%ELO", @@ -37,6 +40,7 @@ impl CrdtType { match &bytes { b"%LOR" => Some(CrdtType::Loro), b"%EPH" => Some(CrdtType::LoroEphemeralStore), + b"%EPS" => Some(CrdtType::LoroEphemeralStorePersisted), b"%YJS" => Some(CrdtType::Yjs), b"%YAW" => Some(CrdtType::YjsAwareness), b"%ELO" => Some(CrdtType::Elo), @@ -145,8 +149,8 @@ impl BatchId { let mut out = [0u8; 8]; for (i, slot) in out.iter_mut().enumerate() { let idx = i * 2; - let byte = u8::from_str_radix(&s[idx..idx + 2], 16) - .map_err(|_| "invalid hex".to_string())?; + let byte = + u8::from_str_radix(&s[idx..idx + 2], 16).map_err(|_| "invalid hex".to_string())?; *slot = byte; } Ok(BatchId(out)) diff --git a/rust/loro-websocket-client/src/lib.rs b/rust/loro-websocket-client/src/lib.rs index af8d281..a9ff929 100644 --- a/rust/loro-websocket-client/src/lib.rs +++ b/rust/loro-websocket-client/src/lib.rs @@ -421,9 +421,10 @@ impl Hash for RoomKey { let tag = match self.crdt { CrdtType::Loro => 0u8, CrdtType::LoroEphemeralStore => 1, - CrdtType::Yjs => 2, - CrdtType::YjsAwareness => 3, - CrdtType::Elo => 4, + CrdtType::LoroEphemeralStorePersisted => 2, + CrdtType::Yjs => 3, + CrdtType::YjsAwareness => 4, + CrdtType::Elo => 5, }; tag.hash(state); self.room.hash(state); diff --git a/rust/loro-websocket-server/examples/simple-server.rs b/rust/loro-websocket-server/examples/simple-server.rs index db9aa25..2d93c0b 100644 --- a/rust/loro-websocket-server/examples/simple-server.rs +++ b/rust/loro-websocket-server/examples/simple-server.rs @@ -8,8 +8,8 @@ //! - Uses a single-thread Tokio runtime to match crate features. //! - Defaults to 127.0.0.1:9000 and ./loro.db if not specified. -use std::{error::Error, path::PathBuf}; use clap::Parser; +use std::{error::Error, path::PathBuf}; use tracing::{debug, error, info, warn}; use tracing_subscriber::EnvFilter; @@ -18,18 +18,37 @@ use loro_websocket_server::{serve_incoming_with_config, ServerConfig}; use tokio::net::TcpListener; #[derive(Parser, Debug)] -#[command(name = "simple-server", about = "Loro WebSocket server with SQLite persistence")] +#[command( + name = "simple-server", + about = "Loro WebSocket server with SQLite persistence" +)] struct Args { #[arg(short = 'a', long, value_name = "ADDR", conflicts_with_all = ["host", "port"], help = "Full socket address to bind, e.g. 0.0.0.0:9000")] addr: Option, - #[arg(short = 'H', long, default_value = "127.0.0.1", help = "Host to bind when --addr not provided")] + #[arg( + short = 'H', + long, + default_value = "127.0.0.1", + help = "Host to bind when --addr not provided" + )] host: String, - #[arg(short = 'p', long, default_value_t = 9000, help = "Port to bind when --addr not provided")] + #[arg( + short = 'p', + long, + default_value_t = 9000, + help = "Port to bind when --addr not provided" + )] port: u16, - #[arg(short = 'd', long = "db", value_name = "PATH", default_value = "loro.db", help = "SQLite database path")] + #[arg( + short = 'd', + long = "db", + value_name = "PATH", + default_value = "loro.db", + help = "SQLite database path" + )] db: PathBuf, } @@ -44,14 +63,18 @@ async fn main() -> Result<(), Box> { .init(); let args = Args::parse(); - let addr = args.addr.unwrap_or_else(|| format!("{}:{}", args.host, args.port)); + let addr = args + .addr + .unwrap_or_else(|| format!("{}:{}", args.host, args.port)); let db_path = args.db; init_db(&db_path).await?; // Wire persistence hooks let db_for_load = db_path.clone(); - type LoadFut = std::pin::Pin>, String>> + Send>>; + type LoadFut = std::pin::Pin< + Box>, String>> + Send>, + >; type SaveFut = std::pin::Pin> + Send>>; let on_load = std::sync::Arc::new(move |workspace: String, room: String, crdt: CrdtType| { @@ -66,16 +89,20 @@ async fn main() -> Result<(), Box> { }); let db_for_save = db_path.clone(); - let on_save = std::sync::Arc::new(move |workspace: String, room: String, crdt: CrdtType, data: Vec| { - let db_path = db_for_save.clone(); - let fut = async move { - tokio::task::spawn_blocking(move || save_snapshot(&db_path, &workspace, &room, crdt, &data)) + let on_save = std::sync::Arc::new( + move |workspace: String, room: String, crdt: CrdtType, data: Vec| { + let db_path = db_for_save.clone(); + let fut = async move { + tokio::task::spawn_blocking(move || { + save_snapshot(&db_path, &workspace, &room, crdt, &data) + }) .await .map_err(|e| format!("task join error: {e}"))? - }; - let fut: SaveFut = Box::pin(fut); - fut - }); + }; + let fut: SaveFut = Box::pin(fut); + fut + }, + ); let cfg = ServerConfig { on_load_document: Some(on_load), @@ -123,13 +150,19 @@ fn crdt_to_str(crdt: CrdtType) -> &'static str { match crdt { CrdtType::Loro => "loro", CrdtType::LoroEphemeralStore => "loro_ephemeral", + CrdtType::LoroEphemeralStorePersisted => "loro_ephemeral_persisted", CrdtType::Yjs => "yjs", CrdtType::YjsAwareness => "yjs_awareness", CrdtType::Elo => "elo", } } -fn load_snapshot(path: &PathBuf, workspace: &str, room: &str, crdt: CrdtType) -> Result>, String> { +fn load_snapshot( + path: &PathBuf, + workspace: &str, + room: &str, + crdt: CrdtType, +) -> Result>, String> { let conn = rusqlite::Connection::open(path).map_err(|e| e.to_string())?; let mut stmt = conn .prepare("SELECT data FROM documents WHERE workspace=?1 AND crdt=?2 AND room=?3") @@ -145,7 +178,13 @@ fn load_snapshot(path: &PathBuf, workspace: &str, room: &str, crdt: CrdtType) -> } } -fn save_snapshot(path: &PathBuf, workspace: &str, room: &str, crdt: CrdtType, data: &[u8]) -> Result<(), String> { +fn save_snapshot( + path: &PathBuf, + workspace: &str, + room: &str, + crdt: CrdtType, + data: &[u8], +) -> Result<(), String> { let conn = rusqlite::Connection::open(path).map_err(|e| e.to_string())?; let now_ms = chrono_like_now_ms(); conn.execute( diff --git a/rust/loro-websocket-server/src/lib.rs b/rust/loro-websocket-server/src/lib.rs index 7365f07..1c066a1 100644 --- a/rust/loro-websocket-server/src/lib.rs +++ b/rust/loro-websocket-server/src/lib.rs @@ -61,9 +61,10 @@ impl Hash for RoomKey { let tag = match self.crdt { CrdtType::Loro => 0u8, CrdtType::LoroEphemeralStore => 1, - CrdtType::Yjs => 2, - CrdtType::YjsAwareness => 3, - CrdtType::Elo => 4, + CrdtType::LoroEphemeralStorePersisted => 2, + CrdtType::Yjs => 3, + CrdtType::YjsAwareness => 4, + CrdtType::Elo => 5, }; tag.hash(state); self.room.hash(state); @@ -104,44 +105,134 @@ pub struct ServerConfig { // CRDT document abstraction to reduce match-based branching trait CrdtDoc: Send { - fn get_version(&self) -> Vec { Vec::new() } - fn compute_backfill(&self, _client_version: &[u8]) -> Vec> { Vec::new() } - fn apply_updates(&mut self, _updates: &[Vec]) -> Result<(), String> { Ok(()) } - fn should_persist(&self) -> bool { false } - fn export_snapshot(&self) -> Option> { None } + fn get_version(&self) -> Vec { + Vec::new() + } + fn compute_backfill(&self, _client_version: &[u8]) -> Vec> { + Vec::new() + } + fn apply_updates(&mut self, _updates: &[Vec]) -> Result<(), String> { + Ok(()) + } + fn should_persist(&self) -> bool { + false + } + fn export_snapshot(&self) -> Option> { + None + } fn import_snapshot(&mut self, _data: &[u8]) {} - fn allow_backfill_when_no_other_clients(&self) -> bool { false } - fn remove_when_last_subscriber_leaves(&self) -> bool { false } + fn allow_backfill_when_no_other_clients(&self) -> bool { + false + } + fn remove_when_last_subscriber_leaves(&self) -> bool { + false + } } -struct LoroRoomDoc { doc: LoroDoc } +struct LoroRoomDoc { + doc: LoroDoc, +} impl LoroRoomDoc { - fn new() -> Self { Self { doc: LoroDoc::new() } } + fn new() -> Self { + Self { + doc: LoroDoc::new(), + } + } } impl CrdtDoc for LoroRoomDoc { fn apply_updates(&mut self, updates: &[Vec]) -> Result<(), String> { - for u in updates { let _ = self.doc.import(u); } + for u in updates { + let _ = self.doc.import(u); + } Ok(()) } - fn should_persist(&self) -> bool { true } - fn export_snapshot(&self) -> Option> { self.doc.export(ExportMode::Snapshot).ok() } - fn import_snapshot(&mut self, data: &[u8]) { let _ = self.doc.import(data); } + fn should_persist(&self) -> bool { + true + } + fn export_snapshot(&self) -> Option> { + self.doc.export(ExportMode::Snapshot).ok() + } + fn import_snapshot(&mut self, data: &[u8]) { + let _ = self.doc.import(data); + } } -struct EphemeralRoomDoc { store: EphemeralStore } +struct EphemeralRoomDoc { + store: EphemeralStore, +} impl EphemeralRoomDoc { - fn new(timeout_ms: i64) -> Self { Self { store: EphemeralStore::new(timeout_ms) } } + fn new(timeout_ms: i64) -> Self { + Self { + store: EphemeralStore::new(timeout_ms), + } + } } impl CrdtDoc for EphemeralRoomDoc { fn compute_backfill(&self, _client_version: &[u8]) -> Vec> { let data = self.store.encode_all(); - if data.is_empty() { Vec::new() } else { vec![data] } + if data.is_empty() { + Vec::new() + } else { + vec![data] + } } fn apply_updates(&mut self, updates: &[Vec]) -> Result<(), String> { - for u in updates { if !u.is_empty() { self.store.apply(u); } } + for u in updates { + if !u.is_empty() { + self.store.apply(u); + } + } Ok(()) } - fn remove_when_last_subscriber_leaves(&self) -> bool { true } + fn remove_when_last_subscriber_leaves(&self) -> bool { + true + } +} + +struct PersistentEphemeralRoomDoc { + store: EphemeralStore, + timeout_ms: i64, +} +impl PersistentEphemeralRoomDoc { + fn new(timeout_ms: i64) -> Self { + Self { + store: EphemeralStore::new(timeout_ms), + timeout_ms, + } + } +} +impl CrdtDoc for PersistentEphemeralRoomDoc { + fn compute_backfill(&self, _client_version: &[u8]) -> Vec> { + let data = self.store.encode_all(); + if data.is_empty() { + Vec::new() + } else { + vec![data] + } + } + fn apply_updates(&mut self, updates: &[Vec]) -> Result<(), String> { + for u in updates { + if !u.is_empty() { + self.store.apply(u); + } + } + Ok(()) + } + fn should_persist(&self) -> bool { + true + } + fn export_snapshot(&self) -> Option> { + Some(self.store.encode_all()) + } + fn import_snapshot(&mut self, data: &[u8]) { + self.store = EphemeralStore::new(self.timeout_ms); + if !data.is_empty() { + self.store.apply(data); + } + } + fn allow_backfill_when_no_other_clients(&self) -> bool { + true + } } // ELO header index entries @@ -156,7 +247,11 @@ struct EloRoomDoc { spans_by_peer: std::collections::HashMap>, } impl EloRoomDoc { - fn new() -> Self { Self { spans_by_peer: std::collections::HashMap::new() } } + fn new() -> Self { + Self { + spans_by_peer: std::collections::HashMap::new(), + } + } fn peer_key_from_bytes(bytes: &[u8]) -> String { // Prefer UTF-8 if valid, else hex @@ -164,7 +259,10 @@ impl EloRoomDoc { Ok(s) => s.to_string(), Err(_) => { let mut out = String::with_capacity(bytes.len() * 2); - for b in bytes { use std::fmt::Write as _; let _ = write!(&mut out, "{:02x}", b); } + for b in bytes { + use std::fmt::Write as _; + let _ = write!(&mut out, "{:02x}", b); + } out } } @@ -174,7 +272,8 @@ impl EloRoomDoc { use loro_protocol::bytes::BytesReader; let mut r = BytesReader::new(buf); let count = usize::try_from(r.read_uleb128().ok()?).ok()?; - let mut map: std::collections::HashMap = std::collections::HashMap::with_capacity(count); + let mut map: std::collections::HashMap = + std::collections::HashMap::with_capacity(count); for _ in 0..count { let peer_bytes = r.read_var_bytes().ok()?; let ctr = r.read_uleb128().ok()?; @@ -187,10 +286,18 @@ impl EloRoomDoc { use loro_protocol::bytes::BytesWriter; let mut entries: Vec<(String, u64)> = Vec::new(); for (peer, spans) in self.spans_by_peer.iter() { - if !peer.as_bytes().iter().all(|b| b.is_ascii_digit()) { continue; } + if !peer.as_bytes().iter().all(|b| b.is_ascii_digit()) { + continue; + } let mut max_end = 0u64; - for s in spans.iter() { if s.end > max_end { max_end = s.end; } } - if max_end > 0 { entries.push((peer.clone(), max_end)); } + for s in spans.iter() { + if s.end > max_end { + max_end = s.end; + } + } + if max_end > 0 { + entries.push((peer.clone(), max_end)); + } } let mut w = BytesWriter::new(); w.push_uleb128(entries.len() as u64); @@ -205,24 +312,38 @@ impl CrdtDoc for EloRoomDoc { fn get_version(&self) -> Vec { // If we have no indexed entries yet, return an empty version to signal // "unknown/empty" baseline so clients may choose to send a snapshot. - if self.spans_by_peer.is_empty() { return Vec::new(); } + if self.spans_by_peer.is_empty() { + return Vec::new(); + } self.encode_current_vv() } fn compute_backfill(&self, client_version: &[u8]) -> Vec> { - let known = self.decode_version_vector(client_version).unwrap_or_default(); + let known = self + .decode_version_vector(client_version) + .unwrap_or_default(); let mut records: Vec> = Vec::new(); for (peer, spans) in self.spans_by_peer.iter() { let k = known.get(peer).copied().unwrap_or(0); - for e in spans { if e.end > k { records.push(e.record.clone()); } } + for e in spans { + if e.end > k { + records.push(e.record.clone()); + } + } + } + if records.is_empty() { + return Vec::new(); } - if records.is_empty() { return Vec::new(); } let mut w = loro_protocol::bytes::BytesWriter::new(); w.push_uleb128(records.len() as u64); - for rec in records.iter() { w.push_var_bytes(rec); } + for rec in records.iter() { + w.push_var_bytes(rec); + } vec![w.finalize()] } fn apply_updates(&mut self, updates: &[Vec]) -> Result<(), String> { - use loro_protocol::elo::{decode_elo_container, parse_elo_record_header, EloRecordKind, EloHeader}; + use loro_protocol::elo::{ + decode_elo_container, parse_elo_record_header, EloHeader, EloRecordKind, + }; for u in updates { let records = decode_elo_container(u.as_slice())?; for rec in records { @@ -230,24 +351,46 @@ impl CrdtDoc for EloRoomDoc { match parsed.kind { EloRecordKind::DeltaSpan => { if let EloHeader::Delta(h) = parsed.header { - if !(h.end > h.start) { return Err("invalid ELO delta span: end must be > start".into()); } - if h.iv.len() != 12 { return Err("invalid ELO delta span: IV must be 12 bytes".into()); } + if !(h.end > h.start) { + return Err("invalid ELO delta span: end must be > start".into()); + } + if h.iv.len() != 12 { + return Err("invalid ELO delta span: IV must be 12 bytes".into()); + } let peer = Self::peer_key_from_bytes(&h.peer_id); let list = self.spans_by_peer.entry(peer).or_default(); // Insert keeping order by start; remove fully covered entries [start, end] - let mut kept: Vec = Vec::with_capacity(list.len() + 1); + let mut kept: Vec = + Vec::with_capacity(list.len() + 1); let mut inserted = false; for e in list.iter() { if !inserted && e.start >= h.start { - kept.push(EloDeltaSpanIndexEntry { start: h.start, end: h.end, key_id: h.key_id.clone(), record: rec.to_vec() }); + kept.push(EloDeltaSpanIndexEntry { + start: h.start, + end: h.end, + key_id: h.key_id.clone(), + record: rec.to_vec(), + }); inserted = true; } // keep entries not fully covered by [start, end] let covered = e.start >= h.start && e.end <= h.end; - if !covered { kept.push(EloDeltaSpanIndexEntry { start: e.start, end: e.end, key_id: e.key_id.clone(), record: e.record.clone() }); } + if !covered { + kept.push(EloDeltaSpanIndexEntry { + start: e.start, + end: e.end, + key_id: e.key_id.clone(), + record: e.record.clone(), + }); + } } if !inserted { - kept.push(EloDeltaSpanIndexEntry { start: h.start, end: h.end, key_id: h.key_id.clone(), record: rec.to_vec() }); + kept.push(EloDeltaSpanIndexEntry { + start: h.start, + end: h.end, + key_id: h.key_id.clone(), + record: rec.to_vec(), + }); } *list = kept; } @@ -260,7 +403,9 @@ impl CrdtDoc for EloRoomDoc { } Ok(()) } - fn allow_backfill_when_no_other_clients(&self) -> bool { true } + fn allow_backfill_when_no_other_clients(&self) -> bool { + true + } } impl Default for ServerConfig { fn default() -> Self { @@ -369,7 +514,9 @@ impl Hub { } async fn ensure_room_loaded(&mut self, room: &RoomKey) { - if self.docs.contains_key(room) { return; } + if self.docs.contains_key(room) { + return; + } match room.crdt { CrdtType::Loro => { let mut d = LoroRoomDoc::new(); @@ -377,20 +524,65 @@ impl Hub { let room_str = String::from_utf8_lossy(&room.room).to_string(); let ws = self.workspace.clone(); match (loader)(ws, room_str, room.crdt).await { - Ok(Some(bytes)) => { d.import_snapshot(&bytes); } + Ok(Some(bytes)) => { + d.import_snapshot(&bytes); + } Ok(None) => {} - Err(e) => { warn!(room=?String::from_utf8_lossy(&room.room), %e, "load document failed"); } + Err(e) => { + warn!(room=?String::from_utf8_lossy(&room.room), %e, "load document failed"); + } } } - self.docs.insert(room.clone(), RoomDocState { doc: Box::new(d), dirty: false }); + self.docs.insert( + room.clone(), + RoomDocState { + doc: Box::new(d), + dirty: false, + }, + ); } CrdtType::LoroEphemeralStore => { let d = EphemeralRoomDoc::new(Self::EPHEMERAL_TIMEOUT_MS); - self.docs.insert(room.clone(), RoomDocState { doc: Box::new(d), dirty: false }); + self.docs.insert( + room.clone(), + RoomDocState { + doc: Box::new(d), + dirty: false, + }, + ); + } + CrdtType::LoroEphemeralStorePersisted => { + let mut d = PersistentEphemeralRoomDoc::new(Self::EPHEMERAL_TIMEOUT_MS); + if let Some(loader) = &self.config.on_load_document { + let room_str = String::from_utf8_lossy(&room.room).to_string(); + let ws = self.workspace.clone(); + match (loader)(ws, room_str, room.crdt).await { + Ok(Some(bytes)) => { + d.import_snapshot(&bytes); + } + Ok(None) => {} + Err(e) => { + warn!(room=?String::from_utf8_lossy(&room.room), %e, "load persisted ephemeral store failed"); + } + } + } + self.docs.insert( + room.clone(), + RoomDocState { + doc: Box::new(d), + dirty: false, + }, + ); } CrdtType::Elo => { let d = EloRoomDoc::new(); - self.docs.insert(room.clone(), RoomDocState { doc: Box::new(d), dirty: false }); + self.docs.insert( + room.clone(), + RoomDocState { + doc: Box::new(d), + dirty: false, + }, + ); } _ => {} } @@ -414,7 +606,14 @@ impl Hub { } fn snapshot_bytes(&self, room: &RoomKey) -> Option> { - self.docs.get(room).and_then(|s| s.doc.export_snapshot()) + let Some(data) = self.docs.get(room).and_then(|s| s.doc.export_snapshot()) else { + return None; + }; + if data.is_empty() { + None + } else { + Some(data) + } } } @@ -457,8 +656,13 @@ impl Hub { ) -> Option> { let key = (room.clone(), batch_id); if let Some(b) = self.fragments.get_mut(&key) { - let idx = match usize::try_from(index) { Ok(i) => i, Err(_) => return None }; - if idx >= b.chunks.len() { return None; } + let idx = match usize::try_from(index) { + Ok(i) => i, + Err(_) => return None, + }; + if idx >= b.chunks.len() { + return None; + } if b.chunks[idx].is_none() { b.chunks[idx] = Some(fragment); b.received += 1; @@ -466,7 +670,9 @@ impl Hub { if b.received == b.fragment_count { let mut out = Vec::with_capacity(b.total_size as usize); for ch in b.chunks.iter() { - if let Some(bytes) = ch.as_ref() { out.extend_from_slice(bytes); } + if let Some(bytes) = ch.as_ref() { + out.extend_from_slice(bytes); + } } self.fragments.remove(&key); return Some(out); @@ -519,7 +725,9 @@ impl HubRegistry { let start = std::time::Instant::now(); if let Some(snapshot) = state.doc.export_snapshot() { let room_str = String::from_utf8_lossy(&room.room).to_string(); - match (saver)(ws.clone(), room_str.clone(), room.crdt, snapshot).await { + match (saver)(ws.clone(), room_str.clone(), room.crdt, snapshot) + .await + { Ok(()) => { state.dirty = false; let elapsed = start.elapsed(); @@ -756,21 +964,40 @@ async fn handle_conn( // send initial state: // - If snapshot available (Loro), send as a DocUpdate. if let Some(snap) = h.snapshot_bytes(&room) { - let du = ProtocolMessage::DocUpdate { crdt, room_id: room.room.clone(), updates: vec![snap] }; + let du = ProtocolMessage::DocUpdate { + crdt, + room_id: room.room.clone(), + updates: vec![snap], + }; if let Ok(bytes) = loro_protocol::encode(&du) { let _ = tx.send(Message::Binary(bytes.into())); debug!(room=?String::from_utf8_lossy(&room.room), "sent initial snapshot after join"); } } else { // Otherwise, attempt backfill if other clients present or the CRDT allows - let others_in_room = h.subs.get(&room).map(|v| v.len()).unwrap_or(0) > 1; - let allow_when_empty = h.docs.get(&room).map(|s| s.doc.allow_backfill_when_no_other_clients()).unwrap_or(false); + let others_in_room = + h.subs.get(&room).map(|v| v.len()).unwrap_or(0) > 1; + let allow_when_empty = h + .docs + .get(&room) + .map(|s| s.doc.allow_backfill_when_no_other_clients()) + .unwrap_or(false); if others_in_room || allow_when_empty { - let backfill = h.docs.get(&room).map(|s| s.doc.compute_backfill(&version)).unwrap_or_default(); + let backfill = h + .docs + .get(&room) + .map(|s| s.doc.compute_backfill(&version)) + .unwrap_or_default(); let backfill_cnt = backfill.len(); for u in backfill { - let du = ProtocolMessage::DocUpdate { crdt, room_id: room.room.clone(), updates: vec![u] }; - if let Ok(bytes) = loro_protocol::encode(&du) { let _ = tx.send(Message::Binary(bytes.into())); } + let du = ProtocolMessage::DocUpdate { + crdt, + room_id: room.room.clone(), + updates: vec![u], + }; + if let Ok(bytes) = loro_protocol::encode(&du) { + let _ = tx.send(Message::Binary(bytes.into())); + } } if backfill_cnt > 0 { debug!(room=?String::from_utf8_lossy(&room.room), cnt=%backfill_cnt, "sent backfill after join"); @@ -778,8 +1005,17 @@ async fn handle_conn( } } } - ProtocolMessage::DocUpdateFragmentHeader { crdt, room_id, batch_id, fragment_count, total_size_bytes } => { - let room = RoomKey { crdt, room: room_id.clone() }; + ProtocolMessage::DocUpdateFragmentHeader { + crdt, + room_id, + batch_id, + fragment_count, + total_size_bytes, + } => { + let room = RoomKey { + crdt, + room: room_id.clone(), + }; if !joined_rooms.contains(&room) { let err = ProtocolMessage::UpdateError { crdt, @@ -789,11 +1025,18 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } continue; } // Permission check - let perm = hub.lock().await.perms.get(&(conn_id, room.clone())).copied(); + let perm = hub + .lock() + .await + .perms + .get(&(conn_id, room.clone())) + .copied(); if !matches!(perm, Some(Permission::Write)) { let err = ProtocolMessage::UpdateError { crdt, @@ -803,11 +1046,16 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } continue; } // Bounds checks - if fragment_count == 0 || fragment_count > MAX_FRAGMENTS || total_size_bytes > MAX_BATCH_BYTES { + if fragment_count == 0 + || fragment_count > MAX_FRAGMENTS + || total_size_bytes > MAX_BATCH_BYTES + { let err = ProtocolMessage::UpdateError { crdt, room_id: room.room.clone(), @@ -816,7 +1064,9 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } continue; } // Initialize batch (guard against hijack by another sender) @@ -832,18 +1082,35 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } continue; } // else: duplicate header from same sender -> accept and broadcast as-is } else { - h.start_fragment_batch(&room, conn_id, batch_id, fragment_count, total_size_bytes); + h.start_fragment_batch( + &room, + conn_id, + batch_id, + fragment_count, + total_size_bytes, + ); } // Broadcast header as-is h.broadcast(&room, conn_id, Message::Binary(data)); } - ProtocolMessage::DocUpdateFragment { crdt, room_id, batch_id, index, fragment } => { - let room = RoomKey { crdt, room: room_id.clone() }; + ProtocolMessage::DocUpdateFragment { + crdt, + room_id, + batch_id, + index, + fragment, + } => { + let room = RoomKey { + crdt, + room: room_id.clone(), + }; if !joined_rooms.contains(&room) { let err = ProtocolMessage::UpdateError { crdt, @@ -853,7 +1120,9 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } continue; } // Validate batch existence and sender binding; also index bounds @@ -869,11 +1138,17 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } // do not broadcast continue; } - if !usize::try_from(index).ok().map(|i| i < b.chunks.len()).unwrap_or(false) { + if !usize::try_from(index) + .ok() + .map(|i| i < b.chunks.len()) + .unwrap_or(false) + { let err = ProtocolMessage::UpdateError { crdt, room_id: room.room.clone(), @@ -882,7 +1157,9 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } continue; } } else { @@ -894,16 +1171,22 @@ async fn handle_conn( batch_id: Some(batch_id), app_code: None, }; - if let Ok(bytes) = loro_protocol::encode(&err) { let _ = tx.send(Message::Binary(bytes.into())); } + if let Ok(bytes) = loro_protocol::encode(&err) { + let _ = tx.send(Message::Binary(bytes.into())); + } continue; } // Broadcast this fragment as-is to others (only after validation) h.broadcast(&room, conn_id, Message::Binary(data.clone())); // Accumulate and possibly finish - if let Some(buf) = h.add_fragment_and_maybe_finish(&room, batch_id, index, fragment) { + if let Some(buf) = + h.add_fragment_and_maybe_finish(&room, batch_id, index, fragment) + { // On completion: parse and apply to stored doc state if applicable match crdt { - CrdtType::Loro | CrdtType::LoroEphemeralStore => { + CrdtType::Loro + | CrdtType::LoroEphemeralStore + | CrdtType::LoroEphemeralStorePersisted => { if let Ok(updates) = parse_docupdate_payload(&buf) { let start = std::time::Instant::now(); h.apply_updates(&room, &updates); @@ -967,7 +1250,9 @@ async fn handle_conn( } let mut h = hub.lock().await; match crdt { - CrdtType::Loro | CrdtType::LoroEphemeralStore => { + CrdtType::Loro + | CrdtType::LoroEphemeralStore + | CrdtType::LoroEphemeralStorePersisted => { let start = std::time::Instant::now(); h.apply_updates(&room, &updates); let elapsed_ms = start.elapsed().as_millis(); @@ -1032,6 +1317,8 @@ fn parse_docupdate_payload(buf: &[u8]) -> Result>, String> { let b = r.read_var_bytes()?.to_vec(); out.push(b); } - if r.remaining() != 0 { return Err("trailing bytes".into()); } + if r.remaining() != 0 { + return Err("trailing bytes".into()); + } Ok(out) }