From 8ed15d86a1a8b70d56f75feb3352742ab4c43564 Mon Sep 17 00:00:00 2001 From: Levi Neuwirth Date: Wed, 10 Jun 2026 10:45:23 -0400 Subject: [PATCH] =?UTF-8?q?optimistic-apply=20daemon=20support=20=E2=80=94?= =?UTF-8?q?=20Loro=20text-delta=20hot=20path=20for=20remote=20inserts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The GPU frontend's per-keystroke edits arrive as FrontendEvent::CrdtOp. The old apply path materialized the whole document and diffed it per op — O(file) per typed character on the daemon main thread. - CrdtState: persistent text-projection subscription (capture gated by an AtomicBool so per-keystroke imports don't register/drop callbacks); import_updates_with_text_deltas returns Loro's deltas; unicode_to_utf8_pos converts the insert point. - Buffer::apply_remote_crdt_op: the common single-insert delta applies straight to the rope; deletes/compound updates keep the conservative materialize+diff fallback. UTF-8 position regression test included. - SyntaxRegistry::has_pending_parse_job_for: the main-thread "parse in flight" bit render producers need for settle-gating. - TextView::pos_to_display: stack buffer for short line prefixes + valid_up_to() boundary trim — removes a per-cursor-move allocation. Co-Authored-By: Claude Fable 5 --- src/buffer.rs | 112 +++++++++++++++++++++++++++++++++++++---------- src/crdt.rs | 91 +++++++++++++++++++++++++++++++++++++- src/syntax.rs | 27 ++++++++++++ src/text_view.rs | 32 +++++++++++--- 4 files changed, 232 insertions(+), 30 deletions(-) diff --git a/src/buffer.rs b/src/buffer.rs index a145981..d45740a 100644 --- a/src/buffer.rs +++ b/src/buffer.rs @@ -684,16 +684,12 @@ impl Buffer { /// Used by the daemon's `FrontendEvent::CrdtOp` handler when a /// replica frontend forwards a CRDT op. The flow: /// - /// 1. Capture the rope's current bytes (rope ≡ CRDT projection - /// invariant — both have the same content pre-import). - /// 2. `crdt.import_updates(op_bytes)` — integrates the remote op - /// into the local CRDT state. CRDT convergence handles - /// concurrent edits. - /// 3. Materialize the post-import CRDT content. - /// 4. Compute the diff between pre- and post-content as a single - /// `Replace` `EditOp` (single insert/delete falls out as - /// Replace with empty inserted or empty range). - /// 5. Apply the rope stages (rope mutation + mark adjustment + + /// 1. `crdt.import_updates_with_text_deltas(op_bytes)` integrates + /// the remote op and captures Loro's text projection delta. + /// 2. Apply the common single-insert shape directly to the rope. + /// Deletes and compound updates conservatively fall back to a + /// post-import materialization + contiguous diff. + /// 3. Apply the rope stages (rope mutation + mark adjustment + /// revision bump + modified flag + `on_edit` broadcast). /// Skips the CRDT-application stage (already done in step 2) /// AND the undo push (remote ops aren't locally undoable per @@ -750,25 +746,37 @@ impl Buffer { }); }; - // Step 1: capture pre-import bytes (rope ≡ CRDT projection - // invariant means rope.slice == crdt.materialize_string here). + // Integrate the remote op and capture Loro's projection diff. + // Optimistic GUI typing produces one Insert delta, so handle + // that shape without copying or materializing the document. + let text_deltas = crdt + .import_updates_with_text_deltas(op_bytes) + .map_err(|e| BufferError::CrdtRejected { + reason: format!("import_updates: {e:?}"), + })?; + if let Some((unicode_pos, inserted)) = single_remote_text_insert(&text_deltas) + && let Some(byte_pos) = crdt.unicode_to_utf8_pos(unicode_pos) + { + let byte_pos = byte_pos as Position; + let mut views = std::mem::take(&mut self.views); + let result = + self.run_remote_rope_stages(&mut views, byte_pos, byte_pos, inserted.as_bytes()); + self.views = views; + return result.map(Some); + } + + // Conservative fallback for deletes, compound updates, and + // already-integrated ops. The rope is still the pre-import + // projection, so it remains the source for the old bytes. let old_len = self.rope.len(); let mut old_bytes = vec![0u8; old_len as usize]; if old_len > 0 { self.rope.slice(0, old_len, &mut old_bytes); } - - // Step 2: integrate the remote op into the CRDT state. - crdt.import_updates(op_bytes) - .map_err(|e| BufferError::CrdtRejected { - reason: format!("import_updates: {e:?}"), - })?; - - // Step 3: materialize the post-import content. let new_content = crdt.materialize_string(); let new_bytes = new_content.as_bytes(); - // Step 4: compute common prefix/suffix at byte level, then + // Compute common prefix/suffix at byte level, then // **back off to UTF-8 char boundaries** in both strings. // // # Post-audit-round-4 F25: char-boundary alignment @@ -829,7 +837,7 @@ impl Buffer { let range_end = (old_bytes.len() - suffix) as Position; let inserted = &new_bytes[prefix..new_bytes.len() - suffix]; - // Step 5: apply rope stages without re-applying to CRDT + // Apply rope stages without re-applying to CRDT // (CRDT was applied above in step 2) and without undo push // (remote ops aren't locally undoable per M10.4). let mut views = std::mem::take(&mut self.views); @@ -1451,6 +1459,37 @@ impl Buffer { #[cfg(feature = "crdt")] type CrdtRoutingResult = (Option>, Option>); +/// Recognize the hot-path projection delta produced by one remote +/// insertion. Loro's retain/delete lengths use Unicode scalar offsets; +/// the caller converts the insertion point through the post-import +/// text container before applying the UTF-8 bytes to the rope. +#[cfg(feature = "crdt")] +fn single_remote_text_insert(deltas: &[Vec]) -> Option<(usize, &str)> { + let [delta] = deltas else { + return None; + }; + let mut cursor = 0usize; + let mut found = None; + for op in delta { + match op { + loro::TextDelta::Retain { retain, .. } => { + cursor = cursor.checked_add(*retain)?; + } + loro::TextDelta::Insert { insert, .. } if insert.is_empty() => {} + loro::TextDelta::Insert { insert, .. } => { + if found.is_some() { + return None; + } + found = Some((cursor, insert.as_str())); + cursor = cursor.checked_add(insert.chars().count())?; + } + loro::TextDelta::Delete { delete } if *delete == 0 => {} + loro::TextDelta::Delete { .. } => return None, + } + } + found +} + /// T M10.4: derive a fine-grained `(range, inserted_len)` Edit /// description for the change from `old_rope` to `new_rope` via /// longest-common-prefix + longest-common-suffix trim. @@ -2884,6 +2923,35 @@ mod tests { assert_eq!(count.get(), 1, "on_edit must fire for remote op"); } + #[cfg(feature = "crdt")] + #[test] + fn apply_remote_crdt_op_insert_after_multibyte_char_uses_utf8_byte_position() { + let donor = crate::crdt::CrdtState::new(2).expect("donor"); + donor.insert(0, "éx").expect("seed"); + + let mut buf = Buffer::new_with_crdt(BufferId::next(), "*utf8-insert*", 1).expect("buf"); + let donor_snap = donor.export_snapshot().expect("snap"); + buf.crdt + .as_ref() + .expect("crdt") + .import_snapshot(&donor_snap) + .expect("init from snap"); + buf.rope = crate::rope::Rope::from_bytes("éx".as_bytes()); + + let v_before = donor.version(); + donor.insert("é".len(), "!").expect("insert"); + let op_bytes = donor.export_updates_since(&v_before).expect("export"); + let edit = buf + .apply_remote_crdt_op(&op_bytes) + .expect("apply") + .expect("non-empty edit"); + + assert_eq!(rope_string(&buf), "é!x"); + assert_eq!(edit.range, Range::new("é".len() as u64, "é".len() as u64)); + assert_eq!(edit.inserted_len, 1); + assert_invariant(&buf); + } + /// F25 (post-audit-round-4): a CRDT update that changes one /// codepoint into another with a shared leading UTF-8 byte /// must produce a char-boundary-aligned diff. Pre-fix, the diff --git a/src/crdt.rs b/src/crdt.rs index c7d493c..8cef9e3 100644 --- a/src/crdt.rs +++ b/src/crdt.rs @@ -40,7 +40,18 @@ //! propagation, the optional `crdt_op` field on [`crate::rope::Edit`], //! and the convergence proptest. -use loro::{ExportMode, LoroDoc, LoroEncodeError, LoroResult, UndoManager, VersionVector}; +use std::sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, +}; + +use loro::{ + ContainerTrait, ExportMode, LoroDoc, LoroEncodeError, LoroResult, TextDelta, UndoManager, + VersionVector, +}; + +type TextDeltaBatches = Arc>>>; +type TextDeltaSubscription = (TextDeltaBatches, Arc, loro::Subscription); /// The CRDT-backed buffer state. /// @@ -54,6 +65,12 @@ use loro::{ExportMode, LoroDoc, LoroEncodeError, LoroResult, UndoManager, Versio /// (foreground worker materializes the initial projection). pub struct CrdtState { doc: LoroDoc, + /// Text projection deltas captured only while a remote import is + /// active. Keeping the subscription alive avoids registering and + /// dropping one callback for every typed character. + text_delta_batches: TextDeltaBatches, + text_delta_capture_enabled: Arc, + _text_delta_subscription: loro::Subscription, /// T M10.4: per-peer undo machinery. Bound to `doc`'s `peer_id` /// at construction; produces inverse ops attributed to that peer. /// @@ -90,13 +107,45 @@ impl CrdtState { // explicit get here ensures the container is registered before // any read or write. let _ = doc.get_text("body"); + let (text_delta_batches, text_delta_capture_enabled, text_delta_subscription) = + Self::subscribe_text_deltas(&doc); let undo = Self::create_undo_manager(&doc); Ok(Self { doc, + text_delta_batches, + text_delta_capture_enabled, + _text_delta_subscription: text_delta_subscription, undo: std::cell::RefCell::new(undo), }) } + fn subscribe_text_deltas(doc: &LoroDoc) -> TextDeltaSubscription { + let text = doc.get_text("body"); + let batches = Arc::new(Mutex::new(Vec::>::new())); + let capture_enabled = Arc::new(AtomicBool::new(false)); + let captured_batches = Arc::clone(&batches); + let captured_enabled = Arc::clone(&capture_enabled); + let subscription = doc.subscribe( + &text.id(), + Arc::new(move |event| { + if !captured_enabled.load(Ordering::Relaxed) { + return; + } + let mut guard = captured_batches + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for event in event.events { + if let Some(delta) = event.diff.as_text() + && !delta.is_empty() + { + guard.push(delta.clone()); + } + } + }), + ); + (batches, capture_enabled, subscription) + } + /// T M10.4: construct a fresh `UndoManager` bound to the given doc. /// Extracted as a helper because `from_bytes` constructs it AFTER /// the initial seed insert (so the seed isn't observable as an @@ -136,9 +185,14 @@ impl CrdtState { // The buffer's starting contents (from file load, scratch // initial text, etc.) shouldn't be undoable from the user's // perspective; only post-construction edits are. + let (text_delta_batches, text_delta_capture_enabled, text_delta_subscription) = + Self::subscribe_text_deltas(&doc); let undo = Self::create_undo_manager(&doc); Ok(Self { doc, + text_delta_batches, + text_delta_capture_enabled, + _text_delta_subscription: text_delta_subscription, undo: std::cell::RefCell::new(undo), }) } @@ -302,6 +356,41 @@ impl CrdtState { self.doc.import(bytes).map(|_| ()) } + /// Import remote updates and capture Loro's text projection deltas. + /// + /// The import callback runs synchronously before `doc.import` + /// returns. Buffer's hot path uses the captured single-insert shape + /// to update its rope projection without materializing the whole + /// document. + pub fn import_updates_with_text_deltas(&self, bytes: &[u8]) -> LoroResult>> { + self.text_delta_batches + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clear(); + self.text_delta_capture_enabled + .store(true, Ordering::Relaxed); + let import_result = self.doc.import(bytes).map(|_| ()); + self.text_delta_capture_enabled + .store(false, Ordering::Relaxed); + import_result?; + let mut guard = self + .text_delta_batches + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Ok(std::mem::take(&mut *guard)) + } + + /// Convert a Unicode scalar offset in the current text projection + /// to its UTF-8 byte offset. + #[must_use] + pub fn unicode_to_utf8_pos(&self, pos: usize) -> Option { + self.doc.get_text("body").convert_pos( + pos, + loro::cursor::PosType::Unicode, + loro::cursor::PosType::Bytes, + ) + } + /// T M10.10 post-audit-round-4 F26 — validate that importing /// the wire bytes `bytes` would attribute every new op to /// `expected_peer_id`. diff --git a/src/syntax.rs b/src/syntax.rs index aa4b538..70106c5 100644 --- a/src/syntax.rs +++ b/src/syntax.rs @@ -579,6 +579,16 @@ impl SyntaxRegistry { self.parse_jobs.borrow().len() } + /// True when a dispatched parse job for `buffer` has not yet been + /// installed or drained. The syntax Lua glue records jobs here at + /// dispatch time and removes them in `_install_settled`, so this + /// is the main-thread "parse in flight" bit for render producers + /// that need to avoid stale whole-file work while typing. + #[must_use] + pub fn has_pending_parse_job_for(&self, buffer: BufferId) -> bool { + self.parse_jobs.borrow().values().any(|&bid| bid == buffer) + } + /// Lazy-compile and cache the bundled `highlights.scm` query for /// `lang_name`. Returns `None` if the language is unknown, the /// language entry has an empty query (no highlights shipped), @@ -869,4 +879,21 @@ mod tests { // Pending list cleared on make_request. assert_eq!(handle.pending_edit_count(), 0); } + + #[test] + fn registry_tracks_inflight_parse_jobs_by_buffer() { + let registry = SyntaxRegistry::new(); + let a = BufferId::next(); + let b = BufferId::next(); + + registry.record_parse_job(11, a); + registry.record_parse_job(12, b); + + assert!(registry.has_pending_parse_job_for(a)); + assert!(registry.has_pending_parse_job_for(b)); + + assert_eq!(registry.take_parse_job(11), Some(a)); + assert!(!registry.has_pending_parse_job_for(a)); + assert!(registry.has_pending_parse_job_for(b)); + } } diff --git a/src/text_view.rs b/src/text_view.rs index 9549031..b552ae4 100644 --- a/src/text_view.rs +++ b/src/text_view.rs @@ -34,6 +34,10 @@ use crate::view::{DisplayCoord, View, Viewport}; /// that is a multiple of this value. const TAB_WIDTH: u32 = 8; +/// Line-prefix lengths up to this many bytes are decoded on the stack in +/// [`TextView::pos_to_display`]; longer prefixes fall back to a heap buffer. +const STACK_CAP: usize = 256; + /// Display width of `ch` when drawn starting at column `current_col`. /// /// Tabs expand to the next [`TAB_WIDTH`]-aligned column, so they need the @@ -180,13 +184,27 @@ impl View for TextView { if take == 0 { return Some(DisplayCoord::new(row_idx as u32, 0)); } - let mut bytes = vec![0u8; take]; - buf.snapshot_rope().slice(line_start, pos, &mut bytes); - // Drop trailing bytes that don't form a complete codepoint. - while !bytes.is_empty() && std::str::from_utf8(&bytes).is_err() { - bytes.pop(); - } - let s = std::str::from_utf8(&bytes).unwrap_or(""); + // Copy [line_start, pos) into a stack buffer for the common short-line + // case, hitting the heap only for unusually long prefixes. This removes + // the per-call allocation that previously ran on every cursor move. + let mut stack_buf = [0u8; STACK_CAP]; + let mut heap_buf: Vec; + let bytes: &mut [u8] = if take <= STACK_CAP { + &mut stack_buf[..take] + } else { + heap_buf = vec![0u8; take]; + &mut heap_buf + }; + buf.snapshot_rope().slice(line_start, pos, bytes); + // If `pos` fell inside a multi-byte codepoint, keep only the bytes up to + // the last complete codepoint. `valid_up_to()` gives that boundary in + // one step, replacing the old pop-one-byte-and-revalidate loop. (Only + // trailing bytes can be invalid here, since the slice is a prefix of + // valid UTF-8 cut at `pos`.) + let s = match std::str::from_utf8(bytes) { + Ok(valid) => valid, + Err(e) => std::str::from_utf8(&bytes[..e.valid_up_to()]).unwrap(), + }; let mut col: u32 = 0; for ch in s.chars() { col += char_display_width(ch, col);