Method handle_worker_msg

components/scheduler/src/vm_session.rs:210–271

Handles a message received from the worker (`None` meaning the channel died), returning the events the guild handler needs to act on.

(&mut self, msg: Option<WorkerMessage>)

    /// Handles a message received from the worker (`None` meaning the channel
    /// died), returning the events the guild handler needs to act on.
    pub fn handle_worker_msg(&mut self, msg: Option<WorkerMessage>) -> Vec<SessionEvent> {
        let Some(msg) = msg else {
            return vec![SessionEvent::Broken];
        };

        match msg {
            WorkerMessage::Shutdown(shutdown) => self.handle_shutdown_msg(shutdown),
            WorkerMessage::Ack(id) => {
                let Some(item) = self.pending_acks.remove(&id) else {
                    return Vec::new();
                };

                if let Some(meta) = &item.dispatched {
                    metrics::histogram!(
                        "dispatch_event_acked_latency",
                        "event_source" => crate::dispatch_metrics::event_source_label(meta.source),
                        "vm_cold" => if meta.vm_cold { "true" } else { "false" },
                    )
                    .record(crate::dispatch_metrics::elapsed_millis(meta.source_timestamp));
                }

                match item.kind {
                    PendingAckType::Dispatch(Some(resp)) => {
                        let _ = resp.send(());
                        Vec::new()
                    }
                    PendingAckType::Dispatch(None) => Vec::new(),
                    PendingAckType::ScheduledTask(task_id) => {
                        vec![SessionEvent::TaskAcked(task_id)]
                    }
                    PendingAckType::IntervalTimer(timer) => vec![SessionEvent::TimerAcked(timer)],
                    PendingAckType::Restart => Vec::new(),
                }
            }
            WorkerMessage::ScriptStarted(meta) => vec![SessionEvent::ScriptStarted(meta)],
            WorkerMessage::ScriptsInit => todo!(),
            WorkerMessage::NonePending => {
                // the vm's event loop has drained, script init is done and
                // future dispatches run against a warm vm
                self.vm_cold = false;

                if self.pending_acks.is_empty() {
                    vec![SessionEvent::VmIdle]
                } else {
                    Vec::new()
                }
            }
            WorkerMessage::TaskScheduled => vec![SessionEvent::TaskScheduled],
            WorkerMessage::GuildLog(entry) => {
                self.logger.log_raw(entry);
                Vec::new()
            }
            WorkerMessage::Hello(_) => {
                // handled when the connection is established, not applicable here
                unreachable!();
            }
            WorkerMessage::Metric(name, m, labels) => {
                self.handle_metric(name, m, labels);