lash_core/runtime/process/
registry.rs

1use crate::plugin::PluginError;
2
3use super::events::{
4    ProcessAwaitOutput, ProcessCompletionAuthority, ProcessEvent, ProcessEventAppendRequest,
5    ProcessEventAppendResult,
6};
7use super::model::{
8    AbandonRequest, ProcessChangeCursor, ProcessExternalRef, ProcessHandleDescriptor,
9    ProcessHandleGrant, ProcessHandleGrantEntry, ProcessLease, ProcessLeaseClaimOutcome,
10    ProcessLeaseCompletion, ProcessListFilter, ProcessRecord, ProcessRegistration,
11    ProcessSessionDeleteReport, ProcessStarted, SessionScope, WaitState,
12};
13use super::references::ProcessLiveReferenceSummary;
14
15/// Outcome of [`ProcessRegistry::prune_terminal_processes`]: how many terminal
16/// process rows and event rows were physically deleted.
17#[derive(Clone, Debug, Default, PartialEq, Eq)]
18pub struct ProcessPruneReport {
19    /// Terminal process rows deleted.
20    pub pruned_processes: usize,
21    /// Event rows deleted across those processes.
22    pub pruned_events: usize,
23}
24
25/// Durability-neutral process registry.
26///
27/// Process waits are coordination behavior and live on
28/// [`ProcessWorkDriver`](crate::ProcessWorkDriver) /
29/// [`ProcessAwaiter`](crate::ProcessAwaiter), not on persistence
30/// implementations. Registry methods are point reads and writes only. See
31/// `docs/adr/0016-process-waits-live-on-the-work-driver-seam.md`.
32#[async_trait::async_trait]
33pub trait ProcessRegistry: Send + Sync {
34    /// Durability tier this process registry provides; defaults to
35    /// [`DurabilityTier`](crate::DurabilityTier)`::Inline`.
36    fn durability_tier(&self) -> crate::DurabilityTier {
37        crate::DurabilityTier::Inline
38    }
39
40    async fn register_process(
41        &self,
42        registration: ProcessRegistration,
43    ) -> Result<ProcessRecord, PluginError>;
44
45    /// Attach a durable backend reference to a registered process.
46    ///
47    /// Implementations must reject unknown process ids. The first assignment
48    /// stores the reference. Repeating the exact same assignment is an
49    /// idempotent no-op that returns the existing record unchanged. Assigning a
50    /// different reference after one has been stored is a registry model error.
51    async fn set_external_ref(
52        &self,
53        process_id: &str,
54        external_ref: ProcessExternalRef,
55    ) -> Result<ProcessRecord, PluginError>;
56
57    async fn grant_handle(
58        &self,
59        session_scope: &SessionScope,
60        process_id: &str,
61        descriptor: ProcessHandleDescriptor,
62    ) -> Result<ProcessHandleGrant, PluginError>;
63
64    async fn revoke_handle(
65        &self,
66        session_scope: &SessionScope,
67        process_id: &str,
68    ) -> Result<(), PluginError>;
69
70    async fn transfer_handle_grants(
71        &self,
72        from_scope: &SessionScope,
73        to_scope: &SessionScope,
74        process_ids: &[String],
75    ) -> Result<(), PluginError>;
76
77    async fn list_handle_grants(
78        &self,
79        session_scope: &SessionScope,
80    ) -> Result<Vec<ProcessHandleGrantEntry>, PluginError>;
81
82    async fn list_live_handle_grants(
83        &self,
84        session_scope: &SessionScope,
85    ) -> Result<Vec<ProcessHandleGrantEntry>, PluginError> {
86        Ok(self
87            .list_handle_grants(session_scope)
88            .await?
89            .into_iter()
90            .filter(|(_, record)| !record.is_terminal())
91            .collect())
92    }
93
94    async fn has_handle_grant(
95        &self,
96        session_scope: &SessionScope,
97        process_id: &str,
98    ) -> Result<bool, PluginError> {
99        Ok(self
100            .list_handle_grants(session_scope)
101            .await?
102            .into_iter()
103            .any(|(grant, _)| grant.process_id == process_id))
104    }
105
106    async fn handle_grants_for_process(
107        &self,
108        process_id: &str,
109    ) -> Result<Vec<ProcessHandleGrant>, PluginError>;
110
111    async fn delete_session_process_state(
112        &self,
113        session_id: &str,
114    ) -> Result<ProcessSessionDeleteReport, PluginError>;
115
116    async fn append_event(
117        &self,
118        process_id: &str,
119        request: ProcessEventAppendRequest,
120    ) -> Result<ProcessEventAppendResult, PluginError>;
121
122    async fn events_after(
123        &self,
124        process_id: &str,
125        after_sequence: u64,
126    ) -> Result<Vec<ProcessEvent>, PluginError>;
127
128    /// Count events of `event_type` with `sequence <= up_to_sequence`.
129    ///
130    /// This is the signal-ordinal query: the Nth occurrence of a signal event
131    /// resolves the Nth durable wait key. The default scans the event log;
132    /// store backends override it with a COUNT so per-signal cost stays flat
133    /// instead of growing with a long-lived process's history.
134    async fn count_events_through(
135        &self,
136        process_id: &str,
137        event_type: &str,
138        up_to_sequence: u64,
139    ) -> Result<u64, PluginError> {
140        Ok(self
141            .events_after(process_id, 0)
142            .await?
143            .into_iter()
144            .filter(|event| event.sequence <= up_to_sequence && event.event_type == event_type)
145            .count() as u64)
146    }
147
148    /// The most recent `limit` events, in ascending sequence order.
149    ///
150    /// Observation snapshots use this to show a bounded activity tail without
151    /// fetching a process's entire history on every poll. The default scans
152    /// the event log; store backends override it with ORDER BY ... LIMIT.
153    async fn recent_events(
154        &self,
155        process_id: &str,
156        limit: usize,
157    ) -> Result<Vec<ProcessEvent>, PluginError> {
158        let mut events = self.events_after(process_id, 0).await?;
159        if events.len() > limit {
160            events.drain(..events.len() - limit);
161        }
162        Ok(events)
163    }
164
165    async fn wake_events_after(
166        &self,
167        process_id: &str,
168        after_sequence: u64,
169    ) -> Result<Vec<ProcessEvent>, PluginError>;
170
171    /// Complete a process without a Lash process lease, under an explicit,
172    /// auditable completion authority.
173    ///
174    /// This path is reserved for writers whose single-writer discipline lives
175    /// *outside* the Lash lease: an external actor closing an externally-owned
176    /// row, a workflow-key-coalesced substrate completing a row it ran, or the
177    /// sweep reconciling an abandon request. The
178    /// [`ProcessCompletionAuthority`] names which of these applies; the
179    /// implementation MUST call
180    /// [`authority.validate`](ProcessCompletionAuthority::validate) against the
181    /// row's declared [`RecoveryDisposition`](super::model::RecoveryDisposition)
182    /// inside this operation, so a mismatched authority is rejected with a typed
183    /// error before any terminal event is appended, and MUST record the
184    /// authority on the terminal event as audit evidence (via
185    /// [`terminal_append_request`](super::events::terminal_append_request)).
186    ///
187    /// Lash-owned workers must instead use
188    /// [`complete_process_with_lease`](Self::complete_process_with_lease), which
189    /// fences the terminal append and lease release in one atomic operation.
190    async fn complete_process(
191        &self,
192        process_id: &str,
193        await_output: ProcessAwaitOutput,
194        authority: ProcessCompletionAuthority,
195    ) -> Result<ProcessRecord, PluginError>;
196
197    /// Atomically append the terminal output while the supplied process lease
198    /// is still current, then release that lease in the same transaction.
199    ///
200    /// Implementations must validate owner incarnation, lease token, fencing
201    /// token, and expiry against the persisted lease. A stale or expired writer
202    /// is rejected without appending any terminal event or clearing a newer
203    /// owner's lease. Replaying the same terminal event after a successful
204    /// completion returns the existing terminal record.
205    async fn complete_process_with_lease(
206        &self,
207        lease: &ProcessLease,
208        await_output: ProcessAwaitOutput,
209    ) -> Result<ProcessRecord, PluginError>;
210
211    /// Record the durable, lease-fenced "execution started" fact (ADR 0019).
212    ///
213    /// First-writer-wins: the first call stores `started`; a later call is an
214    /// idempotent no-op returning the existing record unchanged (the fact is
215    /// immutable once written, so the sweep can prove an OwnerBound row has
216    /// begun executing). Implementations reject unknown process ids.
217    async fn record_first_started(
218        &self,
219        process_id: &str,
220        started: ProcessStarted,
221    ) -> Result<ProcessRecord, PluginError>;
222
223    /// Set the durable, non-terminal Abandon Request marker (ADR 0019).
224    ///
225    /// First-writer-wins: if a marker is already present the call is an
226    /// idempotent no-op returning the existing record unchanged, preserving the
227    /// original recorded authorization rather than letting a later requester
228    /// clobber it. Setting it on a terminal row is a model error — a terminal
229    /// process has already recorded its outcome, so there is nothing to abandon.
230    async fn request_process_abandon(
231        &self,
232        process_id: &str,
233        request: AbandonRequest,
234    ) -> Result<ProcessRecord, PluginError>;
235
236    async fn set_process_wait(
237        &self,
238        process_id: &str,
239        wait: WaitState,
240    ) -> Result<ProcessRecord, PluginError>;
241
242    async fn clear_process_wait(&self, process_id: &str) -> Result<ProcessRecord, PluginError>;
243
244    async fn get_process(&self, process_id: &str) -> Option<ProcessRecord>;
245
246    async fn list_processes(
247        &self,
248        filter: &ProcessListFilter,
249    ) -> Result<Vec<ProcessRecord>, PluginError>;
250
251    /// Return process records whose persisted row changed strictly after
252    /// `cursor`, ordered by the backend's per-store change sequence.
253    ///
254    /// This is a host-level completeness read for trusted projectors. It is not
255    /// scoped by handle grants, and the cursor must be treated as opaque outside
256    /// the store that issued it.
257    async fn processes_changed_since(
258        &self,
259        cursor: ProcessChangeCursor,
260        limit: usize,
261    ) -> Result<(Vec<ProcessRecord>, ProcessChangeCursor), PluginError>;
262
263    async fn ack_wake(&self, process_id: &str, sequence: u64) -> Result<(), PluginError>;
264
265    /// All non-terminal process records, in stable `process_id` order.
266    ///
267    /// This is the recovery sweep's worklist: every process that was started
268    /// but has not reached a terminal event is a candidate for re-execution by
269    /// a [`DurableProcessWorker`](crate::DurableProcessWorker) after a crash.
270    /// Terminal processes are excluded — they are already done and idempotent by
271    /// `process_id`, so re-running them would be wasted work.
272    async fn list_non_terminal(&self) -> Result<Vec<ProcessRecord>, PluginError>;
273
274    /// Count non-terminal process rows by their captured definition and
275    /// execution-environment references.
276    async fn live_reference_summary(&self)
277    -> Result<Vec<ProcessLiveReferenceSummary>, PluginError>;
278
279    /// Claim the durable single-owner lease over a non-terminal process.
280    ///
281    /// An unexpired lease held by a *different* owner returns
282    /// [`ProcessLeaseClaimOutcome::Busy`] carrying the observed holder;
283    /// claiming a free or expired lease succeeds and bumps the
284    /// `fencing_token`, and the same incarnation re-entering its own live
285    /// lease extends it without changing token or fence. The returned
286    /// [`ProcessLease`]'s `(owner, lease_token)` plus `fencing_token` are the
287    /// contract a worker presents on every subsequent renew/complete — a stale
288    /// writer is rejected.
289    async fn claim_process_lease(
290        &self,
291        process_id: &str,
292        owner: &crate::LeaseOwnerIdentity,
293        lease_ttl_ms: u64,
294    ) -> Result<ProcessLeaseClaimOutcome, PluginError>;
295
296    /// Reclaim an unexpired process lease whose observed holder is definitely
297    /// dead according to persisted local-process liveness metadata.
298    ///
299    /// Mirrors
300    /// [`RuntimePersistence::reclaim_session_execution_lease`](crate::RuntimePersistence::reclaim_session_execution_lease):
301    /// backends must CAS on `observed_holder` (owner identity, lease token,
302    /// and fencing token) so a stale claimant cannot clear a newer live lease
303    /// that won the race after the busy observation, and a successful reclaim
304    /// must advance the fencing token monotonically.
305    async fn reclaim_process_lease(
306        &self,
307        process_id: &str,
308        owner: &crate::LeaseOwnerIdentity,
309        observed_holder: &ProcessLease,
310        lease_ttl_ms: u64,
311    ) -> Result<ProcessLeaseClaimOutcome, PluginError>;
312
313    /// Extend the expiry of a live lease the caller still owns.
314    ///
315    /// The lease must match the persisted `(owner, lease_token, fencing_token)`
316    /// and be unexpired, else the renewal is rejected (the lease was superseded
317    /// or expired). Workers renew across long-running effects so a healthy
318    /// process is not swept out from under its live owner.
319    async fn renew_process_lease(
320        &self,
321        lease: &ProcessLease,
322        lease_ttl_ms: u64,
323    ) -> Result<ProcessLease, PluginError>;
324
325    /// Read the current lease row for a process without claiming it.
326    ///
327    /// Returns the persisted lease when one is held (owner and token present),
328    /// or `None` when the row is unleased or released. The returned lease may be
329    /// expired: expiry is a raw fact exposed read-side (ADR 0019) so hosts
330    /// classify staleness themselves; this never mutates the lease. Unknown
331    /// process ids return `None`.
332    async fn get_process_lease(
333        &self,
334        process_id: &str,
335    ) -> Result<Option<ProcessLease>, PluginError>;
336
337    /// Release a lease the caller owns, fenced by the completion's
338    /// `(process_id, lease_token)`.
339    ///
340    /// Mirrors clearing a runtime turn lease: a stale completion (whose token no
341    /// longer matches the live lease) is a no-op so it cannot release a lease a
342    /// newer owner now holds. Idempotent — completing an already-released lease
343    /// succeeds.
344    async fn complete_process_lease(
345        &self,
346        completion: &ProcessLeaseCompletion,
347    ) -> Result<(), PluginError>;
348
349    /// Physically delete terminal process rows whose `updated_at_ms` is older
350    /// than `cutoff_epoch_ms`, match `filter` when one is supplied, and have a
351    /// process change sequence no later than `up_to_change_seq` when supplied,
352    /// together with their events, wake acks, handle grants, lease rows, and
353    /// trigger-delivery reservations whose deterministic process id points at a
354    /// pruned row.
355    /// Host-scheduled retention: hosts that project results/events into their
356    /// own store call this to keep the registry bounded. Non-terminal rows are
357    /// never touched. Callers must choose a retention window comfortably longer
358    /// than any waiter lifetime — a pruned process id becomes "unknown process"
359    /// to late awaits. Re-emitting the same trigger occurrence id after its
360    /// process has aged out of retention may reserve a fresh delivery process
361    /// id; occurrence-level idempotency still holds, and ordinary emit replays
362    /// do not straddle a retention window in practice.
363    ///
364    /// ```no_run
365    /// use std::time::{Duration, SystemTime, UNIX_EPOCH};
366    /// use lash_core::{PluginError, ProcessRegistry};
367    ///
368    /// async fn prune_week_old(registry: &dyn ProcessRegistry) -> Result<(), PluginError> {
369    ///     let now_ms = SystemTime::now()
370    ///         .duration_since(UNIX_EPOCH)
371    ///         .expect("clock after epoch")
372    ///         .as_millis() as u64;
373    ///     // Window must exceed any in-flight await's lifetime (ADR 0017).
374    ///     let cutoff = now_ms - Duration::from_secs(7 * 24 * 60 * 60).as_millis() as u64;
375    ///     let report = registry.prune_terminal_processes(cutoff, None, None).await?;
376    ///     eprintln!(
377    ///         "pruned {} processes, {} events",
378    ///         report.pruned_processes, report.pruned_events
379    ///     );
380    ///     Ok(())
381    /// }
382    /// ```
383    async fn prune_terminal_processes(
384        &self,
385        cutoff_epoch_ms: u64,
386        filter: Option<ProcessListFilter>,
387        up_to_change_seq: Option<ProcessChangeCursor>,
388    ) -> Result<ProcessPruneReport, PluginError>;
389}
lash_core/runtime/process/registry.rs

lash_core/runtime/process/
registry.rs
