strop_core/
worker.rs

1use parking_lot::Mutex;
2use std::sync::{
3    atomic::{AtomicBool, Ordering},
4    Arc,
5};
6
7pub mod cache_record;
8pub mod deploy_policy;
9mod effect;
10pub mod frame_policy;
11pub mod recovery_policy;
12pub mod session;
13pub use effect::spawn_effect;
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
16#[serde(transparent)]
17pub struct WorkerId(u64);
18impl WorkerId {
19    pub const fn new(value: u64) -> Self {
20        Self(value)
21    }
22    pub const fn get(self) -> u64 {
23        self.0
24    }
25}
26#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
27#[serde(transparent)]
28pub struct WorkerIds(u64);
29impl WorkerIds {
30    pub fn allocate(&mut self) -> Result<WorkerId, Failure> {
31        let next = self.0.checked_add(1).ok_or_else(|| {
32            Failure::new(
33                FailureKind::IdentityExhausted,
34                "worker request IDs exhausted",
35            )
36        })?;
37        self.0 = next;
38        Ok(WorkerId::new(next))
39    }
40}
41#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
42pub enum FailureKind {
43    IdentityExhausted,
44    ThreadStart,
45    Panic,
46    Io,
47    Spawn,
48    Wait,
49    Exit,
50    InvalidInput,
51    Unavailable,
52    Protocol,
53    Disconnected,
54}
55#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
56pub struct Failure {
57    pub kind: FailureKind,
58    pub message: String,
59}
60impl Failure {
61    pub fn new(kind: FailureKind, message: impl Into<String>) -> Self {
62        Self {
63            kind,
64            message: message.into(),
65        }
66    }
67}
68#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
69pub enum CancelReason {
70    Superseded,
71    OwnerClosed,
72    Dismissed,
73    Shutdown,
74}
75#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
76pub enum Outcome<T> {
77    Success(T),
78    Failed {
79        failure: Failure,
80        partial: Option<T>,
81    },
82    Cancelled(CancelReason),
83}
84impl<T> Outcome<T> {
85    pub fn failed(kind: FailureKind, message: impl Into<String>) -> Self {
86        Self::Failed {
87            failure: Failure::new(kind, message),
88            partial: None,
89        }
90    }
91}
92#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
93pub struct Ticket<K> {
94    pub request: WorkerId,
95    pub key: K,
96}
97#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
98pub struct Completion<K, T> {
99    pub ticket: Ticket<K>,
100    pub outcome: Outcome<T>,
101}
102#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
103pub enum Load<K> {
104    Idle,
105    Running(Ticket<K>),
106    Ready(K),
107    Failed { key: K, failure: Failure },
108    Cancelled { key: K, reason: CancelReason },
109}
110impl<K: PartialEq> Load<K> {
111    pub fn owns(&self, ticket: &Ticket<K>) -> bool {
112        matches!(self, Self::Running(current) if current == ticket)
113    }
114    pub fn covers(&self, key: &K) -> bool {
115        match self {
116            Self::Idle => false,
117            Self::Running(t) => &t.key == key,
118            Self::Ready(k) | Self::Failed { key: k, .. } | Self::Cancelled { key: k, .. } => {
119                k == key
120            }
121        }
122    }
123    pub fn retry_failed(&mut self) {
124        if matches!(self, Self::Failed { .. } | Self::Cancelled { .. }) {
125            *self = Self::Idle;
126        }
127    }
128}
129
130type Resource = Box<dyn FnOnce() -> Result<(), Failure> + Send>;
131#[derive(Default)]
132struct Cancellation {
133    cancelled: AtomicBool,
134    resource: Mutex<Option<Resource>>,
135}
136#[derive(Clone)]
137pub struct CancelToken(Arc<Cancellation>);
138fn invoke(resource: Resource) -> Result<(), Failure> {
139    match std::panic::catch_unwind(std::panic::AssertUnwindSafe(resource)) {
140        Ok(result) => result,
141        Err(_) => Err(Failure::new(
142            FailureKind::Panic,
143            "cancellation resource panicked",
144        )),
145    }
146}
147impl CancelToken {
148    pub fn is_cancelled(&self) -> bool {
149        self.0.cancelled.load(Ordering::Acquire)
150    }
151
152    /// A token plus the handle that cancels it, for owners running their
153    /// own scheduling loop rather than the engine's worker pool (0058
154    /// WK04: the worker session cancels admitted requests by protocol
155    /// `cancel` and lease teardown, not pool bookkeeping). Dropping the
156    /// handle without cancelling still cancels (OwnerClosed), matching
157    /// the pool's abandon semantics.
158    pub fn standalone() -> (Self, CancelHandle) {
159        let token = Self(Arc::new(Cancellation::default()));
160        let inner = token.clone();
161        let handle = CancelHandle {
162            cancel: Some(Box::new(move |_reason| {
163                let _ = inner.cancel_resource();
164            })),
165        };
166        (token, handle)
167    }
168
169    /// Install the single optional resource hook before acquiring native resources.
170    /// A late installation executes immediately and returns its failure to the
171    /// installer. Hooks must be prompt: they must not join or wait for workers.
172    /// A resource published after cancellation must also be revoked by its owner.
173    pub fn register_cancel_resource(
174        &self,
175        resource: impl FnOnce() -> Result<(), Failure> + Send + 'static,
176    ) -> Result<(), Failure> {
177        let resource: Resource = Box::new(resource);
178        {
179            let mut slot = self.0.resource.lock();
180            if !self.is_cancelled() {
181                if slot.is_some() {
182                    return Err(Failure::new(
183                        FailureKind::Protocol,
184                        "cancellation resource already registered",
185                    ));
186                }
187                *slot = Some(resource);
188                return Ok(());
189            }
190        }
191        invoke(resource)
192    }
193
194    /// Call only after normal resource cleanup. An already detached callback
195    /// may still run; the resource itself must serialize revocation and reuse.
196    pub fn clear_cancel_resource(&self) {
197        let resource = self.0.resource.lock().take();
198        drop(resource);
199    }
200
201    fn cancel_resource(&self) -> Result<(), Failure> {
202        let resource = {
203            let mut slot = self.0.resource.lock();
204            self.0.cancelled.store(true, Ordering::Release);
205            slot.take()
206        };
207        match resource {
208            Some(resource) => invoke(resource),
209            None => Ok(()),
210        }
211    }
212}
213type Emitter<T> = Arc<Mutex<Option<Box<dyn FnOnce(Outcome<T>) + Send>>>>;
214fn finish<T>(emitter: &Emitter<T>, outcome: Outcome<T>) {
215    let emit = emitter.lock().take();
216    if let Some(emit) = emit {
217        emit(outcome);
218    }
219}
220pub struct CancelHandle {
221    cancel: Option<Box<dyn FnOnce(CancelReason) + Send>>,
222}
223impl CancelHandle {
224    pub fn cancel(mut self, reason: CancelReason) {
225        if let Some(cancel) = self.cancel.take() {
226            cancel(reason);
227        }
228    }
229}
230impl Drop for CancelHandle {
231    fn drop(&mut self) {
232        if let Some(cancel) = self.cancel.take() {
233            cancel(CancelReason::OwnerClosed);
234        }
235    }
236}
237/// A registered work envelope without an eagerly spawned thread.
238/// Retain a cancellation handle until it runs or is explicitly cancelled.
239#[must_use]
240pub struct PreparedWork<T> {
241    token: CancelToken,
242    emitter: Emitter<T>,
243}
244
245pub fn prepare<T: Send + 'static>(
246    emit: impl FnOnce(Outcome<T>) + Send + 'static,
247) -> PreparedWork<T> {
248    PreparedWork {
249        token: CancelToken(Arc::new(Cancellation::default())),
250        emitter: Arc::new(Mutex::new(Some(Box::new(emit)))),
251    }
252}
253
254impl<T: Send + 'static> PreparedWork<T> {
255    /// Additional guards share one terminal reservation; dropping any live guard
256    /// cancels the work. A scheduler can retain its own guard while the UI owns one.
257    pub fn cancel_handle(&self) -> CancelHandle {
258        let cancel_token = self.token.clone();
259        let cancel_emitter = self.emitter.clone();
260        CancelHandle {
261            cancel: Some(Box::new(move |reason| {
262                let emit = cancel_emitter.lock().take();
263                if let Some(emit) = emit {
264                    let outcome = match cancel_token.cancel_resource() {
265                        Ok(()) => Outcome::Cancelled(reason),
266                        Err(failure) => Outcome::Failed {
267                            failure,
268                            partial: None,
269                        },
270                    };
271                    emit(outcome);
272                }
273            })),
274        }
275    }
276
277    pub fn run(self, work: impl FnOnce(CancelToken) -> Outcome<T>) {
278        if self.token.is_cancelled() {
279            return;
280        }
281        let outcome = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
282            work(self.token.clone())
283        })) {
284            Ok(outcome) => outcome,
285            Err(_) => match self.token.cancel_resource() {
286                Ok(()) => Outcome::failed(FailureKind::Panic, "worker panicked"),
287                Err(failure) => Outcome::Failed {
288                    failure,
289                    partial: None,
290                },
291            },
292        };
293        self.token.clear_cancel_resource();
294        finish(&self.emitter, outcome);
295    }
296}
297
298pub fn spawn<T: Send + 'static>(
299    name: &'static str,
300    emit: impl FnOnce(Outcome<T>) + Send + 'static,
301    work: impl FnOnce(CancelToken) -> Outcome<T> + Send + 'static,
302) -> CancelHandle {
303    let prepared = prepare(emit);
304    let handle = prepared.cancel_handle();
305    let emitter = prepared.emitter.clone();
306    if let Err(error) = std::thread::Builder::new()
307        .name(name.into())
308        .spawn(move || prepared.run(work))
309    {
310        finish(
311            &emitter,
312            Outcome::failed(FailureKind::ThreadStart, error.to_string()),
313        );
314    }
315    handle
316}
317
318/// Scoped readers use the same envelope, and their scope joins them before a
319/// provider can release physical ownership or start its successor.
320pub fn spawn_scoped<'scope, 'env, T: Send + 'static>(
321    scope: &'scope std::thread::Scope<'scope, 'env>,
322    name: &'static str,
323    emit: impl FnOnce(Outcome<T>) + Send + 'static,
324    work: impl FnOnce(CancelToken) -> Outcome<T> + Send + 'scope,
325) -> CancelHandle {
326    let prepared = prepare(emit);
327    let handle = prepared.cancel_handle();
328    let emitter = prepared.emitter.clone();
329    if let Err(error) = std::thread::Builder::new()
330        .name(name.into())
331        .spawn_scoped(scope, move || prepared.run(work))
332    {
333        finish(
334            &emitter,
335            Outcome::failed(FailureKind::ThreadStart, error.to_string()),
336        );
337    }
338    handle
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344    use std::sync::mpsc::channel;
345
346    #[test]
347    fn cancellation_reserves_terminal_before_callback_and_worker_success() {
348        let (ready_tx, ready_rx) = channel();
349        let (entered_tx, entered_rx) = channel();
350        let (release_tx, release_rx) = channel();
351        let (work_tx, work_rx) = channel();
352        let (done_tx, done_rx) = channel();
353        let (tx, rx) = channel();
354        let handle = spawn(
355            "race",
356            move |result| {
357                tx.send(result).unwrap();
358            },
359            move |token| {
360                token
361                    .register_cancel_resource(move || {
362                        entered_tx.send(()).unwrap();
363                        release_rx.recv().unwrap();
364                        Err(Failure::new(FailureKind::Io, "cleanup failed"))
365                    })
366                    .unwrap();
367                ready_tx.send(()).unwrap();
368                work_rx.recv().unwrap();
369                done_tx.send(()).unwrap();
370                Outcome::Success(())
371            },
372        );
373        ready_rx.recv().unwrap();
374        let cancel = std::thread::spawn(move || handle.cancel(CancelReason::Dismissed));
375        entered_rx.recv().unwrap();
376        work_tx.send(()).unwrap();
377        done_rx.recv().unwrap();
378        assert!(rx.try_recv().is_err());
379        release_tx.send(()).unwrap();
380        cancel.join().unwrap();
381        assert!(
382            matches!(rx.recv().unwrap(), Outcome::Failed { failure, .. } if failure.kind == FailureKind::Io)
383        );
384        assert!(rx.recv().is_err());
385    }
386
387    #[test]
388    fn late_registration_runs_immediately_and_success_wins_when_already_published() {
389        let token = CancelToken(Arc::new(Cancellation::default()));
390        token.cancel_resource().unwrap();
391        let (tx, rx) = channel();
392        token
393            .register_cancel_resource(move || {
394                tx.send(()).unwrap();
395                Ok(())
396            })
397            .unwrap();
398        rx.recv().unwrap();
399        let (tx, rx) = channel();
400        let handle = spawn(
401            "success",
402            move |result| {
403                tx.send(result).unwrap();
404            },
405            |_| Outcome::Success(7),
406        );
407        assert!(matches!(rx.recv().unwrap(), Outcome::Success(7)));
408        drop(handle);
409        assert!(rx.recv().is_err());
410    }
411
412    #[test]
413    fn queued_cancellation_prevents_work_from_starting() {
414        let (tx, rx) = channel();
415        let prepared: PreparedWork<()> = prepare(move |outcome| {
416            tx.send(outcome).unwrap();
417        });
418        prepared.cancel_handle().cancel(CancelReason::Superseded);
419        prepared.run(|_| panic!("cancelled queued work executed"));
420        assert!(matches!(
421            rx.recv().unwrap(),
422            Outcome::Cancelled(CancelReason::Superseded)
423        ));
424        assert!(rx.recv().is_err());
425    }
426
427    #[test]
428    fn scoped_worker_cancellation_does_not_skip_physical_join() {
429        let (control_tx, control_rx) = channel();
430        let (entered_tx, entered_rx) = channel();
431        let (release_tx, release_rx) = channel();
432        let (exited_tx, exited_rx) = channel();
433        let (tx, rx) = channel();
434        let scope_owner = std::thread::spawn(move || {
435            std::thread::scope(|scope| {
436                let handle = spawn_scoped(
437                    scope,
438                    "scoped-reader",
439                    move |outcome| {
440                        tx.send(outcome).unwrap();
441                    },
442                    move |_| {
443                        entered_tx.send(()).unwrap();
444                        release_rx.recv().unwrap();
445                        Outcome::Success(())
446                    },
447                );
448                assert!(control_tx.send(handle).is_ok());
449            });
450            exited_tx.send(()).unwrap();
451        });
452        let handle = control_rx.recv().unwrap();
453        entered_rx.recv().unwrap();
454        handle.cancel(CancelReason::Dismissed);
455        assert!(matches!(
456            rx.recv().unwrap(),
457            Outcome::Cancelled(CancelReason::Dismissed)
458        ));
459        assert!(exited_rx.try_recv().is_err());
460        release_tx.send(()).unwrap();
461        exited_rx.recv().unwrap();
462        scope_owner.join().unwrap();
463    }
464}
strop_core/worker.rs

strop_core/
worker.rs
