Cannot take ownership of stream - client was cloned and othe

Error message

Cannot take ownership of stream - client was cloned and other references exist

What it means

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.

Source

Thrown at crates/adapters/architect_ax/src/websocket/data/client.rs:1029

        self.subscriptions.confirm_unsubscribe(topic);
        self.subscriptions.mark_subscribe(topic);
        if !was_pending {
            self.subscriptions.confirm_subscribe(topic);
        }
    }

    /// Returns a stream of WebSocket messages.
    ///
    /// # Panics
    ///
    /// Panics if called before `connect()` or if the stream has already been taken.
    pub fn stream(&mut self) -> impl futures_util::Stream<Item = AxDataWsMessage> + 'static {
        let rx = self
            .out_rx
            .take()
            .expect("Stream receiver already taken or client not connected - stream() can only be called once");
        let mut rx = Arc::try_unwrap(rx).expect(
            "Cannot take ownership of stream - client was cloned and other references exist",
        );
        async_stream::stream! {
            while let Some(msg) = rx.recv().await {
                yield msg;
            }
        }
    }

    pub(crate) fn begin_shutdown(&self) {
        self.cancellation_token.load().cancel();
        self.signal.store(true, Ordering::Release);
    }

    /// Disconnects the WebSocket connection gracefully.
    pub async fn disconnect(&self) {
        log::debug!("Disconnecting WebSocket");
        let _ = self.send_cmd(HandlerCommand::Disconnect).await;
    }

View on GitHub (pinned to 18893faf8b)

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.

Example fix

// before
let c2 = client.clone();
let stream = client.stream(); // panics: c2 still holds an Arc reference
// after
drop(c2);
let stream = client.stream(); // now try_unwrap succeeds

Defensive patterns

Strategy: validation

Validate before calling

// Arc<...> clones make try_unwrap fail; ensure the client is uniquely owned before streaming
// (no direct public check exists — enforce single ownership by not calling .clone() on the client)

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

AI-assisted analysis of nautechsystems/nautilus_trader@18893faf8b (2026-09-08). Data as JSON: /api/errors/47588dbfd22e4aba. Report an issue: GitHub.