{"record":{"id":"47588dbfd22e4aba","repo":"nautechsystems/nautilus_trader","slug":"cannot-take-ownership-of-stream-client-was-clone","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/data/client.rs","lineNumber":1029,"sourceCode":"        self.subscriptions.confirm_unsubscribe(topic);\n        self.subscriptions.mark_subscribe(topic);\n        if !was_pending {\n            self.subscriptions.confirm_subscribe(topic);\n        }\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 = AxDataWsMessage> + '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":1011,"sourceCodeEnd":1047,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/architect_ax/src/websocket/data/client.rs#L1011-L1047","documentation":"The stream receiver is stored as Arc so cloned clients can share it. stream() must take sole ownership via Arc::try_unwrap to build a 'static stream; if any clone still holds a reference the try_unwrap fails and this expect panics. This enforces the single-consumer invariant of the message stream.","triggerScenarios":"Cloning the AxDataWebSocketClient (e.g. into a task or struct field) and then calling stream() on any handle while the clone(s) are still alive.","commonSituations":"Spawning writer/reader tasks that each hold a client clone; storing a clone for sending orders while streaming data from the original; tests that clone the client for assertions before draining the stream.","solutions":["Drop every clone of the client before calling stream(), or call stream() on the last remaining handle.","Restructure so clones are only used for sending and the original handle is reserved exclusively for stream().","If concurrent consumption is required, add an internal forwarding task with a channel rather than relying on stream().","Use Arc::strong_count debugging (temporarily) to locate which code path still holds a reference."],"exampleFix":"// before\nlet c2 = client.clone();\nlet stream = client.stream(); // panics: c2 still holds an Arc reference\n// after\ndrop(c2);\nlet stream = client.stream(); // now try_unwrap succeeds","handlingStrategy":"validation","validationCode":"// Arc<...> clones make try_unwrap fail; ensure the client is uniquely owned before streaming\n// (no direct public check exists — enforce single ownership by not calling .clone() on the client)","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Treat the streaming handle as unique: use clones only for sending, never alongside stream().","Drop all cloned handles before invoking stream().","Design consumers as one stream owner fanning out 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"}