{"record":{"id":"9294b30f33b55b9b","repo":"nautechsystems/nautilus_trader","slug":"cannot-take-ownership-other-references-exist-9294b3","errorCode":null,"errorMessage":"Cannot take ownership - other references exist","messagePattern":"Cannot take ownership - other references exist","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"crates/adapters/bybit/src/websocket/client.rs","lineNumber":951,"sourceCode":"        let cmd = HandlerCommand::Unsubscribe { topics: payloads };\n        if let Err(e) = self.cmd_tx.read().await.send(cmd) {\n            log::debug!(\"Failed to send unsubscribe command: error={e}\");\n        }\n\n        Ok(())\n    }\n\n    /// Returns a stream of venue-typed [`BybitWsMessage`] items.\n    ///\n    /// # Panics\n    ///\n    /// Panics if called before [`Self::connect`] or if the stream has already been taken.\n    pub fn stream(&mut self) -> impl futures_util::Stream<Item = BybitWsMessage> + use<> {\n        let rx = self\n            .out_rx\n            .take()\n            .expect(\"Stream receiver already taken or client not connected\");\n        let mut rx = Arc::try_unwrap(rx).expect(\"Cannot take ownership - other references exist\");\n        async_stream::stream! {\n            while let Some(msg) = rx.recv().await {\n                yield msg;\n            }\n        }\n    }\n\n    /// Returns the number of currently registered subscriptions.\n    #[must_use]\n    pub fn subscription_count(&self) -> usize {\n        self.subscriptions.len()\n    }\n\n    /// Returns the credential associated with this client, if any.\n    #[must_use]\n    pub fn credential(&self) -> Option<&Credential> {\n        self.credential.as_ref()\n    }","sourceCodeStart":933,"sourceCodeEnd":969,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/bybit/src/websocket/client.rs#L933-L969","documentation":"Inside BybitWebSocketClient::stream(), the receiver is stored in an Arc<Receiver>; Arc::try_unwrap converts it to an owned Receiver only if this is the last reference. The expect fires when other clones of the Arc still exist (e.g. a spawn kept a handle for subscribe/order-response routing), so ownership cannot be moved into the async_stream generator.","triggerScenarios":"Calling stream() while some component (connection task, test harness, subscription router) still holds a clone of the out_rx Arc.","commonSituations":"Tests like test_trade_order_response_preserves_request_id that clone the receiver to assert on responses; refactors that retain an Arc handle for sending requests while also draining via stream(); failing to shut down the reader task before taking the stream.","solutions":["Drop or join all other holders of the receiver Arc before calling stream() (e.g. abort the connection task that cloned it).","Restructure so the stream is the sole consumer and other components send messages into the client instead of cloning the receiver.","If sharing is genuinely required, wrap messages in an Arc and use a broadcast channel rather than relying on try_unwrap.","In tests, clone the Arc only after stream() has been called, or use try_recv on the owned receiver."],"exampleFix":"// before\nlet rx_clone = std::mem::discriminant(&client); // some task still holds out_rx Arc\nlet stream = client.stream(); // panics: other Arc references exist\n// after\ndrop(reader_task_handle); // or reader_task_handle.await; release the clone first\nlet stream = client.stream();","handlingStrategy":"validation","validationCode":"// Ensure no clones of the receiver exist before taking the stream:\n// join or abort tasks that cloned out_rx, then call stream().\nassert!(reader_task.is_finished(), \"receiver holder must be done before stream()\");","typeGuard":null,"tryCatchPattern":"// Unreachable-safety net only:\nstd::panic::catch_unwind(std::panic::AssertUnwindSafe(|| client.stream()))","preventionTips":["Do not clone the out_rx Arc; treat the stream as the single consumer.","Drop/join reader tasks before calling stream().","Use broadcast channels when multiple consumers are required."],"tags":["rust","websocket","panic","bybit","ownership"],"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-14T00:17:10.932Z"}