AsyncPolicy should be able to receive policy state.

Error message

AsyncPolicy should be able to receive policy state.

What it means

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.

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.

Example fix

// before
receiver
    .recv()
    .expect("AsyncPolicy should be able to receive policy state.")
// after
receiver
    .recv_timeout(std::time::Duration::from_secs(30))
    .unwrap_or_else(|err| panic!("Did not receive policy state from inference server (thread dead or busy): {}", err))

Defensive patterns

Strategy: retry

Validate before calling

// Probe server liveness before blocking on a state request
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
    let mut p = policy.clone();
    let _ = p.forward(sample_observation.clone());
})).map_err(|_| anyhow::anyhow!("inference server not alive; state request would hang then panic"))?;

Try / catch

match receiver.recv_timeout(Duration::from_secs(30)) {
    Ok(state) => state,
    Err(RecvTimeoutError::Timeout) => return Err(anyhow::anyhow!("policy state request timed out")),
    Err(RecvTimeoutError::Disconnected) => return Err(anyhow::anyhow!("inference server died before replying with policy state")),
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of tracel-ai/burn@d16f7ba2ed (2026-09-05). Data as JSON: /api/errors/9140d9dfb28a2d29. Report an issue: GitHub.

Appendix: source

Thrown at crates/burn-rl/src/policy/async_policy.rs:310

            .recv()
            .expect("AsyncPolicy should receive queued actions.");
        (action.action, action.context)
    }

    fn update(&mut self, update: Self::PolicyState) {
        self.inference_state_sender
            .send(InferenceMessage::PolicyUpdate(update))
            .expect("AsyncPolicy should be able to send policy state.")
    }

    fn state(&self) -> Self::PolicyState {
        let (sender, receiver) = mpsc::channel();
        self.inference_state_sender
            .send(InferenceMessage::PolicyRequest(sender))
            .expect("should be able to send message to inference_server.");
        receiver
            .recv()
            .expect("AsyncPolicy should be able to receive policy state.")
    }

    fn to_device(self, device: &Device) -> Self {
        self.inference_state_sender
            .send(InferenceMessage::ToDevice(device.clone()))
            .expect("AsyncPolicy should be able to send policy state.");
        self
    }

    fn load_record(self, _record: <Self::PolicyState as PolicyState>::Record) -> Self {
        unimplemented!(
            "Not implemented yet. Please load the record on the inner policy before creating an async policy."
        )
    }
}

#[cfg(test)]
#[allow(clippy::needless_range_loop)]

View on GitHub (pinned to d16f7ba2ed)