{"record":{"id":"454864bc4984a297","repo":"nautechsystems/nautilus_trader","slug":"cannot-take-ownership-of-stream-client-was-clone-454864","errorCode":null,"errorMessage":"Cannot take ownership of stream - client was cloned and other references exist","messagePattern":"Cannot take ownership of stream - client was cloned and other references exist","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/adapters/architect_ax/src/websocket/orders/client.rs","lineNumber":703,"sourceCode":"\n        self.send_cmd(HandlerCommand::GetOpenOrders { request_id })\n            .await?;\n\n        Ok(request_id)\n    }\n\n    /// Returns a stream of WebSocket messages.\n    ///\n    /// # Panics\n    ///\n    /// Panics if called before `connect()` or if the stream has already been taken.\n    pub fn stream(&mut self) -> impl futures_util::Stream<Item = AxOrdersWsMessage> + 'static {\n        let rx = self\n            .out_rx\n            .take()\n            .expect(\"Stream receiver already taken or client not connected - stream() can only be called once\");\n        let mut rx = Arc::try_unwrap(rx).expect(\n            \"Cannot take ownership of stream - client was cloned and other references exist\",\n        );\n        async_stream::stream! {\n            while let Some(msg) = rx.recv().await {\n                yield msg;\n            }\n        }\n    }\n\n    pub(crate) fn begin_shutdown(&self) {\n        self.cancellation_token.load().cancel();\n        self.signal.store(true, Ordering::Release);\n    }\n\n    /// Disconnects the WebSocket connection gracefully.\n    pub async fn disconnect(&self) {\n        log::debug!(\"Disconnecting WebSocket\");\n        let _ = self.send_cmd(HandlerCommand::Disconnect).await;\n    }","sourceCodeStart":685,"sourceCodeEnd":721,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/architect_ax/src/websocket/orders/client.rs#L685-L721","documentation":"In AxOrdersWebSocketClient::stream(), the shared receiver is an Arc and stream() requires unique ownership (Arc::try_unwrap) to return a 'static stream. If clones of the client still exist, try_unwrap returns Err and this expect panics, protecting the single-consumer contract of the orders stream.","triggerScenarios":"Calling stream() while any clone of the AxOrdersWebSocketClient (created with .clone()) is still alive, e.g. a clone held by an order-sending task or a long-lived actor.","commonSituations":"Sharing one client between an order manager and a WebSocket event loop; holding a clone in application state while the stream is consumed elsewhere; tests cloning the client to inspect state before streaming.","solutions":["Drop all clones before calling stream(), or call stream() only on the final remaining handle.","Dedicate one handle exclusively to streaming and use clones only for send operations.","Fan out messages yourself: consume the stream in one task and forward to per-consumer channels.","If the design truly needs concurrent receivers, switch to a broadcast channel in your wrapper layer."],"exampleFix":"// before\nlet clone = client.clone();\nsend_orders(clone); // clone moved into task but another clone retained\nlet stream = client.stream(); // panics: references exist\n// after\n// keep exactly one handle for streaming\nlet stream = client.stream();\n// derive senders from a separate connection or drop clones first","handlingStrategy":"validation","validationCode":"// Ensure no clones of the orders client are alive before streaming\n// (avoid .clone() on the client intended for stream(); keep one unique handle)","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Reserve one client handle exclusively for streaming; use separate clients or channels for senders.","Audit code paths that clone the client into tasks or shared state.","If concurrent receivers are needed, fan out from the single stream via channels."],"tags":["websocket","panic","ownership","arc","rust","architect-ax"],"backgroundTag":"internal-invariant-violation","analyzedSha":"18893faf8b356be3320add8de2f861b0b647cf06","analyzedAt":"2026-09-08T20:49:34.690Z","contentChangedAt":"2026-09-08T20:49:34.690Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}