From 9c006e7a15909af168724f810b3455a572cb501e Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 1 Jul 2026 18:25:07 -0400 Subject: [PATCH 1/6] Fix findings from a full source review (yrby 0.3.1, actioncable 0.2.4, client 0.4.3) Core (Rust): - update_ready? is now EXACT. It only checked the per-client clock lower bound, but yrs's integration gate also requires origin/right-origin/parent blocks (routinely other clients') and post-Skip blocks sit above the lower bound. A cross-client-origin delta on a server missing that client's content passed ready?, then update_advances? misread the parked result as an already-applied retry (pending doesn't move a state vector): the channel ACKED AND DROPPED real content. Reproduced empirically. ready? now trial-integrates on a probe seeded with the doc's integrated state (clock check kept as a cheap pre-filter); advances? gained defense in depth (a parked update reports as advancing, never a duplicate). - read_text deadlock: it opened a second read txn while a chained temporary still held the first; yrs's write-preferring lock made a concurrent writer deadlock the process inside nogvl (uninterruptible). Single txn now. - TOCTOU in integrated_update: pending check and encode ran in separate txns, so a concurrent gappy apply between them could serve pending anyway. One txn now. - update_advances? pre-filter: blocks beyond the doc's SV trivially advance; the common (novel-update) case skips the full O(doc) probe. - read_xml: Lexical linebreak/tab nodes emit \n/\t instead of vanishing. ActionCable concern (Ruby): - A lost-ack retry re-broadcasts before acking :applied. Record-then-crash (or a failed broadcast) previously left live subscribers permanently stale: the retry skipped distribution, and nothing else could reach them. - A missing document key fails closed (Y::Error) instead of silently recording under nil, broadcasting to a dead stream, and still acking (the AnyCable fresh-instance + forgotten-key case). - Floor raised to yrby >= 0.3.1 (the ack-and-drop fix lives in core). Client (TypeScript): - rejected() handler: an auth-rejected subscription surfaces via onError and tears down instead of hanging at "connecting" forever, queueing edits. - Awareness frames are content-validated (dry-run entries) before apply: a partially-malformed payload previously mutated awareness state entry by entry with no event fired. Trailing bytes inside the blob also rejected. - bfcache restore: presence is stashed on pagehide and restored on pageshow(persisted) - a restored page no longer rejoins as a ghost. - Transport sends are guarded: sync throws and promise rejections (@anycable/web) surface via onError instead of unwinding/unhandled. - CJS TypeScript consumers get real CJS-flavored declarations (dist/cjs d.ts + per-condition types exports); fixes TS1479 under node16 resolution. Packaging: - yrby no longer ships the yrby-decoder gem's files (the frozen duplicate could shadow a newer standalone release across the load path). - Cargo.lock ships in the source gem: source builds compile the exact crate graph CI tested. - Demo: the unauthenticated audit-control endpoint (history wipe + per-write delay injection) is no longer mounted in production. New real-Y.js fixture (CrossClientOrigin) + regression tests across all layers, including a read_text-vs-writers thread hammer. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG-actioncable.md | 25 +++ CHANGELOG.md | 39 ++++ examples/actioncable-demo/config/routes.rb | 8 +- ext/yrby/src/lib.rs | 10 +- ext/yrby/src/protocol.rs | 172 ++++++++++++++++-- ext/yrby/src/read.rs | 42 ++++- lib/y/action_cable/sync.rb | 29 ++- lib/y/action_cable/version.rb | 2 +- lib/y/version.rb | 2 +- packages/client/package.json | 35 +++- packages/client/src/actioncable_provider.ts | 69 ++++++- packages/client/src/y_protocol_session.ts | 21 ++- .../client/test/actioncable_provider.test.js | 61 +++++++ .../client/test/y_protocol_session.test.js | 63 +++++++ packages/client/tsconfig.cjs.json | 2 +- test/fixtures/generate_fixtures.mjs | 30 +++ test/fixtures/yjs_fixtures.rb | 10 + test/sync_test.rb | 54 +++++- test/thread_safety_test.rb | 27 +++ yrby-actioncable.gemspec | 14 +- yrby.gemspec | 12 +- 21 files changed, 663 insertions(+), 64 deletions(-) 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/lib.rs b/ext/yrby/src/lib.rs index 63e22334..f334a796 100644 --- a/ext/yrby/src/lib.rs +++ b/ext/yrby/src/lib.rs @@ -160,9 +160,13 @@ 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())) + // One transaction, bound to a local. The previous chained form + // (`doc.transact().get_text(..).map(|t| t.get_string(&doc.transact()))`) + // held the first read guard while acquiring a second; yrs's lock is + // write-preferring, so a writer arriving between the two acquisitions + // deadlocked reader-vs-writer — inside nogvl, uninterruptibly. + 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..5183f4dc 100644 --- a/ext/yrby/src/protocol.rs +++ b/ext/yrby/src/protocol.rs @@ -67,13 +67,40 @@ pub(crate) fn merged_doc_update(bytes: &[u8]) -> Result>, String> } /// 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. +/// dependency the update references is already present. A pure read; does not +/// mutate the doc. When false, applying it would park a pending struct/delete +/// set, the signal that an earlier, causally-prior update is missing. +/// +/// The per-client clock lower bound (`state_vector_lower`) is only a cheap +/// pre-filter: 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 have their own, +/// higher, entry clocks the lower bound doesn't see. An update can pass the +/// clock check and still park as pending. Miss that here and the downstream +/// `update_advances?` probe misreads the parked update as an already-applied +/// retry (pending doesn't move a state vector) — acking and dropping real +/// content. So after the pre-filter we do the exact check: trial-integrate on a +/// throwaway probe seeded with the doc's *integrated* state, and call the update +/// ready only if nothing parks. (Integrated-only seed: if the update depends on +/// content that is itself still pending in `doc`, it can't cleanly integrate +/// yet, and a resync delivers the whole thing as one complete delta.) 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()) + // State vectors are partially ordered; "covered" (>=) is false for both + // strictly-behind AND incomparable vectors — either way, not ready. + let lower_covered = doc.transact().state_vector() >= update.state_vector_lower(); + if !lower_covered { + return Ok(false); // a same-client clock gap: cheap, definitive reject + } + 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 +134,20 @@ 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 +158,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()); + // The pending check and the encode must share ONE transaction: with separate + // transactions, a concurrent gappy apply_update between them could park + // pending that the second transaction's encode then merges back in — serving + // exactly the poison this function exists to exclude (TOCTOU). + let full = { + let txn = doc.transact(); + let store = txn.store(); + // Fast path: with nothing pending the direct encode is already gap-free, + // so the clean common case keeps the zero-copy behavior. + 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 +526,88 @@ 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 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) { diff --git a/ext/yrby/src/read.rs b/ext/yrby/src/read.rs index 67764edb..f6bf9336 100644 --- a/ext/yrby/src/read.rs +++ b/ext/yrby/src/read.rs @@ -82,13 +82,20 @@ fn walk_lexical_block(txn: &T, t: &XmlTextRef, out: &mut Vec match d.insert { Out::Any(Any::String(s)) => line.push_str(&s), 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); + // Soft line break / tab nodes carry no text of their own; emit + // the character they represent so "foo⏎bar" doesn't become + // "foobar". + match ty.as_str() { + "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 @@ -256,6 +263,29 @@ 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) stores the break as an embedded + // XmlText child with __type=linebreak and no text of its own; 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("")); + block.push(&mut txn, "foo"); + let br = block.insert_embed(&mut txn, 3, XmlTextPrelim::new("")); + br.insert_attribute(&mut txn, "__type", "linebreak"); + block.push(&mut txn, "bar"); + let tab = block.insert_embed(&mut txn, 8, XmlTextPrelim::new("")); + tab.insert_attribute(&mut txn, "__type", "tab"); + block.push(&mut txn, "baz"); + } + let txn = doc.transact(); + assert_eq!(xml_blocks_text(&txn, &frag), "foo\nbar\tbaz"); + } + #[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..fb3157cc 100644 --- a/lib/y/action_cable/sync.rb +++ b/lib/y/action_cable/sync.rb @@ -240,6 +240,21 @@ def sync_validate_required_hooks! "that never happened, and a cold load would lose the edit." end + # Fail closed on a missing document key. Without this, a transport that + # doesn't keep the channel instance alive across actions (AnyCable) and an + # app that forgot to pass `key` to sync_receive would silently record + # updates under a nil key, broadcast them to a stream no one subscribes to, + # and still ack them — the client marks the edit delivered while it reached + # zero peers and was filed under the wrong identity. + 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 +265,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 +287,16 @@ 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 the store already has: don't re-record it, but DO + # re-broadcast. If the original attempt recorded and then crashed (or the + # pub/sub broadcast failed) before distributing, this retry is the only + # mechanism that can still reach the live subscribers — skipping it would + # leave them stale until their next full resync. Re-broadcast is safe: + # CRDT apply is idempotent for every receiver. + 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..22a0cc2f 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,14 @@ export class ActionCableProvider { provider.session.onDisconnect(); // pause retransmits, clear remote presence provider.#refreshStatus(); // subscription still set -> "connecting" (retrying) }, + rejected() { + // The channel refused the subscription (authorization, missing doc). + // Without this handler the provider would sit at "connecting" forever, + // silently queueing local edits. Surface it and tear down: the app + // decides whether to re-auth and reconnect. + provider.#onError(new Error("subscription rejected by the server"), "rejected"); + provider.disconnect(); + }, } ); this.#installUnloadHandler(); @@ -229,16 +239,40 @@ export class ActionCableProvider { // 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. + // + // bfcache restore: `pagehide` nulls the local awareness state, so a page + // brought back from the cache would rejoin as a ghost (edits flow, but no + // cursor/presence — editor bindings only set awareness once at setup). Stash + // the state on the way out and restore it on `pageshow` with `persisted`; + // setting it re-fires the awareness update, and onConnect re-broadcasts it + // once the socket is back. #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 +285,28 @@ 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; + // Guard the transport: @anycable/web's send/whisper return promises whose + // rejections would otherwise go unobserved, and a synchronously-throwing + // transport must not unwind into doc/awareness update handlers. A failed + // send is safe to swallow for reliable frames (they stay queued until + // acked) and best-effort anyway for awareness. + 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..b21f5782 100644 --- a/packages/client/src/y_protocol_session.ts +++ b/packages/client/src/y_protocol_session.ts @@ -248,7 +248,26 @@ export class YProtocolSession { break; } case MessageType.Awareness: - decoding.readVarUint8Array(decoder); + // Dry-run the inner payload, not just the envelope: applyAwarenessUpdate + // mutates awareness state one entry at a time inside its decode loop and + // only notifies listeners at the end, so a payload whose entry k is + // malformed would leave entries 0..k-1 applied with no event fired. + // Walking the entries here (count, then per-entry clientID/clock/JSON + // state) makes the real apply infallible — and catches trailing garbage + // *inside* the blob, which applyAwarenessUpdate would silently ignore. + { + 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..3ce5366d 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,63 @@ 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"); +}); 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/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..ebd93d09 100644 --- a/yrby.gemspec +++ b/yrby.gemspec @@ -17,16 +17,22 @@ 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 layer (lib/y/action_cable*) and the decoder (lib/y/decoder*) + # ship in the separate yrby-actioncable / yrby-decoder gems, so both are + # excluded from the core gem here — otherwise the core's frozen snapshot of + # those files would shadow (or be shadowed by) the standalone gems on the load + # path, drifting silently between releases. Cargo.lock IS shipped so source + # builds compile the exact crate graph CI tested, not a fresh resolution. 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"] From d7a518add43e8da272d1291be691361807a60427 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 1 Jul 2026 20:55:27 -0400 Subject: [PATCH 2/6] Close verification gaps: discrimination-proven regressions + missing tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers "did you verify every scenario?" honestly — each fix's regression test was run against the ORIGINAL buggy code where feasible: - read_text deadlock (H1): DISCRIMINATES. Against the reverted buggy form the thread hammer deadlocked so hard the process survived timeout(1)'s SIGTERM for 7+ minutes (threads stuck in uninterruptible nogvl) and needed SIGKILL; with the fix it passes in milliseconds. - awareness half-application (client M8): DISCRIMINATES. Against the reverted envelope-only validation, exactly the two new tests fail — the failure message ("the valid entry was NOT half-applied") empirically proves the old partial-mutation behavior. - TOCTOU (H2): NOT black-box reproducible — 20k racing iterations couldn't hit the nanoseconds-wide window even against the buggy two-transaction form. The new concurrency test is documented honestly as a contention net (catches lock-skipping/fast-path regressions); the fix's guarantee is structural (one transaction is atomic under the doc's lock). New tests this commit: - protocol.rs: integrated_update concurrency net (writer parking/healing gappy updates vs reader encoding; every encode must be pending-free). - test/packaging_test.rb: gemspec file-list regressions (no decoder or actioncable files in the core gem, essentials + Cargo.lock present, no tests/artifacts packaged in any gem). - client: bfcache tests via a window shim — pagehide stashes presence, pageshow(persisted) restores it, a non-persisted pageshow does not resurrect stale presence. (This path previously had NO coverage.) Co-Authored-By: Claude Opus 4.8 --- ext/yrby/src/protocol.rs | 54 +++++++++++++++++++ .../client/test/actioncable_provider.test.js | 49 +++++++++++++++++ test/packaging_test.rb | 41 ++++++++++++++ 3 files changed, 144 insertions(+) create mode 100644 test/packaging_test.rb diff --git a/ext/yrby/src/protocol.rs b/ext/yrby/src/protocol.rs index 5183f4dc..d43606e8 100644 --- a/ext/yrby/src/protocol.rs +++ b/ext/yrby/src/protocol.rs @@ -771,6 +771,60 @@ mod tests { assert!(!has_pending(&peer), "the diff carried no pending"); } + #[test] + fn integrated_update_never_serves_pending_under_concurrent_gappy_applies() { + // Concurrency net for the gap-free invariant: race a writer that parks + // and heals a gappy update against a reader encoding, and assert every + // single encode is pending-free for a fresh peer. + // + // Honest scope: the original TOCTOU (pending check and encode in + // separate transactions) has a nanoseconds-wide window and did NOT + // reproduce here even at 20k iterations — that fix's guarantee is + // structural (one transaction is atomic under the doc's lock), verified + // by construction, not by this test. What this test DOES catch is any + // grosser regression: encoding without the lock, a fast path that skips + // the pending check entirely, or prune logic that leaks under contention. + 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/packages/client/test/actioncable_provider.test.js b/packages/client/test/actioncable_provider.test.js index 3ce5366d..e6c6e4ce 100644 --- a/packages/client/test/actioncable_provider.test.js +++ b/packages/client/test/actioncable_provider.test.js @@ -341,3 +341,52 @@ test("a promise-rejecting transport send surfaces via onError (no unhandled reje 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/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 From a2b0fcf61ede64ac460dcf2ff8537fe8590d2c13 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 1 Jul 2026 21:14:39 -0400 Subject: [PATCH 3/6] read_xml: fix Lexical linebreak extraction against REAL editor bytes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closing the last review-verification caveat surfaced the real bug: driving a live Lexxy editor (agent-browser: type "foo", Shift+Enter, type "barbaz") and running read_xml on the captured durable state returned "foobarbaz" — the hand-built regression test had modeled the structure wrong. A Lexical LineBreakNode is NOT an XmlText child with __type=linebreak; it is an embedded **Y.Map with { __type: "linebreak" }**, the same shape as the per-text-node metadata maps the walker deliberately skips — so the newline was being swallowed by the metadata arm. Fix: Out::YMap embeds now consult their __type — "linebreak" emits '\n', "tab" emits '\t', anything else (the "text" metadata maps) stays silent, in both walk_lexical_block and inline_lexical_text. The XmlText-child handling stays as a defensive path. Tests: - fixtures/lexical_linebreak.bin: ground-truth bytes captured from the live editor via the lexxy-realtime test server (GET /content/:room); the new test asserts they extract as "foo\nbarbaz". - The hand-built test now builds the REAL structure (Y.Map embeds, including a silent metadata map) instead of the imagined one. Verified end-to-end: Doc#read_xml on the captured bytes returns "foo\nbarbaz". Co-Authored-By: Claude Opus 4.8 --- ext/yrby/src/fixtures/lexical_linebreak.bin | Bin 0 -> 441 bytes ext/yrby/src/read.rs | 73 ++++++++++++++++---- 2 files changed, 61 insertions(+), 12 deletions(-) create mode 100644 ext/yrby/src/fixtures/lexical_linebreak.bin diff --git a/ext/yrby/src/fixtures/lexical_linebreak.bin b/ext/yrby/src/fixtures/lexical_linebreak.bin new file mode 100644 index 0000000000000000000000000000000000000000..08c0df4dc894781ecd33ac2a8ba75c5acffc7721 GIT binary patch literal 441 zcmZvX(G9{N6h$FcN~0e({c!{*aRCP)G*BhBr9l#{iGGHEmkM9FyB zuT2SILcy!U=4J~&>Gg2jtj#yw$gq=w9f!SLU~!j@h~a-R3m1~{uRS&(?(<`^u>}9vE*OO!oOk`9Ec?-r7ERo@Bu!0r6T|U literal 0 HcmV?d00001 diff --git a/ext/yrby/src/read.rs b/ext/yrby/src/read.rs index f6bf9336..6d061aaf 100644 --- a/ext/yrby/src/read.rs +++ b/ext/yrby/src/read.rs @@ -62,6 +62,18 @@ fn lexical_type(txn: &T, t: &XmlTextRef) -> String { } } +/// The `__type` of an embedded Lexical `Y.Map`. Lexical embeds two kinds of +/// maps in a block's `Y.XmlText`: per-text-node metadata (`__type: "text"`, +/// carrying format/style/mode) and *node* maps — a `LineBreakNode` is stored as +/// `{ __type: "linebreak" }` (verified against bytes captured from a live +/// Lexxy editor after a Shift+Enter). +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 +81,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,12 +98,20 @@ 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), + // A LineBreakNode (Shift+Enter) is an embedded Y.Map with + // `__type: "linebreak"` — emit the character it represents so + // "foo⏎bar" doesn't extract as "foobar". Every other map embed is + // per-text-node metadata (`__type: "text"`) with no text of its own. + Out::YMap(m) => match lexical_map_type(txn, &m).as_str() { + "linebreak" => line.push('\n'), + "tab" => line.push('\t'), + _ => {} + }, Out::YXmlText(child) => { let ty = lexical_type(txn, &child); - // Soft line break / tab nodes carry no text of their own; emit - // the character they represent so "foo⏎bar" doesn't become - // "foobar". match ty.as_str() { + // Defensive: linebreak/tab as XmlText children (not observed + // in real Lexical docs, which use Y.Map embeds — see above). "linebreak" => line.push('\n'), "tab" => line.push('\t'), _ if is_inline_lexical_type(&ty) => inline_lexical_text(txn, &child, &mut line), @@ -98,7 +123,7 @@ fn walk_lexical_block(txn: &T, t: &XmlTextRef, out: &mut Vec } } } - _ => {} // per-text-node metadata map; embeds we don't read for text + _ => {} // decorator embeds we don't read for text } } if !line.is_empty() { @@ -265,27 +290,51 @@ mod tests { #[test] fn lexical_soft_line_break_and_tab_emit_their_characters() { - // A paragraph "foo⏎bar" (shift-enter) stores the break as an embedded - // XmlText child with __type=linebreak and no text of its own; it must - // come through as '\n', not vanish and glue the words. Same for tab. + // 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 = block.insert_embed(&mut txn, 3, XmlTextPrelim::new("")); - br.insert_attribute(&mut txn, "__type", "linebreak"); + 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 = block.insert_embed(&mut txn, 8, XmlTextPrelim::new("")); - tab.insert_attribute(&mut txn, "__type", "tab"); + 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 From 5125611979c6c2ddaaafc18dd6cc51f171381db8 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 1 Jul 2026 21:57:56 -0400 Subject: [PATCH 4/6] Simplify review-fix comments: purpose-first, history cut Each comment now states the invariant and why it matters, without the incident narrative (that lives in the CHANGELOG and PR). Co-Authored-By: Claude Opus 4.8 --- ext/yrby/src/lib.rs | 8 +- ext/yrby/src/protocol.rs | 82 +++++++++------------ ext/yrby/src/read.rs | 19 ++--- lib/y/action_cable/sync.rb | 21 +++--- packages/client/src/actioncable_provider.ts | 33 ++++----- packages/client/src/y_protocol_session.ts | 13 ++-- yrby.gemspec | 10 +-- 7 files changed, 76 insertions(+), 110 deletions(-) diff --git a/ext/yrby/src/lib.rs b/ext/yrby/src/lib.rs index f334a796..4c340327 100644 --- a/ext/yrby/src/lib.rs +++ b/ext/yrby/src/lib.rs @@ -160,11 +160,9 @@ impl RbDoc { fn read_text(&self, name: String) -> Option { let doc = &self.0; nogvl(move || { - // One transaction, bound to a local. The previous chained form - // (`doc.transact().get_text(..).map(|t| t.get_string(&doc.transact()))`) - // held the first read guard while acquiring a second; yrs's lock is - // write-preferring, so a writer arriving between the two acquisitions - // deadlocked reader-vs-writer — inside nogvl, uninterruptibly. + // 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 d43606e8..93753c8c 100644 --- a/ext/yrby/src/protocol.rs +++ b/ext/yrby/src/protocol.rs @@ -66,31 +66,25 @@ 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. A pure read; does not -/// mutate the doc. When false, applying it would park a pending struct/delete -/// set, 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. /// -/// The per-client clock lower bound (`state_vector_lower`) is only a cheap -/// pre-filter: 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 have their own, -/// higher, entry clocks the lower bound doesn't see. An update can pass the -/// clock check and still park as pending. Miss that here and the downstream -/// `update_advances?` probe misreads the parked update as an already-applied -/// retry (pending doesn't move a state vector) — acking and dropping real -/// content. So after the pre-filter we do the exact check: trial-integrate on a -/// throwaway probe seeded with the doc's *integrated* state, and call the update -/// ready only if nothing parks. (Integrated-only seed: if the update depends on -/// content that is itself still pending in `doc`, it can't cleanly integrate -/// yet, and a resync delivers the whole thing as one complete delta.) +/// 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())?; - // State vectors are partially ordered; "covered" (>=) is false for both - // strictly-behind AND incomparable vectors — either way, not ready. + // 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); // a same-client clock gap: cheap, definitive reject + return Ok(false); } let seed = integrated_update(doc, &StateVector::default())?; let probe = Doc::new(); @@ -134,14 +128,10 @@ pub(crate) fn update_advances_doc(doc: &Doc, update_bytes: &[u8]) -> Result= update.state_vector(); if !covered { return Ok(true); @@ -187,12 +177,10 @@ pub(crate) fn update_advances_doc(doc: &Doc, update_bytes: &[u8]) -> 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> { - // The pending check and the encode must share ONE transaction: with separate - // transactions, a concurrent gappy apply_update between them could park - // pending that the second transaction's encode then merges back in — serving - // exactly the poison this function exists to exclude (TOCTOU). + // 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(); - // Fast path: with nothing pending the direct encode is already gap-free, - // so the clean common case keeps the zero-copy behavior. + // 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)); } @@ -773,17 +759,15 @@ mod tests { #[test] fn integrated_update_never_serves_pending_under_concurrent_gappy_applies() { - // Concurrency net for the gap-free invariant: race a writer that parks - // and heals a gappy update against a reader encoding, and assert every - // single encode is pending-free for a fresh peer. + // 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. // - // Honest scope: the original TOCTOU (pending check and encode in - // separate transactions) has a nanoseconds-wide window and did NOT - // reproduce here even at 20k iterations — that fix's guarantee is - // structural (one transaction is atomic under the doc's lock), verified - // by construction, not by this test. What this test DOES catch is any - // grosser regression: encoding without the lock, a fast path that skips - // the pending check entirely, or prune logic that leaks under contention. + // 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; diff --git a/ext/yrby/src/read.rs b/ext/yrby/src/read.rs index 6d061aaf..fa25a543 100644 --- a/ext/yrby/src/read.rs +++ b/ext/yrby/src/read.rs @@ -62,11 +62,10 @@ fn lexical_type(txn: &T, t: &XmlTextRef) -> String { } } -/// The `__type` of an embedded Lexical `Y.Map`. Lexical embeds two kinds of -/// maps in a block's `Y.XmlText`: per-text-node metadata (`__type: "text"`, -/// carrying format/style/mode) and *node* maps — a `LineBreakNode` is stored as -/// `{ __type: "linebreak" }` (verified against bytes captured from a live -/// Lexxy editor after a Shift+Enter). +/// 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(), @@ -98,10 +97,9 @@ 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), - // A LineBreakNode (Shift+Enter) is an embedded Y.Map with - // `__type: "linebreak"` — emit the character it represents so - // "foo⏎bar" doesn't extract as "foobar". Every other map embed is - // per-text-node metadata (`__type: "text"`) with no text of its own. + // 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'), @@ -110,8 +108,7 @@ fn walk_lexical_block(txn: &T, t: &XmlTextRef, out: &mut Vec Out::YXmlText(child) => { let ty = lexical_type(txn, &child); match ty.as_str() { - // Defensive: linebreak/tab as XmlText children (not observed - // in real Lexical docs, which use Y.Map embeds — see above). + // 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), diff --git a/lib/y/action_cable/sync.rb b/lib/y/action_cable/sync.rb index fb3157cc..92c48c58 100644 --- a/lib/y/action_cable/sync.rb +++ b/lib/y/action_cable/sync.rb @@ -240,12 +240,11 @@ def sync_validate_required_hooks! "that never happened, and a cold load would lose the edit." end - # Fail closed on a missing document key. Without this, a transport that - # doesn't keep the channel instance alive across actions (AnyCable) and an - # app that forgot to pass `key` to sync_receive would silently record - # updates under a nil key, broadcast them to a stream no one subscribes to, - # and still ack them — the client marks the edit delivered while it reached - # zero peers and was filed under the wrong identity. + # 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? @@ -287,12 +286,10 @@ def sync_handle_frame(encoded, bytes) return :gap end - # A lost-ack retry the store already has: don't re-record it, but DO - # re-broadcast. If the original attempt recorded and then crashed (or the - # pub/sub broadcast failed) before distributing, this retry is the only - # mechanism that can still reach the live subscribers — skipping it would - # leave them stale until their next full resync. Re-broadcast is safe: - # CRDT apply is idempotent for every receiver. + # 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 diff --git a/packages/client/src/actioncable_provider.ts b/packages/client/src/actioncable_provider.ts index 22a0cc2f..d9a7da0e 100644 --- a/packages/client/src/actioncable_provider.ts +++ b/packages/client/src/actioncable_provider.ts @@ -183,10 +183,9 @@ export class ActionCableProvider { provider.#refreshStatus(); // subscription still set -> "connecting" (retrying) }, rejected() { - // The channel refused the subscription (authorization, missing doc). - // Without this handler the provider would sit at "connecting" forever, - // silently queueing local edits. Surface it and tear down: the app - // decides whether to re-auth and reconnect. + // 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(); }, @@ -235,17 +234,12 @@ 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. - // - // bfcache restore: `pagehide` nulls the local awareness state, so a page - // brought back from the cache would rejoin as a ghost (edits flow, but no - // cursor/presence — editor bindings only set awareness once at setup). Stash - // the state on the way out and restore it on `pageshow` with `persisted`; - // setting it re-fires the awareness update, and onConnect re-broadcasts it - // once the socket is back. + // 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 = () => { @@ -285,11 +279,10 @@ export class ActionCableProvider { if (!sub) return; const update = toBase64(frame); const isAwareness = frame[0] === MessageType.Awareness; - // Guard the transport: @anycable/web's send/whisper return promises whose - // rejections would otherwise go unobserved, and a synchronously-throwing - // transport must not unwind into doc/awareness update handlers. A failed - // send is safe to swallow for reliable frames (they stay queued until - // acked) and best-effort anyway for awareness. + // 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 })); diff --git a/packages/client/src/y_protocol_session.ts b/packages/client/src/y_protocol_session.ts index b21f5782..149ec2f1 100644 --- a/packages/client/src/y_protocol_session.ts +++ b/packages/client/src/y_protocol_session.ts @@ -248,13 +248,12 @@ export class YProtocolSession { break; } case MessageType.Awareness: - // Dry-run the inner payload, not just the envelope: applyAwarenessUpdate - // mutates awareness state one entry at a time inside its decode loop and - // only notifies listeners at the end, so a payload whose entry k is - // malformed would leave entries 0..k-1 applied with no event fired. - // Walking the entries here (count, then per-entry clientID/clock/JSON - // state) makes the real apply infallible — and catches trailing garbage - // *inside* the blob, which applyAwarenessUpdate would silently ignore. + // 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); diff --git a/yrby.gemspec b/yrby.gemspec index ebd93d09..0da4d42e 100644 --- a/yrby.gemspec +++ b/yrby.gemspec @@ -17,12 +17,10 @@ Gem::Specification.new do |spec| spec.license = "MIT" spec.required_ruby_version = ">= 3.4.0" - # The ActionCable layer (lib/y/action_cable*) and the decoder (lib/y/decoder*) - # ship in the separate yrby-actioncable / yrby-decoder gems, so both are - # excluded from the core gem here — otherwise the core's frozen snapshot of - # those files would shadow (or be shadowed by) the standalone gems on the load - # path, drifting silently between releases. Cargo.lock IS shipped so source - # builds compile the exact crate graph CI tested, not a fresh resolution. + # 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}", From c688d34eae165c87e1cd29980a029ee95dbdb35f Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 1 Jul 2026 22:11:08 -0400 Subject: [PATCH 5/6] Explain the integrated-only probe seed in update_is_ready; test both halves The comment simplification cut the clause explaining why the readiness probe seeds with integrated_update rather than the lossless encode. Restore it compactly, and pin both reasons with tests: - a doc carrying legacy pending must still accept unrelated healthy updates (a lossless seed would veto everything), and - a dependency satisfied only by pending content is NOT ready (recording it would put a gap in the durable log). Co-Authored-By: Claude Opus 4.8 --- ext/yrby/src/protocol.rs | 65 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/ext/yrby/src/protocol.rs b/ext/yrby/src/protocol.rs index 93753c8c..b559aa3e 100644 --- a/ext/yrby/src/protocol.rs +++ b/ext/yrby/src/protocol.rs @@ -86,6 +86,13 @@ pub(crate) fn update_is_ready(doc: &Doc, update_bytes: &[u8]) -> Result> = 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 From eda347febb20a8ebc4eba0896cc6baf207cda80b Mon Sep 17 00:00:00 2001 From: JP Camara Date: Wed, 1 Jul 2026 22:29:28 -0400 Subject: [PATCH 6/6] Clarify the ambiguous-state-vector comment in update_advances_doc Two questions about this comment in a row means it wasn't carrying its weight: name the ambiguity (retry vs parked-as-pending both leave the state vector unchanged) and why it matters (false here = ack-and-drop). Co-Authored-By: Claude Opus 4.8 --- ext/yrby/src/protocol.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/ext/yrby/src/protocol.rs b/ext/yrby/src/protocol.rs index b559aa3e..ef90bb79 100644 --- a/ext/yrby/src/protocol.rs +++ b/ext/yrby/src/protocol.rs @@ -184,10 +184,14 @@ pub(crate) fn update_advances_doc(doc: &Doc, update_bytes: &[u8]) -> Result