{"record":{"id":"ff2a5cc6c2b905ce","repo":"tracel-ai/burn","slug":"asyncpolicy-should-receive-queued-actions","errorCode":null,"errorMessage":"AsyncPolicy should receive queued actions.","messagePattern":"AsyncPolicy should receive queued actions\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/burn-rl/src/policy/async_policy.rs","lineNumber":293,"sourceCode":"    }\n\n    fn action(\n        &mut self,\n        states: Self::Observation,\n        deterministic: bool,\n    ) -> (Self::Action, Vec<Self::ActionContext>) {\n        let (action_sender, action_receiver) = std::sync::mpsc::channel();\n        let item = ActionItem {\n            sender: action_sender,\n            inference_state: states,\n            deterministic,\n        };\n        self.inference_state_sender\n            .send(InferenceMessage::ActionMessage(item))\n            .expect(\"should be able to send message to inference_server.\");\n        let action = action_receiver\n            .recv()\n            .expect(\"AsyncPolicy should receive queued actions.\");\n        (action.action, action.context)\n    }\n\n    fn update(&mut self, update: Self::PolicyState) {\n        self.inference_state_sender\n            .send(InferenceMessage::PolicyUpdate(update))\n            .expect(\"AsyncPolicy should be able to send policy state.\")\n    }\n\n    fn state(&self) -> Self::PolicyState {\n        let (sender, receiver) = mpsc::channel();\n        self.inference_state_sender\n            .send(InferenceMessage::PolicyRequest(sender))\n            .expect(\"should be able to send message to inference_server.\");\n        receiver\n            .recv()\n            .expect(\"AsyncPolicy should be able to receive policy state.\")\n    }","sourceCodeStart":275,"sourceCodeEnd":311,"githubUrl":"https://github.com/tracel-ai/burn/blob/d16f7ba2ed0d41408189384044cc886fb4c8f957/crates/burn-rl/src/policy/async_policy.rs#L275-L311","documentation":"After queueing an ActionMessage, AsyncPolicy::action blocks on action_receiver.recv() and panics on Err, which occurs when the reply Sender was dropped without sending a value. The autobatcher sends exactly one reply per item in flush_actions, so Err means that flush never happened: the server thread died (panic or loop exit) before this item was batched and processed.","triggerScenarios":"An earlier panic in the autobatcher thread (failed reply send to another agent, or inner_policy.action panicking on the batched input such as a device/shape error) dropping all queued senders; or the thread exiting its loop so queued ActionItems are never flushed and their senders are dropped at teardown.","commonSituations":"One agent aborting its request (dropping its receiver) panics flush_actions' reply send, killing the thread and turning every other waiting agent's recv into this panic; num_agents accounting bugs causing batches that never reach the flush threshold, later compounded by a thread shutdown.","solutions":["Diagnose the root cause of the server-thread death from logs; this panic is a symptom, not the cause.","Keep increment_agents/decrement_agents counts in sync with live agents so batches always flush and requests never hang until shutdown.","Make the server flush or reject pending items gracefully on shutdown so waiting callers get a real error instead of a dropped channel.","Use recv_timeout with a descriptive panic/error on the caller side to distinguish 'server dead' from 'slow inference'."],"exampleFix":"// before\nlet action = action_receiver\n    .recv()\n    .expect(\"AsyncPolicy should receive queued actions.\");\n// after\nlet action = action_receiver\n    .recv_timeout(std::time::Duration::from_secs(60))\n    .unwrap_or_else(|err| panic!(\"No action returned by inference server (thread dead or batch stalled): {}\", err));","handlingStrategy":"retry","validationCode":"// Ensure agent accounting is consistent before requesting actions\nassert!(registered_agents.load(Ordering::Relaxed) > 0, \"agents not registered; action batch may never flush\");","typeGuard":null,"tryCatchPattern":"match action_receiver.recv_timeout(Duration::from_secs(60)) {\n    Ok(ac) => (ac.action, ac.context),\n    Err(RecvTimeoutError::Timeout) => return Err(anyhow::anyhow!(\"action request stalled; batch never flushed\")),\n    Err(RecvTimeoutError::Disconnected) => return Err(anyhow::anyhow!(\"inference server died before returning an action\")),\n}","preventionTips":["Use recv_timeout with a descriptive error instead of a bare expect on recv.","Keep increment/decrement agent counts matched to live callers so batches always flush.","Investigate flush_actions/inner_policy panics in logs — they drop all queued reply senders at once.","Ensure orderly shutdown flushes pending ActionItems or replies with errors."],"tags":["rust","mpsc-channel","panics","multi-agent"],"backgroundTag":"channel-sender-dropped","analyzedSha":"d16f7ba2ed0d41408189384044cc886fb4c8f957","analyzedAt":"2026-09-05T13:19:14.260Z","contentChangedAt":"2026-09-05T13:19:14.260Z","schemaVersion":2},"datasetVersion":"2026-09-12T17:17:11.597Z"}