From 808e3920c43debe4be627129088845f27bfb5a0a Mon Sep 17 00:00:00 2001 From: Richard Mathieson <35163478+rlch@users.noreply.github.com> Date: Tue, 23 Jun 2026 15:30:59 +1000 Subject: [PATCH 1/2] fix(executor): drive native lane async tasks cooperatively, not via block_on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ThreadExecutor::execute_async` drove each async task with `futures::executor::block_on` on the lane's single worker thread, so any `.await` inside a `#[frb(thread = ...)]` handler parked that worker until the future resolved — every other call routed to the same lane queued behind it. This is a native-only regression versus `SimpleExecutor` (which spawns onto a multi-thread runtime) and versus the wasm lane (which `spawn_local`s onto the JS event loop, yielding on `.await`). Drive the lane cooperatively instead: - Add `SingleThreadAsyncLane` (native `thread_pool/io.rs`): one dedicated thread running a `current_thread` runtime + `LocalSet`, fed jobs over a channel and running each inside the LocalSet context. - Native `spawn_local_compat` now `tokio::task::spawn_local`s the task future instead of `block_on`-ing it, so `.await` yields the lane to other queued jobs while preserving single-thread FIFO ownership — matching the wasm lane. WASM is unchanged. Adds `single_worker_lane_pinning_tests`, a regression test that deadlocks under `block_on` and passes under cooperative scheduling. Co-Authored-By: Claude Opus 4.8 (1M context) --- frb_rust/src/for_generated/mod.rs | 2 + .../handler/implementation/thread_executor.rs | 128 +++++++++++++++++- frb_rust/src/thread_pool/io.rs | 53 ++++++++ 3 files changed, 179 insertions(+), 4 deletions(-) diff --git a/frb_rust/src/for_generated/mod.rs b/frb_rust/src/for_generated/mod.rs index 5d10316e553..00389e65b5d 100644 --- a/frb_rust/src/for_generated/mod.rs +++ b/frb_rust/src/for_generated/mod.rs @@ -64,6 +64,8 @@ pub use crate::rust_auto_opaque::{inner::RustAutoOpaqueInner, RustAutoOpaqueBase pub use crate::rust_opaque::{dart2rust::decode_rust_opaque_nom, RustOpaqueBase}; pub use crate::stream::stream_sink::StreamSinkBase; pub use crate::thread_pool::{BaseThreadPool, SimpleThreadPool}; +#[cfg(all(feature = "rust-async", not(target_family = "wasm")))] +pub use crate::thread_pool::SingleThreadAsyncLane; #[cfg(target_family = "wasm")] pub use crate::web_transfer::transfer_closure::TransferClosure; #[cfg(feature = "anyhow")] diff --git a/frb_rust/src/handler/implementation/thread_executor.rs b/frb_rust/src/handler/implementation/thread_executor.rs index 57b95d7ec13..730bce70065 100644 --- a/frb_rust/src/handler/implementation/thread_executor.rs +++ b/frb_rust/src/handler/implementation/thread_executor.rs @@ -209,15 +209,19 @@ impl Executor for ThreadExecutor + 'static>(future: F) { wasm_bindgen_futures::spawn_local(future); } #[cfg(not(target_family = "wasm"))] -fn spawn_local_compat>(future: F) { - futures::executor::block_on(future); +fn spawn_local_compat + 'static>(future: F) { + tokio::task::spawn_local(future); } struct ThreadExecuteUtils; @@ -242,3 +246,119 @@ impl ThreadExecuteUtils { }; } } + +// ============================================================================= +// Reproduction: a single-worker lane must not pin on `.await`. +// +// Mirrors modality's `LanePools` (one dedicated worker per `Thread`, FIFO — the +// single-owner invariant that serializes every LoroDoc access). The hazard +// `bounded_loro_wait` bandages is that on native `execute_async` drives the +// task future with `futures::executor::block_on` (see `spawn_local_compat` +// above), which PARKS the lane's only worker on `.await` instead of yielding — +// so an awaiting handler starves every other call routed to the same lane +// (catalog/wizard reads, the drainer reply it is waiting on). On wasm the same +// path uses `spawn_local`, which yields, so wasm is unaffected. +// +// This test routes two async tasks to one single-worker lane with a cross-task +// dependency. Cooperative scheduling completes both; `block_on` deadlocks. It +// FAILS on the current native impl (reproducing the bug) and must PASS once the +// lane drives futures with a cooperative single-threaded runtime +// (current_thread + LocalSet). +#[cfg(all(test, feature = "rust-async", feature = "thread-pool", not(target_family = "wasm")))] +mod single_worker_lane_pinning_tests { + use super::*; + use crate::codec::dco::DcoCodec; + use crate::codec::BaseCodec; + use crate::handler::handler::{FfiCallMode, TaskInfo, Thread}; + use crate::handler::implementation::error_listener::NoOpErrorListener; + use crate::thread_pool::{BaseThreadPool, SingleThreadAsyncLane}; + use std::sync::mpsc; + use std::time::Duration; + + /// One single-owner lane per `Thread` — identical in shape to modality's + /// `frb_handler::LanePools`. Backed by `SingleThreadAsyncLane` so async tasks + /// are driven cooperatively (one thread, one `LocalSet`) instead of + /// `block_on`-parked. + struct SingleWorkerLane { + worker: SingleThreadAsyncLane, + } + + impl ThreadRouter for SingleWorkerLane { + fn execute_on(&self, _thread: Thread, job: ThreadJob) { + self.worker.execute(job); + } + } + + fn loro_task(port: i64) -> TaskInfo { + TaskInfo { + port: Some(port), + debug_name: "repro", + mode: FfiCallMode::Normal, + thread: Thread::Loro, + } + } + + // The result message is irrelevant (it posts to a dead port → just warns); + // we observe scheduling through side channels captured by the task futures. + fn ok_msg() -> Result<::Message, ::Message> { + Ok(::Message::simplest()) + } + + #[test] + fn awaiting_task_must_not_pin_single_worker_lane() { + let exec = ThreadExecutor::new( + NoOpErrorListener, + SingleWorkerLane { + worker: SingleThreadAsyncLane::new("repro-loro-lane"), + }, + ); + + // gate: opened by B, awaited by A. `oneshot` needs only a waker (no + // reactor), so awaiting it is valid on the reactor-less Loro lane. + let (gate_tx, gate_rx) = futures::channel::oneshot::channel::<()>(); + let (done_tx, done_rx) = mpsc::channel::<&'static str>(); + + // Task A — submitted first, so it occupies the single worker first, then + // awaits B's gate. Under cooperative scheduling this await yields the + // lane; under `block_on` it parks the only worker. + let done_a = done_tx.clone(); + exec.execute_async::(loro_task(1), move |_ctx| async move { + let _ = gate_rx.await; + let _ = done_a.send("A"); + ok_msg() + }); + + // Task B — queued behind A on the same lane. Opens the gate so A can + // finish. It can only run if A's await freed the worker. + let done_b = done_tx.clone(); + exec.execute_async::(loro_task(2), move |_ctx| async move { + let _ = gate_tx.send(()); + let _ = done_b.send("B"); + ok_msg() + }); + drop(done_tx); + + let mut done = Vec::new(); + while done.len() < 2 { + match done_rx.recv_timeout(Duration::from_secs(5)) { + Ok(who) => done.push(who), + Err(_) => panic!( + "single-worker lane PINNED: an awaiting task blocked a queued task on the \ + same lane (only completed {done:?}). Native `execute_async` drives the task \ + future with `futures::executor::block_on`, which parks the lane's single \ + worker on `.await` instead of yielding to the queued task — the exact \ + dispatch hazard `bounded_loro_wait` bandages. Fix: drive the lane with a \ + cooperative single-threaded runtime (current_thread + LocalSet) so `.await` \ + yields, matching the wasm `spawn_local` path." + ), + } + } + + done.sort_unstable(); + assert_eq!( + done, + vec!["A", "B"], + "both tasks on the single-worker lane must make progress" + ); + } +} diff --git a/frb_rust/src/thread_pool/io.rs b/frb_rust/src/thread_pool/io.rs index b7a15a5aba4..2658fa323f2 100644 --- a/frb_rust/src/thread_pool/io.rs +++ b/frb_rust/src/thread_pool/io.rs @@ -16,6 +16,59 @@ impl BaseThreadPool for SimpleThreadPool { } } +/// A single-owner worker lane that drives async jobs **cooperatively**. +/// +/// One dedicated OS thread runs a `current_thread` runtime + a `LocalSet`, +/// pulling jobs off a channel and running each inside the LocalSet context. A +/// job that `spawn_local`s a future (what `ThreadExecutor::execute_async` does) +/// therefore yields the lane on `.await` to other queued jobs — unlike a +/// `block_on`-per-job thread pool, which parks the worker until the future +/// completes. This mirrors the wasm `spawn_local` lane on native, preserving +/// single-thread FIFO ownership (one thread, one `LocalSet`) while removing the +/// pinning hazard a single-worker `SimpleThreadPool` has under `block_on`. +#[cfg(feature = "rust-async")] +pub struct SingleThreadAsyncLane { + tx: futures::channel::mpsc::UnboundedSender>, +} + +#[cfg(feature = "rust-async")] +impl SingleThreadAsyncLane { + pub fn new(thread_name: impl Into) -> Self { + use futures::StreamExt; + let (tx, mut rx) = + futures::channel::mpsc::unbounded::>(); + std::thread::Builder::new() + .name(thread_name.into()) + .spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("SingleThreadAsyncLane: failed to build current_thread runtime"); + let local = tokio::task::LocalSet::new(); + // `run_until` drives the receive loop AND every `spawn_local`'d + // task on this one thread, so an awaiting task yields back here. + local.block_on(&runtime, async move { + while let Some(job) = rx.next().await { + job(); + } + }); + }) + .expect("SingleThreadAsyncLane: failed to spawn lane thread"); + Self { tx } + } +} + +#[cfg(feature = "rust-async")] +impl BaseThreadPool for SingleThreadAsyncLane { + fn execute(&self, job: F) + where + F: FnOnce() + Send + 'static, + { + // Process-global lanes outlive every caller; a send failure just means + // the lane thread is gone, so dropping the job is the correct no-op. + let _ = self.tx.unbounded_send(Box::new(job)); + } +} + // TODO migrate this documentation // /// Spawn a task using the internal thread pool. // /// Interprets the parameters as a list of captured transferables to From e8b37254d95360e3761a386e683b6b4a7021bf0f Mon Sep 17 00:00:00 2001 From: Richard Mathieson <35163478+rlch@users.noreply.github.com> Date: Tue, 23 Jun 2026 17:18:26 +1000 Subject: [PATCH 2/2] feat(executor): generic, config-driven lane routing (drop hardcoded domain lanes) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `#[frb(thread = ...)]` routing baked modality's domain into FRB: the `Thread` enum hardcoded `Loro`/`Export` variants, the attribute parser allowlisted those names, and `validate_frb_threads` hardcoded `session_id`/`drive_id`/... param names + a "sync or session-handle ⇒ must annotate" rule. FRB should know none of that — lane identity and policy belong to the embedder. Make lanes opaque and the policy config-driven: - `Thread` is now `{ Main, Worker(&'static str) }`. The lane key is an opaque string the embedder's `ThreadRouter` resolves; FRB assigns it no meaning. - The attribute parser accepts any ident (no allowlist); codegen emits `Thread::Main` for `Main`/unset and `Thread::Worker("")` otherwise. - New optional `lane_routing` config (`flutter_rust_bridge.yaml`): `lanes` (valid keys) + `require_annotation_when { sync, param_named }`. `validate_frb_threads` enforces it against config — erroring on an unknown lane or a function that matches the predicate but isn't annotated. Absent config ⇒ no-op (any key accepted, nothing required), so FRB ships with zero domain knowledge. Runtime/doc comments de-Loro'd. WASM unchanged. Config golden fixtures updated for the new `lane_routing` field (the unrelated `oxidized` MIR-golden drift that predates this branch is left untouched). Co-Authored-By: Claude Opus 4.8 (1M context) --- frb_codegen/src/binary/commands_parser.rs | 2 + .../src/library/codegen/config/config.rs | 31 ++++++ .../config/internal_config_parser/mod.rs | 1 + .../rust/spec_generator/misc/function/mod.rs | 7 +- .../codegen/parser/mir/internal_config.rs | 2 + .../codegen/parser/mir/parser/attribute.rs | 34 +++---- .../codegen/parser/mir/parser/function/mod.rs | 98 ++++++++++++------- frb_codegen/src/library/codegen/parser/mod.rs | 1 + .../single_rust_input/expect_output.json | 4 +- .../wildcard_rust_input/expect_output.json | 4 +- frb_rust/src/handler/handler.rs | 24 +++-- .../handler/implementation/thread_executor.rs | 68 ++++++------- 12 files changed, 168 insertions(+), 108 deletions(-) diff --git a/frb_codegen/src/binary/commands_parser.rs b/frb_codegen/src/binary/commands_parser.rs index 277a4208bcd..ce712fa6bef 100644 --- a/frb_codegen/src/binary/commands_parser.rs +++ b/frb_codegen/src/binary/commands_parser.rs @@ -71,6 +71,8 @@ fn compute_codegen_config_from_naive_command_args(args: GenerateCommandArgsPrima dump_all: positive_bool_arg(args.dump_all), rust_features: args.rust_features, use_oxidized: None, + // CLI args don't carry lane routing; it's a yaml-only config. + lane_routing: None, } } diff --git a/frb_codegen/src/library/codegen/config/config.rs b/frb_codegen/src/library/codegen/config/config.rs index 8f5ad1b5945..6d545538423 100644 --- a/frb_codegen/src/library/codegen/config/config.rs +++ b/frb_codegen/src/library/codegen/config/config.rs @@ -47,6 +47,7 @@ pub struct Config { pub dump_all: Option, pub rust_features: Option>, pub use_oxidized: Option, + pub lane_routing: Option, } #[derive(Debug, Serialize, Deserialize, Default)] @@ -54,6 +55,35 @@ pub struct MetaConfig { pub watch: bool, } +/// Optional policy for the `#[frb(thread = )]` executor-routing feature. +/// +/// Flutter Rust Bridge itself assigns no meaning to worker lanes — it only +/// stamps the lane key onto `TaskInfo.thread` and hands it to the embedder's +/// `ThreadRouter`. This config lets the embedder declare, in +/// `flutter_rust_bridge.yaml`, which lane keys are valid and which functions +/// must be annotated, so codegen can enforce it with no hardcoded domain +/// knowledge. When absent, codegen accepts any lane key and enforces nothing. +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct LaneRoutingConfig { + /// Valid worker-lane keys for `#[frb(thread = )]` (besides the implicit + /// `Main`). When set, codegen errors on any annotation outside this set. + pub lanes: Option>, + /// When set, codegen errors if a function matching the predicate has no + /// `#[frb(thread = ...)]` annotation. + pub require_annotation_when: Option, +} + +/// Predicate selecting functions that must declare a `#[frb(thread = ...)]` lane. +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct RequireAnnotationWhen { + /// Require a lane on every `#[frb(sync)]` function. + pub sync: Option, + /// Require a lane on any function with a parameter named one of these. + pub param_named: Option>, +} + macro_rules! generate_merge { ($($field:ident,)*) => ( impl Config { @@ -111,4 +141,5 @@ generate_merge!( dump_all, rust_features, use_oxidized, + lane_routing, ); diff --git a/frb_codegen/src/library/codegen/config/internal_config_parser/mod.rs b/frb_codegen/src/library/codegen/config/internal_config_parser/mod.rs index 2163dede21b..7ec330545f6 100644 --- a/frb_codegen/src/library/codegen/config/internal_config_parser/mod.rs +++ b/frb_codegen/src/library/codegen/config/internal_config_parser/mod.rs @@ -126,6 +126,7 @@ impl InternalConfig { enable_lifetime: config.enable_lifetime.unwrap_or_default(), type_64bit_int: config.type_64bit_int.unwrap_or_default(), default_dart_async: config.default_dart_async.unwrap_or(true), + lane_routing: config.lane_routing.clone(), }, }, generator, diff --git a/frb_codegen/src/library/codegen/generator/wire/rust/spec_generator/misc/function/mod.rs b/frb_codegen/src/library/codegen/generator/wire/rust/spec_generator/misc/function/mod.rs index 5541e457a39..d9827585c0b 100644 --- a/frb_codegen/src/library/codegen/generator/wire/rust/spec_generator/misc/function/mod.rs +++ b/frb_codegen/src/library/codegen/generator/wire/rust/spec_generator/misc/function/mod.rs @@ -100,7 +100,12 @@ fn generate_wrap_info_obj(func: &MirFunc) -> String { "None" }, mode = ffi_call_mode(func.mode), - thread = func.thread.as_deref().unwrap_or("Main"), + // `Main` (or unset) is the inline lane; any other key is an opaque + // worker-lane string the embedder's `ThreadRouter` resolves. + thread = match func.thread.as_deref() { + None | Some("Main") => "Main".to_owned(), + Some(key) => format!("Worker({key:?})"), + }, ) } diff --git a/frb_codegen/src/library/codegen/parser/mir/internal_config.rs b/frb_codegen/src/library/codegen/parser/mir/internal_config.rs index cbe8d1e637a..ed2aaf4ef1e 100644 --- a/frb_codegen/src/library/codegen/parser/mir/internal_config.rs +++ b/frb_codegen/src/library/codegen/parser/mir/internal_config.rs @@ -1,3 +1,4 @@ +use crate::codegen::config::config::LaneRoutingConfig; use crate::codegen::generator::codec::structs::{CodecMode, CodecModePack}; use crate::codegen::ir::mir::ty::rust_opaque::RustOpaqueCodecMode; use crate::utils::namespace::Namespace; @@ -13,6 +14,7 @@ pub(crate) struct ParserMirInternalConfig { pub enable_lifetime: bool, pub type_64bit_int: bool, pub default_dart_async: bool, + pub lane_routing: Option, } // TODO rename - this is no longer an "input-namespace"-only pack diff --git a/frb_codegen/src/library/codegen/parser/mir/parser/attribute.rs b/frb_codegen/src/library/codegen/parser/mir/parser/attribute.rs index 2be26ea66e1..78a4d3d8f66 100644 --- a/frb_codegen/src/library/codegen/parser/mir/parser/attribute.rs +++ b/frb_codegen/src/library/codegen/parser/mir/parser/attribute.rs @@ -709,30 +709,19 @@ impl Parse for FrbAttributeName { } } -/// Value of `#[frb(thread = )]` — the executor routing lane. -/// Mirrors `flutter_rust_bridge::for_generated::Thread`. Stored as the variant -/// identifier so codegen can emit `Thread::` verbatim. +/// Value of `#[frb(thread = )]` — the executor routing lane key. +/// Stored as the raw identifier; codegen emits `Thread::Main` for `Main` and +/// `Thread::Worker("")` otherwise. FRB assigns no meaning to lanes here: +/// parsing is purely syntactic, and any valid-lane / required-lane policy is +/// enforced later against `lane_routing` config (see `validate_frb_threads`), +/// where the embedder's config is available. #[derive(Clone, Serialize, Eq, PartialEq, Debug)] pub(crate) struct FrbAttributeThread(pub(crate) String); -impl FrbAttributeThread { - const VALID: &'static [&'static str] = &["Main", "Loro", "Export"]; -} - impl Parse for FrbAttributeThread { fn parse(input: ParseStream) -> Result { let ident: syn::Ident = input.parse()?; - let value = ident.to_string(); - if !Self::VALID.contains(&value.as_str()) { - return Err(syn::Error::new( - ident.span(), - format!( - "unknown frb thread `{value}`; expected one of {:?}", - Self::VALID - ), - )); - } - Ok(Self(value)) + Ok(Self(ident.to_string())) } } @@ -828,9 +817,12 @@ mod tests { } #[test] - fn test_thread_invalid() { - let result = parse("#[frb(thread = Nonsense)]"); - assert!(result.is_err()); + fn test_thread_arbitrary_ident_accepted() -> anyhow::Result<()> { + // The parser is purely syntactic: any ident is a valid lane key here; + // lane-set validation happens later against `lane_routing` config. + let parsed = parse("#[frb(thread = AnyLane)]")?; + assert_eq!(parsed.thread(), Some("AnyLane".to_owned())); + Ok(()) } #[test] diff --git a/frb_codegen/src/library/codegen/parser/mir/parser/function/mod.rs b/frb_codegen/src/library/codegen/parser/mir/parser/function/mod.rs index 45dc034bdd6..3c5e9becdb3 100644 --- a/frb_codegen/src/library/codegen/parser/mir/parser/function/mod.rs +++ b/frb_codegen/src/library/codegen/parser/mir/parser/function/mod.rs @@ -30,47 +30,77 @@ pub(crate) fn parse( ]); let (funcs, skips) = IrValueOrSkip::split(items); let funcs = sort_and_add_func_id(funcs); - validate_frb_threads(&funcs)?; + validate_frb_threads(&funcs, config)?; Ok((funcs, skips)) } -/// Hybrid enforcement for `#[frb(thread = ...)]` (the executor routing lane). +/// Optional enforcement for the `#[frb(thread = ...)]` executor-routing lane, +/// driven entirely by the embedder's `lane_routing` config — FRB assigns no +/// meaning to lanes itself. /// -/// A function that may touch live CRDT/session state — it runs on the main -/// thread via `sync`, OR carries a session-handle argument — MUST declare a -/// thread so the custom executor routes it to the right worker. Pure getters -/// may omit it and default to `Thread::Main`. +/// With config present it errors when an annotation names a lane outside the +/// configured set, and when a function matching `require_annotation_when` +/// carries no annotation. Without config it is a no-op (any lane key accepted, +/// nothing required). /// -/// This runs as a post-parse pass (not inside per-function parsing) because an -/// error raised during `Parser::parse_function` with `stop_on_error=false` is -/// swallowed into a silent function *skip*, which would drop the offending -/// functions instead of failing the build. Returning the error here propagates -/// it to the top level. -fn validate_frb_threads(funcs: &[MirFunc]) -> anyhow::Result<()> { - const SESSION_HANDLE_ARGS: &[&str] = - &["session_id", "drive_id", "browser_id", "wizard_id"]; +/// Runs as a post-parse pass (not inside per-function parsing) because an error +/// raised with `stop_on_error=false` is swallowed into a silent function *skip*, +/// which would drop the offending functions instead of failing the build. +/// Returning the error here propagates it to the top level. +fn validate_frb_threads( + funcs: &[MirFunc], + config: &ParserMirInternalConfig, +) -> anyhow::Result<()> { + let Some(lane_routing) = &config.lane_routing else { + return Ok(()); + }; - let offenders = funcs - .iter() - .filter(|func| func.thread.is_none()) - .filter(|func| { - let is_sync = matches!(func.mode, MirFuncMode::Sync); - let has_session_arg = func.inputs.iter().any(|input| { - SESSION_HANDLE_ARGS.contains(&input.inner.name.rust_style(true).as_str()) - }); - is_sync || has_session_arg - }) - .map(|func| func.name.rust_style(true)) - .collect_vec(); + // 1. Lane-name validation: every annotation must name `Main` or a configured lane. + if let Some(lanes) = &lane_routing.lanes { + for func in funcs { + if let Some(lane) = &func.thread { + if lane != "Main" && !lanes.iter().any(|l| l == lane) { + bail!( + "function `{}` is annotated `#[frb(thread = {lane})]`, but `{lane}` \ + is not a configured lane; valid lanes: Main, {}", + func.name.rust_style(true), + lanes.join(", "), + ); + } + } + } + } + + // 2. Required-annotation policy. + if let Some(require) = &lane_routing.require_annotation_when { + let require_sync = require.sync.unwrap_or(false); + let param_named = require.param_named.clone().unwrap_or_default(); + let offenders = funcs + .iter() + .filter(|func| func.thread.is_none()) + .filter(|func| { + let is_sync = require_sync && matches!(func.mode, MirFuncMode::Sync); + let has_named_param = func + .inputs + .iter() + .any(|input| param_named.contains(&input.inner.name.rust_style(true))); + is_sync || has_named_param + }) + .map(|func| func.name.rust_style(true)) + .collect_vec(); - if !offenders.is_empty() { - bail!( - "the following functions may touch live CRDT/session state (they are \ - `sync` or carry a session/drive/browser/wizard id) but have no \ - `#[frb(thread = ...)]` annotation; add \ - `#[frb(thread = Main|Loro|Export)]` to each so the custom \ - executor routes it to the correct worker: {offenders:?}" - ); + if !offenders.is_empty() { + let lanes_hint = lane_routing + .lanes + .as_ref() + .map(|lanes| format!("Main|{}", lanes.join("|"))) + .unwrap_or_else(|| "Main".to_owned()); + bail!( + "these functions match the configured `require_annotation_when` policy but have \ + no `#[frb(thread = ...)]` annotation; add `#[frb(thread = {lanes_hint})]` to \ + each so the custom executor routes it: {offenders:?}" + ); + } } Ok(()) diff --git a/frb_codegen/src/library/codegen/parser/mod.rs b/frb_codegen/src/library/codegen/parser/mod.rs index 225d0ca854c..450878c3a57 100644 --- a/frb_codegen/src/library/codegen/parser/mod.rs +++ b/frb_codegen/src/library/codegen/parser/mod.rs @@ -186,6 +186,7 @@ mod tests { enable_lifetime: false, type_64bit_int: false, default_dart_async: true, + lane_routing: None, }, }; diff --git a/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/single_rust_input/expect_output.json b/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/single_rust_input/expect_output.json index 209d3ef054f..7a66c0c625f 100644 --- a/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/single_rust_input/expect_output.json +++ b/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/single_rust_input/expect_output.json @@ -25,7 +25,8 @@ "web": "{the-working-directory}/my_dart_folder/frb_generated.web.dart" }, "dart_preamble": "", - "dart_type_rename": {} + "dart_type_rename": {}, + "use_oxidized": false }, "wire": { "c": { @@ -110,6 +111,7 @@ "dart2rust": "Pde", "rust2dart": "Pde" }, + "lane_routing": null, "rust_input_namespace_pack": { "rust_input_namespace_prefixes": [ "crate::api" diff --git a/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/wildcard_rust_input/expect_output.json b/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/wildcard_rust_input/expect_output.json index 209d3ef054f..7a66c0c625f 100644 --- a/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/wildcard_rust_input/expect_output.json +++ b/frb_codegen/test_fixtures/library/codegen/config/internal_config_parser/wildcard_rust_input/expect_output.json @@ -25,7 +25,8 @@ "web": "{the-working-directory}/my_dart_folder/frb_generated.web.dart" }, "dart_preamble": "", - "dart_type_rename": {} + "dart_type_rename": {}, + "use_oxidized": false }, "wire": { "c": { @@ -110,6 +111,7 @@ "dart2rust": "Pde", "rust2dart": "Pde" }, + "lane_routing": null, "rust_input_namespace_pack": { "rust_input_namespace_prefixes": [ "crate::api" diff --git a/frb_rust/src/handler/handler.rs b/frb_rust/src/handler/handler.rs index 57fa3694a89..aad1d1cc8a4 100644 --- a/frb_rust/src/handler/handler.rs +++ b/frb_rust/src/handler/handler.rs @@ -76,31 +76,29 @@ pub struct TaskInfo { pub debug_name: &'static str, /// The call mode of this function. pub mode: FfiCallMode, - /// Which executor thread/worker lane this call must run on. - /// A custom [`super::executor::Executor`] uses this as a typed routing key + /// Which executor lane this call must run on. + /// A custom [`super::executor::Executor`] uses this as a routing key /// instead of matching on [`TaskInfo::debug_name`]. The default executor /// ignores it. Stamped by codegen from `#[frb(thread = ...)]`; defaults to /// [`Thread::Main`] when unset. pub thread: Thread, } -/// Typed routing key for a custom [`super::executor::Executor`]. +/// Routing key for a custom [`super::executor::Executor`]. /// /// The default executor ignores this. A custom executor maps each call to a -/// thread/worker lane, so the routing decision is a typed enum match -/// rather than a string comparison on [`TaskInfo::debug_name`]. +/// worker lane. FRB assigns no meaning to a lane beyond "not the caller +/// thread" — the embedder's `ThreadRouter` owns what each key denotes. #[derive(Copy, Clone, PartialEq, Eq, Debug, Default)] pub enum Thread { - /// Pure, side-effect-free getters; safe to run on the caller (main) thread. - /// Also config/global setters that touch no CRDT state, and `!Send` - /// main-thread browser APIs (e.g. the sync WebSocket), which cannot leave - /// the main thread. + /// Run inline on the caller (main thread) — pure getters and config setters + /// that need no dedicated worker lane. The default when `thread` is unset. #[default] Main, - /// Touches live CRDT/session state; must run on the single owner worker. - Loro, - /// Render-from-snapshot (pdf / thumbnail); no live-state handle. - Export, + /// Run on the embedder-defined worker lane identified by this opaque key. + /// Stamped from `#[frb(thread = )]` (the ident becomes the key); the + /// embedder's `ThreadRouter` maps the key to a concrete pool. + Worker(&'static str), } /// The types of return values for a particular Rust function. diff --git a/frb_rust/src/handler/implementation/thread_executor.rs b/frb_rust/src/handler/implementation/thread_executor.rs index 730bce70065..b5c36d12165 100644 --- a/frb_rust/src/handler/implementation/thread_executor.rs +++ b/frb_rust/src/handler/implementation/thread_executor.rs @@ -5,8 +5,8 @@ //! port/sender/panic plumbing is identical — but instead of a single thread //! pool it holds a [`ThreadRouter`] and asks it which pool a call goes //! to, keyed by the typed [`Thread`]. `execute_sync` still runs inline on the -//! caller (the main thread), so a custom routing setup keeps Loro-free getters -//! on main and everything stateful on its lane. +//! caller (the main thread), so a custom routing setup keeps stateless getters +//! on main and everything else on its dedicated lane. //! //! The default generated handler does NOT use this; it is opt-in by providing a //! custom `FLUTTER_RUST_BRIDGE_HANDLER` built from `SimpleHandler, _>`. @@ -77,7 +77,7 @@ macro_rules! thread_job { /// [`SimpleExecutor`](super::executor::SimpleExecutor) builds, then route it /// to the `thread`'s lane via [`ThreadRouter::execute_on`]. /// - `execute_sync`: run inline on the caller (main thread), exactly like -/// `SimpleExecutor`. Callers must only expose Loro-free getters as `sync`. +/// `SimpleExecutor`. Callers must only expose state-free getters as `sync`. pub struct ThreadExecutor { error_listener: EL, pools: P, @@ -147,8 +147,8 @@ impl Executor for ThreadExecutor data, Err(err) => { @@ -248,22 +248,20 @@ impl ThreadExecuteUtils { } // ============================================================================= -// Reproduction: a single-worker lane must not pin on `.await`. +// A single-worker lane must not pin on `.await`. // -// Mirrors modality's `LanePools` (one dedicated worker per `Thread`, FIFO — the -// single-owner invariant that serializes every LoroDoc access). The hazard -// `bounded_loro_wait` bandages is that on native `execute_async` drives the -// task future with `futures::executor::block_on` (see `spawn_local_compat` -// above), which PARKS the lane's only worker on `.await` instead of yielding — -// so an awaiting handler starves every other call routed to the same lane -// (catalog/wizard reads, the drainer reply it is waiting on). On wasm the same -// path uses `spawn_local`, which yields, so wasm is unaffected. +// `ThreadExecutor` routes each call to a worker lane. When a consumer makes a +// lane a single worker (the only way to get FIFO single-owner ordering), an +// async handler that `.await`s must YIELD that worker to other queued calls on +// the same lane — not park it. Native `execute_async` drives the future with +// `tokio::task::spawn_local` onto the lane's `LocalSet` (see `spawn_local_compat` +// above) precisely so it yields; a `block_on` there would park the single worker +// until the future resolved, starving every other call on the lane. The wasm +// path already yields via `spawn_local`. // -// This test routes two async tasks to one single-worker lane with a cross-task -// dependency. Cooperative scheduling completes both; `block_on` deadlocks. It -// FAILS on the current native impl (reproducing the bug) and must PASS once the -// lane drives futures with a cooperative single-threaded runtime -// (current_thread + LocalSet). +// This routes two async tasks to one single-worker lane with a cross-task +// dependency: cooperative scheduling completes both; a parking driver would +// deadlock (A holds the worker awaiting B, which can never start). #[cfg(all(test, feature = "rust-async", feature = "thread-pool", not(target_family = "wasm")))] mod single_worker_lane_pinning_tests { use super::*; @@ -275,10 +273,8 @@ mod single_worker_lane_pinning_tests { use std::sync::mpsc; use std::time::Duration; - /// One single-owner lane per `Thread` — identical in shape to modality's - /// `frb_handler::LanePools`. Backed by `SingleThreadAsyncLane` so async tasks - /// are driven cooperatively (one thread, one `LocalSet`) instead of - /// `block_on`-parked. + /// A router with one single-owner lane, backed by `SingleThreadAsyncLane` so + /// async tasks are driven cooperatively (one thread, one `LocalSet`). struct SingleWorkerLane { worker: SingleThreadAsyncLane, } @@ -289,12 +285,12 @@ mod single_worker_lane_pinning_tests { } } - fn loro_task(port: i64) -> TaskInfo { + fn worker_task(port: i64) -> TaskInfo { TaskInfo { port: Some(port), debug_name: "repro", mode: FfiCallMode::Normal, - thread: Thread::Loro, + thread: Thread::Worker("w"), } } @@ -309,20 +305,20 @@ mod single_worker_lane_pinning_tests { let exec = ThreadExecutor::new( NoOpErrorListener, SingleWorkerLane { - worker: SingleThreadAsyncLane::new("repro-loro-lane"), + worker: SingleThreadAsyncLane::new("repro-lane"), }, ); // gate: opened by B, awaited by A. `oneshot` needs only a waker (no - // reactor), so awaiting it is valid on the reactor-less Loro lane. + // reactor), so awaiting it is valid on the reactor-less worker lane. let (gate_tx, gate_rx) = futures::channel::oneshot::channel::<()>(); let (done_tx, done_rx) = mpsc::channel::<&'static str>(); // Task A — submitted first, so it occupies the single worker first, then - // awaits B's gate. Under cooperative scheduling this await yields the - // lane; under `block_on` it parks the only worker. + // awaits B's gate. Cooperative scheduling yields the lane here; a parking + // driver would hold the only worker. let done_a = done_tx.clone(); - exec.execute_async::(loro_task(1), move |_ctx| async move { + exec.execute_async::(worker_task(1), move |_ctx| async move { let _ = gate_rx.await; let _ = done_a.send("A"); ok_msg() @@ -331,7 +327,7 @@ mod single_worker_lane_pinning_tests { // Task B — queued behind A on the same lane. Opens the gate so A can // finish. It can only run if A's await freed the worker. let done_b = done_tx.clone(); - exec.execute_async::(loro_task(2), move |_ctx| async move { + exec.execute_async::(worker_task(2), move |_ctx| async move { let _ = gate_tx.send(()); let _ = done_b.send("B"); ok_msg() @@ -344,12 +340,10 @@ mod single_worker_lane_pinning_tests { Ok(who) => done.push(who), Err(_) => panic!( "single-worker lane PINNED: an awaiting task blocked a queued task on the \ - same lane (only completed {done:?}). Native `execute_async` drives the task \ - future with `futures::executor::block_on`, which parks the lane's single \ - worker on `.await` instead of yielding to the queued task — the exact \ - dispatch hazard `bounded_loro_wait` bandages. Fix: drive the lane with a \ - cooperative single-threaded runtime (current_thread + LocalSet) so `.await` \ - yields, matching the wasm `spawn_local` path." + same lane (only completed {done:?}). The lane's async driver parked its \ + single worker on `.await` instead of yielding to the queued task. Fix: drive \ + the lane with a cooperative single-threaded runtime (current_thread + \ + LocalSet) so `.await` yields, matching the wasm `spawn_local` path." ), } }