vyre_runtime/
persistent_executor.rs

1//! Persistent resident-work-queue execution over authenticated artifact sessions.
2
3use std::collections::BTreeMap;
4
5use vyre_driver::{BackendError, BackendRegistration, Completion, DeviceIdentity};
6use vyre_foundation::diagnostics::RetryClass;
7use vyre_megakernel::{ArtifactValueId, Digest};
8
9use crate::artifact_admission::{ArtifactSession, ArtifactSessionError, RetainedArtifactSession};
10use crate::recovery::classify_backend_error;
11
12/// Host-visible resident work-queue state.
13#[derive(Clone, Debug, PartialEq, Eq)]
14pub struct ResidentQueueState {
15    /// Queue control words.
16    pub control: Vec<u8>,
17    /// Packed work-slot ring.
18    pub ring: Vec<u8>,
19    /// Device debug-log storage.
20    pub debug_log: Vec<u8>,
21    /// Runtime IO request/completion queue.
22    pub io_queue: Vec<u8>,
23}
24
25/// Completed resident work-queue update.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub struct ResidentQueueCompletion {
28    /// Updated retained queue state.
29    pub state: ResidentQueueState,
30    /// Backend-measured device duration when available.
31    pub device_ns: Option<u64>,
32}
33
34#[derive(Clone, Copy)]
35struct ResidentQueueAbi {
36    control: ArtifactValueId,
37    ring: ArtifactValueId,
38    debug_log: ArtifactValueId,
39    io_queue: ArtifactValueId,
40}
41
42impl ResidentQueueAbi {
43    fn resolve(session: &ArtifactSession) -> Result<Self, ArtifactSessionError> {
44        Ok(Self {
45            control: session.resource("control")?,
46            ring: session.resource("ring_buffer")?,
47            debug_log: session.resource("debug_log")?,
48            io_queue: session.resource("io_queue")?,
49        })
50    }
51
52    fn bind(self, state: ResidentQueueState) -> BTreeMap<ArtifactValueId, Vec<u8>> {
53        BTreeMap::from([
54            (self.control, state.control),
55            (self.ring, state.ring),
56            (self.debug_log, state.debug_log),
57            (self.io_queue, state.io_queue),
58        ])
59    }
60
61    fn completion(
62        self,
63        completion: Completion,
64    ) -> Result<ResidentQueueCompletion, ArtifactSessionError> {
65        let mut retained = completion.retained;
66        let mut take = |value: ArtifactValueId,
67                        name: &str|
68         -> Result<Vec<u8>, ArtifactSessionError> {
69            retained.remove(&value).ok_or_else(|| {
70                BackendError::InvalidProgram {
71                    fix: format!(
72                        "Fix: persistent artifact completion must return retained queue resource `{name}`."
73                    ),
74                }
75                .into()
76            })
77        };
78        let state = ResidentQueueState {
79            control: take(self.control, "control")?,
80            ring: take(self.ring, "ring_buffer")?,
81            debug_log: take(self.debug_log, "debug_log")?,
82            io_queue: take(self.io_queue, "io_queue")?,
83        };
84        if !retained.is_empty() {
85            return Err(BackendError::InvalidProgram {
86                fix: "Fix: persistent queue artifacts must expose exactly control, ring_buffer, debug_log, and io_queue as retained values.".to_string(),
87            }
88            .into());
89        }
90        Ok(ResidentQueueCompletion {
91            state,
92            device_ns: completion.device_ns,
93        })
94    }
95}
96
97/// Authenticated persistent queue session over one immutable artifact identity.
98pub struct PersistentExecutor {
99    session: RetainedArtifactSession,
100    abi: ResidentQueueAbi,
101}
102
103impl PersistentExecutor {
104    /// Authenticate and materialize an artifact envelope, then initialize retained queue state.
105    pub fn from_bytes(
106        registration: &'static BackendRegistration,
107        envelope_bytes: &[u8],
108        initial: ResidentQueueState,
109    ) -> Result<Self, ArtifactSessionError> {
110        let session = ArtifactSession::from_bytes(registration, envelope_bytes)?;
111        let abi = ResidentQueueAbi::resolve(&session)?;
112        let session = RetainedArtifactSession::new(session, abi.bind(initial))?;
113        Ok(Self { session, abi })
114    }
115
116    /// Neutral artifact identity preserved across every queue update and recovery.
117    pub fn artifact(&self) -> Result<Digest, ArtifactSessionError> {
118        self.session.artifact()
119    }
120
121    /// Current acquired device generation.
122    pub fn device(&self) -> Result<DeviceIdentity, ArtifactSessionError> {
123        self.session.device()
124    }
125
126    /// Replace the host queue mirror, submit once, and return the updated retained state.
127    pub fn submit_and_wait(
128        &self,
129        state: ResidentQueueState,
130    ) -> Result<ResidentQueueCompletion, ArtifactSessionError> {
131        self.session.replace_retained(self.abi.bind(state))?;
132        let bindings = self.session.bindings()?;
133        let completion = self.session.submit_and_wait(bindings)?;
134        self.abi.completion(completion)
135    }
136
137    /// Rematerialize authenticated bytes after a structured device-loss failure.
138    ///
139    /// # Errors
140    ///
141    /// Returns the original backend failure for every non-device-loss class.
142    pub fn recover(&self, failure: BackendError) -> Result<DeviceIdentity, ArtifactSessionError> {
143        if classify_backend_error(&failure) != RetryClass::NewDevice {
144            return Err(failure.into());
145        }
146        self.session.rematerialize()
147    }
148}
vyre_runtime/persistent_executor.rs

vyre_runtime/
persistent_executor.rs
