{"record":{"id":"2ccce126544a648c","repo":"tracel-ai/burn","slug":"asyncpolicy-should-be-able-to-send-policy-state","errorCode":null,"errorMessage":"AsyncPolicy should be able to send policy state.","messagePattern":"AsyncPolicy should be able to send policy state\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/burn-rl/src/policy/async_policy.rs","lineNumber":300,"sourceCode":"        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    }\n\n    fn to_device(self, device: &Device) -> Self {\n        self.inference_state_sender\n            .send(InferenceMessage::ToDevice(device.clone()))\n            .expect(\"AsyncPolicy should be able to send policy state.\");\n        self\n    }","sourceCodeStart":282,"sourceCodeEnd":318,"githubUrl":"https://github.com/tracel-ai/burn/blob/d16f7ba2ed0d41408189384044cc886fb4c8f957/crates/burn-rl/src/policy/async_policy.rs#L282-L318","documentation":"AsyncPolicy::update sends a PolicyUpdate message to the autobatcher thread and panics if the send fails. A failed send means the receiving side is closed — the inference thread has already exited, typically because it panicked on an earlier message or saw RecvError after all AsyncPolicy handles were dropped. The policy update is silently lost as a panic instead of being applied.","triggerScenarios":"Calling update() after the server thread died: an earlier panic in the autobatcher loop (e.g. inside update_policy's flush, or inner_policy.update panicking on incompatible records), or the training driver dropped all other AsyncPolicy clones causing thread exit while a learner still calls update.","commonSituations":"Learner/actor split setups where the actor processes die and the learner's update() then panics on the dead channel; optimizer step errors (record mismatch, wrong device) crashing the server thread before subsequent updates arrive.","solutions":["Fix the original autobatcher-thread failure visible in logs before this call.","Recreate the AsyncPolicy (and re-apply the missed update) with AsyncPolicy::new after confirming the thread is dead.","Ensure inner_policy.update/PolicyState records match the running policy's device and architecture to avoid server-thread panics.","For robustness, have the server thread log-and-continue on recoverable message errors rather than unwinding."],"exampleFix":"// before\nself.inference_state_sender\n    .send(InferenceMessage::PolicyUpdate(update))\n    .expect(\"AsyncPolicy should be able to send policy state.\")\n// after\nif let Err(err) = self.inference_state_sender.send(InferenceMessage::PolicyUpdate(update)) {\n    log::error!(\"Failed to deliver policy update; inference server is down: {}\", err);\n    *self = AsyncPolicy::new(self.autobatch_size, self.rebuild_inner_policy());\n    // re-apply the update on the new server\n}","handlingStrategy":"try-catch","validationCode":"// Verify the server is responsive before pushing a policy update\nstd::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {\n    let mut p = policy.clone();\n    let _ = p.forward(sample_observation.clone());\n})).map_err(|_| anyhow::anyhow!(\"inference server dead; cannot apply update\"))?;","typeGuard":null,"tryCatchPattern":"if let Err(_panicked) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| policy.update(new_state))) {\n    log::error!(\"policy update lost: inference server is down; rebuilding and re-applying\");\n    policy = AsyncPolicy::new(autobatch_size, inner_policy.clone());\n    policy.update(new_state);\n}","preventionTips":["Ensure PolicyState records match the running policy's device/architecture to avoid server-thread panics during update_policy.","Snapshot/apply updates before dropping other AsyncPolicy handles.","Fix any earlier autobatcher-thread failure visible in logs before sending updates.","Consider buffering updates so a transient server death does not silently lose training progress."],"tags":["rust","mpsc-channel","panics","training"],"backgroundTag":"mpsc-receiver-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"}