lenso_runner/
lib.rs

1//! Native Tokio Runtime Driver for the Lenso vNext Kernel.
2
3use std::{
4    cell::Cell,
5    panic::AssertUnwindSafe,
6    rc::Rc,
7    time::{Duration, Instant},
8};
9
10use futures::{
11    channel::oneshot,
12    future::{AbortHandle, Abortable, FutureExt},
13};
14use lenso_app_plan::{PlanResolutionError, ResolvedAppPlan};
15use lenso_kernel::{
16    DriverTask, ExecutionAdapterCatalog, LocalTask, PlanValidationError, RuntimeDriver,
17    ShutdownOutcome, TaskOutcome, TerminalOutcome,
18};
19
20mod replicated;
21
22pub use replicated::{
23    CrossLaneRequestCatalog, LaneCancellationToken, LaneDiagnosticsSnapshot, LaneInvocationOptions,
24    ReplicatedNativeApp, ReplicatedRunnerError,
25};
26
27/// Tokio-backed Runtime Driver used by the native App Runner.
28#[derive(Clone, Debug)]
29pub struct TokioDriver {
30    started_at: Instant,
31    shutdown_requested: Rc<Cell<bool>>,
32    jitter_state: Rc<Cell<u64>>,
33}
34
35impl TokioDriver {
36    /// Creates a Driver bound to the current Tokio local task context.
37    pub fn new() -> Self {
38        Self::with_epoch(Instant::now())
39    }
40
41    pub(crate) fn with_epoch(started_at: Instant) -> Self {
42        Self {
43            started_at,
44            shutdown_requested: Rc::new(Cell::new(false)),
45            jitter_state: Rc::new(Cell::new(
46                u64::try_from(started_at.elapsed().as_nanos()).unwrap_or(u64::MAX)
47                    ^ 0x9e37_79b9_7f4a_7c15,
48            )),
49        }
50    }
51
52    /// Requests cooperative Kernel shutdown.
53    pub fn request_shutdown(&self) {
54        self.shutdown_requested.set(true);
55    }
56}
57
58impl Default for TokioDriver {
59    fn default() -> Self {
60        Self::new()
61    }
62}
63
64impl RuntimeDriver for TokioDriver {
65    fn now(&self) -> Duration {
66        self.started_at.elapsed()
67    }
68
69    fn sleep_until(&self, deadline: Duration) -> futures::future::LocalBoxFuture<'static, ()> {
70        let target = self.started_at + deadline;
71        Box::pin(async move {
72            tokio::time::sleep_until(tokio::time::Instant::from_std(target)).await;
73        })
74    }
75
76    fn yield_now(&self) -> futures::future::LocalBoxFuture<'static, ()> {
77        Box::pin(tokio::task::yield_now())
78    }
79
80    fn jitter(&self, maximum: Duration) -> Duration {
81        if maximum.is_zero() {
82            return Duration::ZERO;
83        }
84        let next = self
85            .jitter_state
86            .get()
87            .wrapping_mul(6_364_136_223_846_793_005)
88            .wrapping_add(1_442_695_040_888_963_407);
89        self.jitter_state.set(next);
90        let maximum_nanos = maximum.as_nanos().min(u128::from(u64::MAX));
91        let jitter_nanos = u128::from(next) % maximum_nanos.saturating_add(1);
92        Duration::from_nanos(u64::try_from(jitter_nanos).unwrap_or(u64::MAX))
93    }
94
95    fn spawn_local(&self, task: LocalTask) -> Result<DriverTask, futures::task::SpawnError> {
96        let (abort, registration) = AbortHandle::new_pair();
97        let (completed, completion) = oneshot::channel();
98        tokio::task::spawn_local(async move {
99            let outcome = match AssertUnwindSafe(Abortable::new(task, registration))
100                .catch_unwind()
101                .await
102            {
103                Ok(Ok(())) => TaskOutcome::Completed,
104                Ok(Err(_)) => TaskOutcome::Cancelled,
105                Err(_) => TaskOutcome::Failed,
106            };
107            let _ = completed.send(outcome);
108        });
109        Ok(DriverTask::new(abort, completion))
110    }
111
112    fn shutdown_requested(&self) -> bool {
113        self.shutdown_requested.get()
114    }
115}
116
117/// Runs an App through the Runner-assembled Adapter catalog until shutdown or failure.
118pub async fn run<D: RuntimeDriver>(
119    plan: ResolvedAppPlan,
120    driver: D,
121    adapters: ExecutionAdapterCatalog,
122    shutdown_timeout: Duration,
123) -> Result<TerminalOutcome, PlanValidationError> {
124    if let Err(error) = plan.validate() {
125        return Err(match error {
126            PlanResolutionError::UnsupportedSchemaVersion { expected, actual } => {
127                PlanValidationError::UnsupportedSchemaVersion { expected, actual }
128            }
129            error => PlanValidationError::InvalidResolvedPlan {
130                detail: error.to_string(),
131            },
132        });
133    }
134
135    let app = match lenso_kernel::Kernel::start(plan, driver.clone(), adapters).await {
136        Ok(app) => app,
137        Err(error) => return Ok(TerminalOutcome::StartupFailure { error }),
138    };
139    while !driver.shutdown_requested() && !app.is_failed() {
140        driver.yield_now().await;
141    }
142    if let Some(error) = app.terminal_failure() {
143        return Ok(match app.shutdown(shutdown_timeout).await {
144            ShutdownOutcome::Clean => TerminalOutcome::RuntimeFailure { error },
145            ShutdownOutcome::RuntimeFailure {
146                error: cleanup_error,
147            } => TerminalOutcome::RuntimeFailureDuringShutdown {
148                error,
149                cleanup_error,
150            },
151            ShutdownOutcome::Timeout => {
152                TerminalOutcome::RuntimeFailureWithShutdownTimeout { error }
153            }
154        });
155    }
156    Ok(match app.shutdown(shutdown_timeout).await {
157        ShutdownOutcome::Clean => TerminalOutcome::CleanShutdown,
158        ShutdownOutcome::RuntimeFailure { error } => TerminalOutcome::RuntimeFailure { error },
159        ShutdownOutcome::Timeout => TerminalOutcome::ShutdownTimeout,
160    })
161}
lenso_runner/lib.rs

lenso_runner/
lib.rs
