089401e · oficina/docs

diff --git a/native/apple/Sources/DocsKit/CRDTReplica.swift b/native/apple/Sources/DocsKit/CRDTReplica.swift

index c1b9b7e..f9b987e 100644

--- a/native/apple/Sources/DocsKit/CRDTReplica.swift

+++ b/native/apple/Sources/DocsKit/CRDTReplica.swift

@@ -24,6 +24,7 @@ actor CRDTReplica {

}

private var authority: Sandbox?

+ private var merger: Sandbox?

private var opened = false

private var sources: [UInt64: Sandbox] = [:]

private var sourceSnapshots: [UInt64: Data] = [:]

@@ -34,7 +35,7 @@ actor CRDTReplica {

private let guid: String

enum Failure: Error { case unavailable, rejected, invalidProjection, generationLimit }

init(guid: String) { self.guid = guid }

- func close() { opened = false; authority = nil; sources.removeAll(); sourceSnapshots.removeAll(); lastGood = Data() }

+ func close() { opened = false; authority = nil; merger = nil; sources.removeAll(); sourceSnapshots.removeAll(); lastGood = Data() }

private func fresh(update: Data) throws -> (UInt64, Sandbox) {

let sandbox = try Sandbox()

var client = UInt64(UInt32.random(in: 1...UInt32.max))

@@ -107,6 +108,16 @@ actor CRDTReplica {

// clocks from a partially executed command. UI reports undo loss.

authority = try? fresh(update: lastGood).1

}

+ func mergeUpdates(_ updates: [Data]) throws -> Data {

+ guard let first = updates.first else { throw Failure.rejected }

+ if updates.count == 1 { return first }

+ guard opened else { throw Failure.unavailable }

+ if merger == nil { merger = try Sandbox() }

+ var envelope = Data([253])

+ for update in updates { envelope.append(update) }

+ do { return try merger!.invoke(envelope) }

+ catch { merger = nil; throw error }

+ }

func apply(_ update: Data) throws -> CRDTProjection {

let authority = try currentAuthority()

do {

diff --git a/native/apple/Sources/DocsKit/DocsModel.swift b/native/apple/Sources/DocsKit/DocsModel.swift

index f7a2b86..12b405f 100644

--- a/native/apple/Sources/DocsKit/DocsModel.swift

+++ b/native/apple/Sources/DocsKit/DocsModel.swift

@@ -176,18 +176,36 @@ import RastrilloApple

return !hasUnsavedChanges

}

func flush() async {

- guard !saving, let detail else { return }; saving = true

+ guard !saving, let detail, let replica else { return }; saving = true

let epoch = documentEpoch

defer { if documentEpoch == epoch { saving = false } }

- while let update = pending.first {

+ while !pending.isEmpty {

do {

+ // A slow request must not leave one full round trip per keypress.

+ // Keep the captured prefix until acknowledgement; edits appended

+ // during either await belong to the next batch.

+ var bytes = 0

+ let batch = pending.prefix { update in

+ guard bytes == 0 || bytes + update.count <= 512 * 1024 else { return false }

+ bytes += update.count

+ return true

+ }

+ var count = batch.count

+ let update: Data

+ do { update = try await replica.mergeUpdates(Array(batch)) }

+ catch {

+ // An optimization hitting its sandbox limit must not block saving.

+ update = batch.first!

+ count = 1

+ }

+ guard documentEpoch == epoch else { return }

let (_, status) = try await client.response("/d/\(detail.sqid)/updates", method: "POST", body: update, contentType: "application/octet-stream")

guard documentEpoch == epoch else { return }

guard status == 204 else {

if status == 401 || status == 403 || status == 404 { await refreshAccess(detail.sqid, epoch: epoch) }

throw CRDTReplica.Failure.rejected

}

- pending.removeFirst()

+ pending.removeFirst(count)

} catch { if documentEpoch == epoch { errorKey = "native.pending" }; return }

}

if errorKey == "native.pending" || errorKey == "docs.editor.status.retrying" { errorKey = nil }

diff --git a/native/apple/Tests/DocsKitTests/DocumentSyncIntegrationTests.swift b/native/apple/Tests/DocsKitTests/DocumentSyncIntegrationTests.swift

new file mode 100644

index 0000000..a8c9ddf

--- /dev/null

+++ b/native/apple/Tests/DocsKitTests/DocumentSyncIntegrationTests.swift

@@ -0,0 +1,61 @@

+import Foundation

+import RastrilloApple

+import XCTest

+@testable import DocsKit

+

+final class DocumentSyncIntegrationTests: XCTestCase {

+ func testFailedMergeCannotPoisonEditorOrNextMerge() async throws {

+ let replica = CRDTReplica(guid: "merge-isolation")

+ _ = try await replica.open(update: Data())

+ _ = try await replica.command(["op": .string("insert_element"), "at": .number(0), "type": .string("paragraph")])

+ do {

+ _ = try await replica.mergeUpdates([Data([255]), Data([255])])

+ XCTFail("Malformed updates must fail")

+ } catch { }

+ _ = try await replica.projection()

+ let (first, _) = try await replica.command(["op": .string("display_edit"), "at": .number(0), "length": .number(0), "text": .string("a")])

+ let (second, projection) = try await replica.command(["op": .string("display_edit"), "at": .number(1), "length": .number(0), "text": .string("b")])

+ let merged = try await replica.mergeUpdates([first, second])

+ XCTAssertFalse(merged.isEmpty)

+ let after = try await replica.projection()

+ XCTAssertEqual(after.children.map(\.plainText), projection.children.map(\.plainText))

+ await replica.close()

+ }

+ @MainActor func testTypingDoesNotAccumulateOneRoundTripPerKeystroke() async throws {

+ guard let origin = ProcessInfo.processInfo.environment["DOCS_SYNC_TEST_ORIGIN"] else {

+ throw XCTSkip("Run native/scripts/test-sync-latency.py for the delayed HTTP fixture")

+ }

+ let client = NativeHTTP(origin: URL(string: origin)!)

+ let model = DocsModel(client: client)

+ await model.open("fixture")

+ XCTAssertNotNil(model.projection)

+ model.didRender(model.projectionGeneration)

+ for _ in 0..<20 {

+ XCTAssertTrue(model.enqueue(["op": .string("display_edit"), "at": .number(0), "length": .number(0), "text": .string("X")]))

+ while model.hasPendingInput { try await Task.sleep(for: .milliseconds(1)) }

+ model.didRender(model.projectionGeneration)

+ try await Task.sleep(for: .milliseconds(50))

+ }

+ let finishedTyping = ContinuousClock.now

+ let deadline = finishedTyping.advanced(by: .seconds(15))

+ while model.hasUnsavedChanges && ContinuousClock.now < deadline {

+ await model.flush()

+ try await Task.sleep(for: .milliseconds(10))

+ }

+ XCTAssertFalse(model.hasUnsavedChanges)

+ let lag = finishedTyping.duration(to: .now)

+ struct Stats: Decodable { let updates: [Data] }

+ let stats: Stats = try await client.request("/stats")

+ print("SYNC: 20 edits, 250ms/request, uploads=\(stats.updates.count), catch-up after typing=\(lag)")

+ XCTAssertLessThanOrEqual(stats.updates.count, 10, "Unsent edits must share a request")

+ XCTAssertLessThan(lag, .seconds(1), "Typing must not leave a growing serial-request backlog")

+ let (initial, _) = try await client.response("/d/fixture/updates")

+ let peer = CRDTReplica(guid: "sync-fixture")

+ _ = try await peer.open(update: initial)

+ for update in stats.updates { _ = try await peer.apply(update) }

+ let projection = try await peer.projection()

+ XCTAssertEqual(projection.children.map(\.plainText), model.projection?.children.map(\.plainText), "A receiver must get every edit")

+ await peer.close()

+ await model.stop()

+ }

+}

diff --git a/native/crdt/src/lib.rs b/native/crdt/src/lib.rs

index 2d36459..351d83a 100644

--- a/native/crdt/src/lib.rs

+++ b/native/crdt/src/lib.rs

@@ -12,6 +12,7 @@ use serde_json::{json, Value};

use std::{cell::RefCell, collections::HashSet, sync::Arc};

use yrs::{Doc, OffsetKind, Options, ReadTxn, StateVector, Transact, UndoManager, Update};

use yrs::updates::decoder::{Decode, Decoder, DecoderV1};

+use yrs::updates::encoder::Encode;

pub struct Editor {

doc: Doc,

@@ -34,15 +35,7 @@ impl Editor {

pub fn apply(&self, bytes: &[u8]) -> Result<(), &'static str> { self.apply_origin(bytes, false) }

pub fn apply_local(&self, bytes: &[u8]) -> Result<(), &'static str> { self.apply_origin(bytes, true) }

fn apply_origin(&self, bytes: &[u8], local: bool) -> Result<(), &'static str> {

- let mut remaining = bytes;

- let mut updates = Vec::new();

- while !remaining.is_empty() {

- let mut decoder = DecoderV1::from(remaining);

- updates.push(Update::decode(&mut decoder).map_err(|_| "invalid_update")?);

- let trailing = decoder.read_to_end().map_err(|_| "invalid_update")?.len();

- if trailing >= remaining.len() { return Err("invalid_update"); }

- remaining = &remaining[remaining.len() - trailing..];

- }

+ let updates = decode_updates(bytes)?;

let mut txn = self.doc.transact_mut_with(if local { "native-edit" } else { "remote" });

for update in updates { txn.apply_update(update).map_err(|_| "invalid_update")?; }

Ok(())

@@ -122,7 +115,27 @@ pub extern "C" fn docs_editor_output_ptr() -> u32 { OUTPUT.with(|out| out.borrow

#[no_mangle]

pub extern "C" fn docs_editor_output_len() -> u32 { OUTPUT.with(|out| out.borrow().len() as u32) }

+fn decode_updates(mut remaining: &[u8]) -> Result<Vec<Update>, &'static str> {

+ let mut updates = Vec::new();

+ while !remaining.is_empty() {

+ let mut decoder = DecoderV1::from(remaining);

+ updates.push(Update::decode(&mut decoder).map_err(|_| "invalid_update")?);

+ let trailing = decoder.read_to_end().map_err(|_| "invalid_update")?.len();

+ if trailing >= remaining.len() { return Err("invalid_update"); }

+ remaining = &remaining[remaining.len() - trailing..];

+ }

+ Ok(updates)

+}

+

pub fn dispatch(bytes: &[u8]) -> Result<Vec<u8>, &'static str> {

+ // Merge pending writes without constructing a document: pending structs may

+ // depend on state already held by the server. Snapshotting an empty document

+ // would silently lose those structs and deletes.

+ if bytes.first() == Some(&253) {

+ let updates = decode_updates(&bytes[1..])?;

+ if updates.is_empty() { return Err("invalid_update"); }

+ return Ok(Update::merge_updates(updates).encode_v1());

+ }

// Binary envelope avoids expanding an 8MiB log into JSON byte numbers.

if matches!(bytes.first(), Some(254 | 255)) {

return EDITOR.with(|cell| {

diff --git a/native/crdt/tests/replica.rs b/native/crdt/tests/replica.rs

index f371073..4256d08 100644

--- a/native/crdt/tests/replica.rs

+++ b/native/crdt/tests/replica.rs

@@ -278,3 +278,27 @@ fn input_rules_transform_and_backspace_restore_typed_prefix() {

assert_eq!(plain(&editor.projection()), format!("{prefix} "));

}

+

+#[test]

+fn merged_pending_updates_preserve_dependencies_deletes_and_duplicate_delivery() {

+ use yrs::updates::decoder::{Decode, Decoder, DecoderV1};

+ use yrs::Update;

+ let mut writer = Editor::new("batch", 201);

+ let (_, node) = paragraph(&mut writer, "original");

+ let initial = writer.snapshot();

+ let mut bytes = vec![253];

+ bytes.extend(writer.command(&json!({"op":"replace_text","node":node,"at":8,"length":0,"text":" added"})).unwrap());

+ bytes.extend(writer.command(&json!({"op":"replace_text","node":node,"at":0,"length":3,"text":"new"})).unwrap());

+ bytes.extend(writer.command(&json!({"op":"format","node":node,"at":0,"length":3,"marks":{"strong":{}}})).unwrap());

+ let merged = docs_native_crdt::dispatch(&bytes).unwrap();

+ // A POST must contain exactly one Yjs v1 update, not concatenated messages.

+ let mut decoder = DecoderV1::from(merged.as_slice());

+ Update::decode(&mut decoder).unwrap();

+ assert!(decoder.read_to_end().unwrap().is_empty());

+ let receiver = Editor::new("batch", 202);

+ receiver.apply(&initial).unwrap();

+ receiver.apply(&merged).unwrap();

+ assert_eq!(receiver.projection()["children"], writer.projection()["children"]);

+ assert!(docs_native_crdt::dispatch(&[253, 255]).is_err());

+}

diff --git a/native/scripts/test-sync-latency.py b/native/scripts/test-sync-latency.py

new file mode 100644

index 0000000..ed4ae2a

--- /dev/null

+++ b/native/scripts/test-sync-latency.py

@@ -0,0 +1,61 @@

+#!/usr/bin/env python3

+"""Measure native upload backlog against a delayed loopback server with fixture data."""

+import base64

+import http.server

+import json

+import os

+from pathlib import Path

+import subprocess

+import sys

+import threading

+import time

+

+root = Path(__file__).resolve().parents[2]

+fixture = (root / "internal/docs/seeddata/letter.update.bin").read_bytes()

+updates = []

+lock = threading.Lock()

+fail_once = "--fail-once" in sys.argv

+

+class Handler(http.server.BaseHTTPRequestHandler):

+ def do_GET(self):

+ if self.path.endswith("/stream"):

+ self.send_response(200)

+ self.send_header("Content-Type", "text/event-stream")

+ self.end_headers()

+ self.wfile.write(b"data: 0\n\n")

+ self.wfile.flush()

+ time.sleep(30)

+ return

+ if self.path == "/stats":

+ with lock:

+ data = json.dumps(dict(updates=updates)).encode()

+ elif "/updates" in self.path:

+ data = fixture

+ else:

+ data = json.dumps(dict(sqid="fixture", title="Sync fixture", guid="sync-fixture",

+ can_write=True, can_share=False)).encode()

+ self.send_response(200)

+ self.send_header("Docs-Seq", "0")

+ self.send_header("Content-Length", str(len(data)))

+ self.end_headers()

+ self.wfile.write(data)

+

+ def do_POST(self):

+ body = self.rfile.read(int(self.headers.get("Content-Length", "0")))

+ # Deterministic request latency, not an artificial client debounce.

+ time.sleep(0.25)

+ with lock:

+ updates.append(base64.b64encode(body).decode())

+ self.send_response(503 if fail_once and len(updates) == 1 else 204)

+ self.end_headers()

+

+ def log_message(self, *_):

+ pass

+

+with http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) as server:

+ threading.Thread(target=server.serve_forever, daemon=True).start()

+ env = dict(os.environ, DOCS_SYNC_TEST_ORIGIN=f"http://127.0.0.1:{server.server_port}")

+ result = subprocess.run(["swift", "test", "--package-path", str(root / "native/apple"),

+ "--filter", "DocumentSyncIntegrationTests"], env=env, timeout=240)

+ server.shutdown()

+ raise SystemExit(result.returncode)

oficina / docs Public

Batch pending native document updates without blocking editing