hara_native/work/
scope.rs

1use super::*;
2
3/// Cooperative cancellation token exposed through the active work context.
4#[derive(Clone)]
5pub struct WorkCancellationToken {
6    pub(super) run: Weak<WorkRunInner>,
7}
8
9impl WorkCancellationToken {
10    pub fn cancelled(&self) -> bool {
11        self.run
12            .upgrade()
13            .is_none_or(|run| run.cancellation.borrow().is_some())
14    }
15
16    pub fn reason(&self) -> Option<Value> {
17        self.run.upgrade().and_then(|run| {
18            run.cancellation
19                .borrow()
20                .as_ref()
21                .map(|request| request.reason.clone())
22        })
23    }
24
25    pub fn check(&self) -> Result<(), PromiseRejection> {
26        let Some(run) = self.run.upgrade() else {
27            return Err(cancellation_rejection(Value::Keyword(
28                "scope-closed".into(),
29            )));
30        };
31        let run = WorkRun { inner: run };
32        run.check_deadline();
33        let result = match run.inner.cancellation.borrow().as_ref() {
34            Some(request) => Err(request.rejection.clone()),
35            None => Ok(()),
36        };
37        result
38    }
39}
40
41/// Opaque evaluator-thread context for one native work scope.
42#[derive(Clone)]
43pub struct WorkContext {
44    pub(super) host: WorkHost,
45    pub(super) run: WorkRun,
46}
47
48impl WorkContext {
49    pub fn work_id(&self) -> WorkId {
50        self.run.work_id()
51    }
52
53    pub fn token(&self) -> WorkCancellationToken {
54        self.run.cancellation_token()
55    }
56
57    pub fn cancelled(&self) -> bool {
58        self.token().cancelled()
59    }
60
61    pub fn cancel_reason(&self) -> Option<Value> {
62        self.token().reason()
63    }
64
65    pub fn deadline(&self) -> Option<WorkDeadline> {
66        self.run.deadline()
67    }
68
69    pub fn deadline_nanos(&self) -> Option<u64> {
70        self.deadline().map(WorkDeadline::monotonic_nanos)
71    }
72
73    pub fn check_cancelled(&self) -> Result<(), PromiseRejection> {
74        self.token().check()
75    }
76
77    pub fn emit(&self, kind: Value, data: Value) -> bool {
78        self.run.emit(kind, data)
79    }
80
81    pub fn submit_child<F>(&self, options: WorkOptions, task: F) -> Result<WorkRun, String>
82    where
83        F: FnOnce(WorkContext) -> Result<Value, String> + 'static,
84    {
85        self.check_cancelled().map_err(|error| error.message())?;
86        let parent = if options.detached {
87            None
88        } else {
89            Some(self.run.clone())
90        };
91        self.host.submit_with_parent(
92            parent,
93            options,
94            Box::new(move |context| task(context).map_err(work_failure)),
95        )
96    }
97
98    pub fn submit_child_rejection<F>(
99        &self,
100        options: WorkOptions,
101        task: F,
102    ) -> Result<WorkRun, String>
103    where
104        F: FnOnce(WorkContext) -> Result<Value, PromiseRejection> + 'static,
105    {
106        self.check_cancelled().map_err(|error| error.message())?;
107        let parent = if options.detached {
108            None
109        } else {
110            Some(self.run.clone())
111        };
112        self.host
113            .submit_with_parent(parent, options, Box::new(task))
114    }
115
116    pub fn on_close<F>(&self, finalizer: F) -> bool
117    where
118        F: FnOnce(WorkContext) -> Result<(), PromiseRejection> + 'static,
119    {
120        self.run.register_finalizer(Box::new(finalizer))
121    }
122}
123
124thread_local! {
125    static PROCESS_WORK_HOST: WorkHost = WorkHost::new();
126    static CURRENT_WORK_CONTEXT: RefCell<Option<WorkContext>> = const { RefCell::new(None) };
127}
128
129pub fn monotonic_nanos() -> u64 {
130    u64::try_from(crate::clock::time_ns()).unwrap_or_default()
131}
132
133/// Return the process/evaluator-thread host shared by independent sessions.
134pub fn process_work_host() -> WorkHost {
135    PROCESS_WORK_HOST.with(Clone::clone)
136}
137
138/// Return the currently executing cooperative work context, if any.
139pub fn current_work_context() -> Option<WorkContext> {
140    CURRENT_WORK_CONTEXT.with(|current| current.borrow().clone())
141}
142
143pub(crate) fn with_current_work_context<T>(
144    context: WorkContext,
145    function: impl FnOnce() -> T,
146) -> T {
147    let previous = CURRENT_WORK_CONTEXT.with(|current| current.replace(Some(context)));
148    let result = function();
149    CURRENT_WORK_CONTEXT.with(|current| {
150        current.replace(previous);
151    });
152    result
153}
154
155pub(super) fn install_progress_hooks(host: &WorkHost, run: &WorkRun) {
156    let weak_host = Rc::downgrade(&host.inner);
157    let id = run.work_id();
158    run.inner.result.set_poller(Rc::new(move || {
159        if let Some(inner) = weak_host.upgrade() {
160            WorkHost { inner }.progress(&id);
161        }
162    }));
163
164    let weak_host = Rc::downgrade(&host.inner);
165    let id = run.work_id();
166    run.inner.result.set_waiter(Rc::new(move || {
167        if let Some(inner) = weak_host.upgrade() {
168            WorkHost { inner }.wait_for(&id);
169        }
170    }));
171
172    let weak_run = Rc::downgrade(&run.inner);
173    run.inner.result.set_cancel_hook(Rc::new(move || {
174        if let Some(inner) = weak_run.upgrade() {
175            WorkRun { inner }.cancel(Value::Keyword("result-cancelled".into()));
176        }
177    }));
178}
179
180pub(super) fn resolve_deadline(
181    options: &WorkOptions,
182    parent: Option<&WorkRun>,
183) -> Option<WorkDeadline> {
184    let inherited = parent.and_then(WorkRun::deadline);
185    let relative = options.timeout.map(WorkDeadline::after);
186    [inherited, options.deadline, relative]
187        .into_iter()
188        .flatten()
189        .min()
190}
191
192pub(super) fn deadline_remaining_millis(deadline: WorkDeadline) -> u64 {
193    deadline.remaining_millis()
194}
195
196pub(super) fn next_work_id(host: &mut WorkHostInner) -> Result<WorkId, String> {
197    loop {
198        let id = WorkId::new(format!("run-{}", host.next_id))?;
199        host.next_id = host
200            .next_id
201            .checked_add(1)
202            .ok_or_else(|| "work run identifiers exhausted".to_string())?;
203        if !host.runs.contains_key(&id) {
204            return Ok(id);
205        }
206    }
207}
208
209pub(super) fn work_failure(message: String) -> PromiseRejection {
210    PromiseRejection::Value(Value::Map(
211        [
212            (
213                Value::Keyword("code".into()),
214                Value::Keyword("work/failed".into()),
215            ),
216            (Value::Keyword("message".into()), Value::String(message)),
217            (Value::Keyword("retryable".into()), Value::Bool(false)),
218        ]
219        .into_iter()
220        .collect(),
221    ))
222}
223
224pub(super) fn cancellation_rejection(reason: Value) -> PromiseRejection {
225    PromiseRejection::Cancelled(Value::Map(
226        [
227            (
228                Value::Keyword("code".into()),
229                Value::Keyword("work/cancelled".into()),
230            ),
231            (Value::Keyword("reason".into()), reason),
232            (Value::Keyword("retryable".into()), Value::Bool(false)),
233        ]
234        .into_iter()
235        .collect(),
236    ))
237}
238
239pub(super) fn now_millis() -> u64 {
240    u64::try_from(crate::clock::time_ms()).unwrap_or_default()
241}
hara_native/work/scope.rs

hara_native/work/
scope.rs
