lash_core/runtime/process/registry.rs
1use crate::plugin::PluginError;
2
3use super::events::{
4 ProcessAwaitOutput, ProcessEvent, ProcessEventAppendRequest, ProcessEventAppendResult,
5};
6use super::model::{
7 ProcessExternalRef, ProcessHandleDescriptor, ProcessHandleGrant, ProcessHandleGrantEntry,
8 ProcessLease, ProcessLeaseClaimOutcome, ProcessLeaseCompletion, ProcessListFilter,
9 ProcessRecord, ProcessRegistration, ProcessSessionDeleteReport, SessionScope, WaitState,
10};
11
12/// Outcome of [`ProcessRegistry::prune_terminal_processes`]: how many terminal
13/// process rows and event rows were physically deleted.
14#[derive(Clone, Debug, Default, PartialEq, Eq)]
15pub struct ProcessPruneReport {
16 /// Terminal process rows deleted.
17 pub pruned_processes: usize,
18 /// Event rows deleted across those processes.
19 pub pruned_events: usize,
20}
21
22/// Durability-neutral process registry.
23///
24/// Process waits are coordination behavior and live on
25/// [`ProcessWorkDriver`](crate::ProcessWorkDriver) /
26/// [`ProcessAwaiter`](crate::ProcessAwaiter), not on persistence
27/// implementations. Registry methods are point reads and writes only. See
28/// `docs/adr/0016-process-waits-live-on-the-work-driver-seam.md`.
29#[async_trait::async_trait]
30pub trait ProcessRegistry: Send + Sync {
31 /// Durability tier this process registry provides; defaults to
32 /// [`DurabilityTier`](crate::DurabilityTier)`::Inline`.
33 fn durability_tier(&self) -> crate::DurabilityTier {
34 crate::DurabilityTier::Inline
35 }
36
37 async fn register_process(
38 &self,
39 registration: ProcessRegistration,
40 ) -> Result<ProcessRecord, PluginError>;
41
42 /// Attach a durable backend reference to a registered process.
43 ///
44 /// Implementations must reject unknown process ids. The first assignment
45 /// stores the reference. Repeating the exact same assignment is an
46 /// idempotent no-op that returns the existing record unchanged. Assigning a
47 /// different reference after one has been stored is a registry model error.
48 async fn set_external_ref(
49 &self,
50 process_id: &str,
51 external_ref: ProcessExternalRef,
52 ) -> Result<ProcessRecord, PluginError>;
53
54 async fn grant_handle(
55 &self,
56 session_scope: &SessionScope,
57 process_id: &str,
58 descriptor: ProcessHandleDescriptor,
59 ) -> Result<ProcessHandleGrant, PluginError>;
60
61 async fn revoke_handle(
62 &self,
63 session_scope: &SessionScope,
64 process_id: &str,
65 ) -> Result<(), PluginError>;
66
67 async fn transfer_handle_grants(
68 &self,
69 from_scope: &SessionScope,
70 to_scope: &SessionScope,
71 process_ids: &[String],
72 ) -> Result<(), PluginError>;
73
74 async fn list_handle_grants(
75 &self,
76 session_scope: &SessionScope,
77 ) -> Result<Vec<ProcessHandleGrantEntry>, PluginError>;
78
79 async fn list_live_handle_grants(
80 &self,
81 session_scope: &SessionScope,
82 ) -> Result<Vec<ProcessHandleGrantEntry>, PluginError> {
83 Ok(self
84 .list_handle_grants(session_scope)
85 .await?
86 .into_iter()
87 .filter(|(_, record)| !record.is_terminal())
88 .collect())
89 }
90
91 async fn has_handle_grant(
92 &self,
93 session_scope: &SessionScope,
94 process_id: &str,
95 ) -> Result<bool, PluginError> {
96 Ok(self
97 .list_handle_grants(session_scope)
98 .await?
99 .into_iter()
100 .any(|(grant, _)| grant.process_id == process_id))
101 }
102
103 async fn handle_grants_for_process(
104 &self,
105 process_id: &str,
106 ) -> Result<Vec<ProcessHandleGrant>, PluginError>;
107
108 async fn delete_session_process_state(
109 &self,
110 session_id: &str,
111 ) -> Result<ProcessSessionDeleteReport, PluginError>;
112
113 async fn append_event(
114 &self,
115 process_id: &str,
116 request: ProcessEventAppendRequest,
117 ) -> Result<ProcessEventAppendResult, PluginError>;
118
119 async fn events_after(
120 &self,
121 process_id: &str,
122 after_sequence: u64,
123 ) -> Result<Vec<ProcessEvent>, PluginError>;
124
125 /// Count events of `event_type` with `sequence <= up_to_sequence`.
126 ///
127 /// This is the signal-ordinal query: the Nth occurrence of a signal event
128 /// resolves the Nth durable wait key. The default scans the event log;
129 /// store backends override it with a COUNT so per-signal cost stays flat
130 /// instead of growing with a long-lived process's history.
131 async fn count_events_through(
132 &self,
133 process_id: &str,
134 event_type: &str,
135 up_to_sequence: u64,
136 ) -> Result<u64, PluginError> {
137 Ok(self
138 .events_after(process_id, 0)
139 .await?
140 .into_iter()
141 .filter(|event| event.sequence <= up_to_sequence && event.event_type == event_type)
142 .count() as u64)
143 }
144
145 /// The most recent `limit` events, in ascending sequence order.
146 ///
147 /// Observation snapshots use this to show a bounded activity tail without
148 /// fetching a process's entire history on every poll. The default scans
149 /// the event log; store backends override it with ORDER BY ... LIMIT.
150 async fn recent_events(
151 &self,
152 process_id: &str,
153 limit: usize,
154 ) -> Result<Vec<ProcessEvent>, PluginError> {
155 let mut events = self.events_after(process_id, 0).await?;
156 if events.len() > limit {
157 events.drain(..events.len() - limit);
158 }
159 Ok(events)
160 }
161
162 async fn wake_events_after(
163 &self,
164 process_id: &str,
165 after_sequence: u64,
166 ) -> Result<Vec<ProcessEvent>, PluginError>;
167
168 async fn complete_process(
169 &self,
170 process_id: &str,
171 await_output: ProcessAwaitOutput,
172 ) -> Result<ProcessRecord, PluginError>;
173
174 async fn set_process_wait(
175 &self,
176 process_id: &str,
177 wait: WaitState,
178 ) -> Result<ProcessRecord, PluginError>;
179
180 async fn clear_process_wait(&self, process_id: &str) -> Result<ProcessRecord, PluginError>;
181
182 async fn get_process(&self, process_id: &str) -> Option<ProcessRecord>;
183
184 async fn list_processes(
185 &self,
186 filter: &ProcessListFilter,
187 ) -> Result<Vec<ProcessRecord>, PluginError>;
188
189 async fn ack_wake(&self, process_id: &str, sequence: u64) -> Result<(), PluginError>;
190
191 /// All non-terminal process records, in stable `process_id` order.
192 ///
193 /// This is the recovery sweep's worklist: every process that was started
194 /// but has not reached a terminal event is a candidate for re-execution by
195 /// a [`DurableProcessWorker`](crate::DurableProcessWorker) after a crash.
196 /// Terminal processes are excluded — they are already done and idempotent by
197 /// `process_id`, so re-running them would be wasted work.
198 async fn list_non_terminal(&self) -> Result<Vec<ProcessRecord>, PluginError>;
199
200 /// Claim the durable single-owner lease over a non-terminal process.
201 ///
202 /// An unexpired lease held by a *different* owner returns
203 /// [`ProcessLeaseClaimOutcome::Busy`] carrying the observed holder;
204 /// claiming a free or expired lease succeeds and bumps the
205 /// `fencing_token`, and the same incarnation re-entering its own live
206 /// lease extends it without changing token or fence. The returned
207 /// [`ProcessLease`]'s `(owner, lease_token)` plus `fencing_token` are the
208 /// contract a worker presents on every subsequent renew/complete — a stale
209 /// writer is rejected.
210 async fn claim_process_lease(
211 &self,
212 process_id: &str,
213 owner: &crate::LeaseOwnerIdentity,
214 lease_ttl_ms: u64,
215 ) -> Result<ProcessLeaseClaimOutcome, PluginError>;
216
217 /// Reclaim an unexpired process lease whose observed holder is definitely
218 /// dead according to persisted local-process liveness metadata.
219 ///
220 /// Mirrors
221 /// [`RuntimePersistence::reclaim_session_execution_lease`](crate::RuntimePersistence::reclaim_session_execution_lease):
222 /// backends must CAS on `observed_holder` (owner identity, lease token,
223 /// and fencing token) so a stale claimant cannot clear a newer live lease
224 /// that won the race after the busy observation, and a successful reclaim
225 /// must advance the fencing token monotonically.
226 async fn reclaim_process_lease(
227 &self,
228 process_id: &str,
229 owner: &crate::LeaseOwnerIdentity,
230 observed_holder: &ProcessLease,
231 lease_ttl_ms: u64,
232 ) -> Result<ProcessLeaseClaimOutcome, PluginError>;
233
234 /// Extend the expiry of a live lease the caller still owns.
235 ///
236 /// The lease must match the persisted `(owner, lease_token, fencing_token)`
237 /// and be unexpired, else the renewal is rejected (the lease was superseded
238 /// or expired). Workers renew across long-running effects so a healthy
239 /// process is not swept out from under its live owner.
240 async fn renew_process_lease(
241 &self,
242 lease: &ProcessLease,
243 lease_ttl_ms: u64,
244 ) -> Result<ProcessLease, PluginError>;
245
246 /// Release a lease the caller owns, fenced by the completion's
247 /// `(process_id, lease_token)`.
248 ///
249 /// Mirrors clearing a runtime turn lease: a stale completion (whose token no
250 /// longer matches the live lease) is a no-op so it cannot release a lease a
251 /// newer owner now holds. Idempotent — completing an already-released lease
252 /// succeeds.
253 async fn complete_process_lease(
254 &self,
255 completion: &ProcessLeaseCompletion,
256 ) -> Result<(), PluginError>;
257
258 /// Physically delete terminal process rows whose `updated_at_ms` is older
259 /// than `cutoff_epoch_ms`, together with their events, wake acks, handle
260 /// grants, and lease rows. Host-scheduled retention: hosts that project
261 /// results/events into their own store call this to keep the registry
262 /// bounded. Non-terminal rows are never touched. Callers must choose a
263 /// retention window comfortably longer than any waiter lifetime — a
264 /// pruned process id becomes "unknown process" to late awaits.
265 async fn prune_terminal_processes(
266 &self,
267 cutoff_epoch_ms: u64,
268 ) -> Result<ProcessPruneReport, PluginError>;
269}