lash_core/runtime/
process_work_driver.rs

1use std::sync::Arc;
2
3use super::DurableProcessWorker;
4use super::process::{
5    ProcessAttach, ProcessAwaiter, ProcessChangeHub, ProcessEvent, ProcessEventSink,
6    ProcessRegistry, watch_process_registry_with_sink,
7};
8use crate::{PluginError, ProcessAwaitOutput};
9
10/// Registry and run handle for process work owned outside
11/// [`LashCore`](https://docs.rs/lash/latest/lash/struct.LashCore.html).
12///
13/// The registry non-terminal rows are the durable work queue. Hosts drive that
14/// queue explicitly by calling [`claim_and_run_pending`](Self::claim_and_run_pending)
15/// on each relevant event. Cross-process idempotency belongs to the registry
16/// claim; there is no core-owned polling loop.
17#[derive(Clone)]
18pub struct ProcessWorkDriver {
19    registry: Arc<dyn ProcessRegistry>,
20    run_handle: Arc<dyn ProcessRunHandle>,
21    awaiter: ProcessAwaiter,
22    attach: Option<Arc<dyn ProcessAttach>>,
23    hub: ProcessChangeHub,
24}
25
26impl ProcessWorkDriver {
27    pub fn new(registry: Arc<dyn ProcessRegistry>, run_handle: Arc<dyn ProcessRunHandle>) -> Self {
28        Self::new_with_sink(registry, run_handle, None)
29    }
30
31    /// Like [`new`](Self::new), but installs a host-facing
32    /// [`ProcessEventSink`] on the registry decorator this driver wraps.
33    ///
34    /// The sink receives every appended event, best-effort, after its durable
35    /// write — see [`ProcessEventSink`] for the freshness-not-truth contract.
36    pub fn new_with_sink(
37        registry: Arc<dyn ProcessRegistry>,
38        run_handle: Arc<dyn ProcessRunHandle>,
39        sink: Option<Arc<dyn ProcessEventSink>>,
40    ) -> Self {
41        let (registry, hub) = watch_process_registry_with_sink(registry, sink);
42        Self::from_watched(registry, hub, run_handle)
43    }
44
45    pub fn from_watched(
46        registry: Arc<dyn ProcessRegistry>,
47        hub: ProcessChangeHub,
48        run_handle: Arc<dyn ProcessRunHandle>,
49    ) -> Self {
50        let awaiter = ProcessAwaiter::new(Arc::clone(&registry), hub.clone());
51        Self {
52            registry,
53            run_handle,
54            awaiter,
55            attach: None,
56            hub,
57        }
58    }
59
60    pub fn with_attach(mut self, attach: Arc<dyn ProcessAttach>) -> Self {
61        self.attach = Some(attach);
62        self
63    }
64
65    pub fn inline(registry: Arc<dyn ProcessRegistry>, worker: DurableProcessWorker) -> Self {
66        Self::new(registry, Arc::new(InlineProcessRunHandle::new(worker)))
67    }
68
69    pub fn process_registry(&self) -> Arc<dyn ProcessRegistry> {
70        Arc::clone(&self.registry)
71    }
72
73    pub fn change_hub(&self) -> ProcessChangeHub {
74        self.hub.clone()
75    }
76
77    pub fn awaiter(&self) -> ProcessAwaiter {
78        self.awaiter.clone()
79    }
80
81    pub async fn await_terminal(
82        &self,
83        process_id: &str,
84    ) -> Result<ProcessAwaitOutput, PluginError> {
85        let record = self
86            .registry
87            .get_process(process_id)
88            .await
89            .ok_or_else(|| PluginError::Session(format!("unknown process `{process_id}`")))?;
90        if let Some(output) = record.status.await_output() {
91            return Ok(output.clone());
92        }
93        if let Some(attach) = self.attach.as_ref() {
94            return attach.await_terminal(process_id).await;
95        }
96        self.awaiter.await_terminal(process_id).await
97    }
98
99    pub async fn await_event(
100        &self,
101        process_id: &str,
102        event_type: &str,
103        after_sequence: u64,
104    ) -> Result<ProcessEvent, PluginError> {
105        self.awaiter
106            .await_event(process_id, event_type, after_sequence)
107            .await
108    }
109
110    pub async fn claim_and_run_pending(&self, reason: &str) -> Result<(), PluginError> {
111        if let Err(err) = self.run_handle.claim_and_run_pending().await {
112            tracing::warn!("process work drive ({reason}) failed: {err}");
113            return Err(err);
114        }
115        Ok(())
116    }
117}
118
119/// One lease-protected drive of the registry's pending (non-terminal) processes.
120///
121/// Implementations claim the single-owner [`ProcessLease`](crate::ProcessLease)
122/// per non-terminal row to fence execution, so a concurrent drive on another
123/// owner skips an already-leased process and a process runs exactly once.
124#[async_trait::async_trait]
125pub trait ProcessRunHandle: Send + Sync {
126    /// Claim and run every pending process this owner can claim, driving each to
127    /// a terminal state. Idempotent: leased and terminal rows are skipped.
128    async fn claim_and_run_pending(&self) -> Result<(), PluginError>;
129}
130
131/// Inline run handle: drives the worker's own lease-protected sweep in-process.
132///
133/// Delegates to [`DurableProcessWorker::drive_pending_processes`], the existing
134/// `list_non_terminal -> claim lease -> run -> complete -> release` loop, so the
135/// inline tier reuses the same coordination point as the durable tier.
136pub struct InlineProcessRunHandle {
137    worker: DurableProcessWorker,
138}
139
140impl InlineProcessRunHandle {
141    pub fn new(worker: DurableProcessWorker) -> Self {
142        Self { worker }
143    }
144}
145
146#[async_trait::async_trait]
147impl ProcessRunHandle for InlineProcessRunHandle {
148    async fn claim_and_run_pending(&self) -> Result<(), PluginError> {
149        self.worker.drive_pending_processes().await
150    }
151}
lash_core/runtime/process_work_driver.rs

lash_core/runtime/
process_work_driver.rs
