diff --git a/CHANGELOG-actioncable.md b/CHANGELOG-actioncable.md index a2545bb4..4fa6a3ba 100644 --- a/CHANGELOG-actioncable.md +++ b/CHANGELOG-actioncable.md @@ -6,6 +6,31 @@ this project aims to follow [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.2.4] - 2026-07-01 + +Fixes from a full source review. + +### Fixed + +- **A lost-ack retry now re-broadcasts.** If the original attempt recorded the + update and then crashed (or the pub/sub broadcast failed) before + distributing, the retry was previously settled as `:applied` without + re-broadcasting — live subscribers stayed stale until their next full resync, + and nothing else could reach them. The retry now re-broadcasts before acking; + idempotent CRDT apply makes the duplicate free for every receiver. +- **A missing document key now fails closed.** Under a transport that doesn't + keep the channel instance alive across actions (AnyCable), an app that forgot + to pass `key` to `sync_receive` silently recorded updates under a nil key, + broadcast them to a stream no one subscribes to, and still acked them. The + frame now raises `Y::Error` instead. + +### Changed + +- Raised the `yrby` floor to `>= 0.3.1`, whose `update_ready?` is exact + (trial-integration, not just per-client clocks). With an older core, a + cross-client-origin gap passed the ready check and the `update_advances?` + probe then acked-and-dropped real content. + ## [0.2.3] - 2026-07-01 ### Changed diff --git a/CHANGELOG.md b/CHANGELOG.md index 0adaa762..29193479 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,45 @@ to follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +## [0.3.1] - 2026-07-01 + +Fixes from a full source review. + +### Fixed + +- **`Doc#update_ready?` is now exact.** It previously checked only the + per-client clock lower bound, but yrs's real integration gate also requires + every block referenced by an item's origin / right-origin / parent — which + routinely belong to *other* clients — and post-Skip blocks in a merged update + sit above the lower bound. An update could pass the clock check yet park as + pending; downstream, `update_advances?` then misread the parked update as an + already-applied retry (pending doesn't move a state vector) and the sync + channel **acked and dropped real content**. `update_ready?` now + trial-integrates on a throwaway probe seeded with the doc's integrated state + (the clock check remains as a cheap pre-filter), so a cross-client-origin gap + is correctly rejected for a resync. `update_advances?` also gained defense in + depth: an update that would park reports as advancing, never as a duplicate. +- **`Doc#read_text` could deadlock the process.** It opened a second read + transaction while still holding the first (a chained temporary); yrs's lock is + write-preferring, so a concurrent writer between the two acquisitions + deadlocked reader-vs-writer inside the GVL-released (uninterruptible) region. + Now uses a single transaction. +- **TOCTOU in gap-free encoding.** The pending check and the encode ran in + separate transactions, so a concurrent gappy `apply_update` between them could + make `handle_sync_message`/`compacted_state_update` serve pending structs + anyway. Both now happen under one transaction. +- `read_xml`: Lexical soft line breaks and tabs now come through as `\n`/`\t` + instead of vanishing (`"foo⏎bar"` no longer extracts as `"foobar"`). + +### Changed + +- `update_advances?` skips its full-document probe when the update carries + blocks beyond the doc's state vector (a novel update trivially advances) — + the common case no longer pays O(doc) per frame. +- The gem no longer packages the `yrby-decoder` gem's files (they ship in that + gem; the duplicate copy could shadow a newer standalone release), and now + ships `Cargo.lock` so source builds compile the exact crate graph CI tested. + ## [0.3.0] - 2026-07-01 ### Fixed diff --git a/examples/actioncable-demo/config/routes.rb b/examples/actioncable-demo/config/routes.rb index fbeadc1f..c23fae3e 100644 --- a/examples/actioncable-demo/config/routes.rb +++ b/examples/actioncable-demo/config/routes.rb @@ -10,7 +10,13 @@ get "docs/:id/forms", to: "documents#forms", as: :document_forms get "docs/:id/content", to: "documents#content", as: :document_content get "docs/:id/audit", to: "documents#audit", as: :document_audit - post "docs/:id/audit/control", to: "documents#audit_control", as: :document_audit_control + # DEMO/TEST ONLY — never mount in production. One anonymous POST can wipe a + # document's durable history (reset=1) or inject a per-write sleep (delay_ms) + # that starves the worker pool. The e2e suites depend on it, so it's gated by + # environment rather than removed. + unless Rails.env.production? + post "docs/:id/audit/control", to: "documents#audit_control", as: :document_audit_control + end root to: redirect("/docs/demo") end diff --git a/ext/yrby/src/fixtures/lexical_linebreak.bin b/ext/yrby/src/fixtures/lexical_linebreak.bin new file mode 100644 index 00000000..08c0df4d Binary files /dev/null and b/ext/yrby/src/fixtures/lexical_linebreak.bin differ diff --git a/ext/yrby/src/lib.rs b/ext/yrby/src/lib.rs index 63e22334..4c340327 100644 --- a/ext/yrby/src/lib.rs +++ b/ext/yrby/src/lib.rs @@ -160,9 +160,11 @@ impl RbDoc { fn read_text(&self, name: String) -> Option { let doc = &self.0; nogvl(move || { - doc.transact() - .get_text(name.as_str()) - .map(|t| t.get_string(&doc.transact())) + // Exactly ONE transaction per call. Opening a second while the + // first is still held deadlocks against a waiting writer — and + // inside nogvl that hang can't be interrupted. + let txn = doc.transact(); + txn.get_text(name.as_str()).map(|t| t.get_string(&txn)) }) } diff --git a/ext/yrby/src/protocol.rs b/ext/yrby/src/protocol.rs index 80dde83d..ef90bb79 100644 --- a/ext/yrby/src/protocol.rs +++ b/ext/yrby/src/protocol.rs @@ -66,14 +66,42 @@ pub(crate) fn merged_doc_update(bytes: &[u8]) -> Result>, String> Ok(Some(merged)) } -/// True if applying `update_bytes` to `doc` would integrate cleanly: every -/// dependency the update references is already present (the doc's state vector -/// covers the update's lower bound). A pure read; does not mutate the doc. -/// When false, applying it would park a pending struct, the signal that an -/// earlier, causally-prior update is missing. +/// True if applying `update_bytes` to `doc` would integrate cleanly; false if +/// it would park as pending (a causally-prior update is missing). A pure read. +/// +/// This must be EXACT: the sync layer records on "ready" and resyncs on "not +/// ready", and a parked update that slipped through would look like an +/// already-applied retry downstream — acked and dropped, losing real content. +/// +/// Clocks alone can't decide it. An update can satisfy every per-client clock +/// and still fail to integrate: its items may reference other clients' blocks +/// (origins/parents), and merged updates hide internal gaps behind Skip blocks. +/// So the clock lower bound serves only as a cheap definitive REJECT; "ready" +/// is decided by trial-integrating on a throwaway probe seeded with the doc's +/// integrated state — ready iff nothing parks. pub(crate) fn update_is_ready(doc: &Doc, update_bytes: &[u8]) -> Result { let update = yrs::Update::decode_v1(update_bytes).map_err(|e| e.to_string())?; - Ok(doc.transact().state_vector() >= update.state_vector_lower()) + // Partial order: "not covered" includes incomparable — not ready either way. + let lower_covered = doc.transact().state_vector() >= update.state_vector_lower(); + if !lower_covered { + return Ok(false); + } + // Seed the probe with the doc's INTEGRATED state (gap-free), for two + // reasons. A lossless seed would replant the doc's own pre-existing + // pending in the probe, making has_pending true for EVERY update — the + // verdict must be about this update, not the doc's baggage. And an update + // whose dependency exists only in that pending buffer is genuinely not + // ready: recording it would put a gap in the durable log; a resync heals + // both it and the pending it leans on as one complete delta. + let seed = integrated_update(doc, &StateVector::default())?; + let probe = Doc::new(); + { + let mut txn = probe.transact_mut(); + txn.apply_update(Update::decode_v1(&seed).map_err(|e| e.to_string())?) + .map_err(|e| e.to_string())?; + txn.apply_update(update).map_err(|e| e.to_string())?; + } + Ok(!has_pending(&probe)) } /// True if applying `update_bytes` would actually change `doc`, i.e. it carries @@ -107,6 +135,16 @@ pub(crate) fn update_advances_doc(doc: &Doc, update_bytes: &[u8]) -> Result= update.state_vector(); + if !covered { + return Ok(true); + } + } + // Seed an independent probe with the doc's current state so we can measure the // update's effect without mutating the real doc. let probe = Doc::new(); @@ -117,6 +155,10 @@ pub(crate) fn update_advances_doc(doc: &Doc, update_bytes: &[u8]) -> Result Result bool { /// Non-destructive: the prune happens only on the throwaway copy; `doc` keeps its /// pending, so a genuine gap still heals if its missing dependency later arrives. pub(crate) fn integrated_update(doc: &Doc, sv: &StateVector) -> Result, String> { - // Fast path: with nothing pending the direct encode is already gap-free, so - // the clean common case keeps the zero-copy behavior. - if !has_pending(doc) { - return Ok(doc.transact().encode_state_as_update_v1(sv)); - } - let full = doc - .transact() - .encode_state_as_update_v1(&StateVector::default()); + // Pending check and encode share ONE transaction — with two, a concurrent + // gappy apply_update could slip between them and the encode would serve + // the very pending this function exists to exclude. + let full = { + let txn = doc.transact(); + let store = txn.store(); + // Nothing pending: the direct encode is already gap-free. + if store.pending_update().is_none() && store.pending_ds().is_none() { + return Ok(txn.encode_state_as_update_v1(sv)); + } + txn.encode_state_as_update_v1(&StateVector::default()) + }; let clean = Doc::new(); { let mut txn = clean.transact_mut(); @@ -464,6 +523,146 @@ mod tests { assert!(!has_pending(&doc), "u2 arrived; u3 integrated"); } + // Build a cross-client-origin gap: client C creates "abc"; client A applies + // it and types between C's characters, so A's delta references C's blocks as + // origins. Returns (c_update, a_delta). On a doc missing `c_update`, the + // per-client clock lower bound of `a_delta` is satisfied (A starts at clock + // 0) but integration parks — the case a clock-only readiness check misses. + fn cross_client_origin_gap() -> (Vec, Vec) { + let c = Doc::new(); + let ct = c.get_or_insert_text("t"); + ct.insert(&mut c.transact_mut(), 0, "abc"); + let c_update = c + .transact() + .encode_state_as_update_v1(&yrs::StateVector::default()); + + let a = Doc::new(); + a.transact_mut() + .apply_update(yrs::Update::decode_v1(&c_update).unwrap()) + .unwrap(); + let sv_before = a.transact().state_vector(); + let at = a.get_or_insert_text("t"); + at.insert(&mut a.transact_mut(), 1, "X"); // between C's chars + let a_delta = a.transact().encode_state_as_update_v1(&sv_before); + (c_update, a_delta) + } + + #[test] + fn cross_client_origin_gap_is_not_ready() { + let (c_update, a_delta) = cross_client_origin_gap(); + + // A server that never saw C's content: the clock lower bound passes, but + // the update can't integrate — it must NOT be ready (previously it was, + // and the downstream advances? probe then acked-and-dropped it). + let server = Doc::new(); + assert!( + !update_is_ready(&server, &a_delta).unwrap(), + "a delta with unmet cross-client origins is not ready" + ); + + // Once the server has C's content, the same delta is ready and advances. + server + .transact_mut() + .apply_update(yrs::Update::decode_v1(&c_update).unwrap()) + .unwrap(); + assert!(update_is_ready(&server, &a_delta).unwrap()); + assert!(update_advances_doc(&server, &a_delta).unwrap()); + } + + #[test] + fn merged_update_with_internal_skip_gap_is_not_ready() { + // Merging u1 and u3 (u2 missing) yields one update with a Skip block; its + // clock lower bound is u1's start, but the post-Skip blocks can't + // integrate on a doc that lacks u2. + let src = Doc::new(); + let txt = src.get_or_insert_text("t"); + let mut deltas: Vec> = Vec::new(); + let mut prev = yrs::StateVector::default(); + for (i, ch) in ["A", "B", "C"].into_iter().enumerate() { + txt.insert(&mut src.transact_mut(), i as u32, ch); + deltas.push(src.transact().encode_state_as_update_v1(&prev)); + prev = src.transact().state_vector(); + } + let merged = yrs::merge_updates_v1([deltas[0].as_slice(), deltas[2].as_slice()]).unwrap(); + + let server = Doc::new(); + assert!( + !update_is_ready(&server, &merged).unwrap(), + "the post-Skip blocks depend on the missing u2" + ); + } + + #[test] + fn a_doc_with_legacy_pending_still_accepts_healthy_updates() { + // Why update_is_ready seeds its probe with the INTEGRATED state: with a + // lossless seed, the doc's own pre-existing pending would park in the + // probe and every verdict would come back "not ready" — a server with + // one legacy gap would reject every healthy keystroke forever. + let (_first, dependent) = gap_pair(); + let doc = Doc::new(); + doc.transact_mut() + .apply_update(yrs::Update::decode_v1(&dependent).unwrap()) + .unwrap(); + assert!(has_pending(&doc), "the doc carries a legacy parked gap"); + + // A healthy, self-contained update from an unrelated client. + let healthy = { + let d = Doc::new(); + let t = d.get_or_insert_text("other"); + t.insert(&mut d.transact_mut(), 0, "hello"); + let txn = d.transact(); + txn.encode_state_as_update_v1(&yrs::StateVector::default()) + }; + + assert!( + update_is_ready(&doc, &healthy).unwrap(), + "legacy pending must not veto unrelated healthy updates" + ); + assert!(update_advances_doc(&doc, &healthy).unwrap()); + } + + #[test] + fn an_update_depending_only_on_pending_content_is_not_ready() { + // The other half of the integrated-only seed: a dependency that exists + // solely in the doc's pending buffer doesn't count — recording such an + // update would put a gap in the durable log. Not ready; resync heals + // both as one complete delta. + let src = Doc::new(); + let txt = src.get_or_insert_text("t"); + let mut deltas: Vec> = Vec::new(); + let mut prev = yrs::StateVector::default(); + for (i, ch) in ["A", "B", "C"].into_iter().enumerate() { + txt.insert(&mut src.transact_mut(), i as u32, ch); + deltas.push(src.transact().encode_state_as_update_v1(&prev)); + prev = src.transact().state_vector(); + } + + // The doc holds u2 only as PENDING (u1 never arrived); u3 depends on u2. + let doc = Doc::new(); + doc.transact_mut() + .apply_update(yrs::Update::decode_v1(&deltas[1]).unwrap()) + .unwrap(); + assert!(has_pending(&doc), "u2 parked without u1"); + + assert!( + !update_is_ready(&doc, &deltas[2]).unwrap(), + "a dependency satisfied only by pending content is not ready" + ); + } + + #[test] + fn update_advances_reports_true_when_the_update_would_park() { + // Defense in depth for callers using advances? without the ready gate: a + // gappy update parks pending — that changes the doc, so it advances (it + // must never be misread as an already-applied retry and dropped). + let (_c_update, a_delta) = cross_client_origin_gap(); + let server = Doc::new(); + assert!( + update_advances_doc(&server, &a_delta).unwrap(), + "a parked update is not a duplicate" + ); + } + // Build a causal gap: `first` inserts "a", `dependent` inserts "b" after it, // so `dependent` alone parks as pending on a doc that lacks `first`. fn gap_pair() -> (Vec, Vec) { @@ -627,6 +826,58 @@ mod tests { assert!(!has_pending(&peer), "the diff carried no pending"); } + #[test] + fn integrated_update_never_serves_pending_under_concurrent_gappy_applies() { + // Invariant under contention: while a writer parks and heals a gappy + // update in a loop, every integrated_update encode must be pending-free + // for a fresh peer. + // + // Scope: this can't hit the original check-vs-encode race (its window + // is nanoseconds; never reproduced even at 20k iterations) — that fix + // is guaranteed by using a single transaction. What this catches is + // coarser: encoding outside the lock, or a fast path skipping the + // pending check. + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc as StdArc; + + let (first, dependent) = gap_pair(); + let doc = StdArc::new(Doc::new()); + let stop = StdArc::new(AtomicBool::new(false)); + + let writer = { + let doc = StdArc::clone(&doc); + let stop = StdArc::clone(&stop); + let dependent = dependent.clone(); + let first = first.clone(); + std::thread::spawn(move || { + while !stop.load(Ordering::Relaxed) { + // Park a pending struct, then heal it, over and over — the + // encode below keeps racing both transitions. + doc.transact_mut() + .apply_update(yrs::Update::decode_v1(&dependent).unwrap()) + .unwrap(); + doc.transact_mut() + .apply_update(yrs::Update::decode_v1(&first).unwrap()) + .unwrap(); + } + }) + }; + + for _ in 0..500 { + let encoded = integrated_update(&doc, &yrs::StateVector::default()).unwrap(); + let peer = Doc::new(); + peer.transact_mut() + .apply_update(yrs::Update::decode_v1(&encoded).unwrap()) + .unwrap(); + assert!( + !has_pending(&peer), + "an integrated_update encode leaked pending to a peer" + ); + } + stop.store(true, Ordering::Relaxed); + writer.join().unwrap(); + } + #[test] fn integrated_update_strips_a_pending_delete_set() { // A deletion whose target struct is absent parks as a pending *delete diff --git a/ext/yrby/src/read.rs b/ext/yrby/src/read.rs index 67764edb..fa25a543 100644 --- a/ext/yrby/src/read.rs +++ b/ext/yrby/src/read.rs @@ -62,6 +62,17 @@ fn lexical_type(txn: &T, t: &XmlTextRef) -> String { } } +/// The `__type` of an embedded Lexical `Y.Map`. Two kinds appear inside a +/// block: text-node metadata (`"text"`) and node maps like the LineBreakNode +/// (`"linebreak"`). Structure confirmed from live-editor bytes (see the +/// captured-fixture test). +fn lexical_map_type(txn: &T, m: &MapRef) -> String { + match m.get(txn, "__type") { + Some(Out::Any(Any::String(s))) => s.to_string(), + _ => String::new(), + } +} + /// Gather the text of an inline Lexical element (its text runs and any nested /// inline elements) without introducing block breaks. fn inline_lexical_text(txn: &T, t: &XmlTextRef, buf: &mut String) { @@ -69,7 +80,12 @@ fn inline_lexical_text(txn: &T, t: &XmlTextRef, buf: &mut String) { match d.insert { Out::Any(Any::String(s)) => buf.push_str(&s), Out::YXmlText(child) => inline_lexical_text(txn, &child, buf), - _ => {} // per-text-node metadata map, decorator embeds: no text + Out::YMap(m) => match lexical_map_type(txn, &m).as_str() { + "linebreak" => buf.push('\n'), + "tab" => buf.push('\t'), + _ => {} // per-text-node metadata: no text of its own + }, + _ => {} // decorator embeds: no text } } } @@ -81,17 +97,30 @@ fn walk_lexical_block(txn: &T, t: &XmlTextRef, out: &mut Vec for d in t.diff(txn, YChange::identity) { match d.insert { Out::Any(Any::String(s)) => line.push_str(&s), + // Node maps: linebreak/tab carry no text, so emit the character + // they represent ("foo⏎bar" must not become "foobar"). Metadata + // maps ("text") stay silent. + Out::YMap(m) => match lexical_map_type(txn, &m).as_str() { + "linebreak" => line.push('\n'), + "tab" => line.push('\t'), + _ => {} + }, Out::YXmlText(child) => { - if is_inline_lexical_type(&lexical_type(txn, &child)) { - inline_lexical_text(txn, &child, &mut line); - } else { - if !line.is_empty() { - out.push(std::mem::take(&mut line)); + let ty = lexical_type(txn, &child); + match ty.as_str() { + // Defensive only: real Lexical stores these as Y.Map embeds. + "linebreak" => line.push('\n'), + "tab" => line.push('\t'), + _ if is_inline_lexical_type(&ty) => inline_lexical_text(txn, &child, &mut line), + _ => { + if !line.is_empty() { + out.push(std::mem::take(&mut line)); + } + walk_lexical_block(txn, &child, out); } - walk_lexical_block(txn, &child, out); } } - _ => {} // per-text-node metadata map; embeds we don't read for text + _ => {} // decorator embeds we don't read for text } } if !line.is_empty() { @@ -256,6 +285,53 @@ mod tests { assert_eq!(map_json(&txn, &map), "{}"); } + #[test] + fn lexical_soft_line_break_and_tab_emit_their_characters() { + // A paragraph "foo⏎bar" (shift-enter): Lexical stores the LineBreakNode + // as an embedded Y.Map with __type=linebreak (the same shape as the + // per-text-node metadata maps, which must stay silent). It must come + // through as '\n', not vanish and glue the words. Same for tab. + use yrs::{Text, XmlTextPrelim}; + let doc = Doc::new(); + let frag = doc.get_or_insert_xml_fragment("lex"); + { + let mut txn = doc.transact_mut(); + let block = frag.push_back(&mut txn, XmlTextPrelim::new("")); + let meta: MapPrelim = [("__type", yrs::In::from("text"))].into_iter().collect(); + block.insert_embed(&mut txn, 0, meta); // metadata map: no text + block.push(&mut txn, "foo"); + let br: MapPrelim = [("__type", yrs::In::from("linebreak"))] + .into_iter() + .collect(); + block.insert_embed(&mut txn, 4, br); + block.push(&mut txn, "bar"); + let tab: MapPrelim = [("__type", yrs::In::from("tab"))].into_iter().collect(); + block.insert_embed(&mut txn, 8, tab); + block.push(&mut txn, "baz"); + } + let txn = doc.transact(); + assert_eq!(xml_blocks_text(&txn, &frag), "foo\nbar\tbaz"); + } + + #[test] + fn lexical_real_captured_linebreak_extracts_as_newline() { + // Ground truth: bytes captured from a LIVE Lexxy editor (agent-browser + // typing "foo", pressing Shift+Enter, typing "barbaz"), served by the + // yrby test server's durable store. The hand-built test above models + // this structure; this one IS the structure. Regenerate by driving + // lexxy-realtime's test server and saving GET /content/:room. + use yrs::updates::decoder::Decode; + use yrs::Update; + let bytes = include_bytes!("fixtures/lexical_linebreak.bin"); + let doc = Doc::new(); + doc.transact_mut() + .apply_update(Update::decode_v1(bytes).unwrap()) + .unwrap(); + let txn = doc.transact(); + let frag = txn.get_xml_fragment("root").unwrap(); + assert_eq!(xml_blocks_text(&txn, &frag), "foo\nbarbaz"); + } + #[test] fn lexical_complex_doc_extracts_all_nested_text() { // A real Lexxy/Lexical doc with every block type: headings, formatted diff --git a/lib/y/action_cable/sync.rb b/lib/y/action_cable/sync.rb index 2e54fdeb..92c48c58 100644 --- a/lib/y/action_cable/sync.rb +++ b/lib/y/action_cable/sync.rb @@ -240,6 +240,20 @@ def sync_validate_required_hooks! "that never happened, and a cold load would lose the edit." end + # Fail closed when no document key is set (typically: AnyCable rebuilt the + # channel instance and the app forgot to pass `key` to sync_receive). + # Proceeding would record under nil, broadcast to a stream nobody + # subscribes to, and still ack — the client believes the edit was + # delivered when it reached no one. + def sync_validate_key! + return unless @sync_key.nil? || @sync_key.empty? + + raise Y::Error, + "Y::ActionCable::Sync has no document key. Call sync_subscribed(key) in " \ + "subscribed, and pass the key to sync_receive(data, key) when the transport " \ + "doesn't keep the channel instance alive across actions (e.g. AnyCable)." + end + # Stateless per message: any process can handle any document. A client's # SyncStep1 is answered from the store, document changes are recorded durably # before relay and then broadcast, and awareness is relayed best-effort. @@ -250,6 +264,7 @@ def sync_validate_required_hooks! # rejected for a resync, :noop for everything else. def sync_handle_frame(encoded, bytes) sync_validate_required_hooks! + sync_validate_key! case Y.message_kind(bytes) when MSG_KIND_SYNC_STEP1 @@ -271,9 +286,14 @@ def sync_handle_frame(encoded, bytes) return :gap end - # Skip a lost-ack retry the store already has. Best-effort, not - # cross-process exactly-once (see "Delivery guarantees" in the README). - return :applied unless doc.update_advances?(update) + # A lost-ack retry: already recorded, so skip on_change — but DO + # re-broadcast. If the first attempt died between record and broadcast, + # this retry is the only path left to the live subscribers. Duplicate + # broadcasts are free (CRDT apply is idempotent). + unless doc.update_advances?(update) + sync_distribute(encoded) + return :applied + end sync_record_change(update) # record before relay sync_distribute(encoded) diff --git a/lib/y/action_cable/version.rb b/lib/y/action_cable/version.rb index 530d1b31..98b3f81d 100644 --- a/lib/y/action_cable/version.rb +++ b/lib/y/action_cable/version.rb @@ -2,6 +2,6 @@ module Y module ActionCable - VERSION = "0.2.3" + VERSION = "0.2.4" end end diff --git a/lib/y/version.rb b/lib/y/version.rb index d0cae481..1c509d29 100644 --- a/lib/y/version.rb +++ b/lib/y/version.rb @@ -1,5 +1,5 @@ # frozen_string_literal: true module Y - VERSION = "0.3.0" + VERSION = "0.3.1" end diff --git a/packages/client/package.json b/packages/client/package.json index 4c49dc82..02af1e25 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -1,6 +1,6 @@ { "name": "yrby-client", - "version": "0.4.2", + "version": "0.4.3", "description": "JavaScript client for the yrby y-websocket protocol: a ready-made ActionCable/AnyCable provider, a transport-agnostic protocol session (sync steps, encode/decode, awareness), and a reliable-delivery core (ack-tracked queue, sync-since-last-ack, retransmit + reconnect replay). Written in TypeScript with bundled types; ESM + CommonJS, usable from plain JS.", "type": "module", "main": "./dist/cjs/index.js", @@ -8,19 +8,34 @@ "types": "./dist/index.d.ts", "exports": { ".": { - "types": "./dist/index.d.ts", - "import": "./dist/index.js", - "require": "./dist/cjs/index.js" + "import": { + "types": "./dist/index.d.ts", + "default": "./dist/index.js" + }, + "require": { + "types": "./dist/cjs/index.d.ts", + "default": "./dist/cjs/index.js" + } }, "./reliable": { - "types": "./dist/reliable_sync.d.ts", - "import": "./dist/reliable_sync.js", - "require": "./dist/cjs/reliable_sync.js" + "import": { + "types": "./dist/reliable_sync.d.ts", + "default": "./dist/reliable_sync.js" + }, + "require": { + "types": "./dist/cjs/reliable_sync.d.ts", + "default": "./dist/cjs/reliable_sync.js" + } }, "./base64": { - "types": "./dist/base64.d.ts", - "import": "./dist/base64.js", - "require": "./dist/cjs/base64.js" + "import": { + "types": "./dist/base64.d.ts", + "default": "./dist/base64.js" + }, + "require": { + "types": "./dist/cjs/base64.d.ts", + "default": "./dist/cjs/base64.js" + } } }, "files": [ diff --git a/packages/client/src/actioncable_provider.ts b/packages/client/src/actioncable_provider.ts index e89871a9..d9a7da0e 100644 --- a/packages/client/src/actioncable_provider.ts +++ b/packages/client/src/actioncable_provider.ts @@ -78,6 +78,8 @@ export class ActionCableProvider { #status: ProviderStatus = "disconnected"; #statusListeners = new Set<(event: StatusEvent) => void>(); #onUnload: (() => void) | null = null; + #onRestore: ((event: PageTransitionEvent) => void) | null = null; + #stashedPresence: Record | null = null; constructor( doc: Doc, @@ -180,6 +182,13 @@ export class ActionCableProvider { provider.session.onDisconnect(); // pause retransmits, clear remote presence provider.#refreshStatus(); // subscription still set -> "connecting" (retrying) }, + rejected() { + // The channel refused the subscription (auth, missing doc). Surface + // it and tear down — otherwise the provider sits at "connecting" + // forever, silently queueing edits. The app decides what's next. + provider.#onError(new Error("subscription rejected by the server"), "rejected"); + provider.disconnect(); + }, } ); this.#installUnloadHandler(); @@ -225,20 +234,39 @@ export class ActionCableProvider { for (const listener of this.#statusListeners) listener({ status: next }); } - // Best-effort presence removal when the tab/page goes away (close, navigation, - // bfcache). `pagehide` fires while the socket is still live and is bfcache-safe - // (unlike `beforeunload`, which can block it). Sends are not guaranteed to - // flush on unload, so the server-side awareness timeout remains the backstop. + // Presence teardown/restore around page lifecycle: + // - `pagehide`: remove local presence while the socket is still live so peers + // drop our cursor now (bfcache-safe; the awareness timeout is the backstop). + // - `pageshow` with `persisted`: the user came BACK (bfcache restore), so put + // their presence back — editors set awareness once at setup, so without + // this they'd rejoin as a ghost with no cursor. #installUnloadHandler(): void { if (typeof window === "undefined" || this.#onUnload) return; - this.#onUnload = () => this.session.removeLocalAwareness(); + this.#onUnload = () => { + this.#stashedPresence = this.awareness.getLocalState(); + this.session.removeLocalAwareness(); + }; + this.#onRestore = (event: PageTransitionEvent) => { + if (!event.persisted || !this.#stashedPresence) return; + if (this.awareness.getLocalState() === null) { + this.awareness.setLocalState(this.#stashedPresence); + } + this.#stashedPresence = null; + }; window.addEventListener("pagehide", this.#onUnload); + window.addEventListener("pageshow", this.#onRestore); } #removeUnloadHandler(): void { - if (typeof window === "undefined" || !this.#onUnload) return; - window.removeEventListener("pagehide", this.#onUnload); - this.#onUnload = null; + if (typeof window === "undefined") return; + if (this.#onUnload) { + window.removeEventListener("pagehide", this.#onUnload); + this.#onUnload = null; + } + if (this.#onRestore) { + window.removeEventListener("pageshow", this.#onRestore); + this.#onRestore = null; + } } // Send one raw protocol frame over the cable. Awareness frames are whispered @@ -251,11 +279,27 @@ export class ActionCableProvider { if (!sub) return; const update = toBase64(frame); const isAwareness = frame[0] === MessageType.Awareness; - if (isAwareness && typeof sub.whisper === "function") { - sub.whisper({ awareness: update }); - return; + // Route transport failures (sync throws, or @anycable/web's rejected + // promises) to onError instead of letting them escape into update + // handlers. A failed send is recoverable: reliable frames stay queued + // until acked, and awareness is best-effort anyway. + try { + if (isAwareness && typeof sub.whisper === "function") { + this.#observe(sub.whisper({ awareness: update })); + return; + } + const payload = id === undefined ? { update } : { update, id }; + this.#observe(sub.send(payload)); + } catch (error) { + this.#onError(error, "send"); + } + } + + // Attach a rejection handler when a transport returns a promise, so failures + // surface via onError instead of as unhandled rejections. + #observe(result: unknown): void { + if (result instanceof Promise) { + result.catch((error) => this.#onError(error, "send")); } - const payload = id === undefined ? { update } : { update, id }; - sub.send(payload); } } diff --git a/packages/client/src/y_protocol_session.ts b/packages/client/src/y_protocol_session.ts index 2ffef1d2..149ec2f1 100644 --- a/packages/client/src/y_protocol_session.ts +++ b/packages/client/src/y_protocol_session.ts @@ -248,7 +248,25 @@ export class YProtocolSession { break; } case MessageType.Awareness: - decoding.readVarUint8Array(decoder); + // Validate the payload's CONTENTS, not just the envelope. + // applyAwarenessUpdate mutates state entry by entry and only notifies + // listeners at the end — a bad entry mid-payload would leave earlier + // entries applied with no event fired. Dry-running every entry here + // makes the real apply infallible (and catches trailing garbage + // inside the blob). + { + const payload = decoding.readVarUint8Array(decoder); + const inner = decoding.createDecoder(payload); + const count = decoding.readVarUint(inner); + for (let i = 0; i < count; i++) { + decoding.readVarUint(inner); // clientID + decoding.readVarUint(inner); // clock + JSON.parse(decoding.readVarString(inner)); // state (null on removal) + } + if (decoding.hasContent(inner)) { + throw new Error("awareness payload has trailing bytes"); + } + } break; default: return null; // a y-protocols type yrby doesn't speak: ignore diff --git a/packages/client/test/actioncable_provider.test.js b/packages/client/test/actioncable_provider.test.js index 531d67a4..e6c6e4ce 100644 --- a/packages/client/test/actioncable_provider.test.js +++ b/packages/client/test/actioncable_provider.test.js @@ -15,6 +15,7 @@ function fakeConsumer({ withWhisper } = { withWhisper: false }) { calls, deliverConnected: () => sub.connected(), deliverDisconnected: () => sub.disconnected(), + deliverRejected: () => sub.rejected(), deliverReceived: (msg) => sub.received(msg), // No `subscriptions.remove` -- mirrors @anycable/web (which has none). Teardown // goes through the subscription's own unsubscribe(), the universal path. @@ -280,3 +281,112 @@ test("received awareness envelope rejects non-awareness frames", (t) => { assert.equal(errors[0].context, "received"); assert.match(String(errors[0].err?.message ?? errors[0].err), /non-awareness/); }); + +test("a rejected subscription surfaces via onError and tears down (no infinite 'connecting')", (t) => { + const c = fakeConsumer(); + const errors = []; + const p = makeProvider(t, new Y.Doc(), c, { id: "rej" }, { onError: (err, context) => errors.push({ err, context }) }); + const statuses = []; + p.onStatusChange(({ status }) => statuses.push(status)); + + p.connect(); + c.deliverRejected(); + + assert.equal(errors.length, 1, "rejection is reported"); + assert.equal(errors[0].context, "rejected"); + assert.equal(p.status, "disconnected", "provider tears down instead of hanging at 'connecting'"); + assert.deepEqual(statuses, ["connecting", "disconnected"]); +}); + +test("a throwing transport send is reported, not thrown into update handlers", (t) => { + const c = fakeConsumer(); + const errors = []; + const doc = new Y.Doc(); + const p = makeProvider(t, doc, c, { id: "boom" }, { onError: (err, context) => errors.push({ err, context }) }); + p.connect(); + c.deliverConnected(); + // Sabotage the transport AFTER connect so the handshake went out normally. + const sub = c.subscriptions.create; // (fakeConsumer keeps `sub` internal; sabotage via calls) + c.calls.send.length = 0; + const brokenSend = new Error("socket gone"); + // Replace send on the live subscription through a received-side effect: easiest + // is to monkey-patch through the consumer's stored sub via deliver* closure. + // fakeConsumer exposes no direct handle, so patch through the provider's edit path: + // make every push throw by redefining the calls array's push. + c.calls.send.push = () => { + throw brokenSend; + }; + + doc.getText("t").insert(0, "x"); // triggers a reliable send through the broken transport + + assert.ok(errors.some((e) => e.context === "send" && e.err === brokenSend), "send failure surfaced via onError"); + assert.ok(p.hasPending, "the edit stays queued for retransmit despite the failed send"); +}); + +test("a promise-rejecting transport send surfaces via onError (no unhandled rejection)", async (t) => { + const c = fakeConsumer(); + const errors = []; + const doc = new Y.Doc(); + const p = makeProvider(t, doc, c, { id: "rejp" }, { onError: (err, context) => errors.push({ err, context }) }); + p.connect(); + c.deliverConnected(); + c.calls.send.length = 0; + const rejection = new Error("async transport failure"); + c.calls.send.push = () => Promise.reject(rejection); + // #send observes the transport's return value; fake it by returning from push + // (the fake sub's send returns calls.send.push(...)'s result). + + doc.getText("t").insert(0, "y"); + await new Promise((resolve) => setTimeout(resolve, 0)); // let the rejection propagate + + assert.ok(errors.some((e) => e.context === "send" && e.err === rejection), "promise rejection observed via onError"); +}); + +test("bfcache: presence is stashed on pagehide and restored on pageshow(persisted)", (t) => { + // Shim a window so the unload/restore handlers install in node. + const listeners = new Map(); + globalThis.window = { + addEventListener: (name, fn) => listeners.set(name, fn), + removeEventListener: (name) => listeners.delete(name), + }; + t.after(() => { + delete globalThis.window; + }); + + const c = fakeConsumer(); + const p = makeProvider(t, new Y.Doc(), c, { id: "bf" }); + p.connect(); + c.deliverConnected(); + p.awareness.setLocalStateField("user", "alice"); + assert.deepEqual(p.awareness.getLocalState(), { user: "alice" }); + + // The page goes into the bfcache: presence is removed (peers drop our cursor). + listeners.get("pagehide")(); + assert.equal(p.awareness.getLocalState(), null, "pagehide removed local presence"); + + // Restored from the bfcache: presence must come back, or the returning user + // rejoins as a ghost (editor bindings only set awareness once at setup). + listeners.get("pageshow")({ persisted: true }); + assert.deepEqual(p.awareness.getLocalState(), { user: "alice" }, "pageshow(persisted) restored presence"); +}); + +test("bfcache: a non-persisted pageshow (normal load) does not resurrect stale presence", (t) => { + const listeners = new Map(); + globalThis.window = { + addEventListener: (name, fn) => listeners.set(name, fn), + removeEventListener: (name) => listeners.delete(name), + }; + t.after(() => { + delete globalThis.window; + }); + + const c = fakeConsumer(); + const p = makeProvider(t, new Y.Doc(), c, { id: "bf2" }); + p.connect(); + c.deliverConnected(); + p.awareness.setLocalStateField("user", "bob"); + listeners.get("pagehide")(); + listeners.get("pageshow")({ persisted: false }); // a fresh navigation, not a restore + + assert.equal(p.awareness.getLocalState(), null, "no restore on a normal load"); +}); diff --git a/packages/client/test/y_protocol_session.test.js b/packages/client/test/y_protocol_session.test.js index f31b9a0e..e39cb0d1 100644 --- a/packages/client/test/y_protocol_session.test.js +++ b/packages/client/test/y_protocol_session.test.js @@ -335,3 +335,66 @@ test("awareness: applying a remote update does NOT echo it back out (origin guar awA.destroy(); awB.destroy(); }); + +test("receive: a partially-malformed awareness payload mutates nothing (no half-applied entries)", () => { + const doc = new Y.Doc(); + const awareness = new Awareness(doc); + const errors = []; + const session = new YProtocolSession(doc, { + awareness, + send: () => {}, + onError: (err) => errors.push(err), + }); + + // Craft an awareness payload with TWO entries: entry 0 valid, entry 1 carrying + // invalid JSON. Without content validation, applyAwarenessUpdate would apply + // entry 0, then throw on entry 1 with no event ever fired -- state mutated, + // listeners never told. + const inner = encoding.createEncoder(); + encoding.writeVarUint(inner, 2); // two entries + encoding.writeVarUint(inner, 4242); // clientID + encoding.writeVarUint(inner, 1); // clock + encoding.writeVarString(inner, JSON.stringify({ user: "alice" })); // valid + encoding.writeVarUint(inner, 4343); + encoding.writeVarUint(inner, 1); + encoding.writeVarString(inner, "{"); // invalid JSON + const frame = encoding.createEncoder(); + encoding.writeVarUint(frame, MessageType.Awareness); + encoding.writeVarUint8Array(frame, encoding.toUint8Array(inner)); + + const reply = session.receive(encoding.toUint8Array(frame)); + + assert.equal(reply, null); + assert.equal(errors.length, 1, "the malformed payload is reported"); + assert.equal(awareness.getStates().has(4242), false, "the valid entry was NOT half-applied"); + session.destroy(); + awareness.destroy(); +}); + +test("receive: an awareness payload with trailing bytes inside the blob is rejected", () => { + const doc = new Y.Doc(); + const awareness = new Awareness(doc); + const errors = []; + const session = new YProtocolSession(doc, { + awareness, + send: () => {}, + onError: (err) => errors.push(err), + }); + + const inner = encoding.createEncoder(); + encoding.writeVarUint(inner, 1); + encoding.writeVarUint(inner, 777); + encoding.writeVarUint(inner, 1); + encoding.writeVarString(inner, JSON.stringify({ user: "eve" })); + const padded = new Uint8Array([...encoding.toUint8Array(inner), 0xde, 0xad]); // garbage inside the blob + const frame = encoding.createEncoder(); + encoding.writeVarUint(frame, MessageType.Awareness); + encoding.writeVarUint8Array(frame, padded); + + session.receive(encoding.toUint8Array(frame)); + + assert.equal(errors.length, 1, "trailing bytes inside the awareness blob are rejected"); + assert.equal(awareness.getStates().has(777), false, "nothing was applied"); + session.destroy(); + awareness.destroy(); +}); diff --git a/packages/client/tsconfig.cjs.json b/packages/client/tsconfig.cjs.json index 12a6d607..45a23993 100644 --- a/packages/client/tsconfig.cjs.json +++ b/packages/client/tsconfig.cjs.json @@ -4,7 +4,7 @@ "module": "CommonJS", "moduleResolution": "Node", "outDir": "dist/cjs", - "declaration": false, + "declaration": true, "declarationMap": false, "sourceMap": false } diff --git a/test/fixtures/generate_fixtures.mjs b/test/fixtures/generate_fixtures.mjs index 869c8bcb..4e9e5cbb 100644 --- a/test/fixtures/generate_fixtures.mjs +++ b/test/fixtures/generate_fixtures.mjs @@ -113,6 +113,26 @@ const pendingDelete = (() => { return b64(Y.encodeStateAsUpdate(doc, sv)) // deletion only })() +// Fixture 11: a cross-client-origin gap. Client 3 creates "abc"; client 1 +// applies it and types between client 3's characters, so client 1's delta +// references client 3's blocks as origins. On a doc that lacks CONTENT, the +// per-client clock lower bound of DELTA passes (client 1 starts at clock 0) but +// integration parks -- the readiness case a clock-only check misses. +const crossClientOrigin = (() => { + const c = new Y.Doc() + c.clientID = 3 + c.getText("t").insert(0, "abc") + const content = Y.encodeStateAsUpdate(c) + + const a = new Y.Doc() + a.clientID = 1 + Y.applyUpdate(a, content) + const sv = Y.encodeStateVector(a) + a.getText("t").insert(1, "X") // between client 3's chars + const delta = Y.encodeStateAsUpdate(a, sv) // only client 1's block + return { content: b64(content), delta: b64(delta) } +})() + // Fixture 8: a deletion delivered as its own delta. Insert "hello" (client 1), // snapshot that state, then delete the first char and capture the incremental // update. The deletion diff carries only a delete set (no new structs), so @@ -252,6 +272,16 @@ module YjsFixtures module PendingDelete UPDATE = YjsFixtures.b64("${pendingDelete}") end + + # Fixture 11: a cross-client-origin gap. CONTENT is client 3's "abc"; DELTA is + # client 1's insert BETWEEN client 3's characters, so its origins reference + # client 3's blocks. On a doc lacking CONTENT, DELTA's per-client clock lower + # bound passes but integration parks -- the readiness case a clock-only check + # misses (update_ready? must say false). + module CrossClientOrigin + CONTENT = YjsFixtures.b64("${crossClientOrigin.content}") + DELTA = YjsFixtures.b64("${crossClientOrigin.delta}") + end end ` diff --git a/test/fixtures/yjs_fixtures.rb b/test/fixtures/yjs_fixtures.rb index e910f1a1..eaca7ace 100644 --- a/test/fixtures/yjs_fixtures.rb +++ b/test/fixtures/yjs_fixtures.rb @@ -100,4 +100,14 @@ module Gap module PendingDelete UPDATE = YjsFixtures.b64("AAEDAQAB") end + + # Fixture 11: a cross-client-origin gap. CONTENT is client 3's "abc"; DELTA is + # client 1's insert BETWEEN client 3's characters, so its origins reference + # client 3's blocks. On a doc lacking CONTENT, DELTA's per-client clock lower + # bound passes but integration parks -- the readiness case a clock-only check + # misses (update_ready? must say false). + module CrossClientOrigin + CONTENT = YjsFixtures.b64("AQEDAAQBAXQDYWJjAA==") + DELTA = YjsFixtures.b64("AQEBAMQDAAMBAVgA") + end end diff --git a/test/packaging_test.rb b/test/packaging_test.rb new file mode 100644 index 00000000..2ba333fb --- /dev/null +++ b/test/packaging_test.rb @@ -0,0 +1,41 @@ +# frozen_string_literal: true + +require "test_helper" + +# Gemspec file-list regressions. The repo ships three gems; the core gem must +# not package the files of the other two — a frozen duplicate on the load path +# can shadow (or be shadowed by) the standalone gem and drift silently between +# releases. This is a packaging bug that no runtime test catches, so it's +# asserted here. +class PackagingTest < Minitest::Test + ROOT = File.expand_path("..", __dir__) + + def load_spec(name) + Dir.chdir(ROOT) { Gem::Specification.load(File.join(ROOT, name)) } + end + + def test_core_gem_excludes_the_actioncable_and_decoder_gems_files + files = load_spec("yrby.gemspec").files + + assert_empty files.grep(/action_cable|actioncable/), "actioncable files ship in yrby-actioncable" + assert_empty files.grep(/decoder/), "decoder files ship in yrby-decoder" + end + + def test_core_gem_ships_its_own_essentials + files = load_spec("yrby.gemspec").files + + %w[lib/y.rb lib/yrby.rb lib/y/version.rb ext/yrby/extconf.rb + ext/yrby/src/lib.rs Cargo.toml Cargo.lock].each do |f| + assert_includes files, f + end + end + + def test_no_gem_packages_tests_or_artifacts + %w[yrby.gemspec yrby-actioncable.gemspec yrby-decoder.gemspec].each do |gemspec| + files = load_spec(gemspec).files + + assert_empty files.grep(%r{^(test|bench|examples|pkg|target|tmp)/}), + "#{gemspec} must not package tests/benchmarks/artifacts" + end + end +end diff --git a/test/sync_test.rb b/test/sync_test.rb index 135e7bfd..5e1bac83 100644 --- a/test/sync_test.rb +++ b/test/sync_test.rb @@ -393,14 +393,18 @@ def test_lost_ack_retry_acks_without_double_recording helper.sync_receive(msg, "doc-key") assert_equal [YjsFixtures::TwoDocsMerged::DOC1_UPDATE], store - assert_equal 1, broadcasts.length + # The retry is not re-RECORDED, but it IS re-broadcast: if the original + # attempt recorded and then crashed before distributing, the retry is the + # only mechanism that can still reach live subscribers. Idempotent apply + # makes the duplicate broadcast free. + assert_equal 2, broadcasts.length assert_equal [5, 5], acks_in(transmits) end def test_lost_ack_delete_retry_acks_without_double_recording - # A pure-delete retry the server already integrated must be acked but not - # recorded or re-broadcast. Insert content, delete a char, then replay the - # deletion: the second delivery is a no-op the guard must catch. + # A pure-delete retry the server already integrated must be acked and not + # re-recorded (it IS re-broadcast — see the retry test above). Insert + # content, delete a char, then replay the deletion. content = YjsFixtures::DeleteRetry::CONTENT deletion = YjsFixtures::DeleteRetry::DELETION @@ -414,10 +418,50 @@ def test_lost_ack_delete_retry_acks_without_double_recording helper.sync_receive(update_message(deletion, id: 3), "doc-key") # lost-ack retry assert_equal 2, store.length, "the deletion records once; its retry does not" - assert_equal 2, broadcasts.length, "the retry is not re-broadcast" + assert_equal 3, broadcasts.length, "the retry re-broadcasts (crash-window heal)" assert_equal [1, 2, 3], acks_in(transmits), "every frame is still acked" end + def test_cross_client_origin_gap_is_resynced_not_acked + # DELTA's origins reference client 3's blocks (CONTENT), which this store + # never saw. Its per-client clock lower bound passes, so a clock-only ready + # check used to let it through — and the advances? probe then misread the + # parked update as an already-applied retry: acked :applied and dropped. + # It must instead be rejected as a gap: resynced, never recorded, never + # acked. + store = [] + broadcasts = [] + transmits = [] + helper = helper_for(store: store, transmits: transmits, broadcasts: broadcasts) + + helper.sync_receive(update_message(YjsFixtures::CrossClientOrigin::DELTA, id: 7), "doc-key") + + assert_empty store, "a causally-incomplete update is never recorded" + assert_empty broadcasts + assert_empty acks_in(transmits), "and never acked (the old bug acked it)" + assert_equal 1, transmits.length, "a resync (SyncStep1) was requested" + + # Once the missing content arrives, the same delta is ready and records. + helper.sync_receive(update_message(YjsFixtures::CrossClientOrigin::CONTENT, id: 8), "doc-key") + helper.sync_receive(update_message(YjsFixtures::CrossClientOrigin::DELTA, id: 9), "doc-key") + + assert_equal 2, store.length, "content + delta both recorded once healed" + assert_equal [8, 9], acks_in(transmits) + end + + def test_receive_without_a_key_fails_closed + helper = helper_for + + # No sync_subscribed, no key argument: recording under a nil key and acking + # would silently misfile the update, so the frame must raise instead. + error = assert_raises(Y::Error) do + helper.sync_receive(update_message(YjsFixtures::TwoDocsMerged::DOC1_UPDATE, id: 1)) + end + + assert_match(/document key/, error.message) + assert_empty acks_in(helper.transmits) + end + # -- Store-backed concurrency ------------------------------------------- # # Real MRI threads contend on one document key. Delivery is at-least-once, so a diff --git a/test/thread_safety_test.rb b/test/thread_safety_test.rb index c7de3c7d..19d55658 100644 --- a/test/thread_safety_test.rb +++ b/test/thread_safety_test.rb @@ -75,6 +75,33 @@ def test_concurrent_sync_protocol_between_doc_pairs end end + def test_concurrent_read_text_with_writers_does_not_deadlock + # Regression: read_text used to open a second read transaction while still + # holding the first (a chained temporary). yrs's lock is write-preferring, so + # a writer arriving between the two acquisitions deadlocked reader-vs-writer + # inside nogvl — uninterruptibly. With the fix this completes; without it, + # this test hangs (CI timeout catches it). + doc = Y::Doc.new + doc.apply_update(YjsFixtures::TextHelloWorld::UPDATE) + updates = [ + YjsFixtures::TwoDocsMerged::DOC1_UPDATE, + YjsFixtures::TwoDocsMerged::DOC2_UPDATE + ] + + errors = run_threads do |i| + ITERATIONS.times do + if i.even? + doc.read_text("content") + else + doc.apply_update(updates[(i / 2) % updates.length]) + end + end + end + + assert_empty errors + refute_nil doc.read_text("content") + end + def test_concurrent_fan_in_sync_to_shared_doc # Many threads sync different sources into ONE shared doc concurrently. shared = Y::Doc.new diff --git a/yrby-actioncable.gemspec b/yrby-actioncable.gemspec index 0efd8d37..388828af 100644 --- a/yrby-actioncable.gemspec +++ b/yrby-actioncable.gemspec @@ -33,12 +33,14 @@ Gem::Specification.new do |spec| spec.metadata["rubygems_mfa_required"] = "true" spec.add_dependency "base64", "~> 0.2" - # Floor raised to 0.3.0, whose handle_sync_message answers SyncStep1 with - # integrated-only (gap-free) state. The channel serves the sync response through - # that method, so the floor makes gap-free serving self-enforcing rather than - # dependent on the app updating the core gem. (0.2.3 similarly made - # update_advances? exact for delete-bearing updates.) - spec.add_dependency "yrby", ">= 0.3.0" + # Floor raised to 0.3.1, whose update_ready? is exact (trial-integration, not + # just per-client clocks). The channel gates recording AND the retry-vs-gap + # decision on it; with an older core a cross-client-origin gap passed the ready + # check and the advances? probe then acked-and-dropped real content. The floor + # makes the fix self-enforcing rather than dependent on the app updating the + # core gem. (Earlier floors: 0.3.0 gap-free SyncStep1; 0.2.3 exact + # delete-bearing update_advances?.) + spec.add_dependency "yrby", ">= 0.3.1" # The concern references ActionCable (channels, streaming, broadcasting) and # ActiveSupport (Concern, JSON coder) constants directly. Rails apps already # bundle these, but declaring them makes use outside a full Rails bundle fail diff --git a/yrby.gemspec b/yrby.gemspec index 58d5ab0d..0da4d42e 100644 --- a/yrby.gemspec +++ b/yrby.gemspec @@ -17,16 +17,20 @@ Gem::Specification.new do |spec| spec.license = "MIT" spec.required_ruby_version = ">= 3.4.0" - # The ActionCable layer (lib/y/action_cable*) ships in the separate - # yrby-actioncable gem, so it's excluded from the core gem here. + # The actioncable and decoder files ship in their own gems (yrby-actioncable, + # yrby-decoder) — exclude them here so the core gem can't shadow those gems + # on the load path. Cargo.lock IS shipped so source builds compile the exact + # crate graph CI tested. spec.files = Dir[ "lib/**/*.rb", "ext/**/*.{rb,rs,toml}", "Cargo.toml", + "Cargo.lock", "LICENSE", "README.md", "CHANGELOG.md" - ] - Dir["lib/yrby-actioncable.rb", "lib/y/action_cable.rb", "lib/y/action_cable/**/*"] + ] - Dir["lib/yrby-actioncable.rb", "lib/y/action_cable.rb", "lib/y/action_cable/**/*", + "lib/yrby-decoder.rb", "lib/y/decoder.rb", "lib/y/decoder/**/*"] spec.require_paths = ["lib"] spec.extensions = ["ext/yrby/extconf.rb"]