From c1761aae2bff7f7b6495292e5646485e380d371a Mon Sep 17 00:00:00 2001 From: Alois Date: Sat, 4 Jul 2026 14:06:03 +0200 Subject: [PATCH] (feat): improve wasm/ts-sdk logger (fix): wasm rust runtime error (qol): update comments --- src/sdk/index.ts | 24 ++++++++++++++++++++---- wasm/src/client.rs | 36 +++++++++++++++++++++++++++++++++--- wasm/src/transport.rs | 42 +++++++----------------------------------- 3 files changed, 60 insertions(+), 42 deletions(-) diff --git a/src/sdk/index.ts b/src/sdk/index.ts index eabcc8a..9cd87e9 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -19,8 +19,8 @@ export interface MTPCredentialStorage { export type MTPStorage = MTPCredentialStorage; export type MTPLogEvent = - | { hint: "info" | "warning"; type: string; data: unknown } - | { hint: "error"; type: string | "error"; error: string; data?: unknown }; + | { hint: "info" | "warning"; type: string; data: unknown; direction?: "send" | "recv" } + | { hint: "error"; type: string | "error"; error: string; data?: unknown; direction?: "send" | "recv" }; export type ParsedFrame = RawBindings.ParsedFrame; @@ -605,13 +605,14 @@ export class MTPClient { try { const frame = this.raw.bindings.parse_frame(message); emit(this.#options.logger, isErrorType(frame.type) - ? { hint: "error", type: frame.type, error: errorMessage(frame), data: frame.data } - : { hint: "info", type: frame.type, data: frame.data }); + ? { hint: "error", type: frame.type, error: errorMessage(frame), data: frame.data, direction: "send" } + : { hint: "info", type: frame.type, data: frame.data, direction: "send" }); } catch (error) { emit(this.#options.logger, { hint: "error", type: "error", error: String(error), + direction: "send", }); } @@ -622,6 +623,19 @@ export class MTPClient { async request(type: MTPCommunicationType, data: Record, options?: MTPRequestOptions): Promise; async request(typeOrFrame: Uint8Array | MTPCommunicationType, data?: Record, options: MTPRequestOptions = {}): Promise { const frame = this.#buildFrame(typeOrFrame, data, options); + try { + const parsed = this.raw.bindings.parse_frame(frame); + emit(this.#options.logger, isErrorType(parsed.type) + ? { hint: "error", type: parsed.type, error: errorMessage(parsed), data: parsed.data, direction: "send" } + : { hint: "info", type: parsed.type, data: parsed.data, direction: "send" }); + } catch (error) { + emit(this.#options.logger, { + hint: "error", + type: "error", + error: String(error), + direction: "send", + }); + } return await this.raw.client.request(frame, options.responseType ?? null); } @@ -643,12 +657,14 @@ export class MTPClient { type: frame.type, error: errorMessage(frame), data: frame.data, + direction: "recv", }); } else { emit(this.#options.logger, { hint: "info", type: frame.type, data: frame.data, + direction: "recv", }); } } diff --git a/wasm/src/client.rs b/wasm/src/client.rs index 16c33eb..2a24408 100644 --- a/wasm/src/client.rs +++ b/wasm/src/client.rs @@ -687,9 +687,39 @@ impl WasmClient { fn set_state(&self, new_state: ConnectionState) { self.state.set(new_state); - let _ = self - .on_state_change - .call1(&JsValue::NULL, &JsValue::from(new_state as u8)); + + // Defer the callback to a microtask so re-entrant &mut self calls don't alias. + let cb = self.on_state_change.clone(); + let val = JsValue::from(new_state as u8); + let closure = Closure::wrap(Box::new(move || { + let _ = cb.call1(&JsValue::NULL, &val); + }) as Box); + + let global = js_sys::global(); + let mut closure_opt = Some(closure); + + let qmt = js_sys::Reflect::get(&global, &JsValue::from_str("queueMicrotask")) + .and_then(|f| f.dyn_into::().map_err(Into::into)); + let scheduled = match qmt { + Ok(qmt) => { + if let Some(c) = closure_opt.take() { + let _ = qmt.call1(&global, c.as_ref()); + c.forget(); + } + true + } + Err(_) => false, + }; + if !scheduled { + if let Ok(set_timeout) = js_sys::Reflect::get(&global, &JsValue::from_str("setTimeout")) + .and_then(|f| f.dyn_into::().map_err(Into::into)) + { + if let Some(c) = closure_opt.take() { + let _ = set_timeout.call2(&global, c.as_ref(), &JsValue::from_f64(0.0)); + c.forget(); + } + } + } } fn start_receive_loop(&mut self, transport: WasmTransport) { diff --git a/wasm/src/transport.rs b/wasm/src/transport.rs index 9a8c69d..6860552 100644 --- a/wasm/src/transport.rs +++ b/wasm/src/transport.rs @@ -10,16 +10,7 @@ use crate::frame::parse_frame_value; const CLOSE_FRAME_LEN: u32 = u32::MAX; -/// Inspect a JS error value for a WebTransport **stream-level** error and, if -/// present, log the `streamErrorCode` carried by STOP_SENDING / RESET_STREAM. -/// -/// Per draft-ietf-webtrans-http3-15 §4.4, a WebTransport application MUST -/// provide an error code for those operations. The browser surfaces these as -/// `WebTransportError` with `source = "stream"` and a numeric `streamErrorCode`. -/// -/// Session-level errors (`source = "session"`) are normal connection -/// closures and are **not** logged here — they propagate to `on_error` -/// in the receive loop like any other transport error. +/// Logs the `streamErrorCode` from a stream-level WebTransportError (STOP_SENDING / RESET_STREAM). Session errors are skipped. fn log_stream_error_code(error: &JsValue, context: &str) { let source = js_sys::Reflect::get(error, &JsValue::from_str("source")) .ok() @@ -78,10 +69,7 @@ fn resolve_stream_readable(recv_stream: &JsValue) -> Result { } } -/// Release a `WritableStreamDefaultWriter`'s lock on its stream. Called after -/// `writer.close()` (or on write failure) so the runtime does not interpret an -/// abandoned locked writer as an abort, which would surface as STOP_SENDING to -/// the peer. Errors are ignored — `releaseLock` is best-effort cleanup. +/// Releases a writer's lock so an abandoned writer isn't treated as an abort (which sends STOP_SENDING). fn release_writer_lock(writer: &JsValue) { if let Ok(release) = js_sys::Reflect::get(writer, &JsValue::from_str("releaseLock")) .and_then(|f| f.dyn_into::().map_err(Into::into)) @@ -90,11 +78,7 @@ fn release_writer_lock(writer: &JsValue) { } } -/// Release a `ReadableStreamDefaultReader`'s lock on its stream. Mirrors -/// `release_writer_lock`: abandoning a locked reader can be interpreted by the -/// runtime as a `reader.cancel()` (sending STOP_SENDING to the peer) even on an -/// already-closed or errored stream. Calling `releaseLock` explicitly avoids -/// that. Errors are ignored — best-effort cleanup. +/// Releases a reader's lock so an abandoned reader isn't treated as a cancel (which sends STOP_SENDING). fn release_reader_lock(reader: &JsValue) { if let Ok(release) = js_sys::Reflect::get(reader, &JsValue::from_str("releaseLock")) .and_then(|f| f.dyn_into::().map_err(Into::into)) @@ -256,15 +240,11 @@ impl WasmTransport { .call0(&writer_val) .map_err(|e| js_error(&format!("close failed: {:?}", e)))?; if let Err(e) = JsFuture::from(close_promise.unchecked_into::()).await { - // The write already succeeded; a STOP_SENDING on close just means - // the peer stopped reading before we could send FIN. The data is in - // flight, so this is not a send failure — log and return success. + // Write succeeded; STOP_SENDING on close just means peer stopped reading before FIN. log_stream_error_code(&e, "send_frame close"); } - // Always release the writer's lock on the WritableStream. Abandoning a - // locked writer (e.g. via drop) can be interpreted by the runtime as an - // abort, which may surface as STOP_SENDING to the peer. + // Release the lock so the writer isn't treated as an abort. release_writer_lock(&writer_val); Ok(()) @@ -418,12 +398,7 @@ impl WasmTransport { } } None => { - // Current stream finished; the next frame (if any) is on a - // subsequent stream. Any trailing partial bytes are dropped - // since the host never splits a frame across streams. - // Release the reader's lock explicitly so the runtime does - // not treat the abandoned lock as a cancel (which would send - // STOP_SENDING to the peer on an already-closed stream). + // Stream finished; release the reader's lock to avoid a spurious cancel. if let Some(reader) = self.stream_reader.borrow_mut().take() { release_reader_lock(&reader); } @@ -470,10 +445,7 @@ impl WasmTransport { } pub fn close(&self) { - // Release any held reader locks before tearing down the session, so the - // runtime does not interpret an abandoned locked reader as a cancel - // (which would send STOP_SENDING to the peer). Once the locks are - // released the underlying streams can be torn down cleanly. + // Release reader locks before closing so they aren't treated as cancels. if let Some(reader) = self.stream_reader.borrow_mut().take() { release_reader_lock(&reader); }