App kit: text in elements, and big changes that arrive whole

AGENTS.md

@@ -32,7 +32,7 @@ crates/
   store     craft-cloud-store     blob store: disk, S3, memory; checks every blob in and out
   meta      craft-cloud-meta      metadata on SQLite or Postgres; the batch commit; the change feed; invites and pairings (`join.rs`); live documents' logs and snapshots (`live.rs`)
   sync      craft-cloud-sync      the sync engine over `Fs`/`Transport` traits; `MemFs`; scenario tests and the simulation
-  kit       craft-cloud-kit       the app kit (`docs/kit.md`): an app's model live through the agent; the `LiveDocument` trait and `Session` (`session.rs`: open, the `local_change` hook, `poll`, undo through Loro's undo manager, reconnect and the 64 MiB queue, presence, saves), the change writer that cuts commits under 512 KiB (`writer.rs`), the mappings (`tree.rs`, `text.rs`, `table.rs`, `timeline.rs`, `opaque.rs`, `list.rs`), values, chunk references and blob stores (`value.rs`)
+  kit       craft-cloud-kit       the app kit (`docs/kit.md`): an app's model live through the agent; the `LiveDocument` trait and `Session` (`session.rs`: open, the `local_change` hook, `poll`, undo through Loro's undo manager, reconnect and the 64 MiB queue, presence, saves), the change writer: one commit per change, exported as one update (a bundle of `SLICE_OPS` slices), cut in fragments past 1 MiB by `craft-cloud-app` (`writer.rs`), the mappings (`tree.rs`, `text.rs`, `table.rs`, `timeline.rs`, `opaque.rs`, `list.rs`; stories and tables nest in a node's or a cell's map: `props.rs`), values, chunk references and blob stores (`value.rs`)
   egui      craft-cloud-egui      Craft Cloud's piece of an app's window: the account line, the shared recents, a live document's badge, and `CloudWatch` that keeps them current
   app       craft-cloud-app       what apps link: a blocking client of the agent's local API (status, sign-in, joining, computers, pause, autostart, recents, live documents at the level of bytes: `live.rs`)
 apps/

apps/craft-cloud-agent/src/doc.rs

@@ -105,16 +105,21 @@ fn session(
     };
     let mirror = LoroDoc::new();
     mirror.import(document.state()).map_err(|e| usage(format!("the document's state doesn't read: {e}")))?;
+    // (An update this copy can't take is the other side's; the agent has it, and so will a fresh open. A bundle's
+    // slices go in one after the other.)
+    let import = |mirror: &LoroDoc, bytes: &[u8]| {
+        for part in craft_cloud_proto::fragments::parts(bytes).unwrap_or_default() {
+            let _ = mirror.import(part);
+        }
+    };
     for update in document.state_updates() {
-        // (An update this copy can't take is the other side's; the agent has it, and so will a fresh open.)
-        let _ = mirror.import(update);
+        import(&mirror, update);
     }
     let text = |mirror: &LoroDoc| mirror.get_text(TEXT).to_string();
     // What came right after the state (updates still waiting for their history), before anything else.
     let take = |mirror: &LoroDoc, event: &Event| {
         if let Event::Update(bytes) = event {
-            // (An update this copy can't take is the other side's; the agent has it, and so will a fresh open.)
-            let _ = mirror.import(bytes);
+            import(mirror, bytes);
         }
     };
     while let Some(event) = document.try_recv() {
@@ -148,7 +153,7 @@ fn session(
     loop {
         match document.recv_timeout(Duration::from_secs(1)) {
             Some(Event::Update(bytes)) => {
-                let _ = mirror.import(&bytes);
+                import(&mirror, &bytes);
                 say("--".into());
                 say(text(&mirror));
             }

apps/craft-cloud-agent/src/live/copy.rs

@@ -5,6 +5,7 @@
 
 use std::collections::BTreeMap;
 
+use craft_cloud_proto::fragments::{is_bundle, parts};
 use craft_cloud_proto::live::MAX_UPDATE_BYTES;
 use loro::{ExportMode, IdSpan, LoroDoc, VersionVector};
 
@@ -44,9 +45,16 @@ pub fn frontiers(ids: &[(u64, u32)]) -> Option<loro::Frontiers> {
     ids.iter().map(|(peer, counter)| Some(loro::ID::new(*peer, i32::try_from(*counter).ok()?))).collect::<Option<Vec<_>>>().map(loro::Frontiers::from)
 }
 
-/// The ranges of operations an update (or snapshot) holds, read from its header without importing it; `None` for
-/// bytes Loro can't read.
+/// The ranges of operations an update (or snapshot, or a bundle of updates: its slices', merged) holds, read from its
+/// header without importing it; `None` for bytes Loro can't read.
 pub fn spans_of(bytes: &[u8]) -> Option<Vec<(u64, i32, i32)>> {
+    if is_bundle(bytes) {
+        let mut spans = Spans::new();
+        for part in parts(bytes)? {
+            add_spans(&mut spans, &spans_of(part)?);
+        }
+        return Some(spans.into_iter().flat_map(|(peer, list)| list.into_iter().map(move |(s, e)| (peer, s, e))).collect());
+    }
     let meta = guarded(|| LoroDoc::decode_import_blob_meta(bytes, true).map_err(|e| UpdateError::Malformed(e.to_string()))).ok()?;
     Some(meta.partial_end_vv.iter().map(|(peer, end)| (*peer, meta.partial_start_vv.get(peer).copied().unwrap_or(0), *end)).collect())
 }
@@ -94,6 +102,10 @@ fn usable_peer(peer: u64) -> u64 {
 
 /// The most bytes of updates one of Loro's batch imports takes ([`LiveCopy::import_batch`]).
 const BATCH_GROUP_BYTES: usize = 16 * 1024;
+/// A bundle's slice whose import takes this long: the rest goes in as one update ([`LiveCopy::import_any_size`]).
+/// (The kit takes slices the same way, with the same threshold: `import_slices` in `craft-cloud-kit`; keep the two
+/// alike.)
+const SLOW_SLICE: std::time::Duration = std::time::Duration::from_millis(50);
 
 /// `updates` in consecutive groups of at most `max` bytes (an update bigger than that alone).
 fn groups(updates: &[Vec<u8>], max: usize) -> Vec<&[Vec<u8>]> {
@@ -188,8 +200,57 @@ impl LiveCopy {
         self.import_any_size(bytes)
     }
 
-    /// [`LiveCopy::import`] without the update limit: for a snapshot, or this machine's own stored document.
+    /// [`LiveCopy::import`] without the update limit: for a snapshot, or this machine's own stored document. A bundle
+    /// (one change as its slices) goes in slice by slice, all of it before this returns; one that doesn't read is
+    /// refused whole.
     pub fn import_any_size(&self, bytes: &[u8]) -> Result<Taken, UpdateError> {
+        if is_bundle(bytes) {
+            let slices = parts(bytes).ok_or_else(|| UpdateError::Malformed("a bundle cut short".into()))?;
+            // Every slice is read before any goes in: one that doesn't read refuses the bundle whole, nothing of it in.
+            for slice in &slices {
+                guarded(|| LoroDoc::decode_import_blob_meta(slice, true).map_err(|e| UpdateError::Malformed(e.to_string())))?;
+            }
+            let (mut new, mut needs) = (false, Vec::new());
+            let mut take = |taken: Taken| match taken {
+                Taken::Applied { new: n } => new |= n,
+                Taken::Waiting { needs: more } => needs.extend(more),
+            };
+            let mut i = 0;
+            while let Some(slice) = slices.get(i) {
+                let started = std::time::Instant::now();
+                take(self.import_one(slice)?);
+                i += 1;
+                // A slice this slow: Loro takes its slow way (edits in the same text concurrent with the change, in
+                // one order of the peers, make each slice cost more than the last): the rest goes in as one update.
+                let rest = slices.get(i..).unwrap_or_default();
+                if rest.len() >= 2
+                    && started.elapsed() > SLOW_SLICE
+                    && let Ok(one) = self.joined(rest)
+                {
+                    take(self.import_one(&one)?);
+                    break;
+                }
+            }
+            return Ok(if needs.is_empty() { Taken::Applied { new } } else { Taken::Waiting { needs } });
+        }
+        self.import_one(bytes)
+    }
+
+    /// `updates` (which follow on from this copy) as one update, joined on a detached fork (which only takes their
+    /// operations).
+    fn joined(&self, updates: &[&[u8]]) -> Result<Vec<u8>, UpdateError> {
+        self.guard(|doc| {
+            let fork = doc.fork();
+            fork.detach();
+            let before = fork.oplog_vv();
+            for u in updates {
+                fork.import(u).map_err(|e| UpdateError::Malformed(e.to_string()))?;
+            }
+            fork.export(ExportMode::updates(&before)).map_err(|e| UpdateError::Malformed(e.to_string()))
+        })
+    }
+
+    fn import_one(&self, bytes: &[u8]) -> Result<Taken, UpdateError> {
         let status = self.guard(|doc| {
             #[cfg(test)]
             Self::failing(&[bytes]);
@@ -340,6 +401,9 @@ impl LiveCopy {
 
     /// The longest start of `changes` that fits in `max` bytes as one update (at least one change; `Invalid` if that
     /// one doesn't fit in `hard_max`), with how many changes it holds.
+    ///
+    /// (A Loro change is a run of one peer's commits Loro merged, not an app's change, so a big one goes in pieces
+    /// here: this is history the log lost, sent up again after a server's restore.)
     pub fn export_prefix(&self, changes: &[(u64, i32, i32)], max: usize, hard_max: usize) -> Result<(Vec<u8>, usize), UpdateError> {
         let Some(&(peer, start, end)) = changes.first() else { return Err(UpdateError::Invalid("nothing to send")) };
         // The most operations of the first change that fit, by doubling from one and then halving the gap, so no
@@ -425,11 +489,17 @@ impl LiveCopy {
 /// after a catch-up from the start, or worked out from the copy when nothing else is in it); every update taken after
 /// it adds its ranges. A version with a gap (an update whose history comes later in the log) is no version at all:
 /// no snapshot can be made there.
+///
+/// A change that came in fragments is in the log from its last fragment's row on (its ranges are noted there), and
+/// no version is offered between its first fragment's row and its last: a snapshot there would drop its first
+/// fragments from the log and keep the rest, and a machine catching up later would never get it whole.
 #[derive(Debug, Clone, Default)]
 pub struct LogVersion {
     anchor: Option<(u64, Version)>,
     /// The updates taken since the anchor, in order: number and ranges.
     since: Vec<(u64, Ranges)>,
+    /// The rows of the changes that came in fragments since the anchor: first and last.
+    sets: Vec<(u64, u64)>,
 }
 
 /// The ranges of operations of an update: peer, first and end counter.
@@ -441,7 +511,14 @@ const MAX_SINCE: usize = 100_000;
 impl LogVersion {
     /// Anchored at `seq` with `version`.
     pub fn anchored(seq: u64, version: Version) -> Self {
-        Self { anchor: Some((seq, version)), since: Vec::new() }
+        Self { anchor: Some((seq, version)), since: Vec::new(), sets: Vec::new() }
+    }
+
+    /// A change that came in fragments took the rows `first` to `last`.
+    pub fn note_set(&mut self, first: u64, last: u64) {
+        if self.anchor.is_some() && last > first {
+            self.sets.push((first, last));
+        }
     }
 
     /// Nothing known.
@@ -469,7 +546,7 @@ impl LogVersion {
     /// The version the log's updates up to `upto` make, if it is known and has no gap.
     pub fn at(&self, upto: u64) -> Option<Version> {
         let (at, base) = self.anchor.as_ref()?;
-        if upto < *at {
+        if upto < *at || self.sets.iter().any(|(first, last)| *first <= upto && upto < *last) {
             return None;
         }
         let mut spans = Spans::new();
@@ -495,6 +572,7 @@ impl LogVersion {
     /// The version at `upto` is `version` (a snapshot was made there): what came before is forgotten.
     pub fn rebase(&mut self, upto: u64, version: Version) {
         self.since.retain(|(seq, _)| *seq > upto);
+        self.sets.retain(|(_, last)| *last > upto);
         self.anchor = Some((upto, version));
     }
 }
@@ -579,6 +657,91 @@ mod tests {
         assert_eq!(back.get_text("text").len_utf8(), 5 * 500 * 1024, "all of it");
     }
 
+    /// A 4 MiB paste on top of another peer's text: whole in one update, Loro's import takes seconds (time growing
+    /// with the square of the text); as a bundle of slices (as the kit sends it), the copy takes it in a moment, and
+    /// holds all of it. A bundle's ranges are its slices'.
+    #[test]
+    fn a_big_change_as_a_bundle_of_slices_imports_at_once() {
+        use craft_cloud_proto::fragments::{SLICE_OPS, bundle};
+        let importer = LoroDoc::new();
+        importer.set_peer_id(40).unwrap();
+        text_update(&importer, 0, "import");
+        let import = importer.export(ExportMode::Snapshot).unwrap();
+        let app = LoroDoc::new();
+        app.set_peer_id(50).unwrap();
+        app.import(&import).unwrap();
+        app.get_text("text").insert(0, &"a".repeat(4 * 1024 * 1024)).unwrap();
+        app.commit();
+        let end = *app.oplog_vv().get(&50).unwrap();
+        let mut slices = Vec::new();
+        let mut at = 0;
+        while at < end {
+            let to = (at + SLICE_OPS).min(end);
+            slices.push(app.export(ExportMode::updates_in_range(vec![IdSpan::new(50, at, to)])).unwrap());
+            at = to;
+        }
+        let bundle_bytes = bundle(&slices);
+        assert!(is_bundle(&bundle_bytes));
+        assert_eq!(spans_of(&bundle_bytes), Some(vec![(50, 0, end)]), "its ranges, merged");
+        let copy = LiveCopy::load(&import, 1).unwrap();
+        let started = std::time::Instant::now();
+        assert_eq!(copy.import_any_size(&bundle_bytes).unwrap(), Taken::Applied { new: true });
+        assert!(started.elapsed() < std::time::Duration::from_secs(2), "{:?}", started.elapsed());
+        assert!(copy.covers(&[(50, end)]));
+        let back = LoroDoc::new();
+        back.import(&copy.snapshot().unwrap()).unwrap();
+        assert_eq!(back.get_text("text").len_utf8(), 4 * 1024 * 1024 + "import".len());
+        // A snapshot of the copy as it is: at once too.
+        let started = std::time::Instant::now();
+        let at_now = copy.snapshot_at(&copy.version().unwrap()).unwrap();
+        assert!(!at_now.is_empty());
+        assert!(started.elapsed() < std::time::Duration::from_secs(2), "{:?}", started.elapsed());
+        // A bundle with a slice that doesn't read is refused whole: nothing of its other slices goes in.
+        let fresh = LiveCopy::load(&import, 1).unwrap();
+        let mut bad: Vec<Vec<u8>> = slices.iter().take(3).cloned().collect();
+        bad.push(b"not an update".to_vec());
+        assert!(fresh.import_any_size(&bundle(&bad)).is_err());
+        assert!(!fresh.covers(&[(50, 1)]), "nothing of it in");
+    }
+
+    /// A 600 KB paste's bundle into a copy holding an app's edit of the same text made at the same moment, the edit's
+    /// peer below the paste's (the order Loro is slow in, slice after slice: about a minute here): a few slices one
+    /// by one, then the rest together, in seconds, and the copy holds both.
+    #[test]
+    fn a_bundle_into_a_text_edited_at_the_same_moment_goes_in_in_seconds() {
+        use craft_cloud_proto::fragments::{SLICE_OPS, bundle};
+        let importer = LoroDoc::new();
+        importer.set_peer_id(40).unwrap();
+        text_update(&importer, 0, "import");
+        let import = importer.export(ExportMode::Snapshot).unwrap();
+        let typist = LoroDoc::new();
+        typist.set_peer_id(10).unwrap();
+        typist.import(&import).unwrap();
+        let typed = text_update(&typist, 0, "hello ");
+        let paster = LoroDoc::new();
+        paster.set_peer_id(50).unwrap();
+        paster.import(&import).unwrap();
+        let paste = "p".repeat(600 * 1024);
+        paster.get_text("text").insert(6, &paste).unwrap();
+        paster.commit();
+        let end = *paster.oplog_vv().get(&50).unwrap();
+        let mut slices = Vec::new();
+        let mut at = 0;
+        while at < end {
+            let to = (at + SLICE_OPS).min(end);
+            slices.push(paster.export(ExportMode::updates_in_range(vec![IdSpan::new(50, at, to)])).unwrap());
+            at = to;
+        }
+        let copy = LiveCopy::load(&import, 1).unwrap();
+        copy.import(&typed).unwrap();
+        let started = std::time::Instant::now();
+        assert_eq!(copy.import_any_size(&bundle(&slices)).unwrap(), Taken::Applied { new: true });
+        assert!(started.elapsed() < std::time::Duration::from_secs(15), "{:?}", started.elapsed());
+        let back = LoroDoc::new();
+        back.import(&copy.snapshot().unwrap()).unwrap();
+        assert_eq!(back.get_text("text").to_string(), format!("hello import{paste}"));
+    }
+
     /// An update Loro refuses leaves the copy as it was (Ruling: validation without a scratch copy).
     #[test]
     fn a_refused_update_leaves_the_copy_as_it_was() {