implements PLN-0115

crates/console/prometeu-system/src/lib.rs

@@ -10,6 +10,7 @@ pub use services::fs;
 pub use services::process;
 pub use services::task;
 pub use services::vm_runtime::{
+    RenderWorkerBackend, RenderWorkerConfig, RenderWorkerController, RenderWorkerFrameSink,
     RenderWorkerHandoff, RenderWorkerHandoffWait, VirtualMachineRuntime,
 };
 pub use services::windows;

crates/console/prometeu-system/src/services/vm_runtime/mod.rs

@@ -4,6 +4,7 @@ mod dispatch;
 mod frame_scheduler;
 mod lifecycle;
 pub mod render_manager;
+mod render_worker;
 pub mod render_worker_handoff;
 #[cfg(test)]
 pub(crate) mod render_worker_test_harness;
@@ -20,6 +21,9 @@ use prometeu_hal::telemetry::{AtomicTelemetry, CertificationConfig, Certifier};
 use prometeu_hal::{Gfx2dCommand, GfxUiCommand};
 use prometeu_vm::VirtualMachine;
 pub use render_manager::{RenderManager, RenderRuntimeCapabilities};
+pub use render_worker::{
+    RenderWorkerBackend, RenderWorkerConfig, RenderWorkerController, RenderWorkerFrameSink,
+};
 pub use render_worker_handoff::{RenderWorkerHandoff, RenderWorkerHandoffWait};
 use std::collections::HashMap;
 use std::sync::Arc;

crates/console/prometeu-system/src/services/vm_runtime/render_worker.rs

@@ -0,0 +1,291 @@
+use crate::services::vm_runtime::{RenderWorkerHandoff, RenderWorkerHandoffWait};
+use prometeu_hal::{
+    FrameId, OwnedRgba8888Frame, RenderSubmission, RenderWorkerError, RenderWorkerTelemetry,
+};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
+use std::sync::{Arc, Mutex};
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
+
+pub trait RenderWorkerBackend: Send + 'static {
+    fn render(
+        &self,
+        submission: &RenderSubmission,
+    ) -> Result<OwnedRgba8888Frame, RenderWorkerError>;
+}
+
+pub trait RenderWorkerFrameSink: Send + 'static {
+    fn publish(&self, frame: OwnedRgba8888Frame) -> Result<(), RenderWorkerError>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct RenderWorkerConfig {
+    pub shutdown_timeout: Duration,
+}
+
+impl Default for RenderWorkerConfig {
+    fn default() -> Self {
+        Self { shutdown_timeout: Duration::from_millis(250) }
+    }
+}
+
+#[derive(Debug, Default)]
+struct RenderWorkerState {
+    telemetry: RenderWorkerTelemetry,
+    last_error: Option<RenderWorkerError>,
+}
+
+#[derive(Debug)]
+pub struct RenderWorkerController {
+    config: RenderWorkerConfig,
+    handoff: Arc<RenderWorkerHandoff>,
+    state: Arc<Mutex<RenderWorkerState>>,
+    handle: Option<JoinHandle<()>>,
+    done_rx: Receiver<Result<(), RenderWorkerError>>,
+}
+
+impl RenderWorkerController {
+    pub fn start<B, S>(
+        config: RenderWorkerConfig,
+        handoff: Arc<RenderWorkerHandoff>,
+        backend: B,
+        sink: S,
+    ) -> Self
+    where
+        B: RenderWorkerBackend,
+        S: RenderWorkerFrameSink,
+    {
+        let state = Arc::new(Mutex::new(RenderWorkerState::default()));
+        let worker_handoff = Arc::clone(&handoff);
+        let worker_state = Arc::clone(&state);
+        let panic_state = Arc::clone(&state);
+        let (done_tx, done_rx) = mpsc::channel();
+
+        let handle = thread::spawn(move || {
+            let result = catch_unwind(AssertUnwindSafe(|| {
+                run_worker_loop(worker_handoff, worker_state, backend, sink);
+            }))
+            .map_err(|_| RenderWorkerError::WorkerPanic);
+            if let Err(error) = result {
+                record_error(&panic_state, error, FrameId::ZERO);
+            }
+            let _ = done_tx.send(result);
+        });
+
+        Self { config, handoff, state, handle: Some(handle), done_rx }
+    }
+
+    pub fn stop(&mut self) -> Result<(), RenderWorkerError> {
+        self.handoff.request_shutdown();
+        match self.done_rx.recv_timeout(self.config.shutdown_timeout) {
+            Ok(result) => {
+                if let Some(handle) = self.handle.take() {
+                    if handle.join().is_err() {
+                        record_error(&self.state, RenderWorkerError::WorkerPanic, FrameId::ZERO);
+                        return Err(RenderWorkerError::WorkerPanic);
+                    }
+                }
+                result
+            }
+            Err(RecvTimeoutError::Timeout) => {
+                record_error(&self.state, RenderWorkerError::ShutdownTimeout, FrameId::ZERO);
+                Err(RenderWorkerError::ShutdownTimeout)
+            }
+            Err(RecvTimeoutError::Disconnected) => {
+                record_error(&self.state, RenderWorkerError::InternalFailure, FrameId::ZERO);
+                Err(RenderWorkerError::InternalFailure)
+            }
+        }
+    }
+
+    pub fn telemetry(&self) -> RenderWorkerTelemetry {
+        let mut telemetry = self.handoff.telemetry();
+        let state = self.state.lock().unwrap();
+        telemetry.published_frames = state.telemetry.published_frames;
+        telemetry.render_failures = state.telemetry.render_failures;
+        telemetry.publish_failures = state.telemetry.publish_failures;
+        telemetry.present_failures = state.telemetry.present_failures;
+        telemetry.shutdown_timeouts = state.telemetry.shutdown_timeouts;
+        telemetry.worker_panics = state.telemetry.worker_panics;
+        telemetry.last_published_frame_id = state.telemetry.last_published_frame_id;
+        telemetry.last_error_frame_id = state.telemetry.last_error_frame_id;
+        telemetry
+    }
+
+    pub fn last_error(&self) -> Option<RenderWorkerError> {
+        self.state.lock().unwrap().last_error
+    }
+}
+
+fn run_worker_loop<B, S>(
+    handoff: Arc<RenderWorkerHandoff>,
+    state: Arc<Mutex<RenderWorkerState>>,
+    backend: B,
+    sink: S,
+) where
+    B: RenderWorkerBackend,
+    S: RenderWorkerFrameSink,
+{
+    loop {
+        let submission = match handoff.wait_take() {
+            RenderWorkerHandoffWait::Submission(submission) => submission,
+            RenderWorkerHandoffWait::Shutdown => return,
+        };
+        let frame_id = submission.frame_id;
+
+        let render_result = catch_unwind(AssertUnwindSafe(|| backend.render(&submission)))
+            .map_err(|_| RenderWorkerError::WorkerPanic);
+        let frame = match render_result {
+            Ok(Ok(frame)) => frame,
+            Ok(Err(error)) | Err(error) => {
+                record_error(&state, error, frame_id);
+                continue;
+            }
+        };
+
+        let publish_result = catch_unwind(AssertUnwindSafe(|| sink.publish(frame)))
+            .map_err(|_| RenderWorkerError::WorkerPanic);
+        match publish_result {
+            Ok(Ok(())) => record_published(&state, frame_id),
+            Ok(Err(error)) | Err(error) => record_error(&state, error, frame_id),
+        }
+    }
+}
+
+fn record_published(state: &Arc<Mutex<RenderWorkerState>>, frame_id: FrameId) {
+    state.lock().unwrap().telemetry.record_published(frame_id.get());
+}
+
+fn record_error(
+    state: &Arc<Mutex<RenderWorkerState>>,
+    error: RenderWorkerError,
+    frame_id: FrameId,
+) {
+    let mut state = state.lock().unwrap();
+    state.last_error = Some(error);
+    state.telemetry.record_error(error, frame_id.get());
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::services::vm_runtime::render_worker_test_harness::{
+        FakeRenderBackend, RenderGate, game_submission,
+    };
+    use prometeu_hal::{FrameId, OwnedRgba8888Frame};
+
+    impl RenderWorkerBackend for Arc<FakeRenderBackend> {
+        fn render(
+            &self,
+            submission: &RenderSubmission,
+        ) -> Result<OwnedRgba8888Frame, RenderWorkerError> {
+            FakeRenderBackend::render(self, submission)
+        }
+    }
+
+    impl RenderWorkerFrameSink for Arc<FakeRenderBackend> {
+        fn publish(&self, frame: OwnedRgba8888Frame) -> Result<(), RenderWorkerError> {
+            FakeRenderBackend::publish(self, frame);
+            Ok(())
+        }
+    }
+
+    #[derive(Debug, Default)]
+    struct PanickingBackend;
+
+    impl RenderWorkerBackend for PanickingBackend {
+        fn render(
+            &self,
+            _submission: &RenderSubmission,
+        ) -> Result<OwnedRgba8888Frame, RenderWorkerError> {
+            panic!("backend panic")
+        }
+    }
+
+    #[derive(Debug, Default)]
+    struct NoopSink;
+
+    impl RenderWorkerFrameSink for NoopSink {
+        fn publish(&self, _frame: OwnedRgba8888Frame) -> Result<(), RenderWorkerError> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn render_worker_controller_starts_and_stops_waiting_worker() {
+        let handoff = Arc::new(RenderWorkerHandoff::default());
+        let backend = Arc::new(FakeRenderBackend::default());
+        let mut controller = RenderWorkerController::start(
+            RenderWorkerConfig::default(),
+            Arc::clone(&handoff),
+            Arc::clone(&backend),
+            backend,
+        );
+
+        assert_eq!(controller.stop(), Ok(()));
+        assert_eq!(controller.last_error(), None);
+    }
+
+    #[test]
+    fn render_worker_controller_renders_and_publishes_submission() {
+        let handoff = Arc::new(RenderWorkerHandoff::default());
+        let backend = Arc::new(FakeRenderBackend::default());
+        let mut controller = RenderWorkerController::start(
+            RenderWorkerConfig::default(),
+            Arc::clone(&handoff),
+            Arc::clone(&backend),
+            Arc::clone(&backend),
+        );
+
+        handoff.publish(game_submission(11, 1, 1));
+        assert_eq!(controller.stop(), Ok(()));
+
+        let frames = backend.published_frames();
+        assert_eq!(frames.len(), 1);
+        assert_eq!(frames[0].frame_id, FrameId::new(11));
+        assert_eq!(controller.telemetry().published_frames, 1);
+    }
+
+    #[test]
+    fn render_worker_controller_reports_shutdown_timeout() {
+        let handoff = Arc::new(RenderWorkerHandoff::default());
+        let gate = RenderGate::default();
+        let backend = Arc::new(FakeRenderBackend::with_gate(gate.clone()));
+        let mut controller = RenderWorkerController::start(
+            RenderWorkerConfig { shutdown_timeout: Duration::from_millis(1) },
+            Arc::clone(&handoff),
+            Arc::clone(&backend),
+            Arc::clone(&backend),
+        );
+
+        handoff.publish(game_submission(12, 1, 1));
+        gate.wait_for_entries(1);
+
+        assert_eq!(controller.stop(), Err(RenderWorkerError::ShutdownTimeout));
+        assert_eq!(controller.last_error(), Some(RenderWorkerError::ShutdownTimeout));
+        assert_eq!(controller.telemetry().shutdown_timeouts, 1);
+
+        gate.release_one();
+        controller.config.shutdown_timeout = Duration::from_millis(250);
+        assert_eq!(controller.stop(), Ok(()));
+    }
+
+    #[test]
+    fn render_worker_controller_captures_backend_panic_as_typed_failure() {
+        let handoff = Arc::new(RenderWorkerHandoff::default());
+        let mut controller = RenderWorkerController::start(
+            RenderWorkerConfig::default(),
+            Arc::clone(&handoff),
+            PanickingBackend,
+            NoopSink,
+        );
+
+        handoff.publish(game_submission(13, 1, 1));
+        assert_eq!(controller.stop(), Ok(()));
+
+        assert_eq!(controller.last_error(), Some(RenderWorkerError::WorkerPanic));
+        assert_eq!(controller.telemetry().worker_panics, 1);
+        assert_eq!(controller.telemetry().last_error_frame_id, 13);
+    }
+}

discussion/workflow/plans/PLN-0115-implement-render-worker-controller-lifecycle.md

@@ -2,7 +2,7 @@
 id: PLN-0115
 ticket: real-render-worker-establishment
 title: Implement Render Worker Controller Lifecycle
-status: open
+status: done
 created: 2026-06-15
 completed:
 ref_decisions: [DEC-0033]