From af7b25a32013341192ff398087b1c90cc5f45ad9 Mon Sep 17 00:00:00 2001 From: Nidish Date: Thu, 27 Aug 2026 17:00:35 +0530 Subject: [PATCH] feat(truapi-server): payload-blind wire-debug tap, sinks, and the codegen decode surface --- Cargo.lock | 1 + js/packages/truapi/README.md | 23 +- js/packages/truapi/package.json | 4 + .../truapi/scripts/ensure-generated.sh | 1 + rust/crates/truapi-codegen/src/main.rs | 11 +- rust/crates/truapi-codegen/src/rust.rs | 49 +- .../truapi-codegen/src/rust/wire_table.rs | 22 +- rust/crates/truapi-codegen/src/rustdoc.rs | 119 ++- rust/crates/truapi-codegen/src/ts.rs | 787 ++++++++++++++++- .../truapi-codegen/tests/golden/wire_table.rs | 6 + rust/crates/truapi-macros/src/lib.rs | 49 +- rust/crates/truapi-server/Cargo.toml | 7 +- .../truapi-server/src/generated/wire_table.rs | 6 + rust/crates/truapi-server/src/host_core.rs | 693 ++++++++++++++- rust/crates/truapi-server/src/lib.rs | 11 +- rust/crates/truapi-server/src/native_debug.rs | 813 ++++++++++++++++++ rust/crates/truapi-server/src/wasm.rs | 55 +- rust/crates/truapi/src/api/account.rs | 8 +- rust/crates/truapi/src/api/coin_payment.rs | 6 +- rust/crates/truapi/src/api/entropy.rs | 2 +- rust/crates/truapi/src/api/local_storage.rs | 4 +- rust/crates/truapi/src/api/payment.rs | 2 +- rust/crates/truapi/src/api/signing.rs | 12 +- rust/crates/truapi/src/api/statement_store.rs | 8 +- 24 files changed, 2629 insertions(+), 70 deletions(-) create mode 100644 rust/crates/truapi-server/src/native_debug.rs diff --git a/Cargo.lock b/Cargo.lock index 5a3f46235..795b32114 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5271,6 +5271,7 @@ name = "truapi-server" version = "0.1.0" dependencies = [ "async-trait", + "base64", "blake2b_simd", "chacha20poly1305", "console_error_panic_hook", diff --git a/js/packages/truapi/README.md b/js/packages/truapi/README.md index 46d47676d..3816e1c8e 100644 --- a/js/packages/truapi/README.md +++ b/js/packages/truapi/README.md @@ -70,7 +70,7 @@ sub.unsubscribe(); - **Generated domain clients and types** produced from the Rust API contract. - **SCALE codec helpers** used by the generated code, also re-exported for direct use. - **Sandbox bootstrap** (`@parity/truapi/sandbox`) that detects the host environment, builds the - matching provider, and exposes a cached client — see below. + matching provider, and exposes a cached client - see below. ## Sandbox bootstrap @@ -120,6 +120,27 @@ This is what makes a real host usable from an ordinary browser tab during develo transport without the sandbox's caching and detection, `createWebSocketProvider(url)` from the package root returns the bare `WireProvider`. +## Observability / debugging + +The debugger does not live in this package, and the product transport carries no debug seam - +`@parity/truapi` is genuinely untouched by observability. The host taps every product↔host frame in +its Rust core (`truapi-server`'s `DebugSink`) and streams each one as opaque bytes to a separate +debugger app, which decodes and groups them. + +- The tap: `DebugSink` in `rust/crates/truapi-server/src/host_core.rs`, unset by default. It is read + at two choke points — inbound before the frame is decoded, outbound after the product's copy is + sent — and is fire-and-forget, so an absent or slow debugger loses traces, never a session. +- Topology: the host always dials the debugger, over `ws://` on a loopback host **only**. `wss://`, + certificates, and non-loopback targets are rejected by the native dial gate + (`truapi-server/src/native_debug.rs`), which requires every resolved address to be loopback and + dials the addresses it checked. + +The generated `WIRE_DECODE_TABLE` on the `./wire-decode` subpath (raw SCALE bytes → typed value) +stays here, since it is generated from this package's contract. It is the decode source a debugger +uses to render frame values, kept off the package barrel so importing `@parity/truapi` never pulls a +decoder into a product bundle. `@parity/truapi` itself never decodes payloads — the envelope decode it +does expose (`decodeWireMessage`: `requestId`, frame id) carries no payload value. + ## Wire format Frames are SCALE encoded: diff --git a/js/packages/truapi/package.json b/js/packages/truapi/package.json index 4bcebb412..aa9af6879 100644 --- a/js/packages/truapi/package.json +++ b/js/packages/truapi/package.json @@ -39,6 +39,10 @@ "types": "./dist/generated/wire-table.d.ts", "import": "./dist/generated/wire-table.js" }, + "./wire-decode": { + "types": "./dist/generated/wire-decode.d.ts", + "import": "./dist/generated/wire-decode.js" + }, "./playground/services": { "types": "./dist/playground/codegen/services.d.ts", "import": "./dist/playground/codegen/services.js" diff --git a/js/packages/truapi/scripts/ensure-generated.sh b/js/packages/truapi/scripts/ensure-generated.sh index 807c07166..e32aa3561 100755 --- a/js/packages/truapi/scripts/ensure-generated.sh +++ b/js/packages/truapi/scripts/ensure-generated.sh @@ -9,6 +9,7 @@ codegen_required=( "js/packages/truapi/src/generated/client.ts" "js/packages/truapi/src/generated/types.ts" "js/packages/truapi/src/generated/wire-table.ts" + "js/packages/truapi/src/generated/wire-decode.ts" "js/packages/truapi/src/playground/codegen/services.ts" "js/packages/truapi/src/explorer/codegen/types.ts" "js/packages/truapi/src/explorer/versions.ts" diff --git a/rust/crates/truapi-codegen/src/main.rs b/rust/crates/truapi-codegen/src/main.rs index 66a803751..032982d55 100644 --- a/rust/crates/truapi-codegen/src/main.rs +++ b/rust/crates/truapi-codegen/src/main.rs @@ -152,7 +152,16 @@ fn main() -> Result<()> { println!("Generated client examples in {path}"); } if let Some(path) = &cli.rust_output { - rust::generate(&api, path) + // The Rust routing table (wire_table.rs) is version-*unfiltered* - the + // native host can route any method the crate defines - so its stamp hashes + // the full/latest table, not the client-pinned subset. Otherwise a + // `--client-version`-pinned build would route a newer #[wire(sensitive)] + // frame under an older hash that a same-pinned debugger would accept and + // decode. At the default (latest) client version this equals the TS hash. + let schema_hash = + ts::wire_schema_hash(&api, ts::latest_wire_version(&api), cli.codec_version) + .context("computing wire schema hash")?; + rust::generate(&api, path, &schema_hash) .with_context(|| format!("writing Rust dispatcher to {}", path.display()))?; println!("Wrote Rust dispatcher to {}", path.display()); } diff --git a/rust/crates/truapi-codegen/src/rust.rs b/rust/crates/truapi-codegen/src/rust.rs index 8368a82f7..cec386b35 100644 --- a/rust/crates/truapi-codegen/src/rust.rs +++ b/rust/crates/truapi-codegen/src/rust.rs @@ -23,11 +23,11 @@ pub use wasm_bridge::generate_wasm_bridge; pub use wire_table::generate_wire_table; /// Generates the Rust wire dispatcher and wire-table sources into `output_dir`. -pub fn generate(api: &ApiDefinition, output_dir: &Path) -> Result<()> { +pub fn generate(api: &ApiDefinition, output_dir: &Path, schema_hash: &str) -> Result<()> { fs::create_dir_all(output_dir)?; let dispatcher = generate_dispatcher(api)?; fs::write(output_dir.join("dispatcher.rs"), dispatcher)?; - let wire_table = generate_wire_table(api)?; + let wire_table = generate_wire_table(api, schema_hash)?; fs::write(output_dir.join("wire_table.rs"), wire_table)?; Ok(()) } @@ -158,6 +158,7 @@ mod tests { stop_id: None, interrupt_id: None, receive_id: None, + sensitive: false, }, docs: None, } @@ -180,6 +181,7 @@ mod tests { stop_id: None, interrupt_id: None, receive_id: None, + sensitive: false, }, docs: None, } @@ -197,6 +199,7 @@ mod tests { args: vec![], }]), docs: None, + codec_index: None, }]), docs: None, } @@ -275,9 +278,10 @@ mod tests { }], public_trait_order: vec!["Account".to_string()], types: vec![], + framework_types: Vec::new(), }; - let src = generate_wire_table(&api).expect("generate_wire_table"); + let src = generate_wire_table(&api, "testhash").expect("generate_wire_table"); let entries = parse_entries(&src); assert_eq!( entries, @@ -313,6 +317,7 @@ mod tests { ], public_trait_order: vec!["StatementStore".to_string(), "Preimage".to_string()], types: versioned_request_test_types(), + framework_types: Vec::new(), }; let dispatcher = generate_dispatcher(&api).expect("dispatcher"); @@ -325,7 +330,7 @@ mod tests { "dispatcher missing prefixed Preimage const:\n{dispatcher}" ); - let table = generate_wire_table(&api).expect("wire_table"); + let table = generate_wire_table(&api, "testhash").expect("wire_table"); let entries = parse_entries(&table); assert!( entries @@ -365,8 +370,10 @@ mod tests { ], public_trait_order: vec!["Foo".to_string(), "FooBar".to_string()], types: vec![], + framework_types: Vec::new(), }; - let err = generate_wire_table(&api).expect_err("duplicate wire method name must error"); + let err = generate_wire_table(&api, "testhash") + .expect_err("duplicate wire method name must error"); let msg = format!("{err}"); assert!( msg.contains("wire method name `foo_bar_baz` reused"), @@ -394,14 +401,15 @@ mod tests { }], public_trait_order: vec!["Permissions".to_string()], types: versioned_request_test_types(), + framework_types: Vec::new(), }; let dispatcher_a = generate_dispatcher(&api).expect("dispatcher a"); let dispatcher_b = generate_dispatcher(&api).expect("dispatcher b"); assert_eq!(dispatcher_a, dispatcher_b); - let table_a = generate_wire_table(&api).expect("wire_table a"); - let table_b = generate_wire_table(&api).expect("wire_table b"); + let table_a = generate_wire_table(&api, "testhash").expect("wire_table a"); + let table_b = generate_wire_table(&api, "testhash").expect("wire_table b"); assert_eq!(table_a, table_b); } @@ -423,8 +431,9 @@ mod tests { }], public_trait_order: vec!["Permissions".to_string()], types: vec![], + framework_types: Vec::new(), }; - let err = generate_wire_table(&api).expect_err("duplicate ids must error"); + let err = generate_wire_table(&api, "testhash").expect_err("duplicate ids must error"); let msg = format!("{err}"); assert!( msg.contains("wire id 10 reused"), @@ -486,9 +495,10 @@ mod tests { }], public_trait_order: vec!["Example".to_string()], types: Vec::new(), + framework_types: Vec::new(), }; - let error = generate_wire_table(&api) + let error = generate_wire_table(&api, "testhash") .expect_err(&format!("{method_name} must not allocate wire id 255")); let message = error.to_string(); assert!( @@ -545,8 +555,10 @@ mod tests { }], public_trait_order: vec!["Permissions".to_string()], types: vec![], + framework_types: Vec::new(), }; - let err = generate_wire_table(&api).expect_err("request kind + start_id must error"); + let err = + generate_wire_table(&api, "testhash").expect_err("request kind + start_id must error"); let msg = format!("{err}"); assert!( msg.contains("must not use subscription wire ids"), @@ -568,8 +580,10 @@ mod tests { }], public_trait_order: vec!["Account".to_string()], types: vec![], + framework_types: Vec::new(), }; - let err = generate_wire_table(&api).expect_err("subscription kind + request_id must error"); + let err = generate_wire_table(&api, "testhash") + .expect_err("subscription kind + request_id must error"); let msg = format!("{err}"); assert!( msg.contains("must not use request wire ids"), @@ -592,8 +606,10 @@ mod tests { }], public_trait_order: vec!["Permissions".to_string()], types: vec![], + framework_types: Vec::new(), }; - let err = generate_wire_table(&api).expect_err("missing request_id annotation must error"); + let err = generate_wire_table(&api, "testhash") + .expect_err("missing request_id annotation must error"); let msg = format!("{err}"); assert!( msg.contains("missing #[wire(request_id"), @@ -615,8 +631,10 @@ mod tests { }], public_trait_order: vec!["Account".to_string()], types: vec![], + framework_types: Vec::new(), }; - let err = generate_wire_table(&api).expect_err("missing start_id annotation must error"); + let err = generate_wire_table(&api, "testhash") + .expect_err("missing start_id annotation must error"); let msg = format!("{err}"); assert!( msg.contains("missing #[wire(start_id"), @@ -646,6 +664,7 @@ mod tests { }], public_trait_order: vec!["Permissions".to_string()], types: vec![], + framework_types: Vec::new(), }; let err = generate_dispatcher(&api).expect_err("two-param method must error"); let msg = format!("{err}"); @@ -679,6 +698,7 @@ mod tests { }], public_trait_order: vec!["Permissions".to_string()], types: vec![], + framework_types: Vec::new(), }; let err = generate_dispatcher(&api).expect_err("primitive response must error"); let msg = format!("{err}"); @@ -716,6 +736,7 @@ mod tests { versioned_test_type("ReqWrapper"), versioned_test_type("RespWrapper"), ], + framework_types: Vec::new(), }; let err = generate_dispatcher(&api).expect_err("raw error wrapper must error"); @@ -745,6 +766,7 @@ mod tests { versioned_test_type("RespWrapper"), versioned_test_type("ErrWrapper"), ], + framework_types: Vec::new(), }; let err = generate_dispatcher(&api).expect_err("missing target version must error"); @@ -781,6 +803,7 @@ mod tests { }], public_trait_order: vec!["Account".to_string()], types: vec![versioned_test_type("ItemWrapper")], + framework_types: Vec::new(), }; let err = generate_dispatcher(&api).expect_err("raw result subscription error must error"); diff --git a/rust/crates/truapi-codegen/src/rust/wire_table.rs b/rust/crates/truapi-codegen/src/rust/wire_table.rs index da47e24d4..fd23359cf 100644 --- a/rust/crates/truapi-codegen/src/rust/wire_table.rs +++ b/rust/crates/truapi-codegen/src/rust/wire_table.rs @@ -39,8 +39,9 @@ enum MethodEntry { Subscription(SubEntry), } -/// Emit the contents of `wire_table.rs`. -pub fn generate_wire_table(api: &ApiDefinition) -> Result { +/// Emit the contents of `wire_table.rs`. `schema_hash` is the wire-contract +/// fingerprint emitted as `TRUAPI_WIRE_SCHEMA_HASH`, identical to the TS client's. +pub fn generate_wire_table(api: &ApiDefinition, schema_hash: &str) -> Result { let mut method_entries: Vec<(String, MethodEntry)> = Vec::new(); let mut seen = BTreeMap::from([( RESERVED_PROTOCOL_ERROR_ID, @@ -72,7 +73,7 @@ pub fn generate_wire_table(api: &ApiDefinition) -> Result { MethodEntry::Subscription(SubEntry { start_id, .. }) => *start_id, }); - render(&method_entries) + render(&method_entries, schema_hash) } fn method_entry(trait_def: &TraitDef, method: &MethodDef) -> Result { @@ -173,7 +174,7 @@ fn insert_entry( Ok(()) } -fn render(methods: &[(String, MethodEntry)]) -> Result { +fn render(methods: &[(String, MethodEntry)], schema_hash: &str) -> Result { let mut out = String::new(); writedoc!( out, @@ -229,6 +230,19 @@ fn render(methods: &[(String, MethodEntry)]) -> Result { ) .unwrap(); + writedoc!( + out, + r#" + /// Fingerprint of this build's wire contract: frame ids, method legs, + /// sensitivity, and codec version, identical to the TS client's + /// `TRUAPI_WIRE_SCHEMA_HASH`. A host stamps it on each debug envelope so + /// the debugger refuses to decode a frame whose contract differs from + /// its own, even when the coarse handshake codec version is unchanged. + pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "{schema_hash}"; + "# + ) + .unwrap(); + // Per-method consts: the single source of truth for each method's ids. for (name, entry) in methods { let konst = const_name(name); diff --git a/rust/crates/truapi-codegen/src/rustdoc.rs b/rust/crates/truapi-codegen/src/rustdoc.rs index f84eaff20..63c1748b0 100644 --- a/rust/crates/truapi-codegen/src/rustdoc.rs +++ b/rust/crates/truapi-codegen/src/rustdoc.rs @@ -34,6 +34,11 @@ pub struct Item { pub docs: Option, /// Kind-dependent rustdoc payload, parsed lazily by helpers in this module. pub inner: serde_json::Value, + /// Attributes rustdoc recorded on the item, e.g. `#[codec(index = 0)]`. + /// Needed because the SCALE discriminant a variant ships on is the explicit + /// `codec(index = N)`, not its declaration order. + #[serde(default)] + pub attrs: Vec, } /// Resolves a rustdoc id to its fully-qualified path and item kind. @@ -58,6 +63,12 @@ pub struct ApiDefinition { pub public_trait_order: Vec, /// Data types referenced by the trait surface. pub types: Vec, + /// Framework types that are deliberately not emitted, but whose own shape is + /// still on the wire - `CallError`'s variants are the discriminant of every + /// error response. Kept so the wire schema hash can see them: excluding them + /// from the fingerprint let a variant be inserted, renumbering every error + /// discriminant, with no signal anywhere. + pub framework_types: Vec, } /// Trait extracted from the rustdoc index: name, methods, and rustdoc. @@ -121,6 +132,11 @@ pub struct WireAttrs { pub interrupt_id: Option, /// Subscription item frame discriminant. pub receive_id: Option, + /// Whether the method's payloads carry key material or bearer secrets. + /// Marked by `#[wire(..., sensitive)]`; folded into the wire schema-hash + /// fingerprint so a change in a frame's sensitivity classification is caught + /// as contract drift. + pub sensitive: bool, } /// Wire-shape classification of a trait method. @@ -231,6 +247,13 @@ pub struct VariantDef { pub fields: VariantFields, /// Rustdoc comment on the variant. pub docs: Option, + /// Explicit SCALE discriminant from `#[codec(index = N)]`, when the variant + /// carries one. `None` means the codec falls back to declaration order. + /// + /// This is the byte that actually ships. Fingerprinting the positional index + /// instead cannot see a renumbering that keeps declaration order - which is + /// exactly how RFC-0024 moved `Rejected` from `0x02` to `0x04`. + pub codec_index: Option, } /// Payload shape of an enum variant. @@ -335,9 +358,35 @@ pub fn extract_api(krate: &Crate) -> Result { } let mut types = Vec::new(); + let mut framework_types = Vec::new(); let mut generated_names = BTreeMap::new(); for (name, candidates) in type_candidates { if should_skip_type_name(&name) { + // Not emitted, but still fingerprinted: a shape change here changes + // the wire. Parse failures are ignored - several skipped names are + // markers or lifetimes with no data shape to record. + for candidate in &candidates { + let Some(item) = krate.index.get(&candidate.item_id) else { + continue; + }; + let module_path: Vec = candidate + .path + .iter() + .take(candidate.path.len().saturating_sub(1)) + .cloned() + .collect(); + let extracted = if candidate.kind == "struct" { + extract_struct(&candidate.item_id, item, krate, &names, module_path) + } else if candidate.kind == "enum" { + extract_enum(&candidate.item_id, item, krate, &names, module_path) + } else { + continue; + }; + if let Ok(def) = extracted { + framework_types.push(def); + break; + } + } continue; } @@ -387,10 +436,13 @@ pub fn extract_api(krate: &Crate) -> Result { traits.sort_by(|a, b| a.name.cmp(&b.name)); types.sort_by(|a, b| a.name.cmp(&b.name)); + framework_types.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(ApiDefinition { traits, public_trait_order, types, + framework_types, }) } @@ -819,6 +871,17 @@ fn extract_wire_attrs(docs: &str) -> WireAttrs { if line.starts_with("@wire_host_initiated") { attrs.host_initiated = true; } + if line.starts_with("@wire_sensitive=") { + // Fail CLOSED, and never downgrade. The previous + // `.parse::().ok().unwrap_or(false)` turned every unexpected + // value - `1`, `yes`, a typo - into "not sensitive", which is the wrong + // default for a flag that classifies secret-bearing methods. Only an + // explicit `false` leaves it clear, and `|=` means a later marker + // cannot undo an earlier `true`. + let value = line.trim_end().trim_start_matches("@wire_sensitive="); + attrs.sensitive |= value != "false"; + continue; + } for (needle, target) in [ ("@wire_request_id=", &mut attrs.request_id), ("@wire_response_id=", &mut attrs.response_id), @@ -1042,8 +1105,18 @@ pub(crate) fn resolve_type(ty: &serde_json::Value, names: &NameContext) -> Resul "Option", args, )?))), "Compact" => { - expect_single_arg("Compact", args)?; - Ok(TypeRef::Primitive("compact".to_string())) + // The width is carried in the primitive's NAME, not discarded. + // Emission still keys on the `compact` prefix, so generated + // output is unchanged - but the wire schema hash can now see the + // difference between `Compact` and `Compact`. Dropping + // it made every compact site render identically, so widening one + // left the fingerprint byte-identical while changing which values + // a peer can decode. + let inner = expect_single_arg("Compact", args)?; + let TypeRef::Primitive(width) = &inner else { + bail!("Compact must wrap a primitive integer, found {inner:?}"); + }; + Ok(TypeRef::Primitive(format!("compact<{width}>"))) } "OptionBool" => Ok(TypeRef::Primitive("optionBool".to_string())), "String" => { @@ -1308,6 +1381,7 @@ pub(crate) fn extract_enum( name: variant_name, fields, docs: clean_docs(variant_item.docs.as_deref()), + codec_index: codec_index_attr(&variant_item.attrs), }); } @@ -1320,6 +1394,33 @@ pub(crate) fn extract_enum( }) } +/// Read `#[codec(index = N)]` off a variant's rustdoc attributes. +/// +/// Rustdoc renders each attribute as a JSON object whose `other` key holds the +/// source text, so this matches on that text rather than a structured field. +fn codec_index_attr(attrs: &[serde_json::Value]) -> Option { + for attr in attrs { + let text = attr + .get("other") + .and_then(|value| value.as_str()) + .or_else(|| attr.as_str())?; + let Some(rest) = text.split("codec(index").nth(1) else { + continue; + }; + let digits: String = rest + .trim_start() + .trim_start_matches('=') + .trim_start() + .chars() + .take_while(char::is_ascii_digit) + .collect(); + if let Ok(index) = digits.parse::() { + return Some(index); + } + } + None +} + fn extract_variant_fields( variant_inner: Option<&serde_json::Value>, krate: &Crate, @@ -1484,7 +1585,7 @@ mod tests { #[test] fn clean_docs_strips_wire_markers() { - let docs = "Trait summary.\n\n@wire_request_id=7\n@service_required_execution=Chat\n"; + let docs = "Trait summary.\n\n@wire_request_id=7\n@wire_sensitive=true\n@service_required_execution=Chat\n"; assert_eq!(clean_docs(Some(docs)).as_deref(), Some("Trait summary.")); } @@ -1502,6 +1603,18 @@ mod tests { assert_eq!(trait_def.public_docs().as_deref(), Some("Chat operations.")); } + #[test] + fn extract_wire_attrs_reads_sensitive_flag() { + let sensitive = extract_wire_attrs("@wire_request_id=114\n@wire_sensitive=true"); + assert_eq!(sensitive.request_id, Some(114)); + assert!(sensitive.sensitive); + + // Absent marker ⇒ not sensitive (the default for every unmarked method). + let plain = extract_wire_attrs("@wire_request_id=22"); + assert_eq!(plain.request_id, Some(22)); + assert!(!plain.sensitive); + } + #[test] fn parse_accepts_tested_format_version() { let json = format!(r#"{{ "format_version": {MIN_FORMAT_VERSION}, "index": {{}} }}"#); diff --git a/rust/crates/truapi-codegen/src/ts.rs b/rust/crates/truapi-codegen/src/ts.rs index 41d542e7f..0b5fc6b8d 100644 --- a/rust/crates/truapi-codegen/src/ts.rs +++ b/rust/crates/truapi-codegen/src/ts.rs @@ -490,6 +490,12 @@ pub fn generate( let wire_table_code = generate_wire_table(api, target_version)?; fs::write(Path::new(output_dir).join("wire-table.ts"), wire_table_code)?; + let decode_table_code = generate_decode_table(api, target_version)?; + fs::write( + Path::new(output_dir).join("wire-decode.ts"), + decode_table_code, + )?; + Ok(()) } @@ -656,6 +662,283 @@ fn generate_wire_table(api: &ApiDefinition, target_version: u32) -> Result Result> { + let wrappers = collect_versioned_wrappers(api); + let types = types_by_name(api); + let mut seen: BTreeMap = BTreeMap::new(); + for trait_def in &api.traits { + for method in &trait_def.methods { + if !method_is_included(trait_def, method, &wrappers, target_version)? { + continue; + } + let wire_ids = wire_ids_for_method(trait_def, method)?; + let payload = method_payload_signature(method, &types); + for (id, tag) in wire_ids.entries(&method.name) { + if let Some((existing, _, _, _)) = seen.insert( + id, + ( + tag.clone(), + method.wire.sensitive, + method.wire.host_initiated, + payload.clone(), + ), + ) { + bail!("wire id {id} reused: `{existing}` and `{tag}` collide"); + } + } + } + } + Ok(seen + .into_iter() + .map(|(id, (tag, sensitive, host_initiated, payload))| { + (id, tag, sensitive, host_initiated, payload) + }) + .collect()) +} + +/// Index the API's user-defined types by their emitted name, so a signature walk +/// can resolve a [`TypeRef::Named`] to its actual shape. +fn types_by_name(api: &ApiDefinition) -> HashMap<&str, &TypeDef> { + // Framework types are included even though they are never emitted: their + // shape is still on the wire. `CallError` is the one that matters - it wraps + // every error leg, so its variant list is the discriminant of every error + // response, and leaving it out let a variant be inserted (renumbering every + // discriminant on every error) without moving the fingerprint at all. + api.types + .iter() + .chain(api.framework_types.iter()) + .map(|def| (def.name.as_str(), def)) + .collect() +} + +/// Structural signature of everything a method puts on the wire: its parameters +/// (the request/start payload) and its return shape (the response/item payload). +/// +/// Folded into the wire schema hash so the fingerprint moves when a payload's +/// *layout* changes, not only when a frame id or method name does. +fn method_payload_signature(method: &MethodDef, types: &HashMap<&str, &TypeDef>) -> String { + let mut out = String::new(); + for param in &method.params { + let sig = type_signature(¶m.type_ref, types, &mut Vec::new()); + let _ = write!(out, "{}:{sig},", param.name); + } + out.push_str("->"); + match &method.return_type { + ReturnType::Result { ok, err } => { + let _ = write!( + out, + "res<{},{}>", + type_signature(ok, types, &mut Vec::new()), + type_signature(err, types, &mut Vec::new()) + ); + } + ReturnType::Subscription(item) => { + let _ = write!(out, "sub<{}>", type_signature(item, types, &mut Vec::new())); + } + ReturnType::ResultSubscription { item, err } => { + let _ = write!( + out, + "ressub<{},{}>", + type_signature(item, types, &mut Vec::new()), + type_signature(err, types, &mut Vec::new()) + ); + } + } + out +} + +/// Canonical structural rendering of a type: field order and field types for a +/// struct, positional variant indices and payloads for an enum, resolved +/// transitively. +/// +/// Two layouts that encode differently under SCALE cannot render the same +/// string: field order, field types, variant order, and arity all appear. A type +/// this crate does not own (external or generic) degrades to its name, which is +/// the most that is knowable from rustdoc. `seen` guards recursive types. +fn type_signature( + type_ref: &TypeRef, + types: &HashMap<&str, &TypeDef>, + seen: &mut Vec, +) -> String { + match type_ref { + TypeRef::Primitive(name) => name.clone(), + TypeRef::Unit => "()".to_string(), + TypeRef::Generic(name) => format!("generic:{name}"), + TypeRef::Vec(inner) => format!("vec<{}>", type_signature(inner, types, seen)), + TypeRef::Option(inner) => format!("opt<{}>", type_signature(inner, types, seen)), + TypeRef::Array(inner, len) => { + format!("[{};{len}]", type_signature(inner, types, seen)) + } + TypeRef::Tuple(items) => { + let inner: Vec = items + .iter() + .map(|item| type_signature(item, types, seen)) + .collect(); + format!("({})", inner.join(",")) + } + TypeRef::Named { name, args } => { + let rendered_args: Vec = args + .iter() + .map(|arg| type_signature(arg, types, seen)) + .collect(); + let suffix = if rendered_args.is_empty() { + String::new() + } else { + format!("<{}>", rendered_args.join(",")) + }; + // A type already on the walk stack is recursive; naming it closes the + // cycle without losing that the edge exists. + if seen.iter().any(|entry| entry == name) { + return format!("rec:{name}{suffix}"); + } + let Some(def) = types.get(name.as_str()) else { + // Degrading silently to the bare name is what let a payload's + // shape change without moving the fingerprint - the type's own + // fields or variants simply stop being hashed. Marking it keeps + // the blind spot visible in the canonical string, and + // `every_wire_reachable_type_resolves` fails the build if a new + // one ever appears. + return format!("UNRESOLVED<{name}>{suffix}"); + }; + seen.push(name.clone()); + let body = match &def.kind { + TypeDefKind::Alias(inner) => { + format!("={}", type_signature(inner, types, seen)) + } + TypeDefKind::Struct(fields) => { + let rendered: Vec = fields + .iter() + .map(|field| { + format!( + "{}:{}", + field.name, + type_signature(&field.type_ref, types, seen) + ) + }) + .collect(); + format!("{{{}}}", rendered.join(",")) + } + TypeDefKind::TupleStruct(items) => { + let rendered: Vec = items + .iter() + .map(|item| type_signature(item, types, seen)) + .collect(); + format!("({})", rendered.join(",")) + } + TypeDefKind::Enum(variants) => { + let rendered: Vec = variants + .iter() + .enumerate() + .map(|(index, variant)| { + let payload = match &variant.fields { + VariantFields::Unit => String::new(), + VariantFields::Unnamed(items) => { + let inner: Vec = items + .iter() + .map(|item| type_signature(item, types, seen)) + .collect(); + format!("({})", inner.join(",")) + } + VariantFields::Named(fields) => { + let inner: Vec = fields + .iter() + .map(|field| { + format!( + "{}:{}", + field.name, + type_signature(&field.type_ref, types, seen) + ) + }) + .collect(); + format!("{{{}}}", inner.join(",")) + } + }; + // The SCALE discriminant is the explicit + // `#[codec(index = N)]` when the variant carries one, + // and the positional index otherwise. Hash whichever + // actually ships: fingerprinting position alone is + // blind to a renumbering that leaves declaration order + // untouched, which is how RFC-0024 moved `Rejected` + // from `0x02` to `0x04` without any signal. + let discriminant = variant + .codec_index + .map(|explicit| explicit.to_string()) + .unwrap_or_else(|| index.to_string()); + format!("{discriminant}:{}{payload}", variant.name) + }) + .collect(); + format!("|{}|", rendered.join(";")) + } + }; + seen.pop(); + format!("{name}{suffix}{body}") + } + } +} + +/// A stable fingerprint of the wire contract: every frame id, the method leg it +/// resolves to, and its sensitivity, folded together with the codec version. +/// Two builds whose frame tables differ - a reassigned id, a renamed or +/// added/removed method, or a flipped `#[wire(sensitive)]` - produce different +/// hashes even when the handshake `codec_version` is unchanged, which is the +/// case the coarse codec number cannot see. Emitted as `TRUAPI_WIRE_SCHEMA_HASH` +/// on both the TS and Rust sides so a host stamps it on every debug envelope and +/// the debugger refuses to decode a frame whose contract differs from its own. +pub(crate) fn wire_schema_hash( + api: &ApiDefinition, + target_version: u32, + codec_version: u8, +) -> Result { + let mut canonical = format!("codec={codec_version}\n"); + let mut unresolved: BTreeSet = BTreeSet::new(); + for (id, tag, sensitive, host_initiated, payload) in wire_id_rows(api, target_version)? { + let flag = u8::from(sensitive); + let initiator = u8::from(host_initiated); + for marker in payload.split("UNRESOLVED<").skip(1) { + unresolved.insert(marker.chars().take_while(|c| *c != '>').collect()); + } + canonical.push_str(&format!("{id}:{tag}:{flag}:{initiator}:{payload}\n")); + } + // Fail the BUILD, not a test. A type that does not resolve contributes only + // its name, so its own fields or variants stop being fingerprinted and can + // change undetected - `CallError` sat on every error leg exactly that way, + // and inserting a variant renumbered every error discriminant while the hash + // and the whole generated tree stayed byte-identical. Enforcing it here means + // a future addition to the extractor's skip list cannot re-open the hole, and + // does not depend on a test being wired up to notice. + if !unresolved.is_empty() { + bail!( + "wire schema hash cannot see the shape of {unresolved:?}: these types are \ + reachable from a wire payload but are not in the API definition, so a \ + change to their fields or variants would not move the fingerprint. Add \ + them to `ApiDefinition::framework_types` rather than letting the \ + signature degrade to a bare name." + ); + } + // FNV-1a 64-bit: deterministic across platforms and Rust versions (unlike + // `DefaultHasher`), dependency-free, and ample for a contract fingerprint. + let mut hash: u64 = 0xcbf2_9ce4_8422_2325; + for byte in canonical.as_bytes() { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + Ok(format!("{hash:016x}")) +} + fn method_is_included( trait_def: &TraitDef, method: &MethodDef, @@ -930,6 +1213,7 @@ fn generate_types(api: &ApiDefinition, target_version: u32) -> Result { fn generate_client(api: &ApiDefinition, target_version: u32, codec_version: u8) -> Result { validate_versioned_wrapper_shapes(api)?; + let schema_hash = wire_schema_hash(api, target_version, codec_version)?; let mut out = String::new(); writedoc!( out, @@ -948,6 +1232,7 @@ fn generate_client(api: &ApiDefinition, target_version: u32, codec_version: u8) export type {{ ObservableLike, ObservableSource, Observer, Result, Subscription, TrUApiTransport }}; export const TRUAPI_VERSION = {target_version} as const; export const TRUAPI_CODEC_VERSION = {codec_version} as const; + export const TRUAPI_WIRE_SCHEMA_HASH = "{schema_hash}" as const; function toSubscriptionError(error: unknown): SubscriptionError {{ if (error instanceof SubscriptionError) return error as SubscriptionError; @@ -1081,6 +1366,172 @@ fn generate_client(api: &ApiDefinition, target_version: u32, codec_version: u8) Ok(out) } +/// Generates the dev-only wire decode table (`wire-decode.ts`): a map from wire +/// `frameId` to a decoder that turns a frame's SCALE payload into a plain JS +/// value. It re-derives the exact request/response/subscription codec +/// expressions the client emitter builds (via [`emit_payload`], +/// [`emit_response`], [`emit_error_response`], and +/// [`versioned_result_codec_expr`]), so a debugger decodes wire frames against +/// the same generated codecs. Subscription `start` and `receive` frames are +/// covered; `stop`/`interrupt` frames are intentionally skipped. +fn generate_decode_table(api: &ApiDefinition, target_version: u32) -> Result { + let ctx = codec_context(&[]); + let wrappers = collect_versioned_wrappers(api); + let services = public_services(api)?; + + // (wire id, emitted table line) pairs, sorted by wire id for a stable, + // wire-ordered file that matches the wire-table layout. + let mut entries: Vec<(u8, String)> = Vec::new(); + + for service in &services { + let trait_def = service.trait_def; + for method in included_methods(trait_def, &wrappers, target_version)? { + let wire_const = wire_const_name(&trait_def.name, &method.name); + let wire_version = method_wire_version(method, &wrappers, target_version)?; + let payload = emit_payload(&method.params, &wrappers, &ctx, wire_version)?; + let wire_ids = wire_ids_for_method(trait_def, method)?; + + match (&method.kind, &method.return_type) { + (MethodKind::Request, ReturnType::Result { ok, err }) => { + let ExpandedWireIds::Request { + request_id, + response_id, + } = wire_ids + else { + unreachable!("request method resolved to subscription wire ids"); + }; + let response = emit_response(ok, &wrappers, &ctx, wire_version)?; + let error = emit_error_response(err, &wrappers, &ctx, wire_version)?; + let response_codec = match wire_version { + Some(version) => versioned_result_codec_expr( + version, + &response.inner_codec_expr, + &error.inner_codec_expr, + )?, + None => format!( + "S.Result({}, {})", + response.wire_codec_expr, error.wire_codec_expr + ), + }; + let value_suffix = if wire_version.is_some() { ".value" } else { "" }; + entries.push(( + request_id, + format!( + " [W.{wire_const}.request]: (payload) => {}.dec(payload),", + payload.wire_codec_expr + ), + )); + entries.push(( + response_id, + format!( + " [W.{wire_const}.response]: (payload) => {response_codec}.dec(payload){value_suffix}," + ), + )); + } + (MethodKind::Subscription, ReturnType::Subscription(ty)) => { + let response = emit_response(ty, &wrappers, &ctx, wire_version)?; + push_subscription_entries( + &mut entries, + &wire_const, + &payload, + &response, + wire_ids, + wire_version, + )?; + } + (MethodKind::ResultSubscription, ReturnType::ResultSubscription { item, .. }) => { + let response = emit_response(item, &wrappers, &ctx, wire_version)?; + push_subscription_entries( + &mut entries, + &wire_const, + &payload, + &response, + wire_ids, + wire_version, + )?; + } + (kind, return_type) => { + bail!( + "Generator internal mismatch for method `{}`: kind {:?} does not match return type {:?}", + method.name, + kind, + return_type + ); + } + } + } + } + + entries.sort_by_key(|(id, _)| *id); + + let mut out = String::new(); + writedoc!( + out, + r#" + // Auto-generated by truapi-codegen. Do not edit. + + import * as S from '../scale.js'; + import * as T from './types.js'; + import * as W from './wire-table.js'; + + /** Dev-only: decode a wire frame's SCALE payload to a plain JS value, keyed by frameId. + * Request/response/subscription frames only; unknown ids are absent (caller falls back to bytes). */ + export const WIRE_DECODE_TABLE: Record unknown> = {{ + "# + ) + .unwrap(); + for (_, line) in &entries { + out.push_str(line); + out.push('\n'); + } + out.push_str("};\n"); + + Ok(out) +} + +/// Emits the `.start` (start payload codec) and `.receive` (item codec) decode +/// entries for a subscription method, mirroring the client's `payload` +/// encoding and `decodeItem` expression. `stop`/`interrupt` frames are skipped. +fn push_subscription_entries( + entries: &mut Vec<(u8, String)>, + wire_const: &str, + payload: &PayloadEmission, + response: &ResponseEmission, + wire_ids: ExpandedWireIds, + wire_version: Option, +) -> Result<()> { + let ExpandedWireIds::Subscription { + start_id, + receive_id, + .. + } = wire_ids + else { + unreachable!("subscription method resolved to request wire ids"); + }; + let item_value = if let Some(version) = wire_version { + versioned_value_expr( + &format!("{}.dec(payload)", response.wire_codec_expr), + &response.wire_type_ts, + &response.inner_type_ts, + version, + ) + } else { + format!("{}.dec(payload)", response.wire_codec_expr) + }; + entries.push(( + start_id, + format!( + " [W.{wire_const}.start]: (payload) => {}.dec(payload),", + payload.wire_codec_expr + ), + )); + entries.push(( + receive_id, + format!(" [W.{wire_const}.receive]: (payload) => {item_value},"), + )); + Ok(()) +} + fn write_observable_helper(out: &mut String) { writedoc!( out, @@ -2245,7 +2696,7 @@ fn codec_expr_mode( "u32" => Ok("S.u32".to_string()), "u64" => Ok("S.u64".to_string()), "u128" => Ok("S.u128".to_string()), - "compact" => Ok("S.compact".to_string()), + name if name.starts_with("compact") => Ok("S.compact".to_string()), "optionBool" => Ok("S.OptionBool".to_string()), "i8" => Ok("S.i8".to_string()), "i16" => Ok("S.i16".to_string()), @@ -2330,7 +2781,7 @@ fn ts_type_with_named(ty: &TypeRef, qualified: bool, mode: NameMode<'_>) -> Resu "bool" => Ok("boolean".to_string()), "u8" | "u16" | "u32" | "i8" | "i16" | "i32" | "f32" | "f64" => Ok("number".to_string()), "u64" | "u128" | "i64" | "i128" => Ok("bigint".to_string()), - "compact" => Ok("number | bigint".to_string()), + name if name.starts_with("compact") => Ok("number | bigint".to_string()), "optionBool" => Ok("boolean | undefined".to_string()), "str" => Ok("string".to_string()), _ => bail!("Unsupported primitive type `{name}` in TypeScript type generation"), @@ -2532,6 +2983,267 @@ mod tests { } } + /// Build a one-method API whose request payload is `struct Payload`, with the + /// given named fields, so a test can vary only the payload layout. + fn api_with_payload_fields(fields: Vec<(&str, TypeRef)>) -> ApiDefinition { + let payload = TypeDef { + name: "Payload".to_string(), + module_path: Vec::new(), + generic_params: Vec::new(), + kind: TypeDefKind::Struct( + fields + .into_iter() + .map(|(name, type_ref)| FieldDef { + name: name.to_string(), + type_ref, + docs: None, + }) + .collect(), + ), + docs: None, + }; + let method = MethodDef { + name: "do_thing".to_string(), + kind: MethodKind::Request, + params: vec![ParamDef { + name: "request".to_string(), + type_ref: TypeRef::Named { + name: "Payload".to_string(), + args: Vec::new(), + }, + }], + return_type: ReturnType::Result { + ok: TypeRef::Unit, + err: TypeRef::Unit, + }, + wire: request_wire(Some(7)), + docs: None, + }; + ApiDefinition { + traits: vec![TraitDef { + name: "Thing".to_string(), + module_path: Vec::new(), + methods: vec![method], + docs: None, + }], + public_trait_order: vec!["Thing".to_string()], + types: vec![payload], + framework_types: Vec::new(), + } + } + + #[test] + fn schema_hash_moves_when_a_payload_field_type_changes() { + // The drift class this fingerprint exists to catch: same frame ids, same + // method names, same sensitivity - only a field's width changed. A newer + // host's bytes would otherwise decode on the old table without throwing, + // silently yielding wrong values (the shape of the getAccount P0). + let before = api_with_payload_fields(vec![ + ("ring_index", TypeRef::Primitive("u32".to_string())), + ("ring_revision", TypeRef::Primitive("u32".to_string())), + ]); + let after = api_with_payload_fields(vec![ + ("ring_index", TypeRef::Primitive("u64".to_string())), + ("ring_revision", TypeRef::Primitive("u32".to_string())), + ]); + + assert_ne!( + wire_schema_hash(&before, 1, 1).unwrap(), + wire_schema_hash(&after, 1, 1).unwrap(), + ); + } + + #[test] + fn schema_hash_moves_when_same_width_payload_fields_are_reordered() { + // Nastier than a width change: the frame length is identical, so no + // arithmetic check can see it and the decode cannot fail - the values + // simply swap. + let before = api_with_payload_fields(vec![ + ("ring_index", TypeRef::Primitive("u32".to_string())), + ("ring_revision", TypeRef::Primitive("u32".to_string())), + ]); + let after = api_with_payload_fields(vec![ + ("ring_revision", TypeRef::Primitive("u32".to_string())), + ("ring_index", TypeRef::Primitive("u32".to_string())), + ]); + + assert_ne!( + wire_schema_hash(&before, 1, 1).unwrap(), + wire_schema_hash(&after, 1, 1).unwrap(), + ); + } + + #[test] + fn schema_hash_is_stable_for_an_unchanged_contract() { + // The fingerprint must not be noisy: an identical contract hashes + // identically, or every host would look drifted. + let api = + api_with_payload_fields(vec![("ring_index", TypeRef::Primitive("u32".to_string()))]); + + assert_eq!( + wire_schema_hash(&api, 1, 1).unwrap(), + wire_schema_hash(&api, 1, 1).unwrap(), + ); + } + + #[test] + fn type_signature_terminates_on_a_recursive_type() { + // `struct Node { next: Option }` must not recurse forever. + let node = TypeDef { + name: "Node".to_string(), + module_path: Vec::new(), + generic_params: Vec::new(), + kind: TypeDefKind::Struct(vec![FieldDef { + name: "next".to_string(), + type_ref: TypeRef::Option(Box::new(TypeRef::Named { + name: "Node".to_string(), + args: Vec::new(), + })), + docs: None, + }]), + docs: None, + }; + let types: HashMap<&str, &TypeDef> = [("Node", &node)].into_iter().collect(); + + let sig = type_signature( + &TypeRef::Named { + name: "Node".to_string(), + args: Vec::new(), + }, + &types, + &mut Vec::new(), + ); + + assert!(sig.contains("rec:Node"), "unexpected signature: {sig}"); + } + + #[test] + fn schema_hash_moves_when_a_compact_width_changes() { + // `Compact` and `Compact` encode the same small values the same + // way, so the frame length does not change - but the wider type accepts + // values the narrower decoder rejects. The extractor used to discard the + // argument entirely, collapsing every compact site to one token, so a + // widening left the fingerprint byte-identical. + let build = |width: &str| { + api_with_payload_fields(vec![( + "size", + TypeRef::Primitive(format!("compact<{width}>")), + )]) + }; + + assert_ne!( + wire_schema_hash(&build("u32"), 1, 1).unwrap(), + wire_schema_hash(&build("u64"), 1, 1).unwrap(), + ); + } + + #[test] + fn schema_hash_moves_when_an_enum_variant_is_reordered() { + // Variant position is the SCALE discriminant, so a reorder silently + // renumbers every variant on the wire. + let variant = |name: &str| VariantDef { + name: name.to_string(), + fields: VariantFields::Unit, + docs: None, + codec_index: None, + }; + let build = |names: [&str; 2]| { + let enum_def = TypeDef { + name: "Choice".to_string(), + module_path: Vec::new(), + generic_params: Vec::new(), + kind: TypeDefKind::Enum(names.iter().map(|n| variant(n)).collect()), + docs: None, + }; + let method = MethodDef { + name: "do_thing".to_string(), + kind: MethodKind::Request, + params: vec![ParamDef { + name: "choice".to_string(), + type_ref: TypeRef::Named { + name: "Choice".to_string(), + args: Vec::new(), + }, + }], + return_type: ReturnType::Result { + ok: TypeRef::Unit, + err: TypeRef::Unit, + }, + wire: request_wire(Some(7)), + docs: None, + }; + ApiDefinition { + traits: vec![TraitDef { + name: "Thing".to_string(), + module_path: Vec::new(), + methods: vec![method], + docs: None, + }], + public_trait_order: vec!["Thing".to_string()], + types: vec![enum_def], + framework_types: Vec::new(), + } + }; + + assert_ne!( + wire_schema_hash(&build(["Allow", "Deny"]), 1, 1).unwrap(), + wire_schema_hash(&build(["Deny", "Allow"]), 1, 1).unwrap(), + ); + } + + #[test] + fn an_unresolvable_wire_reachable_type_fails_the_build() { + // The guard that replaced an env-gated test which asserted nothing when + // the variable was unset. `Missing` is referenced by the payload but is + // absent from both `types` and `framework_types`, so its shape cannot be + // fingerprinted - exactly the state `CallError` was in. + let method = MethodDef { + name: "do_thing".to_string(), + kind: MethodKind::Request, + params: vec![ParamDef { + name: "request".to_string(), + type_ref: TypeRef::Named { + name: "Missing".to_string(), + args: Vec::new(), + }, + }], + return_type: ReturnType::Result { + ok: TypeRef::Unit, + err: TypeRef::Unit, + }, + wire: request_wire(Some(7)), + docs: None, + }; + let api = ApiDefinition { + traits: vec![TraitDef { + name: "Thing".to_string(), + module_path: Vec::new(), + methods: vec![method], + docs: None, + }], + public_trait_order: vec!["Thing".to_string()], + types: Vec::new(), + framework_types: Vec::new(), + }; + + let err = wire_schema_hash(&api, 1, 1) + .expect_err("an unresolvable payload type must fail codegen"); + assert!( + format!("{err}").contains("Missing"), + "the error must name the offending type: {err}" + ); + } + + #[test] + fn a_resolvable_payload_hashes_without_complaint() { + // The negative control: the guard must not fire on an ordinary payload, + // or every codegen run would fail. + let api = + api_with_payload_fields(vec![("ring_index", TypeRef::Primitive("u32".to_string()))]); + + assert!(wire_schema_hash(&api, 1, 1).is_ok()); + } + #[test] fn service_display_name_formats_known_acronyms() { let json_rpc = TraitDef { @@ -2586,6 +3298,7 @@ mod tests { }], public_trait_order: Vec::new(), types: Vec::new(), + framework_types: Vec::new(), } } @@ -2630,6 +3343,20 @@ mod tests { } } + /// An empty struct `TypeDef`, so a synthetic fixture's payload types resolve. + /// A fixture that references a name it never defines is not a realistic API, + /// and the schema-hash guard rejects it for the same reason it rejects real + /// drift: an unresolvable type contributes only its name to the fingerprint. + fn empty_struct(name: &str) -> TypeDef { + TypeDef { + name: name.to_string(), + module_path: Vec::new(), + generic_params: Vec::new(), + kind: TypeDefKind::Struct(Vec::new()), + docs: None, + } + } + fn versioned_tuple_wrapper_variants(name: &str, variants: &[(u32, &str)]) -> TypeDef { TypeDef { name: name.to_string(), @@ -2642,6 +3369,7 @@ mod tests { name: format!("V{version}"), fields: VariantFields::Unnamed(vec![named_type(inner)]), docs: None, + codec_index: None, }) .collect(), ), @@ -2695,11 +3423,13 @@ mod tests { name: "V1".to_string(), fields: VariantFields::Named(fields.clone()), docs: None, + codec_index: None, }, VariantDef { name: "V2".to_string(), fields: VariantFields::Named(fields), docs: None, + codec_index: None, }, ]), docs: None, @@ -2720,6 +3450,7 @@ mod tests { args: Vec::new(), }]), docs: None, + codec_index: None, }, VariantDef { name: "V10".to_string(), @@ -2728,6 +3459,7 @@ mod tests { args: Vec::new(), }]), docs: None, + codec_index: None, }, VariantDef { name: "V2".to_string(), @@ -2736,6 +3468,7 @@ mod tests { args: Vec::new(), }]), docs: None, + codec_index: None, }, ]), docs: None, @@ -2779,6 +3512,7 @@ mod tests { traits: Vec::new(), public_trait_order: Vec::new(), types: Vec::new(), + framework_types: Vec::new(), }; assert_eq!(latest_wire_version(&api), 1); } @@ -2793,6 +3527,7 @@ mod tests { versioned_tuple_wrapper_variants("TwoWrapper", &[(1, "Legacy"), (3, "Latest")]), versioned_tuple_wrapper_variants("ThreeWrapper", &[(2, "Middle")]), ], + framework_types: Vec::new(), }; assert_eq!(latest_wire_version(&api), 3); } @@ -2823,6 +3558,37 @@ mod tests { ); } + #[test] + fn generate_decode_table_emits_frame_keyed_decoders() { + let api = ApiDefinition { + traits: vec![TraitDef { + name: "Example".to_string(), + module_path: Vec::new(), + methods: vec![ + request_method("feature_supported", Some(2)), + subscription_method("stream", Some(10)), + ], + docs: None, + }], + public_trait_order: vec!["Example".to_string()], + types: Vec::new(), + framework_types: Vec::new(), + }; + + let source = generate_decode_table(&api, 2).expect("generate decode table"); + + assert!(source.contains("export const WIRE_DECODE_TABLE")); + assert!(source.contains("(payload: Uint8Array) => unknown")); + assert!(source.contains("[W.EXAMPLE_FEATURE_SUPPORTED.request]")); + assert!(source.contains("[W.EXAMPLE_FEATURE_SUPPORTED.response]")); + assert!(source.contains("[W.EXAMPLE_STREAM.start]")); + assert!(source.contains("[W.EXAMPLE_STREAM.receive]")); + assert!(source.contains(".dec(payload)")); + // stop/interrupt subscription frames are intentionally skipped. + assert!(!source.contains(".stop]")); + assert!(!source.contains(".interrupt]")); + } + #[test] fn generate_wire_table_rejects_duplicate_ids() { let err = generate_wire_table( @@ -2911,6 +3677,7 @@ mod tests { docs: None, }], public_trait_order: Vec::new(), + framework_types: Vec::new(), types: vec![ versioned_tuple_wrapper_variants("FutureRequest", &[(2, "FutureRequestV2")]), versioned_tuple_wrapper_variants("FutureResponse", &[(2, "FutureResponseV2")]), @@ -3016,6 +3783,7 @@ mod tests { versioned_tuple_wrapper_variants("FutureError", &[(2, "FutureErrorV2")]), versioned_tuple_wrapper_variants("FutureItem", &[(2, "FutureItemV2")]), ], + framework_types: Vec::new(), }; let source = generate_wire_table(&api, 1).expect("generate wire table"); @@ -3063,7 +3831,11 @@ mod tests { versioned_tuple_wrapper_variants("FutureRequest", &[(2, "FutureRequestV2")]), versioned_tuple_wrapper_variants("FutureResponse", &[(2, "FutureResponseV2")]), versioned_tuple_wrapper_variants("FutureError", &[(2, "FutureErrorV2")]), + empty_struct("LegacyErrorV1"), + empty_struct("LegacyRequestV1"), + empty_struct("LegacyResponseV1"), ], + framework_types: Vec::new(), }; let source = generate_client(&api, 1, 1).expect("generate client"); @@ -3108,7 +3880,12 @@ mod tests { types: vec![ versioned_tuple_wrapper("ExampleRequest", "LegacyRequest", "LatestRequest"), versioned_tuple_wrapper("ExampleResponse", "LegacyResponse", "LatestResponse"), + empty_struct("LatestRequest"), + empty_struct("LatestResponse"), + empty_struct("LegacyRequest"), + empty_struct("LegacyResponse"), ], + framework_types: Vec::new(), }; let client_source = generate_client(&api, 2, 1).expect("generate client"); @@ -3176,6 +3953,7 @@ mod tests { single_field_struct("V01ExampleError", "legacy_code", "u8"), single_field_struct("V02ExampleError", "latest_code", "u32"), ], + framework_types: Vec::new(), }; let source = generate_types(&api, 2).expect("generate types"); @@ -3224,7 +4002,11 @@ mod tests { types: vec![ versioned_tuple_wrapper_variants("ExampleRequest", &[(1, "LegacyRequest")]), versioned_tuple_wrapper("ExampleResponse", "LegacyResponse", "LatestResponse"), + empty_struct("LatestResponse"), + empty_struct("LegacyRequest"), + empty_struct("LegacyResponse"), ], + framework_types: Vec::new(), }; let client_source = generate_client(&api, 2, 1).expect("generate client"); @@ -3271,6 +4053,7 @@ mod tests { named_field_versioned_wrapper("ExampleRequest"), versioned_tuple_wrapper("ExampleResponse", "LegacyResponse", "LatestResponse"), ], + framework_types: Vec::new(), }; let err = generate_client(&api, 2, 1).expect_err("named field wrapper rejected"); diff --git a/rust/crates/truapi-codegen/tests/golden/wire_table.rs b/rust/crates/truapi-codegen/tests/golden/wire_table.rs index 3a3783b85..0affdd39d 100644 --- a/rust/crates/truapi-codegen/tests/golden/wire_table.rs +++ b/rust/crates/truapi-codegen/tests/golden/wire_table.rs @@ -45,6 +45,12 @@ pub enum WireKind { /// Subscription method. Subscription(SubscriptionFrameIds), } +/// Fingerprint of this build's wire contract: frame ids, method legs, +/// sensitivity, and codec version, identical to the TS client's +/// `TRUAPI_WIRE_SCHEMA_HASH`. A host stamps it on each debug envelope so +/// the debugger refuses to decode a frame whose contract differs from +/// its own, even when the coarse handshake codec version is unchanged. +pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "4c260811b036d4ef"; /// Wire discriminants for `system_handshake`. pub const SYSTEM_HANDSHAKE: RequestFrameIds = RequestFrameIds { diff --git a/rust/crates/truapi-macros/src/lib.rs b/rust/crates/truapi-macros/src/lib.rs index 8e27efc4d..649a97575 100644 --- a/rust/crates/truapi-macros/src/lib.rs +++ b/rust/crates/truapi-macros/src/lib.rs @@ -38,6 +38,7 @@ struct WireArgs { stop_id: Option, interrupt_id: Option, receive_id: Option, + sensitive: bool, } struct ServiceArgs { @@ -77,24 +78,32 @@ impl Parse for WireArgs { while !input.is_empty() { let key: Ident = input.parse()?; + if key == "host_initiated" { if args.host_initiated { return Err(syn::Error::new(key.span(), "duplicate `host_initiated`")); } args.host_initiated = true; - if input.is_empty() { - break; + } else if key == "sensitive" { + // `sensitive` is a bare flag with no `= N` value: it classifies + // the method's payloads as carrying key material or bearer + // secrets. The classification is folded into the wire + // schema-hash fingerprint, so a change in a frame's sensitivity + // is caught as contract drift. It suppresses no decoding: it + // reaches neither the generated TS nor any runtime. + if args.sensitive { + return Err(syn::Error::new(key.span(), "duplicate `sensitive`")); } - input.parse::()?; - continue; - } - input.parse::()?; - let lit: LitInt = input.parse()?; - let value = lit.base10_parse().map_err(|err| { - syn::Error::new(lit.span(), format!("wire id must fit in a u8: {err}")) - })?; + args.sensitive = true; + } else { + input.parse::()?; + let lit: LitInt = input.parse()?; + let value = lit.base10_parse().map_err(|err| { + syn::Error::new(lit.span(), format!("wire id must fit in a u8: {err}")) + })?; - set_id(&mut args, &key, value)?; + set_id(&mut args, &key, value)?; + } if input.is_empty() { break; @@ -126,7 +135,7 @@ fn set_id(args: &mut WireArgs, key: &Ident, value: u8) -> syn::Result<()> { } else { return Err(syn::Error::new( key.span(), - "expected one of `request_id`, `response_id`, `start_id`, `stop_id`, `interrupt_id`, `receive_id`", + "expected one of `request_id`, `response_id`, `start_id`, `stop_id`, `interrupt_id`, `receive_id`, `host_initiated`, `sensitive`", )); }; @@ -145,6 +154,15 @@ fn set_id(args: &mut WireArgs, key: &Ident, value: u8) -> syn::Result<()> { /// /// #[wire(start_id = 42)] /// async fn host_account_connection_status_subscribe(...) -> ...; +/// +/// // Classify a method whose payloads carry key material or bearer secrets. +/// // The flag is folded into the wire schema-hash fingerprint, so a change in a +/// // frame's sensitivity classification is caught as contract drift. It is a +/// // classification only, and grants no confidentiality: it reaches neither the +/// // generated TypeScript nor any runtime, and nothing suppresses decoding of +/// // the payload. +/// #[wire(request_id = 114, sensitive)] +/// async fn sign_raw(...) -> ...; /// ``` /// /// Expands to the original method plus hidden doc tags that `truapi-codegen` @@ -177,7 +195,7 @@ pub fn wire(args: TokenStream, item: TokenStream) -> TokenStream { } fn wire_tags(args: &WireArgs) -> Vec { - let mut tags = [ + let mut tags: Vec = [ ("request_id", args.request_id), ("response_id", args.response_id), ("start_id", args.start_id), @@ -187,10 +205,13 @@ fn wire_tags(args: &WireArgs) -> Vec { ] .into_iter() .filter_map(|(name, value)| value.map(|id| format!("@wire_{name}={id}"))) - .collect::>(); + .collect(); if args.host_initiated { tags.push("@wire_host_initiated".to_string()); } + if args.sensitive { + tags.push("@wire_sensitive=true".to_string()); + } tags } diff --git a/rust/crates/truapi-server/Cargo.toml b/rust/crates/truapi-server/Cargo.toml index e941f6c27..390556255 100644 --- a/rust/crates/truapi-server/Cargo.toml +++ b/rust/crates/truapi-server/Cargo.toml @@ -28,7 +28,7 @@ dwarf-debug-info = false [features] default = ["wasm-signing-host"] wasm-signing-host = [] -ws-bridge = ["dep:tokio", "dep:tokio-tungstenite", "dep:rand"] +ws-bridge = ["dep:tokio", "dep:tokio-tungstenite", "dep:rand", "dep:base64"] [dependencies] truapi = { path = "../truapi" } @@ -71,13 +71,14 @@ truapi = { path = "../truapi", features = ["uniffi"] } truapi-platform = { path = "../truapi-platform", features = ["uniffi"] } futures = { version = "0.3", features = ["thread-pool"] } rand = { version = "0.8", optional = true } -tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "macros", "io-util"], optional = true } +tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "macros", "io-util", "time"], optional = true } tokio-tungstenite = { version = "0.21", default-features = false, features = ["handshake"], optional = true } uniffi.workspace = true subxt = { version = "0.50.3", default-features = false, features = ["native"] } subxt-rpcs = { version = "0.50.3", default-features = false, features = ["jsonrpsee", "native"] } frame-metadata = { version = "23", default-features = false, features = ["std", "current", "decode"] } scale-info = { version = "2.11", default-features = false, features = ["decode"] } +base64 = { version = "0.22", optional = true } [target.'cfg(target_arch = "wasm32")'.dependencies] futures-timer = { version = "3", features = ["wasm-bindgen"] } @@ -99,7 +100,7 @@ wasm-bindgen-test = "0.3" [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "macros", "io-util", "time"] } -tokio-tungstenite = { version = "0.21", default-features = false, features = ["connect"] } +tokio-tungstenite = { version = "0.21", default-features = false, features = ["connect", "handshake"] } [lints] workspace = true diff --git a/rust/crates/truapi-server/src/generated/wire_table.rs b/rust/crates/truapi-server/src/generated/wire_table.rs index 3a3783b85..0affdd39d 100644 --- a/rust/crates/truapi-server/src/generated/wire_table.rs +++ b/rust/crates/truapi-server/src/generated/wire_table.rs @@ -45,6 +45,12 @@ pub enum WireKind { /// Subscription method. Subscription(SubscriptionFrameIds), } +/// Fingerprint of this build's wire contract: frame ids, method legs, +/// sensitivity, and codec version, identical to the TS client's +/// `TRUAPI_WIRE_SCHEMA_HASH`. A host stamps it on each debug envelope so +/// the debugger refuses to decode a frame whose contract differs from +/// its own, even when the coarse handshake codec version is unchanged. +pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "4c260811b036d4ef"; /// Wire discriminants for `system_handshake`. pub const SYSTEM_HANDSHAKE: RequestFrameIds = RequestFrameIds { diff --git a/rust/crates/truapi-server/src/host_core.rs b/rust/crates/truapi-server/src/host_core.rs index 7291ecad7..6de872883 100644 --- a/rust/crates/truapi-server/src/host_core.rs +++ b/rust/crates/truapi-server/src/host_core.rs @@ -49,6 +49,137 @@ pub trait FrameSink: Send + Sync { fn emit_frame(&self, frame: Vec); } +/// Dev-only sink that observes host debug events at the core's two frame choke +/// points. A host that does not enable the debugger leaves it unset and the tap +/// is inert. Fire-and-forget by construction: [`DebugSink::emit`] must not block +/// the frame path and must not fail the operation that produced the event, so a +/// slow, absent, or crashed debugger only loses the trace, never a session. +pub trait DebugSink: Send + Sync { + /// Hand one event to the sink. + /// + /// Must not block, and must not panic: `emit` is called from inside the + /// inbound and outbound frame paths, so a panic here would otherwise unwind + /// into a live dispatch. The core contains a panic at both tap sites + /// ([`emit_debug`]) rather than trusting the contract, because the trait is + /// public and implementable out-of-repo, and because the profiles that can + /// unwind are exactly the ones a developer runs: the workspace defines no + /// `[profile.dev]`, so `dev` keeps Cargo's default `panic = "unwind"`, and an + /// out-of-repo or test sink can be installed under it. (The only in-repo + /// installer is the wasm host, which cannot unwind at all; `truapi-host-cli` + /// installs no sink.) Serialize and enqueue only; never do fallible work + /// that can `unwrap`/panic on the caller's thread. + /// + /// The two halves of that contract are NOT equally enforced, and the asymmetry + /// is deliberate rather than an oversight. Panics are contained: both tap sites + /// go through [`emit_debug`], which wraps the call in `catch_unwind`. Blocking + /// is caller-enforced only - nothing here bounds how long `emit` may take. + /// + /// It is not enforced HERE, at the trait boundary, and that is a choice worth + /// stating precisely rather than dressing up. A bounded queue drained by a + /// spawned task does solve the realistic case: it bounds per-frame work to a + /// serialize-and-push and converts an overloaded debugger into counted trace + /// loss. `WsDebugSink` does exactly that, and `services.spawner` is in hand + /// where the sink transport is built, so the core could impose it. + /// + /// What it does not solve is a sink that never yields at all - on wasm32 the + /// drain needs the same single-threaded event loop the tap is blocking, so a + /// truly hung `emit` stalls regardless. The trait therefore requires the sink to + /// own that queue rather than wrapping every sink in one here, which would add a + /// hop to the frame path for every well-behaved implementation to defend against + /// a case it still cannot fix. + /// + /// So the cost is stated rather than papered over. Whatever thread installs a + /// sink is the thread a hung one blocks, and it blocks all of that thread's + /// work: every channel, and the outbound path too, which taps synchronously + /// inside `Transport::send` while a dispatch is live. Tap ordering buys nothing + /// against a hang; it only decides whether a corrupt frame is still observed. + /// + /// In practice the wasm sink is installed from a Web Worker entry point, so the + /// blast radius is that worker rather than the page - but that is CONVENTION, + /// not enforcement: nothing gates sink installation on worker scope, and the + /// raw wasm glue is publicly exported, so a main-thread consumer can install one + /// and hang the page. A sink that may be slow must own its own queue and return + /// immediately. + fn emit(&self, event: DebugEvent); +} + +/// Hand one event to a sink, containing a panic rather than letting it unwind +/// into the frame path that called it. +/// +/// `catch_unwind` is a no-op under `panic = "abort"` (the shipping `release` +/// profile, which `codegen` inherits, and `wasm32`, which cannot unwind at all). +/// It is not dead code, because the profiles that *do* unwind are the ones the +/// debugger is used from: the workspace defines no `[profile.dev]`, so `dev` +/// keeps the default `panic = "unwind"`, and the Makefile builds +/// `truapi-host-cli` without `--release`. It also protects any downstream crate +/// that compiles this one under its own unwinding profile. +/// +/// No in-process test can prove the protection - Cargo ignores the `panic` +/// setting for test profiles, so a test asserting "the guard saved the dispatch" +/// would pass even with the guard removed. The guard is kept because it costs +/// nothing when nothing panics, not because a test can demonstrate it. +fn emit_debug(sink: Arc, event: DebugEvent) { + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || { + sink.emit(event); + // Dropped INSIDE the guard, which is why this takes the `Arc` by value. + // The frame path holds its own clone, so when `set_debug_sink` has + // concurrently replaced the sink this binding is the last reference and + // the out-of-repo destructor runs here - not at the caller's scope end, + // where it would unwind into a live dispatch. + drop(sink); + })); + if result.is_err() { + tracing::warn!("debug sink panicked; frame dropped, dispatch unaffected"); + } +} + +/// Identifies which product channel on a host a debug event belongs to, so one +/// debugger app can demultiplex several channels. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ChannelId(pub String); + +/// Direction of a tapped frame relative to the host core. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameDirection { + /// Product to core (inbound to the host). + In, + /// Core to product (outbound from the host). + Out, +} + +impl FrameDirection { + /// The wire direction string, from the **product's** vantage - the vantage + /// the debugger app and the design doc use: `"out"` = the frame left the + /// product, `"in"` = it arrived at the product. This is the inverse of the + /// enum's host-vantage variants (`In` = product to core, i.e. it *left* the + /// product), so every sink serializes the same product-vantage string + /// instead of re-deriving (and risking inverting) it. + pub fn wire_str(self) -> &'static str { + match self { + FrameDirection::In => "out", + FrameDirection::Out => "in", + } + } +} + +/// One observable host debug event. Frame bytes are the untouched +/// `ProtocolMessage`; the debugger decodes them, so the core never does. The +/// enum leaves room for host-internal events (e.g. SSO) that have no wire frame, +/// so it is `#[non_exhaustive]`: adding a variant is not a breaking change. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum DebugEvent { + /// A SCALE wire frame crossing a product channel. + Frame { + /// Which product channel on this host. + channel_id: ChannelId, + /// Product to core, or core to product. + dir: FrameDirection, + /// Untouched encoded `ProtocolMessage` bytes. + bytes: Vec, + }, +} + /// Errors returned while routing work through a product runtime. #[derive(Debug, Clone, Error)] #[cfg_attr(not(target_arch = "wasm32"), derive(uniffi::Error))] @@ -1086,6 +1217,8 @@ impl ProductRuntime { let transport = Arc::new(SinkTransport { sink, disposed: disposed.clone(), + has_debug: AtomicBool::new(false), + debug: Mutex::new(None), }); let host_subscriptions = Arc::new(HostInitiatedSubscriptionManager::new()); Self { @@ -1114,6 +1247,18 @@ impl ProductRuntime { return Ok(()); } + // Tap inbound before decode, so a corrupt frame is still observed. + if let Some((channel_id, debug)) = self.transport.debug() { + emit_debug( + debug, + DebugEvent::Frame { + channel_id, + dir: FrameDirection::In, + bytes: frame.clone(), + }, + ); + } + let message = ProtocolMessage::decode(&mut frame.as_slice()).map_err(|err| { ProductRuntimeError::InvalidFrame { reason: err.to_string(), @@ -1124,9 +1269,16 @@ impl ProductRuntime { }; let dispatch_id = self.next_dispatch_id.fetch_add(1, Ordering::Relaxed); let (abort_handle, abort_registration) = AbortHandle::new_pair(); + // Same poison recovery as `self.debug`, and for a concrete reason rather than + // symmetry: `dispose` below holds THIS guard across its whole drain loop and + // calls `AbortHandle::abort()` inside it, which wakes the task's waker - i.e. + // arbitrary out-of-repo executor code, under the lock. One panicking waker + // would poison this mutex and every later `receive_frame` would then panic + // here, which is exactly the production-host-killing shape the debug tap + // above was fixed for. self.in_flight .lock() - .expect("host core in-flight dispatch mutex poisoned") + .unwrap_or_else(|poisoned| poisoned.into_inner()) .insert(dispatch_id, abort_handle); let transport: Arc = self.transport.clone(); @@ -1134,7 +1286,7 @@ impl ProductRuntime { self.in_flight .lock() - .expect("host core in-flight dispatch mutex poisoned") + .unwrap_or_else(|poisoned| poisoned.into_inner()) .remove(&dispatch_id); if self.disposed.load(Ordering::Acquire) { self.core.cancel_subscriptions(); @@ -1199,6 +1351,19 @@ impl ProductRuntime { .await } + /// Install a dev-only [`DebugSink`] that observes every product frame in + /// both directions for `channel_id`. Absent by default and inert in + /// production. + /// + /// The sink cannot FAIL a dispatch - a panic is contained at both tap sites - + /// but it can STALL one: `emit` is called synchronously on the frame path, and + /// nothing here bounds how long it may take. Read [`DebugSink::emit`] before + /// implementing one; a sink that may be slow must own its own queue and return + /// immediately. + pub fn set_debug_sink(&self, channel_id: ChannelId, sink: Arc) { + self.transport.set_debug_sink(channel_id, sink); + } + /// Dispose this host core. Idempotent. /// /// Disposal suppresses future outgoing frames, aborts in-flight dispatch @@ -1211,7 +1376,7 @@ impl ProductRuntime { for (_, handle) in self .in_flight .lock() - .expect("host core in-flight dispatch mutex poisoned") + .unwrap_or_else(|poisoned| poisoned.into_inner()) .drain() { handle.abort(); @@ -1225,6 +1390,62 @@ impl ProductRuntime { struct SinkTransport { sink: Arc, disposed: Arc, + /// Fast-path flag: `false` (the production default) lets the per-frame + /// `debug()` return without touching the mutex. Set once when a sink is + /// installed; a reader that races the install just misses one frame. + has_debug: AtomicBool, + debug: Mutex)>>, +} + +impl SinkTransport { + /// The installed debug sink and its channel, if any. Lock-free `None` on the + /// production path (no sink installed); only locks once one is. + fn debug(&self) -> Option<(ChannelId, Arc)> { + if !self.has_debug.load(Ordering::Relaxed) { + return None; + } + // Recover from poisoning rather than panicking. Of the fixes here, moving + // the previous sink's `drop` out of `set_debug_sink`'s critical section + // (below) is what removes the only reachable poisoner: nothing else run + // under this guard can unwind, the body being an + // `Option<(ChannelId, Arc<..>)>` clone. + // + // Two independent reasons that poisoner is already unreachable in what + // ships, neither of them the profile. `wasm.rs` is the ONLY non-test + // `set_debug_sink` caller in the repo (`truapi-host-cli` installs no sink + // at all): the wasm32 target cannot unwind, AND that call site builds a + // fresh `SinkTransport` per `product_runtime()` and installs at most once + // on it, so `previous` is always `None` and there is no destructor to run + // under the lock regardless of profile. + // + // The recovery is kept regardless, because this guard sits on the per-frame + // path in both directions and outside `emit_debug`'s `catch_unwind`, so any + // future in-guard work that can unwind would land in live dispatch. + self.debug + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } + + fn set_debug_sink(&self, channel_id: ChannelId, sink: Arc) { + // Take the previous sink out under the lock, then drop it AFTER releasing. + // `*guard = Some(..)` would drop the old `Arc` in place, running an + // out-of-repo destructor inside the critical section: a panic there + // poisoned the mutex, and every subsequent frame then panicked on the + // lock. Dropping outside keeps the destructor off the critical section. + // It is not containment - this drop is not wrapped in `catch_unwind` - and + // it is not necessarily the last reference either: a frame being tapped + // concurrently holds its own clone and may be the one that drops it. That + // is why `emit_debug` takes its clone by value and drops it inside the + // guard, so the frame path never runs the destructor uncontained. + let previous = self + .debug + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .replace((channel_id, sink)); + self.has_debug.store(true, Ordering::Relaxed); + drop(previous); + } } impl Transport for SinkTransport { @@ -1232,7 +1453,26 @@ impl Transport for SinkTransport { if self.disposed.load(Ordering::Acquire) { return; } - self.sink.emit_frame(message.encode()); + let encoded = message.encode(); + // Forward to the product first, then tap: the debugger is in the path + // but never in the critical path for LATENCY - the product already has the + // frame. That is not a claim about a hung sink: `emit_debug` runs + // synchronously here, so a sink that never returns stalls this dispatch. + // See `DebugSink::emit` for why that is caller-enforced. + match self.debug() { + Some((channel_id, debug)) => { + self.sink.emit_frame(encoded.clone()); + emit_debug( + debug, + DebugEvent::Frame { + channel_id, + dir: FrameDirection::Out, + bytes: encoded, + }, + ); + } + None => self.sink.emit_frame(encoded), + } } fn on_message( @@ -1342,6 +1582,451 @@ mod tests { assert_send(runtime.receive_frame(Vec::new())); } + #[derive(Default)] + struct RecordingDebugSink { + events: Mutex)>>, + } + + impl DebugSink for RecordingDebugSink { + fn emit(&self, event: DebugEvent) { + match event { + DebugEvent::Frame { + channel_id, + dir, + bytes, + } => self + .events + .lock() + .expect("debug events mutex poisoned") + .push((channel_id, dir, bytes)), + } + } + } + + #[test] + fn debug_sink_taps_frames_in_both_directions() { + let platform = Arc::new(StubPlatform::default()); + let sink = Arc::new(RecordingSink::default()); + let debug = Arc::new(RecordingDebugSink::default()); + let (host_config, product) = runtime_config("myapp.dot"); + let runtime = ProductRuntime::from_platform_with_config( + platform, + host_config, + product, + test_spawner(), + sink.clone(), + ); + runtime.set_debug_sink(ChannelId("myapp.dot".to_string()), debug.clone()); + + let ids = subscription_ids("theme_subscribe").expect("known subscription"); + let frame = ProtocolMessage { + request_id: "theme:1".to_string(), + payload: Payload { + id: ids.start_id, + value: Vec::new(), + }, + }; + let raw = frame.encode(); + futures::executor::block_on(runtime.receive_frame(raw.clone())).unwrap(); + + // The subscription's first item is emitted asynchronously; wait for it, + // then let the tap (which runs right after delivery in `send`) settle. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + while sink + .frames + .lock() + .expect("recording sink mutex poisoned") + .is_empty() + && std::time::Instant::now() < deadline + { + std::thread::sleep(std::time::Duration::from_millis(5)); + } + std::thread::sleep(std::time::Duration::from_millis(20)); + + // Snapshot into owned vecs (never hold a lock across an assertion). + let (inbound, outbound, channels): (Vec>, Vec>, Vec) = { + let events = debug.events.lock().expect("debug events mutex poisoned"); + ( + events + .iter() + .filter(|(_, dir, _)| *dir == FrameDirection::In) + .map(|(_, _, bytes)| bytes.clone()) + .collect(), + events + .iter() + .filter(|(_, dir, _)| *dir == FrameDirection::Out) + .map(|(_, _, bytes)| bytes.clone()) + .collect(), + events.iter().map(|(cid, _, _)| cid.clone()).collect(), + ) + }; + let delivered = sink + .frames + .lock() + .expect("recording sink mutex poisoned") + .clone(); + + // Every event carries the installed channel id. + assert!( + channels + .iter() + .all(|c| *c == ChannelId("myapp.dot".to_string())), + "every event carries its channel id" + ); + // Inbound tapped once, untouched, before decode. + assert_eq!( + inbound, + vec![raw], + "inbound frame tapped exactly once, untouched" + ); + // Both directions fire, and every delivered outbound frame is tapped in + // order: the tap is in the path, not a fabricated side channel. + assert!( + !outbound.is_empty(), + "at least one outbound frame is tapped" + ); + assert_eq!( + outbound, delivered, + "every delivered outbound frame is tapped, in order" + ); + } + + /// A sink whose `Drop` panics. `emit` is a no-op: the point is the destructor, + /// which `set_debug_sink` runs when it replaces this sink. + struct PanicOnDropSink; + + impl DebugSink for PanicOnDropSink { + fn emit(&self, _event: DebugEvent) {} + } + + impl Drop for PanicOnDropSink { + fn drop(&mut self) { + panic!("out-of-repo sink destructor"); + } + } + + /// Records how many frames the transport had already delivered at the moment + /// each outbound tap fired, which is what pins the deliver-THEN-tap ordering. + struct DeliveryOrderSink { + transport: Arc, + delivered_at_tap: Mutex>, + } + + impl DebugSink for DeliveryOrderSink { + fn emit(&self, event: DebugEvent) { + let DebugEvent::Frame { dir, .. } = event; + if dir != FrameDirection::Out { + return; + } + let delivered = self + .transport + .frames + .lock() + .expect("recording sink mutex poisoned") + .len(); + self.delivered_at_tap + .lock() + .expect("delivery order mutex poisoned") + .push(delivered); + } + } + + #[test] + fn a_panicking_sink_destructor_does_not_poison_the_tap_for_later_frames() { + // `set_debug_sink` takes the previous sink out under the lock and drops it + // after releasing. If it dropped in place instead, this destructor's unwind + // would poison the debug mutex and - before the recovery below it - every + // subsequent frame would panic on that lock, killing a live host over a + // third-party sink's `Drop`. Both properties are asserted: the unwind + // surfaces to whoever INSTALLS a sink, and the tap keeps working after. + // + // This pins the two fixes as a PAIR, not individually: reverting only the + // drop-outside-the-guard change leaves the poison recovery to absorb it, and + // reverting only the recovery leaves no poisoner to trip it. Restoring both + // (the original code) fails here on the poisoned lock, which is the + // production shape being guarded against. + let platform = Arc::new(StubPlatform::default()); + let transport = Arc::new(RecordingSink::default()); + let (host_config, product) = runtime_config("myapp.dot"); + let runtime = ProductRuntime::from_platform_with_config( + platform, + host_config, + product, + test_spawner(), + transport, + ); + let channel = ChannelId("myapp.dot".to_string()); + runtime.set_debug_sink(channel.clone(), Arc::new(PanicOnDropSink)); + + // Replacing it drops the panicking sink. The unwind lands HERE, on the + // installer, not on a later frame. + let replaced = Arc::new(RecordingDebugSink::default()); + let installed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + runtime.set_debug_sink(channel.clone(), replaced.clone()); + })); + assert!( + installed.is_err(), + "the destructor's panic should surface to the installer" + ); + + // The mutex must not be poisoned: the new sink is reachable and taps. + let ids = subscription_ids("theme_subscribe").expect("known subscription"); + let frame = ProtocolMessage { + request_id: "theme:1".to_string(), + payload: Payload { + id: ids.start_id, + value: Vec::new(), + }, + }; + futures::executor::block_on(runtime.receive_frame(frame.encode())).unwrap(); + let tapped = replaced + .events + .lock() + .expect("debug events mutex poisoned") + .len(); + assert!( + tapped > 0, + "the replacement sink should still receive frames after the panic" + ); + } + + /// The inbound tap runs BEFORE decode, so a frame the codec rejects is still + /// observed. Asserting a well-formed frame reaches the sink cannot see this - + /// it arrives either way. Only an undecodable frame can: the call must fail AND + /// the sink must still hold those exact bytes. + #[test] + fn a_corrupt_inbound_frame_is_tapped_even_though_decode_rejects_it() { + let platform = Arc::new(StubPlatform::default()); + let transport = Arc::new(RecordingSink::default()); + let (host_config, product) = runtime_config("myapp.dot"); + let runtime = ProductRuntime::from_platform_with_config( + platform, + host_config, + product, + test_spawner(), + transport, + ); + let debug = Arc::new(RecordingDebugSink::default()); + runtime.set_debug_sink(ChannelId("myapp.dot".to_string()), debug.clone()); + + // Not a `ProtocolMessage`: the leading compact length claims far more + // bytes than follow, so decode fails before any field is read. + let corrupt = vec![0xff, 0xff, 0xff, 0xff]; + let outcome = futures::executor::block_on(runtime.receive_frame(corrupt.clone())); + assert!( + matches!(outcome, Err(ProductRuntimeError::InvalidFrame { .. })), + "expected decode to reject the frame, got {outcome:?}" + ); + + let events = debug.events.lock().expect("debug events mutex poisoned"); + assert_eq!( + events.len(), + 1, + "a frame rejected by decode must still be tapped" + ); + assert_eq!(events[0].1, FrameDirection::In); + assert_eq!( + events[0].2, corrupt, + "the tap must carry the original bytes, not a decoded form" + ); + } + + /// A sink whose destructor panics must not unwind the FRAME path. + /// + /// `set_debug_sink` drops the previous sink outside the lock, which covers the + /// case where the installer holds the last reference. It does not cover this + /// one: the frame path clones its own `Arc` before tapping, so if the sink is + /// replaced while that tap is in flight, the frame path holds the last + /// reference and runs the destructor itself. `emit_debug` takes the clone by + /// value so that drop lands inside its `catch_unwind`; with `&dyn DebugSink` + /// the clone instead dies at the caller's scope end, outside the guard, and + /// unwinds a live dispatch. + #[test] + fn a_panicking_sink_destructor_does_not_unwind_the_frame_path() { + /// Blocks inside `emit` until the installer has released its reference, so + /// the frame path is provably the one that drops this sink. + struct HandoffSink { + tapped: std::sync::mpsc::SyncSender<()>, + release: Mutex>, + } + impl DebugSink for HandoffSink { + fn emit(&self, _event: DebugEvent) { + let _ = self.tapped.send(()); + let _ = self + .release + .lock() + .expect("release receiver poisoned") + .recv(); + } + } + impl Drop for HandoffSink { + fn drop(&mut self) { + panic!("out-of-repo sink destructor"); + } + } + + let platform = Arc::new(StubPlatform::default()); + let transport = Arc::new(RecordingSink::default()); + let (host_config, product) = runtime_config("myapp.dot"); + let runtime = Arc::new(ProductRuntime::from_platform_with_config( + platform, + host_config, + product, + test_spawner(), + transport, + )); + let channel = ChannelId("myapp.dot".to_string()); + + let (tapped_tx, tapped_rx) = std::sync::mpsc::sync_channel(1); + let (release_tx, release_rx) = std::sync::mpsc::sync_channel(1); + runtime.set_debug_sink( + channel.clone(), + Arc::new(HandoffSink { + tapped: tapped_tx, + release: Mutex::new(release_rx), + }), + ); + + let ids = subscription_ids("theme_subscribe").expect("known subscription"); + let frame = ProtocolMessage { + request_id: "theme:1".to_string(), + payload: Payload { + id: ids.start_id, + value: Vec::new(), + }, + }; + let encoded = frame.encode(); + let frame_runtime = Arc::clone(&runtime); + let frame_thread = std::thread::spawn(move || { + futures::executor::block_on(frame_runtime.receive_frame(encoded)) + }); + + // Wait until the tap is inside `emit`, holding its own clone. + tapped_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("tap never fired"); + // Replace the sink: the installer's reference goes, leaving the frame + // path's clone as the last one. + runtime.set_debug_sink(channel, Arc::new(RecordingDebugSink::default())); + let _ = release_tx.send(()); + + let outcome = frame_thread.join(); + assert!( + outcome.is_ok(), + "a panicking sink destructor unwound the frame path" + ); + } + + #[test] + fn outbound_frames_are_delivered_before_they_are_tapped() { + // `send` hands the frame to the transport and taps afterwards, so no sink can + // delay or drop THIS frame - it is already delivered. It says nothing about + // the next one: `send` returns only after the tap, so a slow sink delays + // every subsequent frame (see `DebugSink::emit`). Asserting the two lists + // match cannot see the ordering at all - they are order-identical either way. + // Counting deliveries AT TAP TIME can: tap N must observe N deliveries, and + // tapping first would make it N-1. + let platform = Arc::new(StubPlatform::default()); + let transport = Arc::new(RecordingSink::default()); + let (host_config, product) = runtime_config("myapp.dot"); + let runtime = ProductRuntime::from_platform_with_config( + platform, + host_config, + product, + test_spawner(), + transport.clone(), + ); + let debug = Arc::new(DeliveryOrderSink { + transport: transport.clone(), + delivered_at_tap: Mutex::new(Vec::new()), + }); + runtime.set_debug_sink(ChannelId("myapp.dot".to_string()), debug.clone()); + + let ids = subscription_ids("theme_subscribe").expect("known subscription"); + let frame = ProtocolMessage { + request_id: "theme:1".to_string(), + payload: Payload { + id: ids.start_id, + value: Vec::new(), + }, + }; + futures::executor::block_on(runtime.receive_frame(frame.encode())).unwrap(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + while debug + .delivered_at_tap + .lock() + .expect("delivery order mutex poisoned") + .is_empty() + && std::time::Instant::now() < deadline + { + std::thread::sleep(std::time::Duration::from_millis(5)); + } + std::thread::sleep(std::time::Duration::from_millis(20)); + + let observed = debug + .delivered_at_tap + .lock() + .expect("delivery order mutex poisoned") + .clone(); + assert!(!observed.is_empty(), "expected at least one outbound tap"); + // Tap i (0-based) must see i+1 frames already delivered. + let expected: Vec = (1..=observed.len()).collect(); + assert_eq!( + observed, expected, + "each outbound tap must run after its own frame was delivered" + ); + } + + /// Every profile that ships a host aborts on panic, so [`emit_debug`]'s + /// `catch_unwind` cannot fire in a shipping build - which is why + /// [`DebugSink::emit`] documents its no-panic rule as caller-enforced. The + /// guard still earns its keep everywhere else: `dev` inherits + /// `panic = "unwind"` (the workspace defines no `[profile.dev]`), the Makefile + /// builds `truapi-host-cli` without `--release`, and a downstream crate may + /// compile this one under its own unwinding profile. + /// + /// No unit test can demonstrate the guard itself - Cargo ignores the `panic` + /// setting for test profiles, so a "the guard protected dispatch" assertion + /// passes with the guard removed. The premise is what is checkable, and this + /// fails if a shipping profile stops aborting, at which point the reasoning on + /// `DebugSink::emit` needs revisiting. + #[test] + fn shipping_profiles_abort_on_panic() { + let workspace_manifest = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .ancestors() + .nth(3) + .expect("crate lives at /rust/crates/truapi-server") + .join("Cargo.toml"); + let manifest = std::fs::read_to_string(&workspace_manifest) + .expect("workspace manifest is readable from the crate directory"); + let release = manifest + .split("[profile.release]") + .nth(1) + .expect("workspace defines [profile.release]") + .split("\n[") + .next() + .expect("release profile section"); + assert!( + release.contains("panic = \"abort\""), + "release no longer aborts on panic: revisit DebugSink::emit's contract docs" + ); + assert!( + manifest.contains("[profile.codegen]") && manifest.contains("inherits = \"release\""), + "codegen no longer inherits release: recheck what the ws-bridge artifacts build with" + ); + } + + #[test] + fn frame_direction_wire_str_is_product_vantage() { + // The wire string is product-vantage (what the debugger and design doc + // use), the inverse of the enum's host-vantage names: a frame the host + // tapped as `In` (product to core) *left the product*, so it serializes + // as `"out"`. This pins the convention so a sink can't re-invert it. + assert_eq!(FrameDirection::In.wire_str(), "out"); + assert_eq!(FrameDirection::Out.wire_str(), "in"); + } + #[test] fn app_connection_rejects_custom_rendering() { let (host_config, product) = runtime_config("myapp.dot"); diff --git a/rust/crates/truapi-server/src/lib.rs b/rust/crates/truapi-server/src/lib.rs index 8b9ef7f53..4e9788e14 100644 --- a/rust/crates/truapi-server/src/lib.rs +++ b/rust/crates/truapi-server/src/lib.rs @@ -18,6 +18,8 @@ //! native WebView hosts (Android/iOS). //! - [`native`]: UniFFI surface exposing the native host runtime + callbacks. //! - `wasm` (wasm32 only): wasm-bindgen surface exposing `WasmProductRuntime`. +//! - `native_debug` (non-wasm32 only): a loopback WebSocket [`DebugSink`] that +//! streams tapped frames to the `@parity/truapi-debugger` app. pub(crate) mod chain_runtime; pub mod core; @@ -49,13 +51,18 @@ pub mod native_renderer; #[cfg(target_arch = "wasm32")] pub mod wasm; +#[cfg(all(not(target_arch = "wasm32"), feature = "ws-bridge"))] +pub mod native_debug; + pub use host_core::{ - FrameSink, HostAdmin, PairingHostRuntime, ProductRuntime, ProductRuntimeControl, - ProductRuntimeError, SigningHostRuntime, + ChannelId, DebugEvent, DebugSink, FrameDirection, FrameSink, HostAdmin, PairingHostRuntime, + ProductRuntime, ProductRuntimeControl, ProductRuntimeError, SigningHostRuntime, }; pub use host_logic::session::{ ExternalPairedSession, SsoSessionInfo, decode_persisted_session, encode_external_paired_session, }; +#[cfg(all(not(target_arch = "wasm32"), feature = "ws-bridge"))] +pub use native_debug::{DebugSinkError, WsDebugSink}; #[cfg(not(target_arch = "wasm32"))] pub use runtime::StatementRenewalTarget; pub use runtime::login_failure::reports_exhausted_period; diff --git a/rust/crates/truapi-server/src/native_debug.rs b/rust/crates/truapi-server/src/native_debug.rs new file mode 100644 index 000000000..cb64eb79c --- /dev/null +++ b/rust/crates/truapi-server/src/native_debug.rs @@ -0,0 +1,813 @@ +//! Native (non-wasm) [`DebugSink`]: streams tapped frames to a loopback +//! `@parity/truapi-debugger` over a WebSocket. +//! +//! The native counterpart of the wasm [`crate::wasm`] `WasmDebugSink`: a dumb, +//! payload-blind byte-forwarder. Each [`DebugEvent::Frame`] is serialized to the +//! debugger's wire envelope - `{channelId, dir, frame}`, where `frame` is the +//! base64 of the untouched SCALE `ProtocolMessage` bytes - and sent as one WS +//! text message. Decoding lives in the debugger app, never here. +//! +//! Fire-and-forget by construction, per the [`DebugSink`] contract: +//! [`WsDebugSink::emit`] never blocks and never fails a dispatch. It only +//! serializes and pushes onto a bounded queue; a background task owns the socket, +//! reconnects with capped backoff, and drops frames (counted) when the queue is +//! full. A slow, absent, or crashed debugger loses traces, never a session. +//! Dropped frames are reported on the wire: the count shed since the previous +//! envelope rides the next one as `dropped`, so the debugger attributes the gap +//! to the link instead of reading it as a host that never answered. +//! +//! Localhost only: the target URL must be `ws://` on a loopback host. No `wss`, +//! no certificates, no LAN. Construct via [`WsDebugSink::connect`] from within a +//! Tokio runtime and install with [`crate::ProductRuntime::set_debug_sink`]; +//! constructing one is a dev-only opt-in, so a host that never calls it leaves +//! the tap inert. + +use core::net::SocketAddr; +use core::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use core::time::Duration; +use std::sync::Arc; + +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use futures::{SinkExt, StreamExt}; +use serde::Serialize; +use thiserror::Error; +use tokio::net::TcpStream; +use tokio::runtime::Handle; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{WebSocketStream, client_async}; +use tracing::debug; + +use crate::generated::wire_table::TRUAPI_WIRE_SCHEMA_HASH; +use crate::host_core::{DebugEvent, DebugSink}; + +/// Bounded so a stalled or absent debugger applies backpressure as counted +/// drops, never unbounded memory growth on the observed session. +const QUEUE_CAPACITY: usize = 4096; + +/// Byte budget alongside [`QUEUE_CAPACITY`]: one `ProtocolMessage` can be MBs, so +/// a count-only cap could still buffer unbounded RSS while the debugger is +/// absent. Whichever ceiling hits first drops the frame (counted), never blocks. +const MAX_QUEUE_BYTES: usize = 8 * 1024 * 1024; + +/// Envelope version, mirroring the debugger's `WIRE_ENVELOPE_VERSION` and the web +/// host's constant. Kept in sync by hand. +const WIRE_ENVELOPE_VERSION: u32 = 1; + +/// The host's wire codec version, mirroring `@parity/truapi`'s +/// `TRUAPI_CODEC_VERSION` (the handshake `codec_version`). Stamped on the +/// envelope so the debugger refuses to decode a frame whose codec differs from +/// its own, rather than resolving `u8` frame ids against the wrong contract. +const WIRE_CODEC_VERSION: u32 = 1; + +/// Port the debugger's server listens on (`@parity/truapi-debugger`'s +/// `npm run serve`), used when the debug URL omits one so `ws://localhost` +/// reaches the debugger instead of HTTP's port 80. +const DEBUGGER_DEFAULT_PORT: u16 = 9231; + +/// Initial reconnect delay; doubles on each failed dial up to [`MAX_BACKOFF`]. +const INITIAL_BACKOFF: Duration = Duration::from_millis(200); + +/// Cap on the reconnect backoff. +const MAX_BACKOFF: Duration = Duration::from_secs(5); + +/// Cap on a single dial + WS handshake; a port that accepts TCP but never +/// completes the upgrade must not park the writer task forever. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(5); + +/// Failure building a [`WsDebugSink`]. +#[derive(Debug, Error)] +pub enum DebugSinkError { + /// The debug URL did not parse. + #[error("invalid debug url: {0}")] + Url(#[from] url::ParseError), + /// The debug URL was not `ws://` on a loopback host. + #[error("debug url must be ws:// on a loopback host, got {0}")] + NotLoopback(String), + /// The debug URL host could not be resolved. + #[error("could not resolve debug url host: {0}")] + Resolve(#[from] std::io::Error), + /// `connect` was called outside a Tokio runtime. + #[error("WsDebugSink::connect must be called from within a Tokio runtime")] + NoRuntime, +} + +/// A dev-only [`DebugSink`] that forwards tapped frames to a loopback debugger +/// over a WebSocket, using the same `{channelId, dir, frame: base64}` envelope +/// the browser host sends. +pub struct WsDebugSink { + outbound: mpsc::Sender, + dropped: Arc, + pending_dropped: Arc, + queued_bytes: Arc, +} + +/// One serialized envelope on its way to the writer task, plus the number of +/// shed frames stamped on it. Carrying the count alongside the line lets the +/// writer put it back if this envelope dies with the socket, so a drop is +/// reported exactly once and never silently swallowed. +struct QueuedFrame { + line: String, + shed: u64, +} + +/// The wire envelope, matching the debugger's `parseWireMessage` / ingest +/// `DebugFrameEnvelope`: `dir` is product-vantage, `frame` is base64 SCALE bytes. +/// `v`/`codec` are the identity the debugger checks before decoding. +#[derive(Serialize)] +struct WireMessage<'a> { + v: u32, + codec: u32, + schema: &'static str, + #[serde(rename = "channelId")] + channel_id: &'a str, + dir: &'a str, + frame: String, + /// Frames this link shed since the previous envelope. Omitted when zero, as + /// the web link omits it, so the common envelope is unchanged; the debugger + /// sums it per channel into `droppedByHost`. + #[serde(skip_serializing_if = "is_zero")] + dropped: u64, +} + +fn is_zero(count: &u64) -> bool { + *count == 0 +} + +/// Validate a debug URL and resolve it to the addresses to dial, in resolver +/// order. +/// +/// Requires `ws://`, then RESOLVES the host and requires *every* resolved +/// address to be loopback. Resolving (rather than string-matching the host) +/// accepts all genuine loopback forms - 127.0.0.0/8, ::1, and a `localhost` that +/// resolves to them - and rejects anything resolving off-loopback, closing the +/// "validate one string, dial another" gap. `Url::socket_addrs` also handles +/// IPv6 bracket-stripping. +/// +/// The port default is applied by hand rather than through `socket_addrs`'s +/// fallback closure: `ws` is a *special* scheme in the URL spec with a known +/// default of 80, so the closure is never consulted and a portless +/// `ws://127.0.0.1` would dial :80 instead of the debugger. +fn resolve_loopback_target(url: &str) -> Result, DebugSinkError> { + let mut parsed = url::Url::parse(url)?; + if parsed.scheme() != "ws" { + return Err(DebugSinkError::NotLoopback(url.to_string())); + } + if parsed.port().is_none() { + parsed + .set_port(Some(DEBUGGER_DEFAULT_PORT)) + .map_err(|()| DebugSinkError::NotLoopback(url.to_string()))?; + } + let addrs = parsed.socket_addrs(|| Some(DEBUGGER_DEFAULT_PORT))?; + if !all_loopback(&addrs) { + return Err(DebugSinkError::NotLoopback(url.to_string())); + } + Ok(addrs) +} + +/// Whether every resolved address is loopback, and there is at least one. +/// +/// `all`, not `any`: a name that resolves to a mix of loopback and routable +/// addresses must be refused outright. Accepting it would let a split-horizon or +/// rebinding resolver hand us one loopback address to pass the check and a +/// routable one to dial. Lifted out of [`resolve_loopback_target`] so the mixed +/// case is testable without a resolver that produces one. +fn all_loopback(addrs: &[SocketAddr]) -> bool { + !addrs.is_empty() && addrs.iter().all(|addr| addr.ip().is_loopback()) +} + +impl WsDebugSink { + /// Build a sink targeting `url` and spawn its writer task. + /// + /// `url` must be `ws://` on `127.0.0.1`, `localhost`, or `[::1]`. Returns + /// immediately even if the debugger is not yet listening; the writer task + /// dials lazily and reconnects. Must be called from within a Tokio runtime. + pub fn connect(url: &str) -> Result, DebugSinkError> { + // Capture *every* resolved loopback address and dial those directly (in + // `writer_loop`), rather than re-resolving the URL string on each dial. + // The WS handshake is therefore only ever sent to a checked loopback + // peer - closing the resolve-then-dial gap where a mid-session resolver + // change could send the handshake off-box. + let addrs = resolve_loopback_target(url)?; + + // Return a Result rather than panicking inside tokio::spawn when called + // outside a runtime. + if Handle::try_current().is_err() { + return Err(DebugSinkError::NoRuntime); + } + + let (outbound, inbox) = mpsc::channel::(QUEUE_CAPACITY); + let dropped = Arc::new(AtomicU64::new(0)); + let pending_dropped = Arc::new(AtomicU64::new(0)); + let queued_bytes = Arc::new(AtomicUsize::new(0)); + tokio::spawn(writer_loop( + url.to_string(), + addrs, + inbox, + Arc::clone(&dropped), + Arc::clone(&pending_dropped), + Arc::clone(&queued_bytes), + )); + Ok(Arc::new(Self { + outbound, + dropped, + pending_dropped, + queued_bytes, + })) + } + + /// Number of frames dropped because the outbound queue was full (debugger + /// absent or slower than the observed session). Never affects the session. + pub fn dropped(&self) -> u64 { + self.dropped.load(Ordering::Relaxed) + } + + /// The count waiting to ride the next envelope. Test-only: the re-add in + /// [`Self::count_drop`] is load-bearing and is otherwise only observable + /// through a live socket, which makes it easy to leave untested. + #[cfg(test)] + fn pending_dropped(&self) -> u64 { + self.pending_dropped.load(Ordering::Relaxed) + } + + /// Account for one lost frame: `carried` drops had been drained onto the + /// envelope that never made it, so they go back on the pending count + /// alongside this one and ride the next envelope instead. Returns the new + /// lifetime total, for logging. + fn count_drop(&self, carried: u64) -> u64 { + self.pending_dropped + .fetch_add(carried + 1, Ordering::Relaxed); + self.dropped.fetch_add(1, Ordering::Relaxed) + 1 + } +} + +impl DebugSink for WsDebugSink { + fn emit(&self, event: DebugEvent) { + let DebugEvent::Frame { + channel_id, + dir, + bytes, + } = event; + // Drain the drops accumulated since the previous envelope and stamp them + // on this one, as the web link does: a shed frame must reach the debugger + // as a counted gap in the link, not as a host that never answered. If + // this envelope is itself lost, `count_drop` puts the count back so it + // rides the next one. + let shed = self.pending_dropped.swap(0, Ordering::Relaxed); + let message = WireMessage { + v: WIRE_ENVELOPE_VERSION, + codec: WIRE_CODEC_VERSION, + schema: TRUAPI_WIRE_SCHEMA_HASH, + channel_id: &channel_id.0, + // Product-vantage string; never hand-mapped, so it cannot invert. + dir: dir.wire_str(), + frame: BASE64.encode(&bytes), + dropped: shed, + }; + let Ok(line) = serde_json::to_string(&message) else { + self.count_drop(shed); + return; + }; + // Byte budget on top of the channel's count cap: one frame can be MBs, so + // a count-only bound could still grow RSS without limit while the debugger + // is absent. Reserve the frame's bytes BEFORE handing the line to the + // channel: the writer task can recv and release (fetch_sub) the instant + // try_send succeeds, so adding *after* would let that sub run first and + // wrap the counter - an overflow panic in debug builds, on the frame path. + // Reserve atomically, then release on any failure. + let len = line.len(); + if self.queued_bytes.fetch_add(len, Ordering::Relaxed) + len > MAX_QUEUE_BYTES { + // This reservation pushed us past the budget: back it out and drop. + self.queued_bytes.fetch_sub(len, Ordering::Relaxed); + let dropped = self.count_drop(shed); + debug!("truapi debug sink: byte budget full, frame dropped (total {dropped})"); + return; + } + if self.outbound.try_send(QueuedFrame { line, shed }).is_err() { + // Not enqueued after all: release the reservation. The frame is lost + // (never the session); count it and log so the gap is attributable to + // the link, not to the host. + self.queued_bytes.fetch_sub(len, Ordering::Relaxed); + let dropped = self.count_drop(shed); + debug!("truapi debug sink: outbound queue full, frame dropped (total {dropped})"); + } + } +} + +/// Dial the pre-validated loopback candidates in resolver order and return the +/// first socket that completes the WS handshake. +/// +/// Trying every candidate is what makes `ws://localhost:9231` work: `localhost` +/// commonly resolves to `::1` first while the debugger binds v4 only, so pinning +/// the first address would retry an address that can never deliver, forever. +/// Every candidate was checked as loopback in [`resolve_loopback_target`], the +/// addresses are not re-resolved, and the handshake runs over the +/// already-connected socket, so it can never reach an off-box peer. Each attempt +/// is bounded so a TCP-accepting but non-upgrading port can't park the task. +async fn dial(url: &str, addrs: &[SocketAddr]) -> Option> { + for addr in addrs { + let dialed = tokio::time::timeout(HANDSHAKE_TIMEOUT, async { + let tcp = TcpStream::connect(addr).await.ok()?; + client_async(url, tcp).await.ok() + }) + .await; + match dialed { + Ok(Some((stream, _response))) => return Some(stream), + Ok(None) => debug!("truapi debug sink: dial/handshake to {addr} failed"), + Err(_) => debug!("truapi debug sink: handshake to {addr} timed out"), + } + } + None +} + +/// Own the socket for the sink's lifetime: dial with capped backoff, then drain +/// the queue to the wire until the sink is dropped. +async fn writer_loop( + url: String, + addrs: Vec, + mut inbox: mpsc::Receiver, + dropped: Arc, + pending_dropped: Arc, + queued_bytes: Arc, +) { + let mut backoff = INITIAL_BACKOFF; + loop { + let Some(stream) = dial(url.as_str(), &addrs).await else { + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + // The sink was dropped while we were retrying: give up. + if inbox.is_closed() { + return; + } + continue; + }; + let (mut write, mut read) = stream.split(); + // Drain queued frames to the wire, and also poll the read half so + // tokio-tungstenite answers server pings and observes a Close; being + // forward-only, any inbound message is ignored. Reset backoff only on a + // *delivered* frame, so an accept-then-close server still backs off + // instead of spinning on zero-delay reconnects. + loop { + tokio::select! { + queued = inbox.recv() => match queued { + Some(QueuedFrame { line, shed }) => { + // Off the queue now: release its bytes from the budget + // before the (moving) send so the counter can't drift. + queued_bytes.fetch_sub(line.len(), Ordering::Relaxed); + match write.send(Message::Text(line)).await { + Ok(()) => backoff = INITIAL_BACKOFF, + Err(_) => { + debug!("truapi debug sink: socket closed, reconnecting"); + // The in-flight line is lost across this reconnect. + dropped.fetch_add(1, Ordering::Relaxed); + // It carried `shed` earlier drops that therefore + // never reached the debugger: make them pending + // again (with this frame) so the next delivered + // envelope still reports the whole gap. + pending_dropped.fetch_add(shed + 1, Ordering::Relaxed); + break; + } + } + } + // All senders dropped: the sink is gone, so is the host. Done. + None => return, + }, + inbound = read.next() => match inbound { + Some(Ok(_)) => {} // forward-only: ignore any inbound message + Some(Err(_)) | None => { + debug!("truapi debug sink: read side closed, reconnecting"); + break; + } + }, + } + } + // Reconnect after an established socket dropped: back off here too. + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + if inbox.is_closed() { + return; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::host_core::{ChannelId, FrameDirection}; + + use tokio::net::TcpListener; + use tokio::sync::oneshot; + use tokio_tungstenite::accept_async; + + #[tokio::test] + async fn emits_base64_envelope_with_product_vantage_dir() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + + // Server side: accept one connection, capture the first text message. + let (tx, rx) = oneshot::channel::(); + tokio::spawn(async move { + let (stream, _peer) = listener.accept().await.unwrap(); + let ws = accept_async(stream).await.unwrap(); + let (_write, mut read) = ws.split(); + let message = read.next().await.unwrap().unwrap(); + tx.send(message.into_text().unwrap()).unwrap(); + }); + + let sink = WsDebugSink::connect(&format!("ws://127.0.0.1:{port}")).unwrap(); + // `In` = product→core, i.e. the frame *left* the product → product-vantage "out". + sink.emit(DebugEvent::Frame { + channel_id: ChannelId("myapp.dot".to_string()), + dir: FrameDirection::In, + bytes: vec![1, 2, 3, 4], + }); + + let text = tokio::time::timeout(Duration::from_secs(5), rx) + .await + .expect("debugger did not receive a frame") + .unwrap(); + let value: serde_json::Value = serde_json::from_str(&text).unwrap(); + + assert_eq!(value["channelId"], "myapp.dot"); + // Identity the debugger checks before decoding. Asserted as literals: a + // constant on both sides would agree with itself even if the value the + // debugger expects changed. + assert_eq!(value["v"], 1); + assert_eq!(value["codec"], 1); + assert_eq!(value["v"], WIRE_ENVELOPE_VERSION); + assert_eq!(value["codec"], WIRE_CODEC_VERSION); + assert_eq!(value["schema"], TRUAPI_WIRE_SCHEMA_HASH); + // Guard against re-inversion: In must serialize as product-vantage "out". + assert_eq!(value["dir"], FrameDirection::In.wire_str()); + assert_eq!(value["dir"], "out"); + assert_eq!(value["frame"], BASE64.encode([1, 2, 3, 4])); + // Nothing was shed, so the envelope stays exactly as the web link's: + // `dropped` is absent rather than a noisy zero. + assert!( + value.get("dropped").is_none(), + "a frame with no preceding drops must not carry a dropped count" + ); + } + + /// A drop that happens while an earlier drop is still pending must ADD to the + /// pending count, not replace it. `count_drop(carried)` re-adds the `carried` + /// drops that had already been drained onto the envelope now being discarded - + /// without that, the first drop is silently swallowed and the wire reports one + /// loss where two happened. + /// + /// Asserted on the pending counter rather than the wire because the wire only + /// shows the total, and a total of "1 then 1" is indistinguishable from + /// "1 then 2" unless you catch both envelopes. + /// + /// Covers the BUDGET path only. Two sibling invariants remain uncovered, stated + /// here rather than left to be rediscovered: + /// - `writer_loop`'s reconnect re-add (`pending_dropped.fetch_add(shed + 1)`) + /// is the same invariant on the socket-death path. Reaching it needs a live + /// socket killed while an envelope carrying a shed count is in flight; + /// mutating that `shed + 1` to `1` passes today. + /// - the byte budget's `>` boundary. Hitting exactly `MAX_QUEUE_BYTES` needs a + /// frame sized against base64 plus JSON overhead; `>` vs `>=` passes today. + /// - the ATOMICITY of the re-add. `emit` always swaps pending to 0 before + /// calling `count_drop`, so this test never observes a non-zero pending and + /// `fetch_add` -> `store` passes. `store` would clobber a concurrent re-add + /// from `writer_loop`, which is the whole reason it is `fetch_add`. + + #[tokio::test] + async fn a_drop_while_one_is_pending_accumulates_rather_than_replaces() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + drop(listener); + + let sink = WsDebugSink::connect(&format!("ws://127.0.0.1:{port}")).unwrap(); + let big = vec![0u8; 4 * 1024 * 1024]; + let emit = |sink: &WsDebugSink| { + sink.emit(DebugEvent::Frame { + channel_id: ChannelId("myapp.dot".to_string()), + dir: FrameDirection::Out, + bytes: big.clone(), + }); + }; + + // 1st fills the queue's byte budget; 2nd is shed with nothing pending. + emit(&sink); + emit(&sink); + assert_eq!(sink.pending_dropped(), 1, "the first shed is pending"); + + // 3rd drains that 1 onto its envelope, is itself shed, and must put the + // drained count BACK alongside its own. + emit(&sink); + assert_eq!( + sink.pending_dropped(), + 2, + "the drained count must be re-added, not replaced" + ); + assert_eq!(sink.dropped(), 2, "and the lifetime total counts both"); + } + + /// The mixed-resolution case, which no URL-level test reaches: every input a + /// real resolver produces here is homogeneous (all loopback, or all not), so + /// `all` vs `any` is indistinguishable through `resolve_loopback_target`. A + /// name resolving to both must be refused - that is the split-horizon and + /// DNS-rebinding shape the predicate exists for. + #[test] + fn a_mixed_resolution_is_not_loopback() { + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + + let loopback_v4 = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9231); + let loopback_v6 = SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 9231); + let routable = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), 9231); + + assert!(all_loopback(&[loopback_v4])); + assert!(all_loopback(&[loopback_v4, loopback_v6])); + assert!(!all_loopback(&[]), "an empty resolution is not loopback"); + assert!(!all_loopback(&[routable])); + // The two orderings that `any` would wrongly accept. + assert!( + !all_loopback(&[loopback_v4, routable]), + "loopback first must not launder a routable sibling" + ); + assert!( + !all_loopback(&[routable, loopback_v4]), + "a routable address must be refused whatever its position" + ); + } + + /// A shed frame must reach the debugger as a counted gap in the link. The + /// debugger sums `dropped` per channel into `/stats.droppedByHost`, so + /// without it a 4096-frame or 8 MiB shed reads as a host that never answered. + #[tokio::test] + async fn a_shed_frame_is_reported_as_dropped_on_the_next_envelope() { + // Reserve a loopback port, then free it: with nothing listening the queue + // cannot drain, so the byte budget sheds a frame deterministically. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + drop(listener); + + let sink = WsDebugSink::connect(&format!("ws://127.0.0.1:{port}")).unwrap(); + // 4 MiB → ~5.6 MiB of base64 per envelope: the first fits the 8 MiB + // budget, the second pushes past it and is shed. + let big = vec![0u8; 4 * 1024 * 1024]; + for _ in 0..2 { + sink.emit(DebugEvent::Frame { + channel_id: ChannelId("myapp.dot".to_string()), + dir: FrameDirection::Out, + bytes: big.clone(), + }); + } + assert_eq!(sink.dropped(), 1, "the byte budget must shed exactly one"); + + // Bring the debugger up on that port and let the writer connect. + let listener = TcpListener::bind(("127.0.0.1", port)).await.unwrap(); + // The reader reports a pair: (count on the envelope that carried it, count + // on the NEXT envelope). + // The second is what pins the reset: `dropped` is + // `skip_serializing_if = "is_zero"`, so once the pending count is drained + // the following envelope must OMIT the field entirely. Without asserting + // that, dropping the reset leaves this test green - the first envelope + // looks identical either way. + let (tx, rx) = oneshot::channel::<(u64, Option)>(); + tokio::spawn(async move { + let (stream, _peer) = listener.accept().await.unwrap(); + let ws = accept_async(stream).await.unwrap(); + let (_write, mut read) = ws.split(); + let mut carried: Option = None; + while let Some(Ok(message)) = read.next().await { + let Ok(text) = message.into_text() else { + continue; + }; + let value: serde_json::Value = serde_json::from_str(&text).unwrap(); + let dropped = value["dropped"].as_u64(); + match carried { + // Still waiting for the envelope that carries the shed count. + None => { + if let Some(d) = dropped { + carried = Some(d); + } + } + // Got it; this is the next envelope after it. + Some(first) => { + tx.send((first, dropped)).unwrap(); + return; + } + } + } + }); + + // The shed happened while the queue held an already-serialized envelope, + // so the count rides the next frame emitted after it - exactly the web + // link's "piggyback onto the next live frame". + let deadline = tokio::time::Instant::now() + Duration::from_secs(20); + let mut rx = rx; + loop { + sink.emit(DebugEvent::Frame { + channel_id: ChannelId("myapp.dot".to_string()), + dir: FrameDirection::Out, + bytes: vec![7], + }); + match tokio::time::timeout(Duration::from_millis(250), &mut rx).await { + Ok(received) => { + let (first, next) = received.unwrap(); + assert_eq!( + first, 1, + "the shed frame must be reported once, on the wire" + ); + assert_eq!( + next, None, + "the pending count must reset after it rides an envelope, \ + so the next one omits `dropped` entirely" + ); + return; + } + Err(_) => assert!( + tokio::time::Instant::now() < deadline, + "no envelope ever carried the shed frame's drop count" + ), + } + } + } + + #[test] + fn rejects_non_loopback_and_non_ws_urls() { + // 192.0.2.1 (TEST-NET-1) is a non-loopback IP literal, so no DNS is hit. + assert!(WsDebugSink::connect("wss://127.0.0.1:9231").is_err()); + assert!(WsDebugSink::connect("ws://192.0.2.1:9231").is_err()); + assert!(WsDebugSink::connect("http://127.0.0.1:9231").is_err()); + assert!(WsDebugSink::connect("not a url").is_err()); + } + + #[tokio::test] + async fn accepts_loopback_forms_at_validation() { + for url in [ + "ws://127.0.0.1:9231", + "ws://localhost:9231", + "ws://[::1]:9231", + ] { + assert!(WsDebugSink::connect(url).is_ok(), "should accept {url}"); + } + } + + /// Accepting a URL is not the same as being able to deliver on it: on macOS + /// `localhost` resolves to `::1` first while the debugger binds v4 only, so a + /// sink that pins the first resolved address retries an address that can + /// never deliver, forever. Every candidate must be tried. + #[tokio::test] + async fn delivers_through_localhost_to_a_v4_only_debugger() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + // The bug only exists when the name resolves to something before the v4 + // address; on a v4-only resolver this still passes, it just proves less. + let resolved = resolve_loopback_target(&format!("ws://localhost:{port}")).unwrap(); + assert!( + !resolved.is_empty(), + "localhost must resolve to at least one loopback address" + ); + + let (tx, rx) = oneshot::channel::(); + tokio::spawn(async move { + let (stream, _peer) = listener.accept().await.unwrap(); + let ws = accept_async(stream).await.unwrap(); + let (_write, mut read) = ws.split(); + let message = read.next().await.unwrap().unwrap(); + tx.send(message.into_text().unwrap()).unwrap(); + }); + + let sink = WsDebugSink::connect(&format!("ws://localhost:{port}")).unwrap(); + sink.emit(DebugEvent::Frame { + channel_id: ChannelId("myapp.dot".to_string()), + dir: FrameDirection::Out, + bytes: vec![9], + }); + + let text = tokio::time::timeout(Duration::from_secs(20), rx) + .await + .expect("a v4-only debugger never received the frame via localhost") + .unwrap(); + let value: serde_json::Value = serde_json::from_str(&text).unwrap(); + assert_eq!(value["frame"], BASE64.encode([9])); + } + + /// A port-less debug URL must target the debugger, not HTTP's port 80. + #[test] + fn a_url_without_a_port_targets_the_debugger_port() { + let addrs = resolve_loopback_target("ws://127.0.0.1").unwrap(); + assert_eq!(addrs.first().unwrap().port(), 9231); + for addr in resolve_loopback_target("ws://localhost").unwrap() { + assert_eq!(addr.port(), 9231, "every candidate uses the default port"); + } + } + + /// The codec version stamped on the envelope is hand-mirrored from the + /// generated TS `TRUAPI_CODEC_VERSION` (codegen emits only the schema hash to + /// Rust). Bind it to the Rust-side authority on the same number: the codec + /// version this host accepts in the handshake. A `--codec-version` bump that + /// forgets this constant then fails here instead of stamping a frame the + /// debugger reads as a foreign contract. + #[test] + fn stamped_codec_version_is_the_one_the_host_negotiates() { + use truapi::api::System; + use truapi::versioned::system::{ + HostFeatureSupportedError, HostFeatureSupportedRequest, HostFeatureSupportedResponse, + HostHandshakeRequest, HostInfoError, HostInfoRequest, HostInfoResponse, + HostNavigateToError, HostNavigateToRequest, HostNavigateToResponse, + }; + use truapi::{CallContext, CallError, v01}; + + /// Exercises only `System::handshake`'s default (host-side) codec check. + struct HandshakeOnly; + + #[truapi::async_trait] + impl System for HandshakeOnly { + async fn feature_supported( + &self, + _cx: &CallContext, + _request: HostFeatureSupportedRequest, + ) -> Result> + { + unreachable!("handshake-only host") + } + + async fn navigate_to( + &self, + _cx: &CallContext, + _request: HostNavigateToRequest, + ) -> Result> { + unreachable!("handshake-only host") + } + + async fn host_info( + &self, + _cx: &CallContext, + _request: HostInfoRequest, + ) -> Result> { + unreachable!("handshake-only host") + } + } + + let handshake = |codec: u32| { + let cx = CallContext::with_request_id("codec:1".to_string()); + let codec_version = u8::try_from(codec).expect("codec version fits a u8"); + futures::executor::block_on(HandshakeOnly.handshake( + &cx, + HostHandshakeRequest::V1(v01::HostHandshakeRequest { codec_version }), + )) + }; + + assert!( + handshake(WIRE_CODEC_VERSION).is_ok(), + "the host must accept the codec version its debug envelopes stamp" + ); + assert!( + handshake(WIRE_CODEC_VERSION + 1).is_err(), + "the stamped codec version must be the newest one the host accepts" + ); + } + + #[tokio::test] + async fn emit_is_nonblocking_and_counts_drops_when_debugger_absent() { + // A loopback port with nothing listening: dials never succeed, so the + // bounded queue fills and further frames are dropped, never blocking emit. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + drop(listener); // free the port; nothing is listening now + + let sink = WsDebugSink::connect(&format!("ws://127.0.0.1:{port}")).unwrap(); + for _ in 0..(QUEUE_CAPACITY + 50) { + sink.emit(DebugEvent::Frame { + channel_id: ChannelId("myapp.dot".to_string()), + dir: FrameDirection::Out, + bytes: vec![1], + }); + } + assert!( + sink.dropped() > 0, + "a full queue must count drops, not block" + ); + } + + #[tokio::test] + async fn byte_budget_drops_large_frames_before_the_count_cap() { + // Nothing listening: the writer never drains, so queued bytes accumulate. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + drop(listener); + + let sink = WsDebugSink::connect(&format!("ws://127.0.0.1:{port}")).unwrap(); + // ~2 MiB per frame; a handful blows past the 8 MiB byte budget long before + // the 4096-frame count cap, so the BYTE cap is what drops here. Also + // exercises reserve-before-send: emit must never panic on the counter even + // as the writer task races it. + let big = vec![0u8; 2 * 1024 * 1024]; + for _ in 0..8 { + sink.emit(DebugEvent::Frame { + channel_id: ChannelId("myapp.dot".to_string()), + dir: FrameDirection::Out, + bytes: big.clone(), + }); + } + assert!( + sink.dropped() > 0, + "the byte budget must drop large frames well under the count cap" + ); + } +} diff --git a/rust/crates/truapi-server/src/wasm.rs b/rust/crates/truapi-server/src/wasm.rs index 8b849fb0f..e6a212c0d 100644 --- a/rust/crates/truapi-server/src/wasm.rs +++ b/rust/crates/truapi-server/src/wasm.rs @@ -38,8 +38,8 @@ use wasm_bindgen::prelude::*; use crate::SigningHostRuntime; use crate::subscription::Spawner; use crate::{ - FrameSink, PairingHostRuntime, PermissionAuthorizationRequest, PermissionAuthorizationStatus, - ProductRuntime, + ChannelId, DebugEvent, DebugSink, FrameSink, PairingHostRuntime, + PermissionAuthorizationRequest, PermissionAuthorizationStatus, ProductRuntime, }; mod generated_bridge; @@ -74,6 +74,47 @@ impl FrameSink for WasmFrameSink { } } +/// This core's wire-contract fingerprint, for a host to stamp on each debug +/// envelope it forwards to the debugger. +/// +/// The frames a web host taps are encoded by *this* core, so the identity the +/// debugger checks has to come from here. A host that stamped its JS client's +/// hash instead would attest to a table it did not encode with: `dist/wasm/web/` +/// is a hand-built, gitignored artifact, so a stale core paired with a fresh +/// client would pass the identity check while emitting frames from a different +/// contract - exactly the silent mis-decode the fingerprint exists to stop. +#[wasm_bindgen(js_name = wireSchemaHash)] +pub fn wire_schema_hash() -> String { + crate::generated::wire_table::TRUAPI_WIRE_SCHEMA_HASH.to_string() +} + +/// Streams tapped debug frames out to a JS `debugEmit(channelId, dir, frame)` +/// callback so the host worker can forward them to the debugger it dials. +/// Dev-only: installed only when the host provides the callback, and +/// fire-and-forget - a failing callback is logged, never propagated. +struct WasmDebugSink { + emit: SendWrapper, +} + +impl DebugSink for WasmDebugSink { + fn emit(&self, event: DebugEvent) { + let DebugEvent::Frame { + channel_id, + dir, + bytes, + } = event; + let frame = Uint8Array::from(bytes.as_slice()); + if let Err(err) = self.emit.call3( + &JsValue::NULL, + &JsValue::from_str(&channel_id.0), + &JsValue::from_str(dir.wire_str()), + &frame, + ) { + web_sys::console::error_1(&err); + } + } +} + struct WasmPlatform { bridge: SendWrapper>, } @@ -837,10 +878,20 @@ impl WasmPairingHostRuntime { ) -> Result { let product = product_context_from_js(&product)?; let channel = CoreChannel::from_js(&core_callbacks)?; + let debug_emit = get_optional_function(&core_callbacks, "debugEmit")?; + let channel_id = product.product_id.clone(); let sink = Arc::new(WasmFrameSink { emit_frame: SendWrapper::new(channel.emit_frame), }); let runtime = self.runtime.product_runtime(product, sink); + if let Some(debug_emit) = debug_emit { + runtime.set_debug_sink( + ChannelId(channel_id), + Arc::new(WasmDebugSink { + emit: SendWrapper::new(debug_emit), + }), + ); + } Ok(WasmProductRuntime::from_parts(runtime, channel.dispose)) } diff --git a/rust/crates/truapi/src/api/account.rs b/rust/crates/truapi/src/api/account.rs index f7ee2b758..f3e18bcde 100644 --- a/rust/crates/truapi/src/api/account.rs +++ b/rust/crates/truapi/src/api/account.rs @@ -151,7 +151,7 @@ pub trait Account: Send + Sync { /// ); /// console.log("foreign account proof refused without prompting"); /// ``` - #[wire(request_id = 26)] + #[wire(request_id = 26, sensitive)] async fn create_account_proof( &self, _cx: &CallContext, @@ -185,7 +185,7 @@ pub trait Account: Send + Sync { /// assert(result.isOk(), "signVrf failed:", result); /// console.log("vrf signature:", result.value); /// ``` - #[wire(request_id = 164)] + #[wire(request_id = 164, sensitive)] async fn sign_vrf( &self, _cx: &CallContext, @@ -298,7 +298,7 @@ pub trait Account: Send + Sync { /// assert(result.isOk(), "getUserId failed:", result); /// console.log("user id:", result.value); /// ``` - #[wire(request_id = 110)] + #[wire(request_id = 110, sensitive)] async fn get_user_id( &self, _cx: &CallContext, @@ -319,7 +319,7 @@ pub trait Account: Send + Sync { /// assert(result.isOk(), "requestLogin failed:", result); /// console.log("login completed:", result.value); /// ``` - #[wire(request_id = 112)] + #[wire(request_id = 112, sensitive)] async fn request_login( &self, _cx: &CallContext, diff --git a/rust/crates/truapi/src/api/coin_payment.rs b/rust/crates/truapi/src/api/coin_payment.rs index 5839b8e3c..90baf8417 100644 --- a/rust/crates/truapi/src/api/coin_payment.rs +++ b/rust/crates/truapi/src/api/coin_payment.rs @@ -141,7 +141,7 @@ pub trait CoinPayment: Send + Sync { /// assert(result.isOk(), "createCheque failed:", result); /// console.log("cheque created:", result.value.cheque); /// ``` - #[wire(request_id = 150)] + #[wire(request_id = 150, sensitive)] async fn create_cheque( &self, _cx: &CallContext, @@ -168,7 +168,7 @@ pub trait CoinPayment: Send + Sync { /// ); /// console.log("deposit status:", status); /// ``` - #[wire(start_id = 152)] + #[wire(start_id = 152, sensitive)] async fn deposit( &self, _cx: &CallContext, @@ -222,7 +222,7 @@ pub trait CoinPayment: Send + Sync { /// ); /// console.log("payment received:", item); /// ``` - #[wire(start_id = 160)] + #[wire(start_id = 160, sensitive)] async fn listen_for_payment( &self, _cx: &CallContext, diff --git a/rust/crates/truapi/src/api/entropy.rs b/rust/crates/truapi/src/api/entropy.rs index 32f510b9b..36176db6c 100644 --- a/rust/crates/truapi/src/api/entropy.rs +++ b/rust/crates/truapi/src/api/entropy.rs @@ -18,7 +18,7 @@ pub trait Entropy: Send + Sync { /// assert(result.isOk(), "derive failed:", result); /// console.log("entropy derived:", result.value); /// ``` - #[wire(request_id = 108)] + #[wire(request_id = 108, sensitive)] async fn derive( &self, _cx: &CallContext, diff --git a/rust/crates/truapi/src/api/local_storage.rs b/rust/crates/truapi/src/api/local_storage.rs index ec0bc6343..5c2057858 100644 --- a/rust/crates/truapi/src/api/local_storage.rs +++ b/rust/crates/truapi/src/api/local_storage.rs @@ -18,7 +18,7 @@ pub trait LocalStorage: Send + Sync { /// assert(result.isOk(), "read failed:", result); /// console.log("storage value read:", result.value.value); /// ``` - #[wire(request_id = 12)] + #[wire(request_id = 12, sensitive)] async fn read( &self, cx: &CallContext, @@ -35,7 +35,7 @@ pub trait LocalStorage: Send + Sync { /// assert(result.isOk(), "write failed:", result); /// console.log("storage write succeeded"); /// ``` - #[wire(request_id = 14)] + #[wire(request_id = 14, sensitive)] async fn write( &self, cx: &CallContext, diff --git a/rust/crates/truapi/src/api/payment.rs b/rust/crates/truapi/src/api/payment.rs index eab781c5f..f1740cc59 100644 --- a/rust/crates/truapi/src/api/payment.rs +++ b/rust/crates/truapi/src/api/payment.rs @@ -112,7 +112,7 @@ pub trait Payment: Send + Sync { /// assert(result.isOk(), "topUp failed:", result); /// console.log("balance topped up"); /// ``` - #[wire(request_id = 122)] + #[wire(request_id = 122, sensitive)] async fn top_up( &self, _cx: &CallContext, diff --git a/rust/crates/truapi/src/api/signing.rs b/rust/crates/truapi/src/api/signing.rs index 010d39a31..aa13b3f9b 100644 --- a/rust/crates/truapi/src/api/signing.rs +++ b/rust/crates/truapi/src/api/signing.rs @@ -61,7 +61,7 @@ pub trait Signing: Send + Sync { /// console.log(`${version} transaction created:`, result.value); /// } /// ``` - #[wire(request_id = 30)] + #[wire(request_id = 30, sensitive)] async fn create_transaction( &self, _cx: &CallContext, @@ -118,7 +118,7 @@ pub trait Signing: Send + Sync { /// assert(result.isOk(), "createTransactionWithLegacyAccount failed:", result); /// console.log("transaction created:", result.value); /// ``` - #[wire(request_id = 32)] + #[wire(request_id = 32, sensitive)] async fn create_transaction_with_legacy_account( &self, _cx: &CallContext, @@ -150,7 +150,7 @@ pub trait Signing: Send + Sync { /// assert(result.isOk(), "signRawWithLegacyAccount failed:", result); /// console.log("raw bytes signed:", result.value); /// ``` - #[wire(request_id = 34)] + #[wire(request_id = 34, sensitive)] async fn sign_raw_with_legacy_account( &self, _cx: &CallContext, @@ -196,7 +196,7 @@ pub trait Signing: Send + Sync { /// assert(result.isOk(), "signPayloadWithLegacyAccount failed:", result); /// console.log("payload signed:", result.value); /// ``` - #[wire(request_id = 36)] + #[wire(request_id = 36, sensitive)] async fn sign_payload_with_legacy_account( &self, _cx: &CallContext, @@ -226,7 +226,7 @@ pub trait Signing: Send + Sync { /// assert(result.isOk(), "signRaw failed:", result); /// console.log("raw bytes signed:", result.value); /// ``` - #[wire(request_id = 114)] + #[wire(request_id = 114, sensitive)] async fn sign_raw( &self, _cx: &CallContext, @@ -263,7 +263,7 @@ pub trait Signing: Send + Sync { /// assert(result.isOk(), "signPayload failed:", result); /// console.log("payload signed:", result.value); /// ``` - #[wire(request_id = 116)] + #[wire(request_id = 116, sensitive)] async fn sign_payload( &self, _cx: &CallContext, diff --git a/rust/crates/truapi/src/api/statement_store.rs b/rust/crates/truapi/src/api/statement_store.rs index ae8bdbef8..66a693e0a 100644 --- a/rust/crates/truapi/src/api/statement_store.rs +++ b/rust/crates/truapi/src/api/statement_store.rs @@ -57,7 +57,7 @@ pub trait StatementStore: Send + Sync { /// const page = await waitForStatement(); /// console.log("subscribe received", page); /// ``` - #[wire(start_id = 56)] + #[wire(start_id = 56, sensitive)] async fn subscribe( &self, _cx: &CallContext, @@ -99,7 +99,7 @@ pub trait StatementStore: Send + Sync { /// console.log("proof created:", result.value); /// } /// ``` - #[wire(request_id = 60)] + #[wire(request_id = 60, sensitive)] async fn create_proof( &self, _cx: &CallContext, @@ -126,7 +126,7 @@ pub trait StatementStore: Send + Sync { /// assert(result.isOk(), "createProof failed:", result); /// console.log("proof created:", result.value); /// ``` - #[wire(request_id = 132)] + #[wire(request_id = 132, sensitive)] async fn create_proof_authorized( &self, _cx: &CallContext, @@ -158,7 +158,7 @@ pub trait StatementStore: Send + Sync { /// assert(result.isOk(), "submit failed:", result); /// console.log("statement submitted"); /// ``` - #[wire(request_id = 62)] + #[wire(request_id = 62, sensitive)] async fn submit( &self, _cx: &CallContext,