{"record":{"id":"9140d9dfb28a2d29","repo":"tracel-ai/burn","slug":"asyncpolicy-should-be-able-to-receive-policy-state","errorCode":null,"errorMessage":"AsyncPolicy should be able to receive policy state.","messagePattern":"AsyncPolicy should be able to receive policy state\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/burn-rl/src/policy/async_policy.rs","lineNumber":310,"sourceCode":"            .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    }\n\n    fn load_record(self, _record: <Self::PolicyState as PolicyState>::Record) -> Self {\n        unimplemented!(\n            \"Not implemented yet. Please load the record on the inner policy before creating an async policy.\"\n        )\n    }\n}\n\n#[cfg(test)]\n#[allow(clippy::needless_range_loop)]","sourceCodeStart":292,"sourceCodeEnd":328,"githubUrl":"https://github.com/tracel-ai/burn/blob/d16f7ba2ed0d41408189384044cc886fb4c8f957/crates/burn-rl/src/policy/async_policy.rs#L292-L328","documentation":"After sending a PolicyRequest, AsyncPolicy::state blocks on receiver.recv() and panics if it returns Err — the reply Sender was dropped without sending a PolicyState. The autobatcher replies in its PolicyRequest arm, so Err means that arm never completed: the server thread panicked (e.g. inside autobatcher.state() or earlier in the loop) or exited before servicing the request.","triggerScenarios":"The autobatcher thread panics while computing state (e.g. inner_policy.state() moving tensors across devices fails) or on any earlier queued message, dropping the reply sender; or the thread exited its recv loop so the PolicyRequest sits unprocessed until teardown drops the sender held in the queued message.","commonSituations":"Periodic policy-state synchronization between learner and actors where a server-thread crash converts all waiters' recv into panics; snapshot logic combined with device migration (to_device) triggering errors inside state(); stalls from miscounted agents leaving requests queued while the process shuts down.","solutions":["Find the root panic in the autobatcher thread from logs and fix it (often in inner_policy.state() or an earlier message).","Use recv_timeout on the caller side with a clear error so a dead or stalled server produces a diagnosable failure instead of a bare channel panic.","Keep the server alive: replace expects in the loop with logged errors, and consider catch_unwind around autobatcher.state().","Take state snapshots before device moves/shutdown, and keep agent counts correct so requests are serviced promptly."],"exampleFix":"// before\nreceiver\n    .recv()\n    .expect(\"AsyncPolicy should be able to receive policy state.\")\n// after\nreceiver\n    .recv_timeout(std::time::Duration::from_secs(30))\n    .unwrap_or_else(|err| panic!(\"Did not receive policy state from inference server (thread dead or busy): {}\", err))","handlingStrategy":"retry","validationCode":"// Probe server liveness before blocking on a state request\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 not alive; state request would hang then panic\"))?;","typeGuard":null,"tryCatchPattern":"match receiver.recv_timeout(Duration::from_secs(30)) {\n    Ok(state) => state,\n    Err(RecvTimeoutError::Timeout) => return Err(anyhow::anyhow!(\"policy state request timed out\")),\n    Err(RecvTimeoutError::Disconnected) => return Err(anyhow::anyhow!(\"inference server died before replying with policy state\")),\n}","preventionTips":["Use recv_timeout to turn dead-server hangs into clear, catchable errors.","Investigate autobatcher.state()/inner_policy.state() panics (device moves, record mismatches) that drop the reply sender.","Keep the server thread alive by replacing expects with logged errors in the message loop.","Schedule state synchronization away from shutdown windows and stalled batch conditions."],"tags":["rust","mpsc-channel","panics","concurrency"],"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"}