{"record":{"id":"3fc9dacdedca87a9","repo":"nautechsystems/nautilus_trader","slug":"broadcast-semaphore-closed-e","errorCode":null,"errorMessage":"Broadcast semaphore closed: {e}","messagePattern":"Broadcast semaphore closed: (.+?)","errorType":"exception","errorClass":"DydxError::Nautilus","httpStatus":null,"severity":"error","filePath":"crates/adapters/dydx/src/execution/broadcaster.rs","lineNumber":180,"sourceCode":"    ///\n    /// # Returns\n    ///\n    /// The transaction hash on success.\n    ///\n    /// # Errors\n    ///\n    /// Returns error if all retries are exhausted or a non-retryable error occurs.\n    pub async fn broadcast_with_retry(\n        &self,\n        tx_manager: &TransactionManager,\n        msgs: Vec<Any>,\n        operation_name: &str,\n    ) -> Result<String, DydxError> {\n        // Acquire semaphore to serialize broadcasts.\n        // This ensures sequence N is fully broadcast before sequence N+1 is allocated.\n        let _permit =\n            self.broadcast_semaphore.acquire().await.map_err(|e| {\n                DydxError::Nautilus(anyhow::anyhow!(\"Broadcast semaphore closed: {e}\"))\n            })?;\n\n        log::debug!(\"Acquired broadcast permit for {operation_name}\");\n\n        // Flag to track if we need to resync sequence before the next attempt.\n        // Set by should_retry when a sequence mismatch is detected.\n        let needs_resync = Arc::new(AtomicBool::new(false));\n        let needs_resync_for_retry = Arc::clone(&needs_resync);\n\n        // Clone values that need to be moved into closures\n        let grpc_client = self.grpc_client.clone();\n        let rate_limiter = Arc::clone(&self.rate_limiter);\n        let op_name = operation_name.to_string();\n\n        let operation = || {\n            // Clone captures for the async block\n            let needs_resync = Arc::clone(&needs_resync);\n            let grpc_client = grpc_client.clone();","sourceCodeStart":162,"sourceCodeEnd":198,"githubUrl":"https://github.com/nautechsystems/nautilus_trader/blob/18893faf8b356be3320add8de2f861b0b647cf06/crates/adapters/dydx/src/execution/broadcaster.rs#L162-L198","documentation":"Before broadcasting, `broadcast_with_retry` acquires a tokio semaphore permit to serialize transaction broadcasts (ensuring sequence N is broadcast before N+1 is allocated). If `semaphore.acquire()` returns `AcquireError` — which happens only when the semaphore has been closed — the error is wrapped as 'Broadcast semaphore closed'. A closed semaphore means the broadcaster is shutting down or was closed via `close()`, so no further broadcasts can proceed.","triggerScenarios":"Calling `broadcast_with_retry` after another task called `broadcast_semaphore.close()` — typically during adapter shutdown/stop, or a raced shutdown where a broadcast is attempted concurrently with teardown.","commonSituations":"Sending a final order/cancel during engine shutdown, a task holding a reference to the broadcaster after `disconnect()`/`stop()`, or reconnect logic racing with a close.","solutions":["Ensure no broadcasts are submitted after the adapter's stop/disconnect lifecycle has begun","Check application code for calling `close()` on the semaphore/broadcaster while workers are still active","Gracefully drain pending broadcasts before initiating shutdown","Treat this error as terminal: do not retry the broadcast; re-establish the execution client if needed"],"exampleFix":"// before\n// shutdown path\ntokio::spawn(async move { broadcaster.broadcast_with_retry(\"order\", op).await; }); // races close\n// after\nbroadcaster.broadcast_with_retry(\"order\", op).await?; // complete work first\nbroadcaster.shutdown().await; // then close semaphore","handlingStrategy":"try-catch","validationCode":"// Guard against broadcasting during shutdown\nif shutting_down.load(std::sync::atomic::Ordering::Acquire) { anyhow::bail!(\"shutting down, skipping broadcast\"); }","typeGuard":"fn broadcaster_open(b: &Broadcaster) -> bool { !b.is_closed() }","tryCatchPattern":"match broadcaster.broadcast_with_retry(\"order\", op).await {\n    Err(e) if e.to_string().contains(\"semaphore closed\") => log::warn!(\"broadcast skipped: shutting down\"),\n    other => other?,\n}","preventionTips":["Drain in-flight broadcasts before closing the semaphore","Never submit broadcasts after stop()/disconnect() begins","Use lifecycle events to gate order submission during shutdown"],"tags":["concurrency","shutdown","semaphore","broadcast","dydx"],"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"}