Share one localhost WS listener across native product executions

.changeset/shared-ws-listener-teardown.md

@@ -0,0 +1,9 @@
+---
+"@parity/truapi-host": patch
+---
+
+Product executions under one host runtime share a single localhost WebSocket listener, each with its own token. Closing a
+connection disposes the runtime that served it, so its host-core subscriptions and chat state are released rather than
+held for the life of the listener. Admitted connections are bounded per execution and listener-wide, with the
+per-execution cap a share of the listener-wide one; handshakes in flight are bounded separately, and a full backlog
+evicts its oldest entry so a stalled peer cannot lock out other executions.

android/truapi-host/README.md

@@ -67,7 +67,7 @@ The public surface lives in [`src/main/kotlin/io/parity/truapi/TrUAPIHost.kt`](s
 - `HostStorage` - product-scoped read/write/clear interface the host backs with its own persistence.
 - `HostCoreStorage` - core-owned read/write/clear interface for auth session, pairing identity, and persisted permission decisions (`key` is a SCALE-encoded `CoreStorageKey`).
 - `LocalhostBridgeBootstrap` - JS snippet that publishes the WS bridge endpoint (`window.__truapi_localhost`) to the product page so it can dial back in.
-- `TrUAPIHostRuntime` - process-owned runtime whose product executions share one authentication session. Open a connection per executable with `openProductExecution`, which returns a `TrUAPIProductExecution` carrying that connection's own WS bridge, permission authorization, theme/preimage/chain notifications, and the Chat controls below.
+- `TrUAPIHostRuntime` - process-owned runtime whose product executions share one authentication session. Open a connection per executable with `openProductExecution`, which returns a `TrUAPIProductExecution` holding its own token on the runtime's shared WS bridge, permission authorization, theme/preimage/chain notifications, and the Chat controls below.
 - `ChatHostBridge` - native Chat storage and UI, implemented by hosts that serve the Chat modality and passed to `openProductExecution`. Hosts without it pass nothing and Chat calls answer unsupported.
 - `PocketHostBridge` - the host's Pocket card collection, implemented by hosts with a Pocket surface and passed as `pocket` to `openProductExecution`. The execution then offers `notifyPocketCardsChanged`. `removeCard` decides and removes together, returning `NativePocketRemoval.Removed`, `Absent` or `Privileged`, so a card cannot be pinned between the check and the removal. Like Chat, Pocket is reachable only from a Worker execution with an active session, so without `activateLocalSession` every Pocket call answers `Denied`. Hosts without the bridge pass nothing and Pocket calls answer unsupported.
 

android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt

@@ -974,11 +974,15 @@ class TrUAPIProductExecution internal constructor(
 ) : AutoCloseable {
     private val shutDown = AtomicBoolean(false)
 
-    /** Start this execution's independently authenticated localhost bridge. */
+    /**
+     * Register this execution against the host runtime's shared localhost
+     * bridge, minting an independent authentication token. Every execution
+     * under the same host runtime connects through the same port.
+     */
     @Throws(WsBridgeStartException::class)
     fun startWsBridge(bindPort: UShort = 0u): WsBridgeEndpoint = inner.startWsBridge(bindPort)
 
-    /** Stop the active bridge while leaving the execution reusable. */
+    /** Revoke this execution's bridge registration while leaving it reusable. */
     fun stopWsBridge() {
         inner.stopWsBridge()
     }

rust/crates/truapi-server/src/host_core.rs

@@ -1408,10 +1408,19 @@ impl ProductRuntime {
         // would poison this mutex and every later `receive_frame` would then panic
         // here, which is exactly the production-host-killing shape the debug tap
         // above was fixed for.
-        self.in_flight
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
-            .insert(dispatch_id, abort_handle);
+        //
+        // Re-check under the disposal lock so a racing dispatch cannot register
+        // after `dispose` has drained the active requests.
+        {
+            let mut in_flight = self
+                .in_flight
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if self.disposed.load(Ordering::Acquire) {
+                return Ok(());
+            }
+            in_flight.insert(dispatch_id, abort_handle);
+        }
 
         let transport: Arc<dyn Transport> = self.transport.clone();
         let _ = Abortable::new(self.core.dispatch(message, transport), abort_registration).await;
@@ -1502,16 +1511,21 @@ impl ProductRuntime {
     /// futures, and cancels active subscriptions.
     #[instrument(skip_all, fields(runtime.method = "product_runtime.dispose"))]
     pub fn dispose(&self) {
-        if self.disposed.swap(true, Ordering::AcqRel) {
+        // Aborting under the lock can wake code that re-enters disposal.
+        if self.disposed.load(Ordering::Acquire) {
             return;
         }
-        for (_, handle) in self
-            .in_flight
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
-            .drain()
         {
-            handle.abort();
+            let mut in_flight = self
+                .in_flight
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if self.disposed.swap(true, Ordering::AcqRel) {
+                return;
+            }
+            for (_, handle) in in_flight.drain() {
+                handle.abort();
+            }
         }
         self.admin.product_runtime.detach_chat();
         self.admin.product_runtime.detach_renderer();
@@ -1628,7 +1642,7 @@ impl Transport for SinkTransport {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::frame::{Payload, ProtocolMessage, subscription_ids};
+    use crate::frame::{Payload, ProtocolMessage, request_ids, subscription_ids};
     use crate::host_logic::product_account::derive_identity_keypair;
     use crate::host_logic::sso::messages::{
         RemoteMessage, RemoteMessageData, decode_incoming_sso_request, v1,
@@ -2765,6 +2779,92 @@ mod tests {
         assert_eq!(response.payload.value, expected);
     }
 
+    // The debug tap deliberately blocks to expose the registration race reliably.
+    #[test]
+    fn a_dispatch_racing_dispose_does_not_reach_the_platform() {
+        struct ParkingDebugSink {
+            entered: std::sync::mpsc::SyncSender<()>,
+            release: Mutex<Option<std::sync::mpsc::Receiver<()>>>,
+        }
+
+        impl DebugSink for ParkingDebugSink {
+            fn emit(&self, _event: DebugEvent) {
+                let _ = self.entered.try_send(());
+                if let Some(release) = self
+                    .release
+                    .lock()
+                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .take()
+                {
+                    let _ = release.recv();
+                }
+            }
+        }
+
+        let navigations = Arc::new(Mutex::new(Vec::new()));
+        let platform = Arc::new(StubPlatform {
+            navigations: navigations.clone(),
+            ..Default::default()
+        });
+        let (host_config, product) = runtime_config("myapp.dot");
+        let runtime = Arc::new(ProductRuntime::from_platform_with_config(
+            platform,
+            host_config,
+            product,
+            test_spawner(),
+            Arc::new(RecordingSink::default()),
+        ));
+
+        let (entered_tx, entered_rx) = std::sync::mpsc::sync_channel::<()>(1);
+        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+        runtime.set_debug_sink(
+            ChannelId("race".to_string()),
+            Arc::new(ParkingDebugSink {
+                entered: entered_tx,
+                release: Mutex::new(Some(release_rx)),
+            }),
+        );
+
+        let ids = request_ids("system_navigate_to").expect("known request method");
+        let frame = ProtocolMessage {
+            request_id: "nav:1".to_string(),
+            payload: Payload {
+                trait_id: ids.trait_id,
+                method_id: ids.method_id,
+                message_type: crate::frame::MESSAGE_TYPE_REQUEST,
+                value: truapi::versioned::system::HostNavigateToRequest::V1(
+                    v01::HostNavigateToRequest {
+                        url: "https://example.invalid/".to_string(),
+                    },
+                )
+                .encode(),
+            },
+        }
+        .encode();
+
+        let dispatching = {
+            let runtime = runtime.clone();
+            std::thread::spawn(move || {
+                futures::executor::block_on(runtime.receive_frame(frame)).expect("receive frame");
+            })
+        };
+
+        entered_rx
+            .recv_timeout(std::time::Duration::from_secs(5))
+            .expect("frame never reached the debug tap");
+        runtime.dispose();
+        let _ = release_tx.send(());
+        dispatching.join().expect("dispatch thread panicked");
+
+        assert!(
+            navigations
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .is_empty(),
+            "a dispatch that lost the race with dispose still reached the platform"
+        );
+    }
+
     #[test]
     fn dispose_cancels_active_subscriptions() {
         let theme_stream_dropped = Arc::new(AtomicBool::new(false));