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<Instant> {
66        self.run.deadline()
67    }
68
69    pub fn deadline_nanos(&self) -> Option<u64> {
70        let deadline = self.deadline()?;
71        let now = Instant::now();
72        let remaining =
73            u64::try_from(deadline.saturating_duration_since(now).as_nanos()).unwrap_or(u64::MAX);
74        Some(monotonic_nanos().saturating_add(remaining))
75    }
76
77    pub fn check_cancelled(&self) -> Result<(), PromiseRejection> {
78        self.token().check()
79    }
80
81    pub fn emit(&self, kind: Value, data: Value) -> bool {
82        self.run.emit(kind, data)
83    }
84
85    pub fn submit_child<F>(&self, options: WorkOptions, task: F) -> Result<WorkRun, String>
86    where
87        F: FnOnce(WorkContext) -> Result<Value, String> + 'static,
88    {
89        self.check_cancelled().map_err(|error| error.message())?;
90        let parent = if options.detached {
91            None
92        } else {
93            Some(self.run.clone())
94        };
95        self.host.submit_with_parent(
96            parent,
97            options,
98            Box::new(move |context| task(context).map_err(work_failure)),
99        )
100    }
101
102    pub fn submit_child_rejection<F>(
103        &self,
104        options: WorkOptions,
105        task: F,
106    ) -> Result<WorkRun, String>
107    where
108        F: FnOnce(WorkContext) -> Result<Value, PromiseRejection> + 'static,
109    {
110        self.check_cancelled().map_err(|error| error.message())?;
111        let parent = if options.detached {
112            None
113        } else {
114            Some(self.run.clone())
115        };
116        self.host
117            .submit_with_parent(parent, options, Box::new(task))
118    }
119
120    pub fn on_close<F>(&self, finalizer: F) -> bool
121    where
122        F: FnOnce(WorkContext) -> Result<(), PromiseRejection> + 'static,
123    {
124        self.run.register_finalizer(Box::new(finalizer))
125    }
126}
127
128thread_local! {
129    static PROCESS_WORK_HOST: WorkHost = WorkHost::new();
130    static CURRENT_WORK_CONTEXT: RefCell<Option<WorkContext>> = const { RefCell::new(None) };
131    static MONOTONIC_ORIGIN: Instant = Instant::now();
132}
133
134pub fn monotonic_nanos() -> u64 {
135    MONOTONIC_ORIGIN.with(|origin| origin.elapsed().as_nanos() as u64)
136}
137
138/// Return the process/evaluator-thread host shared by independent sessions.
139pub fn process_work_host() -> WorkHost {
140    PROCESS_WORK_HOST.with(Clone::clone)
141}
142
143/// Return the currently executing cooperative work context, if any.
144pub fn current_work_context() -> Option<WorkContext> {
145    CURRENT_WORK_CONTEXT.with(|current| current.borrow().clone())
146}
147
148pub(crate) fn with_current_work_context<T>(
149    context: WorkContext,
150    function: impl FnOnce() -> T,
151) -> T {
152    let previous = CURRENT_WORK_CONTEXT.with(|current| current.replace(Some(context)));
153    let result = function();
154    CURRENT_WORK_CONTEXT.with(|current| {
155        current.replace(previous);
156    });
157    result
158}
159
160pub(super) fn install_progress_hooks(host: &WorkHost, run: &WorkRun) {
161    let weak_host = Rc::downgrade(&host.inner);
162    let id = run.work_id();
163    run.inner.result.set_poller(Rc::new(move || {
164        if let Some(inner) = weak_host.upgrade() {
165            WorkHost { inner }.progress(&id);
166        }
167    }));
168
169    let weak_host = Rc::downgrade(&host.inner);
170    let id = run.work_id();
171    run.inner.result.set_waiter(Rc::new(move || {
172        if let Some(inner) = weak_host.upgrade() {
173            WorkHost { inner }.wait_for(&id);
174        }
175    }));
176
177    let weak_run = Rc::downgrade(&run.inner);
178    run.inner.result.set_cancel_hook(Rc::new(move || {
179        if let Some(inner) = weak_run.upgrade() {
180            WorkRun { inner }.cancel(Value::Keyword("result-cancelled".into()));
181        }
182    }));
183}
184
185pub(super) fn resolve_deadline(options: &WorkOptions, parent: Option<&WorkRun>) -> Option<Instant> {
186    let inherited = parent.and_then(WorkRun::deadline);
187    let relative = options
188        .timeout
189        .and_then(|timeout| Instant::now().checked_add(timeout));
190    [inherited, options.deadline, relative]
191        .into_iter()
192        .flatten()
193        .min()
194}
195
196pub(super) fn deadline_remaining_millis(deadline: Instant) -> u64 {
197    u64::try_from(
198        deadline
199            .saturating_duration_since(Instant::now())
200            .as_millis(),
201    )
202    .unwrap_or(u64::MAX)
203}
204
205pub(super) fn next_work_id(host: &mut WorkHostInner) -> Result<WorkId, String> {
206    loop {
207        let id = WorkId::new(format!("run-{}", host.next_id))?;
208        host.next_id = host
209            .next_id
210            .checked_add(1)
211            .ok_or_else(|| "work run identifiers exhausted".to_string())?;
212        if !host.runs.contains_key(&id) {
213            return Ok(id);
214        }
215    }
216}
217
218pub(super) fn work_failure(message: String) -> PromiseRejection {
219    PromiseRejection::Value(Value::Map(
220        [
221            (
222                Value::Keyword("code".into()),
223                Value::Keyword("work/failed".into()),
224            ),
225            (Value::Keyword("message".into()), Value::String(message)),
226            (Value::Keyword("retryable".into()), Value::Bool(false)),
227        ]
228        .into_iter()
229        .collect(),
230    ))
231}
232
233pub(super) fn cancellation_rejection(reason: Value) -> PromiseRejection {
234    PromiseRejection::Cancelled(Value::Map(
235        [
236            (
237                Value::Keyword("code".into()),
238                Value::Keyword("work/cancelled".into()),
239            ),
240            (Value::Keyword("reason".into()), reason),
241            (Value::Keyword("retryable".into()), Value::Bool(false)),
242        ]
243        .into_iter()
244        .collect(),
245    ))
246}
247
248pub(super) fn now_millis() -> u64 {
249    let millis = SystemTime::now()
250        .duration_since(UNIX_EPOCH)
251        .unwrap_or_default()
252        .as_millis();
253    u64::try_from(millis).unwrap_or(u64::MAX)
254}
hara_native/work/scope.rs

hara_native/work/
scope.rs
