diff --git a/.envrc b/.envrc deleted file mode 100644 index 3550a30..0000000 --- a/.envrc +++ /dev/null @@ -1 +0,0 @@ -use flake diff --git a/.forgejo/workflows/ci.yml b/.forgejo/workflows/ci.yml index 970eece..70e6c39 100644 --- a/.forgejo/workflows/ci.yml +++ b/.forgejo/workflows/ci.yml @@ -7,12 +7,16 @@ on: env: CARGO_TERM_COLOR: always + NIX_CONFIG: experimental-features = nix-command flakes jobs: checks: name: checks runs-on: nixos steps: + - name: Install node + run: nix profile add nixpkgs#nodejs_24 + - name: Checkout uses: https://data.forgejo.org/actions/checkout@v7 @@ -29,6 +33,8 @@ jobs: cargo machete pnpm install --frozen-lockfile + pnpm add --save-dev --save-exact --workspace-root jscpd-linux-x64-gnu@5.0.14 + pnpm run dup RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack build wasm --target web diff --git a/.forgejo/workflows/release.yml b/.forgejo/workflows/release.yml index 5996da0..05bc6e2 100644 --- a/.forgejo/workflows/release.yml +++ b/.forgejo/workflows/release.yml @@ -14,10 +14,16 @@ on: required: true type: string +env: + NIX_CONFIG: experimental-features = nix-command flakes + jobs: release: runs-on: nixos steps: + - name: Install node & bun + run: nix profile add nixpkgs#nodejs_24 nixpkgs#bun + - name: Check out repo uses: https://data.forgejo.org/actions/checkout@v7 with: @@ -26,6 +32,9 @@ jobs: - name: Install dependencies run: bun install + - name: Install cc linker, sed & jq + run: nix profile add nixpkgs#stdenv.cc nixpkgs#gnused nixpkgs#jq + - name: Build all run: bun build:all diff --git a/.gitignore b/.gitignore index f600624..6dff05e 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,3 @@ dist/ *.tgz wasm/pkg/ web_client/ -.direnv diff --git a/Cargo.lock b/Cargo.lock index 8e8f6ba..88e9b74 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -256,9 +256,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.2" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1326,6 +1326,9 @@ dependencies = [ "mtp-transport", "mtp-type-map", "mtp-webserver", + "rand", + "rcgen", + "tokio", ] [[package]] @@ -1360,6 +1363,7 @@ name = "mtp-common" version = "0.3.0" dependencies = [ "quinn", + "rustls", "thiserror 2.0.20", "wtransport", ] @@ -1800,7 +1804,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.2", + "chacha20 0.10.1", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/Cargo.toml b/Cargo.toml index 2c1b1ec..ee1c93a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -113,5 +113,10 @@ tls = ["crypto", "mtp-crypto?/tls"] # Requires MTP_INSECURE_TLS=1 at runtime. insecure-tls = ["dep:mtp-transport", "mtp-transport?/insecure-tls"] +[dev-dependencies] +tokio = { version = "1", features = ["full"] } +rcgen = "0.14" +rand = "0.10.1" + [package.metadata.cargo-machete] ignored = ["mtp-transport"] diff --git a/client/src/pipe.rs b/client/src/pipe.rs index 136fe01..8e839dd 100644 --- a/client/src/pipe.rs +++ b/client/src/pipe.rs @@ -78,41 +78,9 @@ pub struct PipeRequest { pub(crate) pipe_id: u32, pub(crate) description: String, pub(crate) sender: Sender, - pub(crate) receiver: Receiver, pub(crate) dispatcher: Arc, } -#[cfg(feature = "pipes")] -struct ExpectedPipeGuard { - receiver: Receiver, - pipe_id: u32, - armed: bool, -} - -#[cfg(feature = "pipes")] -impl ExpectedPipeGuard { - fn new(receiver: Receiver, pipe_id: u32) -> Self { - Self { - receiver, - pipe_id, - armed: true, - } - } - - fn disarm(&mut self) { - self.armed = false; - } -} - -#[cfg(feature = "pipes")] -impl Drop for ExpectedPipeGuard { - fn drop(&mut self) { - if self.armed { - self.receiver.cancel_expected_pipe(self.pipe_id); - } - } -} - #[cfg(feature = "pipes")] impl PipeRequest { pub fn id(&self) -> u32 { @@ -124,10 +92,6 @@ impl PipeRequest { } pub async fn accept(self) -> Result { - self.receiver - .expect_pipe(self.pipe_id) - .map_err(PipeError::from)?; - let mut expected_pipe = ExpectedPipeGuard::new(self.receiver.clone(), self.pipe_id); let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel(); { let mut pending = self.dispatcher.pending_pipes.lock().await; @@ -151,10 +115,7 @@ impl PipeRequest { let timeout = self.dispatcher.policy.read_timeout; match tokio::time::timeout(timeout, pipe_rx).await { - Ok(Ok(reader)) => { - expected_pipe.disarm(); - Ok(reader) - } + Ok(Ok(reader)) => Ok(reader), Ok(Err(_)) => { self.dispatcher .pending_pipes @@ -452,7 +413,6 @@ pub(crate) async fn run_dispatcher( pipe_id, description, sender: sender.clone(), - receiver: receiver.clone(), dispatcher: dispatcher.clone(), }; let _ = pipe_req_tx.send(req).await; diff --git a/codec/Cargo.lock b/codec/Cargo.lock index a898cf4..5ed6f53 100644 --- a/codec/Cargo.lock +++ b/codec/Cargo.lock @@ -187,9 +187,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.2" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1151,7 +1151,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.2", + "chacha20 0.10.1", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/common/Cargo.lock b/common/Cargo.lock index 8d0f18c..338d35d 100644 --- a/common/Cargo.lock +++ b/common/Cargo.lock @@ -139,9 +139,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" [[package]] name = "chacha20" -version = "0.10.2" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", "cpufeatures", diff --git a/common/Cargo.toml b/common/Cargo.toml index 5fae68e..24d4777 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -15,6 +15,7 @@ wtransport = { version = "0.7.1", default-features = false, features = [ "quinn", "self-signed", ] } +rustls = { version = "0.23.41" } quinn = { version = "0.11.11", default-features = false, features = [ "rustls-aws-lc-rs", "rustls", diff --git a/common/src/lib.rs b/common/src/lib.rs index 71e87f2..1a7fe66 100644 --- a/common/src/lib.rs +++ b/common/src/lib.rs @@ -164,9 +164,6 @@ pub enum CommunicationError { #[error("Stream Error")] StreamError, - #[error("Stream failed after delivery may have started")] - DeliveryUnknown, - #[error("Stream Error: {0}")] #[cfg(not(target_arch = "wasm32"))] StreamWriteError(#[from] wtransport::error::StreamWriteError), @@ -185,38 +182,6 @@ pub enum CommunicationError { Other(String), } -/// How the protocol layer should handle the first frame on a receive stream. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum FirstFrameDisposition { - Message, - Pipe(u32), -} - -/// Classify a first frame without tying the decision to a WebTransport backend. -/// -/// `PipeRequest` is used both as a control message and as the header of the raw -/// stream opened after that request is accepted. Only the protocol layer knows -/// which raw stream IDs are currently expected. -pub fn classify_first_frame( - is_pipe_request: bool, - pipe_id: Option, - pipe_is_expected: bool, -) -> Result { - if !is_pipe_request { - return Ok(FirstFrameDisposition::Message); - } - - let pipe_id = pipe_id.filter(|id| *id != 0).ok_or_else(|| { - CommunicationError::Other("PipeRequest frame must contain a non-zero id".into()) - })?; - - if pipe_is_expected { - Ok(FirstFrameDisposition::Pipe(pipe_id)) - } else { - Ok(FirstFrameDisposition::Message) - } -} - // ---- manual PartialEq (quinn / wtransport types don't impl PartialEq) ---- impl PartialEq for CommunicationError { @@ -247,7 +212,6 @@ impl PartialEq for CommunicationError { (Self::ReadExactError(_), Self::ReadExactError(_)) => true, (Self::StreamClosed, Self::StreamClosed) => true, (Self::StreamError, Self::StreamError) => true, - (Self::DeliveryUnknown, Self::DeliveryUnknown) => true, #[cfg(not(target_arch = "wasm32"))] (Self::StreamWriteError(_), Self::StreamWriteError(_)) => true, #[cfg(not(target_arch = "wasm32"))] diff --git a/crypto/Cargo.lock b/crypto/Cargo.lock index 5b86012..e536096 100644 --- a/crypto/Cargo.lock +++ b/crypto/Cargo.lock @@ -181,9 +181,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.2" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -853,7 +853,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.2", + "chacha20 0.10.1", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/docs/WASM-CLIENT.md b/docs/WASM-CLIENT.md index 8a56bbd..02be900 100644 --- a/docs/WASM-CLIENT.md +++ b/docs/WASM-CLIENT.md @@ -104,9 +104,6 @@ if (!MTPClient.isSupported()) { | `requestTimeoutMs` | 30 seconds | Default `request()` timeout. | | `pings` | `false` | Protocol pings, or an object with `intervalMs`. | | `logger` | No-op | Receives SDK state and error events. | -| `schemas` | None | Client-wide request and response schema registry. | -| `throwProtocolErrors` | `false` | Reject requests whose correlated response is an `Error*` frame. | -| `onValidationError` | No-op | Receives subscription validation failures. | | `sessionStorage` | In-memory | E2EE session state storage. | | `encryptedSecretProvider` | In-memory | Independent caller-managed encrypted secret storage. | | `defaultSignatureVerificationPolicy` | `"ed25519"` | Receiver policy for protected signatures. | @@ -474,60 +471,6 @@ const unsubscribe = client.subscribe("SomeType", (message) => { unsubscribe(); ``` -### Zod request and response schemas - -Applications can provide their request and response schemas once when creating -the client. MTP uses `parseAsync`, so synchronous schemas, async refinements, -defaults, coercions, and transforms all work. MTP has no runtime dependency on -Zod; the application supplies its preferred Zod version. - -```typescript -import { z } from "zod"; -import { MTPClient, MTPValidationError } from "mtp"; - -const schemas = { - GetUser: { - request: z.object({ UserId: z.number().int().positive() }), - response: z.object({ - UserId: z.number().int().positive(), - Display: z.string(), - }), - }, -}; - -const client = await MTPClient.create({ - url, - schemas, - throwProtocolErrors: true, - onValidationError(error) { - console.error(error.messageType, error.cause); - }, -}); - -const response = await client.request("GetUser", { UserId: 42 }); -console.log(response.data.Display); -``` - -Request schemas run before frame encoding and transmission. Their transformed -output is sent. Response schemas run after request correlation, and their -transformed output replaces `frame.data`; `frame.raw`, when present, remains the -original wire frame. Invalid requests and responses reject with -`MTPValidationError`. Invalid subscription messages do not reach the handler -and are reported through `onValidationError`. - -`throwProtocolErrors: true` converts correlated `Error*` frames into -`MTPProtocolError`. It defaults to `false` for compatibility. - -`MTPProxyConnection` applies the same schema registry to another TypeScript -request/subscription transport, such as a Tauri command and event proxy: - -```typescript -const connection = new MTPProxyConnection(adapter, { - schemas, - throwProtocolErrors: true, -}); -``` - Protocol ping behavior is defined in [Protocol Reference](PROTOCOL-REFERENCE.md#protocol-keepalive). The SDK configuration is: ```typescript diff --git a/example/Cargo.lock b/example/Cargo.lock index f81b014..8d2fb30 100644 --- a/example/Cargo.lock +++ b/example/Cargo.lock @@ -225,9 +225,9 @@ dependencies = [ [[package]] name = "chacha20" -version = "0.10.2" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1305,6 +1305,7 @@ name = "mtp-common" version = "0.3.0" dependencies = [ "quinn", + "rustls", "thiserror 2.0.20", "wtransport", ] @@ -1701,7 +1702,7 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "chacha20 0.10.2", + "chacha20 0.10.1", "getrandom 0.4.3", "rand_core 0.10.1", ] diff --git a/example/server/src/main.rs b/example/server/src/main.rs index 08969f4..17b4164 100644 --- a/example/server/src/main.rs +++ b/example/server/src/main.rs @@ -38,7 +38,6 @@ async fn handle_pipe_loopback( conn: &mtp::webserver::WebMTPConnection, request: mtp::host::PipeRequest< mtp::webserver::WebMtpSender, - mtp::webserver::WebMtpReceiver, mtp::webserver::H3TransportReceiver, >, ) -> Result> { diff --git a/flake.nix b/flake.nix index 0a40973..7951305 100644 --- a/flake.nix +++ b/flake.nix @@ -5,30 +5,26 @@ rust-overlay.url = "github:oxalica/rust-overlay"; }; - outputs = - { - self, - nixpkgs, - rust-overlay, - }: - let - systems = [ - "aarch64-darwin" - "aarch64-linux" - "x86_64-darwin" - "x86_64-linux" - ]; - eachSystem = - f: - nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate { } ( - map (system: nixpkgs.lib.mapAttrs (_: value: { ${system} = value; }) (f system)) systems - ); - in + outputs = { + self, + nixpkgs, + rust-overlay, + }: let + systems = [ + "aarch64-darwin" + "aarch64-linux" + "x86_64-darwin" + "x86_64-linux" + ]; + eachSystem = f: + nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate {} ( + map (system: nixpkgs.lib.mapAttrs (_: value: {${system} = value;}) (f system)) systems + ); + in eachSystem ( - system: - let - overlays = [ rust-overlay.overlays.default ]; - pkgs = import nixpkgs { inherit system overlays; }; + system: let + overlays = [rust-overlay.overlays.default]; + pkgs = import nixpkgs {inherit system overlays;}; rustToolchain = pkgs.rust-bin.stable.latest.default.override { extensions = [ @@ -36,12 +32,12 @@ "clippy" "rustfmt" ]; - targets = [ "wasm32-unknown-unknown" ]; + targets = ["wasm32-unknown-unknown"]; }; clippyCheck = pkgs.writeShellApplication { name = "mtp-clippy"; - runtimeInputs = [ rustToolchain ]; + runtimeInputs = [rustToolchain]; text = '' export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}" cargo clippy --workspace --exclude mtp-wasm --all-targets --all-features -- -D warnings -W unreachable-pub @@ -50,7 +46,7 @@ macheteCheck = pkgs.writeShellApplication { name = "mtp-machete"; - runtimeInputs = [ pkgs.cargo-machete ]; + runtimeInputs = [pkgs.cargo-machete]; text = '' cargo machete "$@" ''; @@ -58,15 +54,7 @@ buildAll = pkgs.writeShellApplication { name = "mtp-build-all"; - runtimeInputs = [ - rustToolchain - pkgs.cargo-deny - pkgs.wasm-pack - pkgs.pnpm - pkgs.coreutils - clippyCheck - macheteCheck - ]; + runtimeInputs = [rustToolchain pkgs.cargo-deny pkgs.wasm-pack pkgs.pnpm pkgs.coreutils clippyCheck macheteCheck]; text = '' export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}" @@ -78,6 +66,7 @@ cargo check --manifest-path example/Cargo.toml --workspace --all-targets --all-features mtp-clippy mtp-machete + pnpm run dup pnpm run build RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm pnpm run test:e2e @@ -90,17 +79,13 @@ healthCheck = pkgs.writeShellApplication { name = "mtp-health"; - runtimeInputs = [ - clippyCheck - macheteCheck - ]; + runtimeInputs = [clippyCheck macheteCheck]; text = '' mtp-clippy mtp-machete ''; }; - in - { + in { devShells = { default = pkgs.mkShell { name = "mtp-dev"; diff --git a/host/src/config.rs b/host/src/config.rs index 99d62a2..0cef52a 100644 --- a/host/src/config.rs +++ b/host/src/config.rs @@ -251,8 +251,6 @@ pub struct HostConfig { #[cfg(feature = "crypto")] pub authentication_policy: AuthenticationPolicy, #[cfg(feature = "crypto")] - authentication_policy_explicit: bool, - #[cfg(feature = "crypto")] pub auth_timeout: Duration, #[cfg(feature = "crypto")] pub require_pq: bool, @@ -290,8 +288,6 @@ impl HostConfig { #[cfg(feature = "crypto")] authentication_policy: AuthenticationPolicy::Unauthenticated, #[cfg(feature = "crypto")] - authentication_policy_explicit: false, - #[cfg(feature = "crypto")] auth_timeout: Duration::from_secs(30), #[cfg(feature = "crypto")] require_pq: true, @@ -345,9 +341,7 @@ impl HostConfig { get_existing_client: GetExistingClient, complete_register: CompleteRegister, ) -> Self { - if !self.authentication_policy_explicit { - self.authentication_policy = AuthenticationPolicy::ForceAuthentication; - } + self.authentication_policy = AuthenticationPolicy::ForceAuthentication; self.host_keyring = host_keyring; self.get_existing_client = Box::new(get_existing_client); self.complete_register = Box::new(complete_register); @@ -357,7 +351,6 @@ impl HostConfig { #[cfg(feature = "crypto")] pub fn with_authentication_policy(mut self, policy: AuthenticationPolicy) -> Self { self.authentication_policy = policy; - self.authentication_policy_explicit = true; self } @@ -448,78 +441,4 @@ mod tests { .expect("repeated registration decision") ); } - - fn test_keyring() -> mtp_crypto::Keyring { - mtp_crypto::Keyring::new( - mtp_crypto::KemPublicKey::new(Vec::new()), - mtp_crypto::KemPrivateKey::new(Vec::new()), - mtp_crypto::SignaturePqPublicKey::new(Vec::new()), - mtp_crypto::SignaturePqPrivateKey::new(Vec::new()), - mtp_crypto::SignaturePublicKey::new(Vec::new()), - mtp_crypto::SignaturePrivateKey::new(Vec::new()), - ) - } - - fn test_get_existing_client() -> GetExistingClient { - Box::new(|_, _| Box::pin(async { None })) - } - - fn test_complete_register() -> CompleteRegister { - Box::new(|_, _| Box::pin(async { 1 })) - } - - fn test_config() -> HostConfig { - HostConfig::new( - IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), - 4433, - Vec::new(), - Vec::new(), - ) - } - - #[test] - fn with_authentication_defaults_to_force_authentication() { - let config = test_config().with_authentication( - test_keyring(), - test_get_existing_client(), - test_complete_register(), - ); - - assert_eq!( - config.authentication_policy, - AuthenticationPolicy::ForceAuthentication - ); - } - - #[test] - fn explicit_authentication_policy_before_with_authentication_is_preserved() { - let config = test_config() - .with_authentication_policy(AuthenticationPolicy::AllowAuthentication) - .with_authentication( - test_keyring(), - test_get_existing_client(), - test_complete_register(), - ); - - assert_eq!( - config.authentication_policy, - AuthenticationPolicy::AllowAuthentication - ); - } - - #[test] - fn explicit_authentication_policy_after_with_authentication_is_preserved() { - let config = test_config() - .with_authentication( - test_keyring(), - test_get_existing_client(), - test_complete_register(), - ) - .with_authentication_policy(AuthenticationPolicy::AllowAuthentication); - - assert_eq!( - config.authentication_policy, - AuthenticationPolicy::AllowAuthentication - ); - } } diff --git a/host/src/connection.rs b/host/src/connection.rs index 7871d55..71c60db 100644 --- a/host/src/connection.rs +++ b/host/src/connection.rs @@ -73,7 +73,7 @@ pub struct MTPConnection< #[cfg(feature = "pipes")] pub(crate) app_rx: Mutex>>, #[cfg(feature = "pipes")] - pub(crate) pipe_req_rx: Mutex>>, + pub(crate) pipe_req_rx: Mutex>>, #[cfg(feature = "pipes")] pub(crate) pipe_dispatcher: Arc>, #[cfg(not(feature = "pipes"))] @@ -381,7 +381,7 @@ where }) } - pub async fn receive_pipe(&self) -> Result, CommunicationError> { + pub async fn receive_pipe(&self) -> Result, CommunicationError> { self.pipe_req_rx .lock() .await diff --git a/host/src/engine.rs b/host/src/engine.rs old mode 100755 new mode 100644 index 08ca87f..38e8422 --- a/host/src/engine.rs +++ b/host/src/engine.rs @@ -217,12 +217,6 @@ impl HandshakeEngine { _authentication_context: &AuthenticationContext, ) -> Result { let mut first_msg = receiver.receive().await.map_err(AcceptError::Receive)?; - tracing::debug!( - message_type = ?first_msg.get_type(), - version = ?first_msg.get_str(DataType::Version), - client_id = ?first_msg.get_data(DataType::Id), - "received MTP opening message" - ); let version_str = match first_msg.get_data(DataType::Version) { Some(DataValue::Str(s)) => s.clone(), @@ -277,11 +271,6 @@ impl HandshakeEngine { return Err(AcceptError::UnsupportedVersion(client_version)); } }; - tracing::debug!( - client_version = %client_version, - negotiated_version = %negotiated, - "MTP protocol version negotiated" - ); let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) .ok_or_else(|| AcceptError::UnsupportedVersion(negotiated.clone()))?; @@ -309,12 +298,6 @@ impl HandshakeEngine { ) || registration || first_msg.get_data(DataType::PublicKeys).is_some() || claimed_client_id.is_some_and(|client_id| client_id != 0); - tracing::info!( - claimed_client_id = ?claimed_client_id, - registration, - authentication_requested, - "classified MTP opening authentication mode" - ); if authentication_requested { let attempt = crate::config::AuthenticationAttempt { peer_network_identity: _authentication_context.peer_network_identity.clone(), @@ -572,7 +555,6 @@ impl HandshakeEngine { } // Unknown or zero ID: fall back to guest - tracing::info!("allocating MTP guest identity"); let guest_id_lease = match self.assign_guest_id().await { Ok(lease) => lease, Err(error) => { @@ -581,7 +563,6 @@ impl HandshakeEngine { } }; let guest_id = guest_id_lease.id; - tracing::info!(guest_id, "allocated MTP guest identity"); send_accepted_generic(sender, &negotiated, tm, Some(guest_id)) .await .map_err(AcceptError::Send)?; @@ -1124,12 +1105,6 @@ async fn send_rejection_generic( .add_typed_default(DataType::Connected, DataValue::BoolFalse) .add_typed_default(DataType::ErrorMessage, DataValue::Str(reason.to_string())), }; - tracing::debug!( - reason = %reason, - response_type = ?response.get_type(), - has_version = response.get_data(DataType::Version).is_some(), - "sending MTP handshake rejection" - ); let _ = sender.send(&response).await; } @@ -1163,11 +1138,6 @@ async fn send_accepted_generic( if let Some(id) = assigned_id { response = response.add_typed_default(DataType::Id, DataValue::UnsignedNumber(id as u128)); } - tracing::debug!( - version = %version, - assigned_id = ?assigned_id, - "sending accepted MTP handshake response" - ); sender.send(&response).await?; sender.finish_stream().await } diff --git a/host/src/pipe.rs b/host/src/pipe.rs index 192e3d7..eae1383 100644 --- a/host/src/pipe.rs +++ b/host/src/pipe.rs @@ -27,10 +27,6 @@ pub trait PipeReceiver

: Clone + Send + Sync + 'static where P: tokio::io::AsyncRead + Send + Unpin + 'static, { - fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError>; - - fn cancel_expected_pipe(&self, pipe_id: u32); - fn receive_pipe_event( &self, ) -> impl std::future::Future, CommunicationError>> + Send; @@ -56,14 +52,6 @@ impl PipeSender for mtp_transport::Sender { } impl PipeReceiver for mtp_transport::Receiver { - fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { - self.expect_pipe(pipe_id) - } - - fn cancel_expected_pipe(&self, pipe_id: u32) { - self.cancel_expected_pipe(pipe_id); - } - async fn receive_pipe_event( &self, ) -> Result, CommunicationError> { @@ -99,14 +87,6 @@ where C: mtp_transport::TransportConnection, C::RecvStream: tokio::io::AsyncRead + Send + Unpin + 'static, { - fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { - self.expect_pipe(pipe_id) - } - - fn cancel_expected_pipe(&self, pipe_id: u32) { - self.cancel_expected_pipe(pipe_id); - } - async fn receive_pipe_event( &self, ) -> Result, CommunicationError> { @@ -172,60 +152,16 @@ where } } -pub struct PipeRequest { +pub struct PipeRequest { pub(crate) pipe_id: u32, pub(crate) description: String, pub(crate) sender: S, - pub(crate) receiver: R, pub(crate) dispatcher: Arc>, } -struct ExpectedPipeGuard -where - R: PipeReceiver

, - P: tokio::io::AsyncRead + Send + Unpin + 'static, -{ - receiver: R, - pipe_id: u32, - armed: bool, - _stream: std::marker::PhantomData

, -} - -impl ExpectedPipeGuard -where - R: PipeReceiver

, - P: tokio::io::AsyncRead + Send + Unpin + 'static, -{ - fn new(receiver: R, pipe_id: u32) -> Self { - Self { - receiver, - pipe_id, - armed: true, - _stream: std::marker::PhantomData, - } - } - - fn disarm(&mut self) { - self.armed = false; - } -} - -impl Drop for ExpectedPipeGuard -where - R: PipeReceiver

, - P: tokio::io::AsyncRead + Send + Unpin + 'static, -{ - fn drop(&mut self) { - if self.armed { - self.receiver.cancel_expected_pipe(self.pipe_id); - } - } -} - -impl PipeRequest +impl PipeRequest where S: PipeSender, - R: PipeReceiver

, P: tokio::io::AsyncRead + Send + Unpin + 'static, { pub fn id(&self) -> u32 { @@ -237,10 +173,6 @@ where } pub async fn accept(self) -> Result, PipeError> { - self.receiver - .expect_pipe(self.pipe_id) - .map_err(PipeError::from)?; - let mut expected_pipe = ExpectedPipeGuard::::new(self.receiver.clone(), self.pipe_id); let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel(); self.dispatcher .pending_pipes @@ -264,10 +196,7 @@ where } match tokio::time::timeout(self.dispatcher.policy.read_timeout, pipe_rx).await { - Ok(Ok(reader)) => { - expected_pipe.disarm(); - Ok(reader) - } + Ok(Ok(reader)) => Ok(reader), Ok(Err(_)) => { self.dispatcher .pending_pipes @@ -432,7 +361,7 @@ pub(crate) async fn run_dispatcher( receiver: R, sender: S, app_tx: mpsc::Sender>, - pipe_req_tx: mpsc::Sender>, + pipe_req_tx: mpsc::Sender>, dispatcher: Arc>, ) where S: PipeSender, @@ -459,7 +388,6 @@ pub(crate) async fn run_dispatcher( .unwrap_or("") .to_owned(), sender: sender.clone(), - receiver: receiver.clone(), dispatcher: dispatcher.clone(), }; let _ = pipe_req_tx.send(request).await; @@ -518,7 +446,6 @@ pub(crate) async fn run_dispatcher( pipe_id, description: reader.description().to_owned(), sender: sender.clone(), - receiver: receiver.clone(), dispatcher: dispatcher.clone(), }; let _ = pipe_req_tx.send(request).await; diff --git a/mtp-webserver/src/transport.rs b/mtp-webserver/src/transport.rs index 9dcce5f..31d8376 100644 --- a/mtp-webserver/src/transport.rs +++ b/mtp-webserver/src/transport.rs @@ -57,14 +57,14 @@ impl TransportSendStream for H3TransportSender { self.stream .write_all(buf) .await - .map_err(|_| CommunicationError::DeliveryUnknown)?; + .map_err(|_| CommunicationError::StreamError)?; // Control/authentication frames use a persistent stream. h3 keeps // those writes buffered until flushed; without this the peer can wait // for the challenge while the server waits for its proof. self.stream .flush() .await - .map_err(|_| CommunicationError::DeliveryUnknown) + .map_err(|_| CommunicationError::StreamError) } async fn finish(&mut self) -> Result<(), CommunicationError> { @@ -73,11 +73,6 @@ impl TransportSendStream for H3TransportSender { .await .map_err(|_| CommunicationError::StreamError) } - - fn reset(&mut self, code: u32) -> Result<(), CommunicationError> { - h3::quic::SendStream::reset(&mut self.stream, code as u64); - Ok(()) - } } #[async_trait::async_trait] @@ -93,7 +88,6 @@ impl TransportRecvStream for H3TransportReceiver { tracing::debug!( remote = %self.quinn.remote_address(), bytes = buf.len(), - header = ?buf, "received first bytes from WebTransport MTP stream" ); } @@ -145,11 +139,6 @@ impl TransportRecvStream for H3TransportReceiver { } } } - - fn stop(mut self, code: u32) -> Result<(), CommunicationError> { - h3::quic::RecvStream::stop_sending(&mut self.stream, code as u64); - Ok(()) - } } impl tokio::io::AsyncWrite for H3TransportSender { diff --git a/package.json b/package.json index 72e0b12..dd4e7a1 100644 --- a/package.json +++ b/package.json @@ -61,6 +61,7 @@ "pack": "pnpm run release:web", "release:web": "node create-web-release.mjs", "build:all": "nix run .#build-all", + "dup": "jscpd --pattern '**/*.{rs,ts}' --ignore 'target/**' --ignore 'wasm/pkg/**' --min-lines 8 --min-tokens 80 --threshold 4 --reporters console --no-tips .", "test:e2e": "tsc && node test/e2ee.mjs", "test:secrets": "tsc && node --test --test-isolation=none test/encrypted-secret.mjs", "test:wasm-init": "tsc && node --test test/wasm-init.mjs", diff --git a/src/sdk/client.ts b/src/sdk/client.ts index e0307cd..d3c31be 100644 --- a/src/sdk/client.ts +++ b/src/sdk/client.ts @@ -10,15 +10,6 @@ import * as bindings from "mtp/raw"; import { unixTimeMillis, utf8Encode } from "./utils.js"; import type * as RawBindings from "../raw/index"; import type { MTPCommunicationType } from "../type-map/index"; -import { MTPProtocol } from "./schema.js"; -import type { - MTPMessageType, - MTPFrame, - MTPNoSchemas, - MTPRequestData, - MTPResponseFrame, - MTPSchemaRegistry, -} from "./schema.js"; import type { MTPSessionStorage, MTPSessionState } from "./session"; import { MTPSessionManager } from "./session.js"; import { @@ -258,9 +249,7 @@ export interface MTPPublicKeyBundleKeys { sigClPublicKey: Uint8Array; } -export interface MTPClientOptions< - Registry extends MTPSchemaRegistry = MTPNoSchemas, -> { +export interface MTPClientOptions { url: string; descriptor?: string; hostPublicKey?: MTPKeyMaterialInput; @@ -293,12 +282,6 @@ export interface MTPClientOptions< securityProfile?: MTPSecurityProfile; /** One receive resource policy shared by frame and protected-value opening. */ receiveLimits?: MTPReceiveLimits; - /** Application request and response schemas, keyed by communication type. */ - schemas?: Registry; - /** Reject `request()` when the correlated response is an `Error*` frame. */ - throwProtocolErrors?: boolean; - /** Receives subscription validation failures. Request failures reject normally. */ - onValidationError?: (error: import("./schema.js").MTPValidationError) => void; } export interface MTPSecurityProfile { @@ -579,8 +562,8 @@ export interface MTPAcceptEncryptedPipeOptions { signaturePolicy?: MTPSignatureVerificationPolicy; } -type NormalizedMTPClientOptions = Omit< - MTPClientOptions, +type NormalizedMTPClientOptions = Omit< + MTPClientOptions, "hostPublicKey" | "receiveLimits" > & { hostPublicKey?: Uint8Array; @@ -931,34 +914,14 @@ function validateOptions(options) { ) { throw new TypeError("requestTimeoutMs must be a positive safe integer"); } - if (options.schemas != null) { - if (typeof options.schemas !== "object" || Array.isArray(options.schemas)) { - throw new TypeError("schemas must be an object"); - } - for (const [type, pair] of Object.entries(options.schemas)) { - if ( - !pair || - typeof pair !== "object" || - typeof (pair as { request?: { parseAsync?: unknown } }).request - ?.parseAsync !== "function" || - typeof (pair as { response?: { parseAsync?: unknown } }).response - ?.parseAsync !== "function" - ) { - throw new TypeError( - `schemas.${type} must contain request and response schemas with parseAsync()`, - ); - } - } - } } -export class MTPClient { +export class MTPClient { static readonly crypto = crypto; static readonly codec = codec; #credentials: InternalCredentials | null; - #options: NormalizedMTPClientOptions; - readonly #protocol: MTPProtocol | undefined; + #options: NormalizedMTPClientOptions; readonly #protectedReplayGuard = new InMemoryReplayGuard(); readonly #relayReplayGuard = new InMemoryReplayGuard(); readonly raw: MTPRaw; @@ -971,17 +934,10 @@ export class MTPClient { readonly encryptedSecretProvider: MTPEncryptedSecretProvider; private constructor( - options: NormalizedMTPClientOptions, + options: NormalizedMTPClientOptions, client: RawBindings.WasmClient, ) { this.#options = options; - this.#protocol = options.schemas - ? new MTPProtocol({ - schemas: options.schemas, - throwProtocolErrors: options.throwProtocolErrors, - onValidationError: options.onValidationError, - }) - : undefined; this.#credentials = deserializeCredentials(options.credentials); this.raw = { client, bindings }; this.encryptedSecretProvider = @@ -991,11 +947,7 @@ export class MTPClient { ); } - static async create< - const Registry extends MTPSchemaRegistry = MTPNoSchemas, - >( - options: MTPClientOptions, - ): Promise> { + static async create(options: MTPClientOptions): Promise { validateOptions(options); await MTPClient.init(options.wasm); @@ -1016,7 +968,7 @@ export class MTPClient { securityProfile: resolveSecurityProfile(options), }; - let sdk: MTPClient | undefined; + let sdk: MTPClient | undefined; const client = new WasmClient( (state) => emit(normalizedOptions.logger, { @@ -1052,7 +1004,7 @@ export class MTPClient { setReceiveLimits.call(rawClient, normalizedOptions.receiveLimits); } - sdk = new MTPClient(normalizedOptions, client); + sdk = new MTPClient(normalizedOptions, client); await sdk.#loadStoredCredentials(); if (!sdk.#credentials) { sdk.#credentials = { @@ -1286,35 +1238,6 @@ export class MTPClient { }; } - async #parseRequestData( - type: MTPCommunicationType, - data: unknown, - ): Promise> { - if (!this.#protocol || !this.#protocol.schemas[type]) { - return (data ?? {}) as Record; - } - const parsed = await this.#protocol.parseRequest( - type as MTPMessageType, - data as never, - ); - return (parsed ?? {}) as Record; - } - - async #parseResponseData( - requestedType: MTPCommunicationType, - frame: ParsedFrame, - phase: "response" | "subscription" = "response", - ): Promise> { - if (!this.#protocol || !this.#protocol.schemas[requestedType]) { - return frame; - } - return await this.#protocol.parseResponse( - requestedType as MTPMessageType, - frame, - phase, - ); - } - #buildFrame(typeOrFrame, data, options) { if (typeOrFrame instanceof Uint8Array) { if ( @@ -1356,11 +1279,6 @@ export class MTPClient { } async send(message: Uint8Array): Promise; - async send>( - type: Type, - data?: MTPRequestData, - options?: MTPSendOptions, - ): Promise; async send( type: MTPCommunicationType, data: Record, @@ -1368,14 +1286,10 @@ export class MTPClient { ): Promise; async send( typeOrFrame: Uint8Array | MTPCommunicationType, - data?: unknown, + data?: Record, options?: MTPSendOptions, ): Promise { - const parsedData = - typeof typeOrFrame === "string" - ? await this.#parseRequestData(typeOrFrame, data) - : data; - const message = this.#buildFrame(typeOrFrame, parsedData, options); + const message = this.#buildFrame(typeOrFrame, data, options); try { const frame = this.raw.bindings.parse_frame(message); @@ -1431,11 +1345,6 @@ export class MTPClient { data?: never, options?: MTPRequestOptions, ): Promise; - async request>( - type: Type, - data?: MTPRequestData, - options?: MTPRequestOptions, - ): Promise>; async request( type: MTPCommunicationType, data: Record, @@ -1443,19 +1352,15 @@ export class MTPClient { ): Promise; async request( typeOrFrame: Uint8Array | MTPCommunicationType, - data?: unknown, + data?: Record, options: MTPRequestOptions = {}, - ): Promise> { + ): Promise { const timeoutMs = options.timeoutMs ?? this.#options.requestTimeoutMs ?? 30_000; if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) { throw new TypeError("request timeoutMs must be a positive safe integer"); } - const parsedData = - typeof typeOrFrame === "string" - ? await this.#parseRequestData(typeOrFrame, data) - : data; - const frame = this.#buildFrame(typeOrFrame, parsedData, options); + const frame = this.#buildFrame(typeOrFrame, data, options); try { const parsed = this.raw.bindings.parse_frame(frame); emit( @@ -1486,29 +1391,16 @@ export class MTPClient { // The WASM client owns request expiry and its late-response tombstones. // Keeping a second Promise timer here can reject the SDK call while the // protocol request is still allowed to complete successfully. - const response = await this.raw.client.request( + return await this.raw.client.request( frame, options.responseType ?? null, timeoutMs, ); - return typeof typeOrFrame === "string" - ? await this.#parseResponseData(typeOrFrame, response) - : response; } - subscribe>( - type: Type, - handler: ( - message: MTPResponseFrame, - ) => void | Promise, - ): Unsubscribe; subscribe( type: MTPCommunicationType, - handler: (message: ParsedFrame) => void | Promise, - ): Unsubscribe; - subscribe( - type: MTPCommunicationType, - handler: (message: any) => void | Promise, + handler: (message: ParsedFrame) => void, ): Unsubscribe { if (typeof type !== "string" || !type) { throw new TypeError("subscription type must be a non-empty string"); @@ -1516,25 +1408,8 @@ export class MTPClient { if (typeof handler !== "function") { throw new TypeError("subscription handler must be a function"); } - let active = true; - const id = this.raw.client.subscribe(type, (message) => { - if (!this.#protocol || !this.#protocol.schemas[type]) { - void handler(message); - return; - } - void this.#parseResponseData(type, message, "subscription").then( - (parsed) => { - if (active) void handler(parsed); - }, - (error) => { - this.#protocol?.reportValidationError(error); - }, - ); - }); - return () => { - active = false; - this.raw.client.unsubscribe(id); - }; + const id = this.raw.client.subscribe(type, handler); + return () => this.raw.client.unsubscribe(id); } #handleFrame(frame) { diff --git a/src/sdk/index.ts b/src/sdk/index.ts index 5e3a1e9..e93ca7e 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -5,4 +5,3 @@ * keeps the package's historical exports stable. */ export * from "./client.js"; -export * from "./schema.js"; diff --git a/src/sdk/schema.ts b/src/sdk/schema.ts deleted file mode 100644 index 9b9aaf6..0000000 --- a/src/sdk/schema.ts +++ /dev/null @@ -1,234 +0,0 @@ -import type { MTPRequestOptions, ParsedFrame, Unsubscribe } from "./client.js"; -import type { MTPCommunicationType } from "../type-map/index.js"; - -export interface MTPSchema { - readonly _input: Input; - readonly _output: Output; - parseAsync(value: unknown): Promise; -} - -export interface MTPSchemaPair< - Request extends MTPSchema = MTPSchema, - Response extends MTPSchema = MTPSchema, -> { - request: Request; - response: Response; -} - -export type MTPSchemaRegistry = Record; -export type MTPNoSchemas = Record; - -export type MTPSchemaInput = Schema["_input"]; -export type MTPSchemaOutput = Schema["_output"]; -export type MTPMessageType = - keyof Registry & string; - -export type MTPFrame = { - id?: number; - type: string; - data: Data; - sender?: ParsedFrame["sender"]; - receiver?: ParsedFrame["receiver"]; - raw?: ParsedFrame["raw"]; -}; - -export type MTPTypedFrame = MTPFrame; - -export type MTPResponseFrame< - Registry extends MTPSchemaRegistry, - Type extends MTPMessageType, -> = MTPTypedFrame>; - -export type MTPRequestData< - Registry extends MTPSchemaRegistry, - Type extends MTPMessageType, -> = MTPSchemaInput; - -export type MTPRequestFunction = < - Type extends MTPMessageType, ->( - type: Type, - data?: MTPRequestData, - options?: MTPRequestOptions, -) => Promise>; - -export type MTPSubscriptionFunction = < - Type extends MTPMessageType, ->( - type: Type, - handler: (message: MTPResponseFrame) => void | Promise, -) => Unsubscribe; - -export class MTPValidationError extends Error { - readonly phase: "request" | "response" | "subscription"; - readonly messageType: string; - readonly frame?: MTPFrame; - - constructor( - phase: MTPValidationError["phase"], - messageType: string, - cause: unknown, - frame?: MTPFrame, - ) { - super(`${phase} validation failed for ${messageType}`, { cause }); - this.name = "MTPValidationError"; - this.phase = phase; - this.messageType = messageType; - this.frame = frame; - } -} - -export class MTPProtocolError extends Error { - readonly type: string; - readonly id: number | undefined; - readonly communicationType: string; - readonly requestId: number | undefined; - readonly errorType: string | undefined; - readonly frame: MTPFrame; - - constructor(frame: MTPFrame) { - const errorType = - frame.data && - typeof frame.data === "object" && - !Array.isArray(frame.data) && - typeof (frame.data as Record).ErrorType === "string" - ? ((frame.data as Record).ErrorType as string) - : undefined; - super(errorType ? `${frame.type}: ${errorType}` : frame.type); - this.name = "MTPProtocolError"; - this.type = frame.type; - this.id = frame.id; - this.communicationType = frame.type; - this.requestId = frame.id; - this.errorType = errorType; - this.frame = frame; - } -} - -export interface MTPProtocolOptions { - schemas: Registry; - throwProtocolErrors?: boolean; - onValidationError?: (error: MTPValidationError) => void; -} - -function isErrorFrame(frame: MTPFrame): boolean { - return frame.type.startsWith("Error"); -} - -export class MTPProtocol { - readonly schemas: Registry; - readonly #throwProtocolErrors: boolean; - readonly #onValidationError: - | ((error: MTPValidationError) => void) - | undefined; - - constructor(options: MTPProtocolOptions) { - this.schemas = options.schemas; - this.#throwProtocolErrors = options.throwProtocolErrors ?? false; - this.#onValidationError = options.onValidationError; - } - - async parseRequest>( - type: Type, - data: MTPRequestData | undefined, - ): Promise> { - try { - return await this.schemas[type].request.parseAsync(data); - } catch (error) { - throw new MTPValidationError("request", type, error); - } - } - - async parseResponse>( - requestedType: Type, - frame: MTPFrame, - phase: "response" | "subscription" = "response", - ): Promise> { - if (isErrorFrame(frame)) { - if (phase === "response" && this.#throwProtocolErrors) { - throw new MTPProtocolError(frame); - } - return frame as MTPResponseFrame; - } - - const schema = - this.schemas[frame.type]?.response ?? - this.schemas[requestedType].response; - try { - const data = await schema.parseAsync(frame.data); - return { ...frame, data } as MTPResponseFrame; - } catch (error) { - throw new MTPValidationError( - phase, - frame.type || requestedType, - error, - frame, - ); - } - } - - reportValidationError(error: unknown): void { - if (error instanceof MTPValidationError) { - this.#onValidationError?.(error); - } - } -} - -export interface MTPProxyAdapter { - request( - type: MTPCommunicationType, - data: Record, - options?: MTPRequestOptions, - ): Promise; - subscribe( - type: MTPCommunicationType, - handler: (message: MTPFrame) => void, - ): Unsubscribe; -} - -export class MTPProxyConnection { - readonly #adapter: MTPProxyAdapter; - readonly #protocol: MTPProtocol; - - constructor(adapter: MTPProxyAdapter, options: MTPProtocolOptions) { - this.#adapter = adapter; - this.#protocol = new MTPProtocol(options); - } - - async request>( - type: Type, - data?: MTPRequestData, - options?: MTPRequestOptions, - ): Promise> { - const parsed = await this.#protocol.parseRequest(type, data); - const response = await this.#adapter.request( - type, - (parsed ?? {}) as Record, - options, - ); - return await this.#protocol.parseResponse(type, response); - } - - subscribe>( - type: Type, - handler: ( - message: MTPResponseFrame, - ) => void | Promise, - ): Unsubscribe { - let active = true; - const unsubscribe = this.#adapter.subscribe(type, (message) => { - void this.#protocol.parseResponse(type, message, "subscription").then( - (parsed) => { - if (active) void handler(parsed); - }, - (error) => { - this.#protocol.reportValidationError(error); - }, - ); - }); - return () => { - active = false; - unsubscribe(); - }; - } -} diff --git a/transport/src/connection.rs b/transport/src/connection.rs index 7db09e8..dece604 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -4,10 +4,6 @@ use crate::framing::RetryClassifier; use crate::pipe::PipeReader; use mtp_codec::{CommunicationValue, DecodeError, DecodeLimits, EncodeLimits, TypeMap}; use mtp_common::CommunicationError; -#[cfg(feature = "pipes")] -use mtp_common::{FirstFrameDisposition, classify_first_frame}; -#[cfg(feature = "pipes")] -use std::collections::HashSet; use std::ops::Deref; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -292,15 +288,15 @@ impl Sender { Ok(Ok(())) => Ok(()), Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { warn!("[Sender] write failed: peer sent STOP_SENDING (error code {code})"); - Err(CommunicationError::DeliveryUnknown) + Err(CommunicationError::StreamClosed) } Ok(Err(other)) => { warn!("[Sender] write failed: {other}"); - Err(CommunicationError::DeliveryUnknown) + Err(CommunicationError::StreamError) } Err(_) => { warn!("[Sender] write timed out (len={})", bytes.len()); - Err(CommunicationError::DeliveryUnknown) + Err(CommunicationError::StreamError) } } } @@ -402,15 +398,15 @@ impl Sender { Ok(Ok(())) => Ok(()), Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { warn!("[Sender] finish failed: peer sent STOP_SENDING (error code {code})"); - Err(CommunicationError::DeliveryUnknown) + Err(CommunicationError::StreamClosed) } Ok(Err(other)) => { warn!("[Sender] finish failed: {other}"); - Err(CommunicationError::DeliveryUnknown) + Err(CommunicationError::StreamError) } Err(_) => { warn!("[Sender] finish timed out"); - Err(CommunicationError::DeliveryUnknown) + Err(CommunicationError::StreamError) } } } @@ -749,8 +745,6 @@ struct ReceiverInner { max_message_size: Arc, type_map: Arc>, decode_rejections: Arc, - #[cfg(feature = "pipes")] - expected_pipes: Arc>>, } impl Clone for Receiver { @@ -837,10 +831,6 @@ impl Receiver { let accept_type_map = type_map.clone(); let decode_rejections = Arc::new(DecodeRejectionCounters::default()); let accept_decode_rejections = decode_rejections.clone(); - #[cfg(feature = "pipes")] - let expected_pipes = Arc::new(std::sync::Mutex::new(HashSet::new())); - #[cfg(feature = "pipes")] - let accept_expected_pipes = expected_pipes.clone(); let stream_limit = Arc::new(Semaphore::new(policy.max_concurrent_stream_tasks.max(1))); let accept_stream_limit = stream_limit.clone(); debug!( @@ -910,8 +900,6 @@ impl Receiver { let stream_max_message_size = accept_max_message_size.clone(); let stream_type_map = accept_type_map.clone(); let stream_decode_rejections = accept_decode_rejections.clone(); - #[cfg(feature = "pipes")] - let stream_expected_pipes = accept_expected_pipes.clone(); tokio::spawn(async move { let _permit = permit; @@ -947,54 +935,38 @@ impl Receiver { #[cfg(feature = "pipes")] { - if frame_count == 1 { - let is_pipe_request = msg.is_type( - mtp_codec::CommunicationType::PipeRequest, - ); - let pipe_id = msg.id().filter(|id| *id != 0); - let pipe_is_expected = is_pipe_request && pipe_id.is_some_and(|pipe_id| { - stream_expected_pipes - .lock() - .is_ok_and(|mut expected| expected.remove(&pipe_id)) - }); - let disposition = match classify_first_frame( - is_pipe_request, - msg.id(), - pipe_is_expected, - ) { - Ok(disposition) => disposition, - Err(error) => { - let _ = msg_tx_stream - .send(Err(error.clone())) - .await; - stream_handle.close(Some(error)); - break; - } + if msg.is_type(mtp_codec::CommunicationType::PipeRequest) + && frame_count == 1 + { + let Some(pipe_id) = msg.id().filter(|id| *id != 0) else { + let error = CommunicationError::Other( + "PipeRequest frame must contain a non-zero id".into(), + ); + let _ = msg_tx_stream.send(Err(error.clone())).await; + stream_handle.close(Some(error)); + break; + }; + let description = msg + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); + + let pipe_reader = crate::pipe::PipeReader { + stream: s, + description, + pipe_id, }; - if let FirstFrameDisposition::Pipe(pipe_id) = disposition { - let description = msg - .get_str(mtp_codec::DataType::Description) - .unwrap_or("") - .to_string(); - - let pipe_reader = crate::pipe::PipeReader { - stream: s, - description, - pipe_id, - }; - - if pipe_tx_stream - .send(pipe_reader) - .await - .is_err() - { - stream_handle.close(Some( - CommunicationError::StreamClosed, - )); - } - break; + if pipe_tx_stream + .send(pipe_reader) + .await + .is_err() + { + stream_handle.close(Some( + CommunicationError::StreamClosed, + )); } + break; } } @@ -1139,8 +1111,6 @@ impl Receiver { max_message_size, type_map, decode_rejections, - #[cfg(feature = "pipes")] - expected_pipes, }), } } @@ -1157,26 +1127,6 @@ impl Receiver { *self.inner.type_map.write().await = type_map.clone(); } - #[cfg(feature = "pipes")] - pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { - if pipe_id == 0 { - return Err(CommunicationError::Other("pipe id must be non-zero".into())); - } - self.inner - .expected_pipes - .lock() - .map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))? - .insert(pipe_id); - Ok(()) - } - - #[cfg(feature = "pipes")] - pub fn cancel_expected_pipe(&self, pipe_id: u32) { - if let Ok(mut expected) = self.inner.expected_pipes.lock() { - expected.remove(&pipe_id); - } - } - /// Return local counts for frames rejected by the structured decoder. /// /// These counters are intentionally local-only; peers continue to receive diff --git a/transport/src/framing.rs b/transport/src/framing.rs index ac4e005..9fa1d80 100644 --- a/transport/src/framing.rs +++ b/transport/src/framing.rs @@ -82,10 +82,6 @@ mod tests { async fn finish(&mut self) -> Result<(), CommunicationError> { Ok(()) } - - fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> { - Ok(()) - } } #[tokio::test] diff --git a/transport/src/generic_connection.rs b/transport/src/generic_connection.rs index 4361855..4fa6433 100644 --- a/transport/src/generic_connection.rs +++ b/transport/src/generic_connection.rs @@ -9,13 +9,9 @@ use crate::{ connection::{DecodeRejectionCounters, RuntimePolicy, classify_decode_error}, framing::{RetryClassifier, write_frame}, }; -use mtp_codec::{CommunicationValue, DataType, DecodeLimits, TypeMap}; -use mtp_common::{CommunicationError, FirstFrameDisposition, classify_first_frame}; -#[cfg(feature = "pipes")] -use std::collections::HashSet; +use mtp_codec::{CommunicationValue, DecodeLimits, TypeMap}; +use mtp_common::CommunicationError; use std::sync::Arc; -#[cfg(feature = "pipes")] -use std::sync::Mutex as StdMutex; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::sync::{Mutex, Notify, RwLock, Semaphore, mpsc}; use tokio::time::{Instant, timeout, timeout_at}; @@ -71,15 +67,6 @@ impl GenericSender { if self.connection.close_reason().is_some() { return Err(CommunicationError::StreamClosed); } - if let Some(version) = value.get_str(DataType::Version) { - tracing::debug!( - message_type = ?value.get_type(), - version, - connected = ?value.get_data(DataType::Connected), - client_id = ?value.get_data(DataType::Id), - "sending MTP handshake response frame" - ); - } match self.policy.send_mode { crate::SendMode::SingleStreamPerMessage => { let mut stream = self.open().await?; @@ -89,10 +76,9 @@ impl GenericSender { ) .await .map_err(|_| CommunicationError::StreamError)??; - match timeout(self.policy.write_timeout, stream.finish()).await { - Ok(Ok(())) => Ok(()), - Ok(Err(_)) | Err(_) => Err(CommunicationError::DeliveryUnknown), - } + timeout(self.policy.write_timeout, stream.finish()) + .await + .map_err(|_| CommunicationError::StreamError)? } crate::SendMode::PersistentStream => { let mut stream = self.persistent.lock().await; @@ -209,8 +195,6 @@ pub struct GenericReceiver { type_map: Arc>, queue_notify: Arc, decode_rejections: Arc, - #[cfg(feature = "pipes")] - expected_pipes: Arc>>, _accept_task: Arc>, } @@ -226,8 +210,6 @@ impl Clone for GenericReceiver { type_map: self.type_map.clone(), queue_notify: self.queue_notify.clone(), decode_rejections: self.decode_rejections.clone(), - #[cfg(feature = "pipes")] - expected_pipes: self.expected_pipes.clone(), _accept_task: self._accept_task.clone(), } } @@ -263,10 +245,6 @@ impl GenericReceiver { let task_queue_notify = queue_notify.clone(); let decode_rejections = Arc::new(DecodeRejectionCounters::default()); let task_decode_rejections = decode_rejections.clone(); - #[cfg(feature = "pipes")] - let expected_pipes = Arc::new(StdMutex::new(HashSet::new())); - #[cfg(feature = "pipes")] - let task_expected_pipes = expected_pipes.clone(); let task_accept_task_tx = tx.clone(); #[cfg(feature = "pipes")] let task_accept_task_pipe_tx = pipe_tx.clone(); @@ -325,8 +303,6 @@ impl GenericReceiver { let connection = task_connection.clone(); let type_map = task_type_map.clone(); let decode_rejections = task_decode_rejections.clone(); - #[cfg(feature = "pipes")] - let expected_pipes = task_expected_pipes.clone(); tokio::spawn(async move { let _permit = permit; let mut stream = stream; @@ -447,62 +423,42 @@ impl GenericReceiver { break; } }; - tracing::debug!( - frames, - frame_len, - message_type = ?message.get_type(), - "decoded MTP receive frame" - ); let negotiated_type_map = type_map.read().await.clone(); message.set_type_map(&negotiated_type_map); #[cfg(feature = "pipes")] { - if frames == 1 { - let is_pipe_request = - message.is_type(mtp_codec::CommunicationType::PipeRequest); - let pipe_id = message.id().filter(|id| *id != 0); - let pipe_is_expected = is_pipe_request - && pipe_id.is_some_and(|pipe_id| { - expected_pipes - .lock() - .is_ok_and(|mut expected| expected.remove(&pipe_id)) - }); - let disposition = match classify_first_frame( - is_pipe_request, - message.id(), - pipe_is_expected, - ) { - Ok(disposition) => disposition, - Err(error) => { - let _ = tx.send(Err(error.clone())).await; - connection.close( - policy.application_close_code, - b"pipe request missing id", - ); - break; - } + if message.is_type(mtp_codec::CommunicationType::PipeRequest) + && frames == 1 + { + let Some(pipe_id) = message.id().filter(|id| *id != 0) else { + let error = CommunicationError::Other( + "PipeRequest frame must contain a non-zero id".into(), + ); + let _ = tx.send(Err(error.clone())).await; + connection.close( + policy.application_close_code, + b"pipe request missing id", + ); + break; + }; + let description = message + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); + + let pipe_reader = PipeReader { + stream, + description, + pipe_id, }; - if let FirstFrameDisposition::Pipe(pipe_id) = disposition { - let description = message - .get_str(mtp_codec::DataType::Description) - .unwrap_or("") - .to_string(); + tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream"); - let pipe_reader = PipeReader { - stream, - description, - pipe_id, - }; - - tracing::debug!(pipe_id, description = %pipe_reader.description, "classified incoming pipe stream"); - - if pipe_tx.send(pipe_reader).await.is_err() { - break; - } - return; + if pipe_tx.send(pipe_reader).await.is_err() { + break; } + return; } } @@ -546,8 +502,6 @@ impl GenericReceiver { type_map, queue_notify, decode_rejections, - #[cfg(feature = "pipes")] - expected_pipes, _accept_task: Arc::new(accept_task), } } @@ -555,25 +509,6 @@ impl GenericReceiver { *self.ping_sender.write().await = Some(sender); } - #[cfg(feature = "pipes")] - pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> { - if pipe_id == 0 { - return Err(CommunicationError::Other("pipe id must be non-zero".into())); - } - self.expected_pipes - .lock() - .map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))? - .insert(pipe_id); - Ok(()) - } - - #[cfg(feature = "pipes")] - pub fn cancel_expected_pipe(&self, pipe_id: u32) { - if let Ok(mut expected) = self.expected_pipes.lock() { - expected.remove(&pipe_id); - } - } - /// Switch from the handshake frame limit to the application frame limit. pub fn set_max_message_size(&self, max_message_size: u64) { self.max_message_size diff --git a/transport/src/transport_traits.rs b/transport/src/transport_traits.rs index 70395af..63c0268 100644 --- a/transport/src/transport_traits.rs +++ b/transport/src/transport_traits.rs @@ -17,7 +17,6 @@ use mtp_common::CommunicationError; pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync { async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError>; async fn finish(&mut self) -> Result<(), CommunicationError>; - fn reset(&mut self, code: u32) -> Result<(), CommunicationError>; } /// A readable unidirectional stream suitable for MTP frames. @@ -29,9 +28,6 @@ pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync { pub trait TransportRecvStream: tokio::io::AsyncRead + Send + Sync { async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError>; async fn read_chunk(&mut self, max: usize) -> Result>, CommunicationError>; - fn stop(self, code: u32) -> Result<(), CommunicationError> - where - Self: Sized; } /// A QUIC/WebTransport connection that provides MTP's unidirectional streams. @@ -51,7 +47,7 @@ impl TransportSendStream for wtransport::SendStream { async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError> { wtransport::SendStream::write_all(self, buf) .await - .map_err(|_| CommunicationError::DeliveryUnknown) + .map_err(|_| CommunicationError::StreamError) } async fn finish(&mut self) -> Result<(), CommunicationError> { @@ -59,23 +55,14 @@ impl TransportSendStream for wtransport::SendStream { .await .map_err(|_| CommunicationError::StreamError) } - - fn reset(&mut self, code: u32) -> Result<(), CommunicationError> { - wtransport::SendStream::reset(self, wtransport::VarInt::from_u32(code)) - .map_err(|_| CommunicationError::StreamClosed) - } } #[async_trait] impl TransportRecvStream for wtransport::RecvStream { async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> { - match wtransport::RecvStream::read_exact(self, buf).await { - Ok(()) => Ok(()), - Err(wtransport::error::StreamReadExactError::FinishedEarly(0)) => { - Err(CommunicationError::StreamClosed) - } - Err(_) => Err(CommunicationError::StreamError), - } + wtransport::RecvStream::read_exact(self, buf) + .await + .map_err(|_| CommunicationError::StreamError) } async fn read_chunk(&mut self, max: usize) -> Result>, CommunicationError> { @@ -89,11 +76,6 @@ impl TransportRecvStream for wtransport::RecvStream { Err(_) => Err(CommunicationError::StreamError), } } - - fn stop(self, code: u32) -> Result<(), CommunicationError> { - wtransport::RecvStream::stop(self, wtransport::VarInt::from_u32(code)); - Ok(()) - } } #[async_trait] diff --git a/transport/tests/generic_pipe.rs b/transport/tests/generic_pipe.rs index 7dfb538..81a875f 100644 --- a/transport/tests/generic_pipe.rs +++ b/transport/tests/generic_pipe.rs @@ -4,7 +4,7 @@ use async_trait::async_trait; use mtp_codec::CommunicationValue; use mtp_common::CommunicationError; use mtp_transport::{ - GenericReceiver, GenericSender, Policy, SendMode, TransportConnection, TransportEvent, + GenericReceiver, GenericSender, Policy, TransportConnection, TransportEvent, TransportRecvStream, TransportSendStream, }; use std::sync::Arc; @@ -53,10 +53,6 @@ impl TransportSendStream for MockSendStream { .await .map_err(|_| CommunicationError::StreamError) } - - fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> { - Ok(()) - } } struct MockRecvStream { @@ -93,10 +89,6 @@ impl TransportRecvStream for MockRecvStream { Err(_) => Err(CommunicationError::StreamError), } } - - fn stop(self, _code: u32) -> Result<(), CommunicationError> { - Ok(()) - } } #[derive(Clone)] @@ -165,7 +157,6 @@ async fn test_open_pipe_and_receive_reader() -> Result<(), Box Result<(), Box let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); - receiver.expect_pipe(1)?; let mut pipe_writer = sender.open_pipe(1, "data-pipe").await?; let data = b"hello through the pipe"; @@ -205,7 +195,6 @@ async fn test_pipe_large_payload() -> Result<(), Box> { let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); - receiver.expect_pipe(7)?; let mut pipe_writer = sender.open_pipe(7, "big-pipe").await?; let data: Vec = (0..256 * 1024).map(|i| (i % 256) as u8).collect(); @@ -234,7 +223,6 @@ async fn test_receive_event_dispatches_pipe() -> Result<(), Box Result<(), Box Result<(), Box> { let (conn_a, conn_b) = mock_connected_pair().await; - let policy = Arc::new(Policy::default().with_send_mode(SendMode::SingleStreamPerMessage)); + let policy = Arc::new(Policy::default()); let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); - let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest) - .with_id(1) - .add_typed_default( - mtp_codec::DataType::Description, - mtp_codec::DataValue::Str("mixed-pipe".into()), - ); - sender.send(&request).await?; + let msg = CommunicationValue::new(mtp_codec::CommunicationType::BadRequest); + sender.send(&msg).await?; + + let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?; let received = receiver.receive().await?; - assert!(received.is_type(mtp_codec::CommunicationType::PipeRequest)); - - receiver.expect_pipe(1)?; - let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?; + assert_eq!( + received.get_type(), + mtp_codec::CommunicationType::BadRequest + .try_to_id(&mtp_codec::TypeMap::latest()) + .unwrap() + ); let pipe_reader = receiver.receive_pipe().await?; assert_eq!(pipe_reader.pipe_id(), 1); @@ -304,8 +291,6 @@ async fn test_multiple_pipes() -> Result<(), Box> { let sender = GenericSender::new(conn_a, policy.clone()); let receiver = GenericReceiver::new(conn_b, policy); - receiver.expect_pipe(10)?; - receiver.expect_pipe(20)?; let mut pw1 = sender.open_pipe(10, "first").await?; let mut pw2 = sender.open_pipe(20, "second").await?; diff --git a/type-map/Cargo.toml b/type-map/Cargo.toml index 1cacaa0..ad811ae 100644 --- a/type-map/Cargo.toml +++ b/type-map/Cargo.toml @@ -16,7 +16,3 @@ pipes = [] [build-dependencies] serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" - -[package.metadata.cargo-machete] -# cargo-machete does not inspect build.rs, where both build dependencies are used. -ignored = ["serde", "serde_yaml"] diff --git a/wasm/src/client/authentication.rs b/wasm/src/client/authentication.rs index 73bfc9f..df7989e 100644 --- a/wasm/src/client/authentication.rs +++ b/wasm/src/client/authentication.rs @@ -99,19 +99,10 @@ impl WasmClient { return Err(js_error(message)); } - let missing_version = || { - js_error(format!( - "host omitted a valid negotiated protocol version (response_type={:?}, connected={:?}, frame_len={})", - outcome.get_type(), - outcome.get_data(DataType::Connected), - outcome_bytes.len(), - )) - }; - let negotiated_version = match outcome.get_data(DataType::Version) { Some(DataValue::Str(version)) => mtp_codec::Version::parse(version) - .ok_or_else(|| missing_version())?, - _ => return Err(missing_version()), + .ok_or_else(|| js_error("host omitted a valid negotiated protocol version"))?, + _ => return Err(js_error("host omitted a valid negotiated protocol version")), }; if negotiated_version != PROTOCOL_VERSION { return Err(js_error( diff --git a/wasm/src/client/receive.rs b/wasm/src/client/receive.rs index 51a48bf..35ceb02 100644 --- a/wasm/src/client/receive.rs +++ b/wasm/src/client/receive.rs @@ -179,7 +179,6 @@ impl WasmClient { let expired_pipe_creations = self.expired_pipe_creations.clone(); let pending_pipes = self.pending_pipes.clone(); let loop_pending_pipes = pending_pipes.clone(); - let expected_pending_pipes = pending_pipes.clone(); let on_pipe_request = self.on_pipe_request.clone(); let loop_pipe_creations = pending_pipe_creations.clone(); let loop_expired_pipe_creations = expired_pipe_creations.clone(); @@ -294,12 +293,6 @@ impl WasmClient { let _ = entry.sender.send(Ok(pipe_reader)); } }, - move |pipe_id| { - expected_pending_pipes - .borrow() - .get(&pipe_id) - .is_some_and(|entry| entry.generation == loop_generation) - }, ) .await; if connection_generation.get() != generation { diff --git a/wasm/src/pipe.rs b/wasm/src/pipe.rs index 88783ae..efa4f07 100644 --- a/wasm/src/pipe.rs +++ b/wasm/src/pipe.rs @@ -1,6 +1,9 @@ +use wasm_bindgen::JsCast; use wasm_bindgen::prelude::*; +use wasm_bindgen_futures::JsFuture; -use crate::transport::{BrowserRecvStream, BrowserSendStream, log_stream_error_code}; +use crate::error::js_error; +use crate::transport::release_writer_lock; #[wasm_bindgen(typescript_custom_section)] const PIPE_TS: &str = r#" @@ -20,41 +23,54 @@ export interface PipeReader { #[wasm_bindgen] pub struct PipeWriter { - stream: BrowserSendStream, + writer: JsValue, pipe_id: u32, } impl PipeWriter { - pub(crate) fn new(stream: BrowserSendStream, pipe_id: u32) -> Self { - Self { stream, pipe_id } - } -} - -impl Drop for PipeWriter { - fn drop(&mut self) { - self.stream.release(); + pub fn new(writer: JsValue, pipe_id: u32) -> Self { + Self { writer, pipe_id } } } #[wasm_bindgen] impl PipeWriter { pub async fn write(&mut self, data: &[u8]) -> Result<(), JsValue> { - self.stream.write_all(data).await + let chunk = js_sys::Uint8Array::from(data); + let write_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write")) + .map_err(|_| js_error("missing write"))? + .dyn_into::() + .map_err(|_| js_error("write not a function"))?; + let write_promise = write_fn + .call1(&self.writer, &chunk) + .map_err(|e| js_error(format!("write failed: {:?}", e)))?; + JsFuture::from(write_promise.unchecked_into::()).await?; + Ok(()) } - pub async fn close(mut self) -> Result<(), JsValue> { - let result = self.stream.finish().await; - if let Err(error) = &result { - log_stream_error_code(error, "pipe writer close"); + pub async fn close(self) -> Result<(), JsValue> { + let close_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close")) + .map_err(|_| js_error("missing close"))? + .dyn_into::() + .map_err(|_| js_error("close not a function"))?; + let close_promise = close_fn + .call0(&self.writer) + .map_err(|e| js_error(format!("close failed: {:?}", e)))?; + if let Err(e) = JsFuture::from(close_promise.unchecked_into::()).await { + crate::transport::log_stream_error_code(&e, "pipe writer close"); } - self.stream.release(); - result + release_writer_lock(&self.writer); + Ok(()) } pub fn abort(&mut self) -> Result<(), JsValue> { - let result = self.stream.reset(0); - self.stream.release(); - result + let abort_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort")) + .map_err(|_| js_error("missing abort"))? + .dyn_into::() + .map_err(|_| js_error("abort not a function"))?; + let _ = abort_fn.call0(&self.writer); + release_writer_lock(&self.writer); + Ok(()) } pub fn pipe_id(&self) -> u32 { @@ -64,26 +80,19 @@ impl PipeWriter { #[wasm_bindgen] pub struct PipeReader { - stream: BrowserRecvStream, + reader: JsValue, description: String, pipe_id: u32, pending: Vec, - finished: bool, } impl PipeReader { - pub(crate) fn new( - stream: BrowserRecvStream, - pipe_id: u32, - description: String, - pending: Vec, - ) -> Self { + pub fn new(reader: JsValue, pipe_id: u32, description: String, pending: Vec) -> Self { Self { - stream, + reader, pipe_id, description, pending, - finished: false, } } } @@ -96,18 +105,27 @@ impl PipeReader { return Ok(js_sys::Uint8Array::from(&data[..]).into()); } - if self.finished { + let read_fn = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read")) + .map_err(|_| js_error("missing read"))? + .dyn_into::() + .map_err(|_| js_error("read not a function"))?; + let promise = read_fn + .call0(&self.reader) + .map_err(|_| js_error("read call failed"))? + .unchecked_into::(); + let result = JsFuture::from(promise).await?; + + let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) + .ok() + .and_then(|v| v.as_bool()) + .unwrap_or(true); + if done { return Ok(JsValue::NULL); } - match self.stream.read_chunk().await? { - Some(value) => Ok(js_sys::Uint8Array::from(&value[..]).into()), - None => { - self.stream.release(); - self.finished = true; - Ok(JsValue::NULL) - } - } + let value = js_sys::Reflect::get(&result, &JsValue::from_str("value")) + .map_err(|_| js_error("missing value"))?; + Ok(js_sys::Uint8Array::new(&value).into()) } pub fn pipe_id(&self) -> u32 { @@ -118,9 +136,3 @@ impl PipeReader { self.description.clone() } } - -impl Drop for PipeReader { - fn drop(&mut self) { - self.stream.release(); - } -} diff --git a/wasm/src/transport.rs b/wasm/src/transport.rs index b7ffebe..a4deea6 100644 --- a/wasm/src/transport.rs +++ b/wasm/src/transport.rs @@ -9,7 +9,6 @@ use wasm_bindgen_futures::JsFuture; use crate::error::js_error; use crate::frame::parse_frame_value_with_limits; use mtp_codec::{DecodeLimits, EncodeLimits, TypeMap}; -use mtp_common::{FirstFrameDisposition, classify_first_frame}; const CLOSE_FRAME_LEN: u32 = u32::MAX; @@ -26,6 +25,12 @@ pub(crate) fn log_stream_error_code(error: &JsValue, context: &str) { let stream_error_code = js_sys::Reflect::get(error, &JsValue::from_str("streamErrorCode")) .ok() .and_then(|v| v.as_f64()); + if matches!(stream_error_code, Some(0.0)) { + // WebTransport reports peer-driven stream shutdown as code 0 in this + // environment. For one-frame handshake streams, that is expected and + // should not be surfaced as a warning. + return; + } let message = error .as_string() .or_else(|| { @@ -71,233 +76,6 @@ fn resolve_stream_readable(recv_stream: &JsValue) -> Result { } } -#[derive(Clone)] -struct BrowserConnection { - inner: JsValue, - incoming_reader: Rc>>, -} - -pub(crate) struct BrowserSendStream { - writer: JsValue, -} - -pub(crate) struct BrowserRecvStream { - reader: JsValue, -} - -impl BrowserConnection { - async fn connect(url: &str, cert_hashes: Option>) -> Result { - let constructor = - js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))? - .dyn_into::() - .map_err(|_| js_error("WebTransport not available"))?; - let args = js_sys::Array::new(); - args.push(&JsValue::from_str(url)); - - if let Some(hashes) = cert_hashes { - let webtransport_hashes = js_sys::Array::new(); - for hash in hashes { - let (algorithm, value) = hash.split_once(':').unwrap_or(("sha-256", hash.as_str())); - if let Ok(value) = hex::decode(value) { - let entry = js_sys::Object::new(); - js_sys::Reflect::set( - &entry, - &JsValue::from_str("algorithm"), - &JsValue::from_str(algorithm), - )?; - js_sys::Reflect::set( - &entry, - &JsValue::from_str("value"), - &js_sys::Uint8Array::from(&value[..]), - )?; - webtransport_hashes.push(&entry); - } - } - if webtransport_hashes.length() > 0 { - let options = js_sys::Object::new(); - js_sys::Reflect::set( - &options, - &JsValue::from_str("serverCertificateHashes"), - &webtransport_hashes, - )?; - args.push(&options); - } - } - - let inner = js_sys::Reflect::construct(&constructor, &args)?; - let ready = js_sys::Reflect::get(&inner, &JsValue::from_str("ready"))? - .dyn_into::() - .map_err(|_| js_error("WebTransport.ready is not a Promise"))?; - JsFuture::from(ready) - .await - .map_err(|error| js_error(format!("WebTransport ready failed: {error:?}")))?; - Ok(Self { - inner, - incoming_reader: Rc::new(RefCell::new(None)), - }) - } - - async fn open_uni(&self) -> Result { - let create_stream = js_sys::Reflect::get( - &self.inner, - &JsValue::from_str("createUnidirectionalStream"), - )? - .dyn_into::() - .map_err(|_| js_error("createUnidirectionalStream not a function"))?; - let stream_promise = create_stream - .call0(&self.inner)? - .dyn_into::() - .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?; - let stream = JsFuture::from(stream_promise).await?; - let writable = resolve_stream_writable(&stream)?; - let writer = js_sys::Reflect::get(&writable, &JsValue::from_str("getWriter")) - .map_err(|_| js_error("missing getWriter"))? - .dyn_into::() - .map_err(|_| js_error("getWriter not a function"))? - .call0(&writable) - .map_err(|_| js_error("getWriter call failed"))?; - Ok(BrowserSendStream { writer }) - } - - async fn accept_uni(&self) -> Result, JsValue> { - let streams_reader = if let Some(reader) = self.incoming_reader.borrow().clone() { - reader - } else { - let incoming = js_sys::Reflect::get( - &self.inner, - &JsValue::from_str("incomingUnidirectionalStreams"), - )?; - let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader")) - .map_err(|_| js_error("missing getReader"))? - .dyn_into::() - .map_err(|_| js_error("getReader not a function"))? - .call0(&incoming) - .map_err(|_| js_error("getReader call failed"))?; - *self.incoming_reader.borrow_mut() = Some(reader.clone()); - reader - }; - - let read = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read")) - .map_err(|_| js_error("missing read"))? - .dyn_into::() - .map_err(|_| js_error("read not a function"))?; - let promise = read - .call0(&streams_reader) - .map_err(|_| js_error("read call failed"))? - .unchecked_into::(); - let result = JsFuture::from(promise).await.map_err(|error| { - log_stream_error_code(&error, "accept_uni"); - js_error(format!("accept stream failed: {error:?}")) - })?; - if js_sys::Reflect::get(&result, &JsValue::from_str("done")) - .ok() - .and_then(|value| value.as_bool()) - .unwrap_or(false) - { - return Ok(None); - } - - let stream = js_sys::Reflect::get(&result, &JsValue::from_str("value")) - .map_err(|_| js_error("missing value"))?; - let readable = resolve_stream_readable(&stream)?; - let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader")) - .map_err(|_| js_error("missing stream getReader"))? - .dyn_into::() - .map_err(|_| js_error("stream getReader not a function"))? - .call0(&readable) - .map_err(|_| js_error("stream getReader call failed"))?; - Ok(Some(BrowserRecvStream { reader })) - } - - fn close(&self) { - if let Some(reader) = self.incoming_reader.borrow_mut().take() { - release_reader_lock(&reader); - } - if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close")) - .and_then(|value| value.dyn_into::()) - { - let _ = close.call1(&self.inner, &js_sys::Object::new()); - } - } -} - -impl BrowserSendStream { - pub(crate) async fn write_all(&mut self, bytes: &[u8]) -> Result<(), JsValue> { - let write = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write")) - .map_err(|_| js_error("missing write"))? - .dyn_into::() - .map_err(|_| js_error("write not a function"))?; - let promise = write - .call1(&self.writer, &js_sys::Uint8Array::from(bytes)) - .map_err(|error| js_error(format!("write failed: {error:?}")))? - .unchecked_into::(); - JsFuture::from(promise).await.map(|_| ()) - } - - pub(crate) async fn finish(&mut self) -> Result<(), JsValue> { - let close = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close")) - .map_err(|_| js_error("missing close"))? - .dyn_into::() - .map_err(|_| js_error("close not a function"))?; - let promise = close - .call0(&self.writer) - .map_err(|error| js_error(format!("close failed: {error:?}")))? - .unchecked_into::(); - JsFuture::from(promise).await.map(|_| ()) - } - - pub(crate) fn reset(&mut self, code: u32) -> Result<(), JsValue> { - let abort = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort")) - .map_err(|_| js_error("missing abort"))? - .dyn_into::() - .map_err(|_| js_error("abort not a function"))?; - let _ = abort.call1(&self.writer, &JsValue::from_f64(code as f64))?; - Ok(()) - } - - pub(crate) fn release(&self) { - release_writer_lock(&self.writer); - } -} - -impl BrowserRecvStream { - pub(crate) async fn read_chunk(&mut self) -> Result>, JsValue> { - let read = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read")) - .map_err(|_| js_error("missing read"))? - .dyn_into::() - .map_err(|_| js_error("read not a function"))?; - let promise = read - .call0(&self.reader) - .map_err(|_| js_error("read call failed"))? - .unchecked_into::(); - let result = JsFuture::from(promise).await?; - if js_sys::Reflect::get(&result, &JsValue::from_str("done")) - .ok() - .and_then(|value| value.as_bool()) - .unwrap_or(true) - { - return Ok(None); - } - let value = js_sys::Reflect::get(&result, &JsValue::from_str("value")) - .map_err(|_| js_error("missing value"))?; - Ok(Some(js_sys::Uint8Array::new(&value).to_vec())) - } - - #[allow(dead_code)] - pub(crate) fn stop(self, code: u32) -> Result<(), JsValue> { - let cancel = js_sys::Reflect::get(&self.reader, &JsValue::from_str("cancel")) - .map_err(|_| js_error("missing cancel"))? - .dyn_into::() - .map_err(|_| js_error("cancel not a function"))?; - let _ = cancel.call1(&self.reader, &JsValue::from_f64(code as f64))?; - Ok(()) - } - - pub(crate) fn release(&self) { - release_reader_lock(&self.reader); - } -} - /// Releases a writer's lock so an abandoned writer isn't treated as an abort (which sends STOP_SENDING). pub(crate) fn release_writer_lock(writer: &JsValue) { if let Ok(release) = js_sys::Reflect::get(writer, &JsValue::from_str("releaseLock")) @@ -340,14 +118,18 @@ enum FrameOutcome { */ #[derive(Clone)] pub struct WasmTransport { - connection: BrowserConnection, + inner: JsValue, max_message_size: u32, - /// Current incoming unidirectional stream, shared across handshake and receive loops. - stream_reader: Rc>>, + /// Reader over `incoming_unidirectional_streams()` (a singleton stream of streams). + streams_reader: Rc>>, + /// Reader over the host's current uni-directional stream, if one is open. + stream_reader: Rc>>, /// Bytes already read from the current stream but not yet consumed as a frame. buffer: Rc>>, /// Set to `true` when `open_next_stream` succeeds; cleared after the first frame is parsed. new_stream_frame: Rc>, + /// A single ordered browser send stream shared by all cloned transports. + outgoing_writer: Rc>>, /// Serializes stream creation and writes across concurrent callers. send_lock: Rc>, type_map: Rc>, @@ -369,17 +151,65 @@ impl WasmTransport { max_message_size: u32, configured_limits: Option, ) -> Result { - let connection = BrowserConnection::connect(url, cert_hashes).await?; + let ctor = js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))? + .dyn_into::() + .map_err(|_| js_error("WebTransport not available"))?; + let args = js_sys::Array::new(); + args.push(&JsValue::from_str(url)); + + if let Some(hashes) = cert_hashes { + let wt_hashes = js_sys::Array::new(); + for h in hashes { + let (algo, hex_val) = match h.split_once(':') { + Some((algo, hex_val)) => (algo, hex_val), + None => ("sha-256", h.as_str()), + }; + + if let Ok(bytes) = hex::decode(hex_val) { + let hash = js_sys::Object::new(); + js_sys::Reflect::set( + &hash, + &JsValue::from_str("algorithm"), + &JsValue::from_str(algo), + )?; + js_sys::Reflect::set( + &hash, + &JsValue::from_str("value"), + &js_sys::Uint8Array::from(&bytes[..]), + )?; + wt_hashes.push(&hash); + } + } + if wt_hashes.length() > 0 { + let opts = js_sys::Object::new(); + js_sys::Reflect::set( + &opts, + &JsValue::from_str("serverCertificateHashes"), + &wt_hashes, + )?; + args.push(&opts); + } + }; + + let transport = js_sys::Reflect::construct(&ctor, &args)?; + let ready = js_sys::Reflect::get(&transport, &JsValue::from_str("ready"))? + .dyn_into::() + .map_err(|_| js_error("WebTransport.ready is not a Promise"))?; + JsFuture::from(ready) + .await + .map_err(|e| js_error(format!("WebTransport ready failed: {:?}", e)))?; let transport_limits = DecodeLimits::for_transport_message_size(max_message_size as u64); let decode_limits = configured_limits .map(|limits| restrict_decode_limits(limits, transport_limits)) .unwrap_or(transport_limits); Ok(Self { - connection, + inner: transport, max_message_size, + streams_reader: Rc::new(RefCell::new(None)), stream_reader: Rc::new(RefCell::new(None)), buffer: Rc::new(RefCell::new(Vec::new())), new_stream_frame: Rc::new(Cell::new(false)), + outgoing_writer: Rc::new(RefCell::new(None)), send_lock: Rc::new(AsyncMutex::new(())), type_map: Rc::new(RefCell::new(TypeMap::latest())), decode_limits: Rc::new(RefCell::new(decode_limits)), @@ -387,7 +217,7 @@ impl WasmTransport { } pub fn inner(&self) -> &JsValue { - &self.connection.inner + &self.inner } pub fn set_type_map(&self, type_map: &TypeMap) { @@ -417,61 +247,147 @@ impl WasmTransport { return Err(js_error("message too large")); } - // Use one WebTransport uni-stream per MTP frame. Chromium reliably - // publishes a browser-created uni-stream to the peer when it is - // closed; leaving a shared stream open can leave the server waiting - // in accept_uni() until the authentication deadline. The bytes are - // already the canonical MTP self-framed value, so no extra stream - // length prefix is added here. - let mut stream = self.connection.open_uni().await?; - if let Err(e) = stream.write_all(frame).await { + let writer_val = if let Some(writer) = self.outgoing_writer.borrow().clone() { + writer + } else { + let create_stream = js_sys::Reflect::get( + &self.inner, + &JsValue::from_str("createUnidirectionalStream"), + )? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream not a function"))?; + let stream_promise = create_stream + .call0(&self.inner)? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?; + let stream = JsFuture::from(stream_promise).await?; + let writable_or_stream = resolve_stream_writable(&stream)?; + let writer = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter")) + .map_err(|_| js_error("missing getWriter"))? + .dyn_into::() + .map_err(|_| js_error("getWriter not a function"))? + .call0(&writable_or_stream) + .map_err(|_| js_error("getWriter call failed"))?; + *self.outgoing_writer.borrow_mut() = Some(writer.clone()); + writer + }; + + let chunk = js_sys::Uint8Array::from(frame); + + let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write")) + .map_err(|_| js_error("missing write"))? + .dyn_into::() + .map_err(|_| js_error("write not a function"))?; + let write_promise = write_fn + .call1(&writer_val, &chunk) + .map_err(|e| js_error(format!("write failed: {:?}", e)))?; + if let Err(e) = JsFuture::from(write_promise.unchecked_into::()).await { log_stream_error_code(&e, "send_frame write"); - stream.release(); + self.outgoing_writer.borrow_mut().take(); + release_writer_lock(&writer_val); return Err(e); } - if let Err(e) = stream.finish().await { - // The frame was already written; do not retry it merely because - // FIN failed, as that would duplicate the MTP frame. - log_stream_error_code(&e, "send_frame close"); - } - stream.release(); - Ok(()) } + /// Get (creating once) the reader over `incoming_unidirectional_streams()`. + fn ensure_streams_reader(&self) -> Result { + if let Some(reader) = self.streams_reader.borrow().clone() { + return Ok(reader); + } + let incoming = js_sys::Reflect::get( + &self.inner, + &JsValue::from_str("incomingUnidirectionalStreams"), + )?; + let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader")) + .map_err(|_| js_error("missing getReader"))? + .dyn_into::() + .map_err(|_| js_error("getReader not a function"))? + .call0(&incoming) + .map_err(|_| js_error("getReader call failed"))?; + *self.streams_reader.borrow_mut() = Some(reader.clone()); + Ok(reader) + } + /// Accept the next incoming uni-directional stream and make it current. /// Returns `false` if the incoming-streams readable has ended. async fn open_next_stream(&self) -> Result { - let Some(stream) = self.connection.accept_uni().await? else { - return Ok(false); + let streams_reader = self.ensure_streams_reader()?; + + let read_fn = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read")) + .map_err(|_| js_error("missing read"))? + .dyn_into::() + .map_err(|_| js_error("read not a function"))?; + let promise = read_fn + .call0(&streams_reader) + .map_err(|_| js_error("read call failed"))? + .unchecked_into::(); + let result = match JsFuture::from(promise).await { + Ok(r) => r, + Err(e) => { + log_stream_error_code(&e, "open_next_stream accept"); + return Err(js_error(format!("accept stream failed: {:?}", e))); + } }; - *self.stream_reader.borrow_mut() = Some(stream); + let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) + .ok() + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if done { + return Ok(false); + } + + let recv_stream = js_sys::Reflect::get(&result, &JsValue::from_str("value")) + .map_err(|_| js_error("missing value"))?; + let readable = resolve_stream_readable(&recv_stream)?; + let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader")) + .map_err(|_| js_error("missing stream getReader"))? + .dyn_into::() + .map_err(|_| js_error("stream getReader not a function"))? + .call0(&readable) + .map_err(|_| js_error("stream getReader call failed"))?; + + *self.stream_reader.borrow_mut() = Some(reader); self.new_stream_frame.set(true); Ok(true) } /// Read one chunk from the current stream. `Ok(None)` means the stream ended. async fn read_chunk(&self) -> Result>, JsValue> { - let mut stream = match self.stream_reader.borrow_mut().take() { - Some(stream) => stream, + let reader = match self.stream_reader.borrow().clone() { + Some(r) => r, None => return Ok(None), }; - let result = match stream.read_chunk().await { - Ok(result) => result, + + let read_fn = js_sys::Reflect::get(&reader, &JsValue::from_str("read")) + .map_err(|_| js_error("missing read"))? + .dyn_into::() + .map_err(|_| js_error("read not a function"))?; + let promise = read_fn + .call0(&reader) + .map_err(|_| js_error("read call failed"))? + .unchecked_into::(); + let result = match JsFuture::from(promise).await { + Ok(r) => r, Err(e) => { log_stream_error_code(&e, "read_chunk"); - stream.release(); return Err(js_error(format!("read failed: {:?}", e))); } }; - if result.is_some() { - *self.stream_reader.borrow_mut() = Some(stream); - } else { - stream.release(); + + let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) + .ok() + .and_then(|v| v.as_bool()) + .unwrap_or(true); + if done { + return Ok(None); } - Ok(result) + + let value = js_sys::Reflect::get(&result, &JsValue::from_str("value")) + .map_err(|_| js_error("missing value"))?; + Ok(Some(js_sys::Uint8Array::new(&value).to_vec())) } /// Try to pull one complete frame out of the buffer without reading more. @@ -533,7 +449,9 @@ impl WasmTransport { } None => { // Stream finished; release the reader's lock to avoid a spurious cancel. - // `read_chunk` releases the raw stream lock on clean FIN. + if let Some(reader) = self.stream_reader.borrow_mut().take() { + release_reader_lock(&reader); + } // A frame is never allowed to span stream boundaries. The // native persistent-stream sender packs frames on one // stream, while the WASM sender uses one stream per frame; @@ -592,17 +510,15 @@ impl WasmTransport { /// Pipe-aware receive loop. Identical to `receive_loop` but detects /// `PipeRequest` as the first frame on a new incoming stream and routes /// the stream to `on_pipe` instead of `on_message`. - pub async fn receive_loop_with_pipes( + pub async fn receive_loop_with_pipes( &self, mut on_message: F, mut on_error: H, mut on_pipe: G, - mut pipe_is_expected: I, ) where F: FnMut(JsValue), G: FnMut(crate::pipe::PipeReader), H: FnMut(JsValue), - I: FnMut(u32) -> bool, { loop { match self.next_frame(self.max_message_size).await { @@ -624,44 +540,36 @@ impl WasmTransport { if is_first { self.new_stream_frame.set(false); - if let Some(comm) = comm.as_ref() { - let is_pipe_request = Some(comm.get_type()) == pipe_request_type; - let pipe_id = comm.id().filter(|id| *id != 0); - let is_expected = - is_pipe_request && pipe_id.is_some_and(&mut pipe_is_expected); - let disposition = - match classify_first_frame(is_pipe_request, comm.id(), is_expected) - { - Ok(disposition) => disposition, - Err(error) => { - on_error(JsValue::from_str(&error.to_string())); - self.close(); - break; - } - }; + if let Some(comm) = comm.as_ref() + && Some(comm.get_type()) == pipe_request_type + { + let Some(pipe_id) = comm.id().filter(|id| *id != 0) else { + on_error(JsValue::from_str( + "PipeRequest frame must contain a non-zero id", + )); + self.close(); + break; + }; + let description = comm + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); - if let FirstFrameDisposition::Pipe(pipe_id) = disposition { - let description = comm - .get_str(mtp_codec::DataType::Description) - .unwrap_or("") - .to_string(); + let pending = { + let mut buf = self.buffer.borrow_mut(); + std::mem::take(&mut *buf) + }; - let pending = { - let mut buf = self.buffer.borrow_mut(); - std::mem::take(&mut *buf) - }; - - if let Some(reader) = self.stream_reader.borrow_mut().take() { - let pipe_reader = crate::pipe::PipeReader::new( - reader, - pipe_id, - description, - pending, - ); - on_pipe(pipe_reader); - } - continue; + if let Some(reader) = self.stream_reader.borrow_mut().take() { + let pipe_reader = crate::pipe::PipeReader::new( + reader, + pipe_id, + description, + pending, + ); + on_pipe(pipe_reader); } + continue; } } @@ -717,7 +625,25 @@ impl WasmTransport { description: &str, ) -> Result { let _send_guard = self.send_lock.lock().await; - let mut stream = self.connection.open_uni().await?; + let create_stream = js_sys::Reflect::get( + &self.inner, + &JsValue::from_str("createUnidirectionalStream"), + )? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream not a function"))?; + let stream_promise = create_stream + .call0(&self.inner)? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?; + let stream = JsFuture::from(stream_promise).await?; + + let writable_or_stream = resolve_stream_writable(&stream)?; + let writer_val = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter")) + .map_err(|_| js_error("missing getWriter"))? + .dyn_into::() + .map_err(|_| js_error("getWriter not a function"))? + .call0(&writable_or_stream) + .map_err(|_| js_error("getWriter call failed"))?; let type_map = self.type_map(); let request = mtp_codec::CommunicationValue::new_with_type_map( @@ -733,21 +659,43 @@ impl WasmTransport { .to_bytes() .map_err(|e| js_error(format!("encode failed: {}", e)))?; - if let Err(e) = stream.write_all(&frame_bytes).await { + let chunk = js_sys::Uint8Array::from(&frame_bytes[..]); + let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write")) + .map_err(|_| js_error("missing write"))? + .dyn_into::() + .map_err(|_| js_error("write not a function"))?; + let write_promise = write_fn + .call1(&writer_val, &chunk) + .map_err(|e| js_error(format!("write failed: {:?}", e)))?; + if let Err(e) = JsFuture::from(write_promise.unchecked_into::()).await { log_stream_error_code(&e, "open_pipe write"); - stream.release(); + release_writer_lock(&writer_val); return Err(e); } - Ok(crate::pipe::PipeWriter::new(stream, pipe_id)) + Ok(crate::pipe::PipeWriter::new(writer_val, pipe_id)) } pub fn close(&self) { + if let Some(writer) = self.outgoing_writer.borrow_mut().take() { + // The WebTransport session close below terminates the stream. The + // lock must be released first so dropping it is not interpreted as + // an application abort. + release_writer_lock(&writer); + } // Release reader locks before closing so they aren't treated as cancels. if let Some(reader) = self.stream_reader.borrow_mut().take() { - reader.release(); + release_reader_lock(&reader); + } + if let Some(reader) = self.streams_reader.borrow_mut().take() { + release_reader_lock(&reader); + } + + if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close")) + .and_then(|value| value.dyn_into::()) + { + let _ = close.call1(&self.inner, &js_sys::Object::new()); } - self.connection.close(); } }