authorgravatar for git@paperclover.netclover caruso <git@paperclover.net> 2026-10-05 21:44:54-07:00
committergravatar for git@paperclover.netclover caruso <git@paperclover.net> 2026-10-05 22:21:54-07:00
loga31dfda0e9162d43429aa87a8f7f99cc2273a2cc
tree601928ef5fd760808f4cce638f34af76498eadd2
parent0df51c3995b5c6dc655b42eaff33529f17b56d52
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU

fix: keep large Live Share section updates within relay limits

Bound differential section data by the encoded chunk budget; larger updates retain the existing notification and chunked read path. The regression reproduces a relay disconnect on the previous writer. fixes #94 Assisted-by: gpt-6.1-sol

2 files changed, 108 insertions(+), 2 deletions(-)

crates/notebook/src/live/share.rs+18-2
...@@ -822,7 +822,7 @@ fn digest(stamp: &Stamp) -> u64 {...@@ -822,7 +822,7 @@ fn digest(stamp: &Stamp) -> u64 {
822}822}
823823
824/// The writes that make `after` of `before`, a commit's appended bytes, patches and header,824/// The writes that make `after` of `before`, a commit's appended bytes, patches and header,
825/// where they are much less than `after` itself.825/// where their encoding fits a chunk and they are much less than `after` itself.
826fn delta(before: &[u8], after: &[u8]) -> Option<Vec<Written>> {826fn delta(before: &[u8], after: &[u8]) -> Option<Vec<Written>> {
827 const BLOCK: usize = 4096;827 const BLOCK: usize = 4096;
828 if before.len() < 1024 || after.len() < before.len() {828 if before.len() < 1024 || after.len() < before.len() {
...@@ -858,7 +858,8 @@ fn delta(before: &[u8], after: &[u8]) -> Option<Vec<Written>> {...@@ -858,7 +858,8 @@ fn delta(before: &[u8], after: &[u8]) -> Option<Vec<Written>> {
858 });858 });
859 }859 }
860 let sent: usize = writes.iter().map(|write| write.bytes.len()).sum();860 let sent: usize = writes.iter().map(|write| write.bytes.len()).sum();
861 (sent <= after.len() / 2).then_some(writes)861 (sent <= after.len() / 2 && sent <= CHUNK && minicbor::to_vec(&writes).ok()?.len() <= CHUNK)
862 .then_some(writes)
862}863}
863864
864/// `image` with `writes`, `length` long: none where a write falls outside it.865/// `image` with `writes`, `length` long: none where a write falls outside it.
...@@ -2060,4 +2061,19 @@ mod tests {...@@ -2060,4 +2061,19 @@ mod tests {
2060 }];2061 }];
2061 assert!(written(&before, after.len() as u64, &outside).is_none());2062 assert!(written(&before, after.len() as u64, &outside).is_none());
2062 }2063 }
2064
2065 #[test]
2066 fn differential_reads_include_metadata_in_the_chunk_budget() {
2067 let before = vec![0; CHUNK * 4];
2068 let mut after = before.clone();
2069 after.resize(before.len() + CHUNK - 1024, 1);
2070 assert!(delta(&before, &after).is_none());
2071 after.truncate(before.len() + CHUNK / 2);
2072 let writes = delta(&before, &after).unwrap();
2073 assert!(minicbor::to_vec(&writes).unwrap().len() <= CHUNK);
2074 assert_eq!(
2075 written(&before, after.len() as u64, &writes).unwrap(),
2076 after
2077 );
2078 }
2063}2079}
crates/notebook/tests/live_share.rs+90
...@@ -305,6 +305,96 @@ fn large_files_travel_in_chunks() {...@@ -305,6 +305,96 @@ fn large_files_travel_in_chunks() {
305 );305 );
306}306}
307307
308#[test]
309fn large_section_appends_keep_the_room_connected_and_guests_converge() {
310 use notebook::{EditStatus, Replica, live::share::HostedRemote};
311 use onestore::op::{Edit, Op, PageOp};
312 use std::time::{Duration, Instant};
313
314 let directory = tempfile::tempdir().unwrap();
315 let folder = notebook(directory.path());
316 let file = folder.join("Garden.one");
317 let mut original = "A".repeat(512 << 10);
318 let image = onestore::create_section("Garden.one", &original, "Fixture").unwrap();
319 std::fs::write(&file, &image).unwrap();
320 let (space, text, _) = server::text(&image);
321 let arena = Arena::default();
322 let mut section = onestore::Section::open(&arena, image).unwrap();
323 section
324 .apply(
325 "Fixture",
326 &Edit {
327 at: 134_000_000_000_000_000,
328 ops: vec![Op::Page {
329 space,
330 op: PageOp::Text {
331 text,
332 range: 0..1,
333 with: "a".into(),
334 },
335 }],
336 },
337 )
338 .unwrap();
339 section.seal().unwrap().unwrap().commit_file(&file).unwrap();
340 original.replace_range(0..1, "a");
341 let url = relay(Default::default());
342 let sharing = Sharing::new("").unwrap();
343 let host = host(&folder, &directory.path().join("host"), &sharing, &url);
344 let code = code(&host);
345 let (alice, alice_notebook) = guest("Alice", &code, &url, &directory.path().join("alice"));
346 let (bob, bob_notebook) = guest("Bob", &code, &url, &directory.path().join("bob"));
347 until("the guests never joined the room", || {
348 host.guests().len() == 2
349 });
350 let image = bob_notebook.read_section("Garden.one").unwrap();
351 assert_eq!(alice_notebook.read_section("Garden.one").unwrap(), image);
352 let (space, text, _) = server::text(&image);
353 let writer = Replica::create(directory.path().join("writer.sqlite"), &image).unwrap();
354 let reader = Replica::create(directory.path().join("reader.sqlite"), &image).unwrap();
355 let append = "B".repeat(192 << 10);
356 let expected = format!("{original}{append}");
357 let id = writer
358 .apply(
359 "Alice",
360 Edit {
361 at: 134_000_000_000_000_000,
362 ops: vec![Op::Page {
363 space,
364 op: PageOp::Text {
365 text,
366 range: original.len() as u32..original.len() as u32,
367 with: append,
368 },
369 }],
370 },
371 )
372 .unwrap();
373 writer
374 .sync_once(&mut HostedRemote::new(&alice, "Garden.one"))
375 .unwrap();
376 let stored = std::fs::read(&file).unwrap();
377 let added = stored.len() - image.len();
378 assert!(added > relay::server::Config::default().max_message);
379 assert!(added + 1024 < stored.len() / 2);
380 assert_eq!(server::text(&stored).2, expected);
381 let Some(EditStatus::Published { revision }) = writer.status(id).unwrap() else {
382 panic!("the append has no durable receipt");
383 };
384 let store = onestore::Store::parse(&stored).unwrap();
385 let index = onestore::RevisionIndex::parse(&store).unwrap();
386 assert!(index.spaces[&space].revisions.contains_key(&revision));
387 assert!(writer.pending().unwrap().is_empty());
388 let mut remote = HostedRemote::new(&bob, "Garden.one");
389 reader.sync_once(&mut remote).unwrap();
390 assert_eq!(reader.snapshot().unwrap(), stored);
391 let deadline = Instant::now() + Duration::from_millis(1200);
392 while Instant::now() < deadline {
393 assert_eq!(host.guests().len(), 2, "the section append broke the room");
394 std::thread::sleep(Duration::from_millis(10));
395 }
396}
397
308/// A guest that floods its host with requests is hung up on once too many wait, having had398/// A guest that floods its host with requests is hung up on once too many wait, having had
309/// answers to few of them, and the host goes on serving the others.399/// answers to few of them, and the host goes on serving the others.
310#[test]400#[test]