nautechsystems/nautilus_trader · error

Command receiver already taken

Error message

Command receiver already taken

What it means

BlockchainDataClient::spawn_process_task takes its oneshot command receiver (command_rx) via Option::take because it can only be consumed once. If connect() is invoked a second time after the receiver was already moved into the spawned handler task, the Option is None and the client logs and bails with "Command receiver already taken". This enforces that a live data client has at most one active command-processing task.

Source

Thrown at crates/adapters/blockchain/src/data/client.rs:126

            session_tasks,
        }
    }

    /// Spawns the main processing task that handles commands and blockchain data.
    ///
    /// This method creates a background task that:
    /// 1. Processes subscription/unsubscription commands from the command channel
    /// 2. Handles incoming blockchain data from HyperSync
    /// 3. Processes RPC messages if RPC client is configured
    /// 4. Routes processed data to subscribers
    fn spawn_process_task(
        &mut self,
    ) -> anyhow::Result<tokio::sync::oneshot::Receiver<anyhow::Result<()>>> {
        let command_rx = if let Some(r) = self.command_rx.take() {
            r
        } else {
            log::error!("Command receiver already taken, not spawning handler");
            anyhow::bail!("Command receiver already taken");
        };

        let cancellation_token = self.cancellation_token.clone();

        let data_tx = nautilus_common::live::runner::get_data_event_sender();

        let mut hypersync_rx = self.hypersync_rx.take().unwrap();
        let hypersync_tx = self.hypersync_tx.take();

        let mut core_client = BlockchainDataClientCore::new(
            self.config.clone(),
            hypersync_tx,
            Some(data_tx),
            cancellation_token.clone(),
        );
        core_client.set_socket_control(self.socket_factory.control("blockchain-rpc"));

        let (startup_tx, startup_rx) = tokio::sync::oneshot::channel();

View on GitHub (pinned to 18893faf8b)

Solutions

1. Recreate the BlockchainDataClient (fresh instance, new command_rx) before calling connect() again instead of reusing the old one.
2. Ensure connect() is only called once per client lifetime — guard with a connected/started flag or an async OnceCell in the owning code.
3. If reconnects are needed, implement them inside the spawned handler task or tear down and rebuild the whole client on each reconnect.
4. Check for racing callers: serialize startup so only one component invokes connect().

Example fix

// before
// after disconnect
client.connect().await?; // panics into bail: receiver already taken
// after
drop(client);
let client = BlockchainDataClient::new(config).await?;
client.connect().await?;

use std::sync::atomic::{AtomicBool, Ordering};
static CONNECTED: AtomicBool = AtomicBool::new(false);
if CONNECTED.swap(true, Ordering::SeqCst) {
    return Err(anyhow::anyhow!("BlockchainDataClient already connected; recreate the client to reconnect"));
}
client.connect().await?;

match client.connect().await {
    Ok(rx) => { /* handle shutdown via rx */ }
    Err(e) if e.to_string().contains("Command receiver already taken") => {
        log::warn!("client already started; rebuilding client for reconnect");
        let client = BlockchainDataClient::new(config).await?;
        client.connect().await?;
    }
    Err(e) => return Err(e),
}

When it happens

Trigger: Calling connect() (which calls spawn_process_task) twice on the same BlockchainDataClient instance without recreating it — e.g. a reconnect loop that reuses the client object instead of rebuilding it, or concurrent connect() calls racing on the same instance.

Common situations: Automatic reconnect logic in a live trading node calls connect again after a disconnect; a node restart path forgets to drop/recreate the data client; two subsystems both attempt to start the same client.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Failed to start Polymarket task generation: {e}
• Active execution intent {intent_id} was not found
• DataActor {} must be registered before calling `clock_mut()`
• DataActor {} must be registered before calling `clock()` - t
• DataActor {} must be registered before calling `clock_ref()`

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