{"record":{"id":"dc62edefa90c4e64","repo":"nautechsystems/nautilus_trader","slug":"command-receiver-already-taken","errorCode":null,"errorMessage":"Command receiver already taken","messagePattern":"Command receiver already taken","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/adapters/blockchain/src/data/client.rs","lineNumber":126,"sourceCode":"            session_tasks,\n        }\n    }\n\n    /// Spawns the main processing task that handles commands and blockchain data.\n    ///\n    /// This method creates a background task that:\n    /// 1. Processes subscription/unsubscription commands from the command channel\n    /// 2. Handles incoming blockchain data from HyperSync\n    /// 3. Processes RPC messages if RPC client is configured\n    /// 4. Routes processed data to subscribers\n    fn spawn_process_task(\n        &mut self,\n    ) -> anyhow::Result<tokio::sync::oneshot::Receiver<anyhow::Result<()>>> {\n        let command_rx = if let Some(r) = self.command_rx.take() {\n            r\n        } else {\n            log::error!(\"Command receiver already taken, not spawning handler\");\n            anyhow::bail!(\"Command receiver already taken\");\n        };\n\n        let cancellation_token = self.cancellation_token.clone();\n\n        let data_tx = nautilus_common::live::runner::get_data_event_sender();\n\n        let mut hypersync_rx = self.hypersync_rx.take().unwrap();\n        let hypersync_tx = self.hypersync_tx.take();\n\n        let mut core_client = BlockchainDataClientCore::new(\n            self.config.clone(),\n            hypersync_tx,\n            Some(data_tx),\n            cancellation_token.clone(),\n        );\n        core_client.set_socket_control(self.socket_factory.control(\"blockchain-rpc\"));\n\n        let (startup_tx, startup_rx) = tokio::sync::oneshot::channel();","sourceCodeStart":108,"sourceCodeEnd":144,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/blockchain/src/data/client.rs#L108-L144","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Recreate the BlockchainDataClient (fresh instance, new command_rx) before calling connect() again instead of reusing the old one.","Ensure connect() is only called once per client lifetime — guard with a connected/started flag or an async OnceCell in the owning code.","If reconnects are needed, implement them inside the spawned handler task or tear down and rebuild the whole client on each reconnect.","Check for racing callers: serialize startup so only one component invokes connect()."],"exampleFix":"// before\n// after disconnect\nclient.connect().await?; // panics into bail: receiver already taken\n// after\ndrop(client);\nlet client = BlockchainDataClient::new(config).await?;\nclient.connect().await?;","handlingStrategy":"validation","validationCode":"use std::sync::atomic::{AtomicBool, Ordering};\nstatic CONNECTED: AtomicBool = AtomicBool::new(false);\nif CONNECTED.swap(true, Ordering::SeqCst) {\n    return Err(anyhow::anyhow!(\"BlockchainDataClient already connected; recreate the client to reconnect\"));\n}\nclient.connect().await?;","typeGuard":null,"tryCatchPattern":"match client.connect().await {\n    Ok(rx) => { /* handle shutdown via rx */ }\n    Err(e) if e.to_string().contains(\"Command receiver already taken\") => {\n        log::warn!(\"client already started; rebuilding client for reconnect\");\n        let client = BlockchainDataClient::new(config).await?;\n        client.connect().await?;\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Treat BlockchainDataClient as single-use: drop and rebuild it for every reconnect.","Centralize client startup in one place guarded by a flag or OnceCell.","Never call connect() from multiple tasks concurrently.","Implement reconnection inside the handler task so the receiver stays alive across reconnects."],"tags":["rust","anyhow","tokio","lifecycle","double-initialization"],"backgroundTag":"invalid-state-transition","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"}