projectatlas/runtime/
optional_parser_runtime.rs

1//! Process-wide admission and scheduling for the optional parser-pack worker.
2
3use super::{
4    SymbolBuildOptions, SymbolParseJob, SymbolParseOutcome, admit_symbol_job_source,
5    parse_admitted_symbol_job,
6};
7use crate::CliError;
8use projectatlas_cli::optional_parser_lifecycle::{
9    OptionalParserPackLifecycle, OptionalParserPackLifecycleError,
10    OptionalParserPackProjectSelection, OptionalParserPackSelectionKey,
11    VerifiedOptionalParserPackSelection,
12};
13use projectatlas_cli::parser_supervisor::ParserSupervisorError;
14use projectatlas_core::language::{BROAD_PARSER_PACK_ID, language_capability};
15use projectatlas_core::optional_parser_protocol::{
16    PARSER_MAX_NODE_COUNT, PARSER_MAX_OUTPUT_BYTES, PARSER_MAX_TREE_DEPTH, ParserRequestLimits,
17};
18use projectatlas_core::{IndexWorkControl, IndexWorkFailure, IndexWorkStage};
19use rayon::prelude::*;
20use std::path::Path;
21use std::sync::atomic::{AtomicBool, Ordering};
22use std::sync::{Mutex, MutexGuard, OnceLock, TryLockError};
23use std::thread;
24use std::time::{Duration, Instant};
25
26/// Per-file ceiling proven by the fresh-artifact release verifier.
27const OPTIONAL_PARSE_TIMEOUT: Duration = Duration::from_secs(15);
28/// Maximum interval without identity-validated worker progress.
29const OPTIONAL_PARSE_NO_PROGRESS_TIMEOUT: Duration = Duration::from_secs(5);
30/// Cooperative polling interval while another project owns optional parsing.
31const OPTIONAL_RUNTIME_ADMISSION_POLL: Duration = Duration::from_millis(10);
32/// Maximum built-in jobs retained in one Rayon result batch.
33const BUILT_IN_PARSE_BATCH_SIZE: usize = 64;
34
35/// One process-wide optional worker owner shared by CLI, watcher, and MCP paths.
36static OPTIONAL_PARSER_RUNTIME: OnceLock<Mutex<OptionalParserRuntime>> = OnceLock::new();
37/// Whether an inactive project requested cleanup while another project held the worker lease.
38static OPTIONAL_PARSER_DEACTIVATION_REQUESTED: AtomicBool = AtomicBool::new(false);
39/// Whether the process-global coordinator is inside one optional parsing group.
40static OPTIONAL_PARSER_GROUP_ACTIVE: AtomicBool = AtomicBool::new(false);
41
42/// Closed process-wide optional parser state.
43enum OptionalParserRuntimeState<T = Box<VerifiedOptionalParserPackSelection>> {
44    /// No optional worker has been admitted.
45    Inactive,
46    /// One verified selection owns the current grammar-affined supervisor.
47    Resident {
48        /// Verified process owner retaining immutable-slot execution authority.
49        verified: T,
50    },
51    /// Cleanup could not be proved; retained authority blocks cross-process slot mutation.
52    UnavailableAfterCleanupFailure {
53        /// Verified selection retained until process exit when child cleanup is uncertain.
54        retained: Option<T>,
55    },
56}
57
58impl<T> OptionalParserRuntimeState<T> {
59    /// Quarantine the runtime without dropping an execution owner whose cleanup is uncertain.
60    fn retain_after_cleanup_failure(&mut self) {
61        let previous = std::mem::replace(
62            self,
63            Self::UnavailableAfterCleanupFailure { retained: None },
64        );
65        let retained = match previous {
66            Self::Resident { verified } => Some(verified),
67            Self::UnavailableAfterCleanupFailure { retained } => retained,
68            Self::Inactive => None,
69        };
70        *self = Self::UnavailableAfterCleanupFailure { retained };
71    }
72}
73
74/// Synchronous coordinator protected for exactly one optional job group at a time.
75struct OptionalParserRuntime {
76    /// Current process-wide worker state.
77    state: OptionalParserRuntimeState,
78}
79
80/// Panic-safe process-global optional-group activity signal.
81struct OptionalParserGroupLease {
82    /// Whether dropping this lease still needs to clear the activity signal.
83    active: bool,
84}
85
86impl OptionalParserGroupLease {
87    /// Publish one active group after the coordinator mutex has been acquired.
88    fn acquire() -> Self {
89        OPTIONAL_PARSER_GROUP_ACTIVE.store(true, Ordering::Release);
90        Self { active: true }
91    }
92
93    /// Clear activity before servicing a pending transition under the same mutex lease.
94    fn release(&mut self) {
95        OPTIONAL_PARSER_GROUP_ACTIVE.store(false, Ordering::Release);
96        self.active = false;
97    }
98}
99
100impl Drop for OptionalParserGroupLease {
101    fn drop(&mut self) {
102        if self.active {
103            OPTIONAL_PARSER_GROUP_ACTIVE.store(false, Ordering::Release);
104        }
105    }
106}
107
108impl OptionalParserRuntime {
109    /// Create the process-global coordinator without launching a worker.
110    const fn new() -> Self {
111        Self {
112            state: OptionalParserRuntimeState::Inactive,
113        }
114    }
115
116    /// Activate one verified selection, reusing only the exact same immutable artifact.
117    fn activate(
118        &mut self,
119        verified: VerifiedOptionalParserPackSelection,
120    ) -> Result<(), ParserSupervisorError> {
121        let selection = verified.selection_key().clone();
122        if matches!(
123            &self.state,
124            OptionalParserRuntimeState::Resident {
125                verified: current,
126            } if current.selection_key() == &selection
127        ) {
128            return Ok(());
129        }
130        if matches!(
131            self.state,
132            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { .. }
133        ) {
134            return Err(ParserSupervisorError::Cleanup {
135                message: "optional parser runtime is unavailable after an earlier cleanup failure"
136                    .to_owned(),
137            });
138        }
139        let previous = std::mem::replace(&mut self.state, OptionalParserRuntimeState::Inactive);
140        if let OptionalParserRuntimeState::Resident { mut verified } = previous
141            && let Err(error) = verified.supervisor_mut().shutdown()
142        {
143            self.state = OptionalParserRuntimeState::UnavailableAfterCleanupFailure {
144                retained: Some(verified),
145            };
146            return Err(error);
147        }
148        self.state = OptionalParserRuntimeState::Resident {
149            verified: Box::new(verified),
150        };
151        Ok(())
152    }
153
154    /// Shut down a resident selection while keeping default-core work available.
155    fn deactivate(&mut self) -> Result<(), ParserSupervisorError> {
156        let previous = std::mem::replace(&mut self.state, OptionalParserRuntimeState::Inactive);
157        match previous {
158            OptionalParserRuntimeState::Resident { mut verified } => {
159                match verified.supervisor_mut().shutdown() {
160                    Ok(()) => Ok(()),
161                    Err(error) => {
162                        self.state = OptionalParserRuntimeState::UnavailableAfterCleanupFailure {
163                            retained: Some(verified),
164                        };
165                        Err(error)
166                    }
167                }
168            }
169            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained } => {
170                self.state =
171                    OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained };
172                Ok(())
173            }
174            OptionalParserRuntimeState::Inactive => Ok(()),
175        }
176    }
177
178    /// Mark poisoned or cleanup-uncertain state unavailable after one best-effort shutdown.
179    fn quarantine(&mut self) -> Result<(), ParserSupervisorError> {
180        let cleanup = self.deactivate();
181        if cleanup.is_ok() {
182            self.state =
183                OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained: None };
184        }
185        cleanup
186    }
187}
188
189/// Parse built-in jobs in Rayon and accepted optional jobs in deterministic grammar/path order.
190pub(super) fn parse_symbol_jobs_controlled(
191    project_root: &Path,
192    project_selection: &OptionalParserPackProjectSelection,
193    pool: &rayon::ThreadPool,
194    jobs: &[SymbolParseJob],
195    options: &SymbolBuildOptions,
196    control: &IndexWorkControl,
197) -> Result<Vec<SymbolParseOutcome>, CliError> {
198    if matches!(
199        project_selection,
200        OptionalParserPackProjectSelection::Inactive
201    ) {
202        deactivate_if_initialized(control)?;
203        let built_in = jobs.iter().collect::<Vec<_>>();
204        return parse_built_in_jobs(pool, &built_in, options, control);
205    }
206
207    let lifecycle = OptionalParserPackLifecycle::new(project_root, None)?;
208    let verified = lifecycle.resolve_selected_pack()?.ok_or_else(|| {
209        CliError::ParserPack(OptionalParserPackLifecycleError::InvalidData {
210            reason: "optional parser selection disappeared before source staging".to_owned(),
211        })
212    })?;
213    if project_selection.selection_key() != Some(verified.selection_key()) {
214        return Err(CliError::ParserPack(
215            OptionalParserPackLifecycleError::InvalidData {
216                reason: "optional parser selection changed before source staging".to_owned(),
217            },
218        ));
219    }
220    let expected_selection = verified.selection_key().clone();
221
222    let mut built_in = Vec::with_capacity(jobs.len());
223    let mut optional = Vec::new();
224    for job in jobs {
225        match optional_language_id(job) {
226            Some(language) if verified.accepts_language(language) => {
227                optional.push((language, job));
228            }
229            _ => built_in.push(job),
230        }
231    }
232    sort_optional_jobs(&mut optional);
233
234    let mut outcomes = parse_built_in_jobs(pool, &built_in, options, control)?;
235    if outcomes.iter().any(terminal_parse_outcome) {
236        return Ok(outcomes);
237    }
238
239    let mut runtime = lock_optional_runtime(control)?;
240    let mut group_lease = OptionalParserGroupLease::acquire();
241    let operation = (|| {
242        service_pending_deactivation(&mut runtime)?;
243        ensure_selection_current(&lifecycle, &expected_selection)?;
244        runtime.activate(verified).map_err(supervisor_error)?;
245        let limits = ParserRequestLimits::new(
246            PARSER_MAX_OUTPUT_BYTES,
247            PARSER_MAX_NODE_COUNT,
248            PARSER_MAX_TREE_DEPTH,
249        )
250        .map_err(ParserSupervisorError::from)
251        .map_err(supervisor_error)?;
252
253        for (language, job) in optional {
254            control.check(IndexWorkStage::SymbolParsing)?;
255            let content = match admit_symbol_job_source(job, options, control) {
256                Ok(content) => content,
257                Err(outcome) if matches!(&*outcome, SymbolParseOutcome::BinaryOrNonUtf8 { .. }) => {
258                    outcomes.push(*outcome);
259                    continue;
260                }
261                Err(outcome) => {
262                    outcomes.push(*outcome);
263                    return Ok(outcomes);
264                }
265            };
266            ensure_selection_current(&lifecycle, &expected_selection)?;
267            let deadline = optional_parse_deadline(control)?;
268            let result = match &mut runtime.state {
269                OptionalParserRuntimeState::Resident { verified } => {
270                    verified.supervisor_mut().parse(
271                        language,
272                        content.as_bytes(),
273                        limits,
274                        deadline,
275                        OPTIONAL_PARSE_NO_PROGRESS_TIMEOUT,
276                        control.cancellation(),
277                    )
278                }
279                OptionalParserRuntimeState::Inactive
280                | OptionalParserRuntimeState::UnavailableAfterCleanupFailure { .. } => {
281                    return Err(runtime_unavailable_error());
282                }
283            };
284            if let Err(error) = result {
285                retain_runtime_after_parser_failure(&mut runtime.state, &error);
286                return Err(supervisor_error(error));
287            }
288            let outcome = parse_admitted_symbol_job(
289                job,
290                &content,
291                Some(projectatlas_core::symbols::ParserKind::TreeSitter),
292                options,
293                control,
294            );
295            let terminal = terminal_parse_outcome(&outcome);
296            outcomes.push(outcome);
297            if terminal {
298                break;
299            }
300        }
301        Ok(outcomes)
302    })();
303    group_lease.release();
304    let cleanup = service_pending_deactivation(&mut runtime);
305    combine_optional_operation_and_cleanup(operation, cleanup)
306}
307
308/// Quarantine one concrete execution owner only when the supervisor reports uncertain cleanup.
309fn retain_runtime_after_parser_failure<T>(
310    state: &mut OptionalParserRuntimeState<T>,
311    error: &ParserSupervisorError,
312) {
313    if error.has_mandatory_cleanup_failure() {
314        state.retain_after_cleanup_failure();
315    }
316}
317
318/// Return the canonical optional grammar identity for one registry-owned job.
319fn optional_language_id(job: &SymbolParseJob) -> Option<&str> {
320    let capability = language_capability(job.language.as_deref()?)?;
321    (capability.optional_pack == Some(BROAD_PARSER_PACK_ID)).then_some(capability.id)
322}
323
324/// Keep each grammar contiguous and paths stable inside its worker session.
325fn sort_optional_jobs(jobs: &mut [(&str, &SymbolParseJob)]) {
326    jobs.sort_by(|(left_language, left_job), (right_language, right_job)| {
327        left_language
328            .cmp(right_language)
329            .then_with(|| left_job.path.cmp(&right_job.path))
330    });
331}
332
333/// Parse non-optional jobs without changing the existing Rayon ownership.
334fn parse_built_in_jobs(
335    pool: &rayon::ThreadPool,
336    jobs: &[&SymbolParseJob],
337    options: &SymbolBuildOptions,
338    control: &IndexWorkControl,
339) -> Result<Vec<SymbolParseOutcome>, CliError> {
340    let mut outcomes = Vec::with_capacity(jobs.len());
341    for batch in jobs.chunks(BUILT_IN_PARSE_BATCH_SIZE) {
342        control.check(IndexWorkStage::SymbolParsing)?;
343        outcomes.extend(pool.install(|| {
344            batch
345                .par_iter()
346                .map(|job| super::parse_symbol_job_controlled(job, options, control))
347                .collect::<Vec<_>>()
348        }));
349    }
350    Ok(outcomes)
351}
352
353/// Return whether later jobs must not start after this outcome.
354fn terminal_parse_outcome(outcome: &SymbolParseOutcome) -> bool {
355    matches!(
356        outcome,
357        SymbolParseOutcome::SourceChanged { .. }
358            | SymbolParseOutcome::Io { .. }
359            | SymbolParseOutcome::IndexWork(_)
360    )
361}
362
363/// Acquire the one process-global owner without hiding cancellation behind a blocking lock.
364fn lock_optional_runtime(
365    control: &IndexWorkControl,
366) -> Result<MutexGuard<'static, OptionalParserRuntime>, CliError> {
367    let runtime = OPTIONAL_PARSER_RUNTIME.get_or_init(|| Mutex::new(OptionalParserRuntime::new()));
368    lock_runtime(runtime, control)
369}
370
371/// Poll one concrete coordinator lock under the caller's cancellation and deadline.
372fn lock_runtime<'a>(
373    runtime: &'a Mutex<OptionalParserRuntime>,
374    control: &IndexWorkControl,
375) -> Result<MutexGuard<'a, OptionalParserRuntime>, CliError> {
376    loop {
377        control.check(IndexWorkStage::SymbolParsing)?;
378        match runtime.try_lock() {
379            Ok(guard) => return Ok(guard),
380            Err(TryLockError::WouldBlock) => thread::park_timeout(OPTIONAL_RUNTIME_ADMISSION_POLL),
381            Err(TryLockError::Poisoned(poisoned)) => {
382                let mut guard = poisoned.into_inner();
383                if let Err(error) = guard.quarantine() {
384                    return Err(supervisor_error(error));
385                }
386                return Err(runtime_unavailable_error());
387            }
388        }
389    }
390}
391
392/// Request cleanup without making default-core wait behind an active optional group.
393fn deactivate_if_initialized(control: &IndexWorkControl) -> Result<(), CliError> {
394    let Some(runtime) = OPTIONAL_PARSER_RUNTIME.get() else {
395        return Ok(());
396    };
397    OPTIONAL_PARSER_DEACTIVATION_REQUESTED.store(true, Ordering::Release);
398    loop {
399        if OPTIONAL_PARSER_GROUP_ACTIVE.load(Ordering::Acquire) {
400            return Ok(());
401        }
402        control.check(IndexWorkStage::SymbolParsing)?;
403        match runtime.try_lock() {
404            Ok(mut guard) => return service_pending_deactivation(&mut guard),
405            // A selected group that begins while this caller polls owns the pending request and
406            // will service it before releasing its lease. Otherwise the lock release is imminent.
407            Err(TryLockError::WouldBlock) => thread::park_timeout(OPTIONAL_RUNTIME_ADMISSION_POLL),
408            Err(TryLockError::Poisoned(poisoned)) => {
409                let mut guard = poisoned.into_inner();
410                OPTIONAL_PARSER_DEACTIVATION_REQUESTED.store(false, Ordering::Release);
411                return guard.quarantine().map_err(supervisor_error);
412            }
413        }
414    }
415}
416
417/// Apply one pending cleanup transition while the caller owns the process-global state.
418fn service_pending_deactivation(runtime: &mut OptionalParserRuntime) -> Result<(), CliError> {
419    if !OPTIONAL_PARSER_DEACTIVATION_REQUESTED.swap(false, Ordering::AcqRel) {
420        return Ok(());
421    }
422    runtime.deactivate().map_err(supervisor_error)
423}
424
425/// Revalidate the project-local selection immediately before optional source admission.
426fn ensure_selection_current(
427    lifecycle: &OptionalParserPackLifecycle,
428    expected: &OptionalParserPackSelectionKey,
429) -> Result<(), CliError> {
430    let current = lifecycle.derive_project_selection()?;
431    if current.selection_key() == Some(expected) {
432        return Ok(());
433    }
434    Err(CliError::ParserPack(
435        OptionalParserPackLifecycleError::InvalidData {
436            reason: "optional parser selection changed before source transfer".to_owned(),
437        },
438    ))
439}
440
441/// Preserve both an operation failure and a mandatory process-cleanup failure.
442fn combine_optional_operation_and_cleanup<T>(
443    operation: Result<T, CliError>,
444    cleanup: Result<(), CliError>,
445) -> Result<T, CliError> {
446    match (operation, cleanup) {
447        (Ok(value), Ok(())) => Ok(value),
448        (Err(operation), Ok(())) => Err(operation),
449        (Ok(_), Err(cleanup)) => Err(cleanup),
450        (Err(operation), Err(cleanup)) => Err(CliError::OptionalParserOperationAndCleanup {
451            operation: Box::new(operation),
452            cleanup: Box::new(cleanup),
453        }),
454    }
455}
456
457/// Bound one worker request by both the operation deadline and the proven file ceiling.
458fn optional_parse_deadline(control: &IndexWorkControl) -> Result<Instant, CliError> {
459    control.check(IndexWorkStage::SymbolParsing)?;
460    let now = Instant::now();
461    let per_file = now.checked_add(OPTIONAL_PARSE_TIMEOUT).ok_or_else(|| {
462        CliError::IndexWork(IndexWorkFailure::DeadlineExceeded {
463            stage: IndexWorkStage::SymbolParsing,
464        })
465    })?;
466    Ok(control
467        .deadline()
468        .map_or(per_file, |deadline| deadline.min(per_file)))
469}
470
471/// Preserve task cancellation and deadline state across the supervisor boundary.
472fn supervisor_error(error: ParserSupervisorError) -> CliError {
473    match error {
474        ParserSupervisorError::Cancelled { .. } => {
475            CliError::IndexWork(IndexWorkFailure::Cancelled {
476                stage: IndexWorkStage::SymbolParsing,
477            })
478        }
479        ParserSupervisorError::DeadlineExceeded { .. } => {
480            CliError::IndexWork(IndexWorkFailure::DeadlineExceeded {
481                stage: IndexWorkStage::SymbolParsing,
482            })
483        }
484        other => CliError::ParserPack(OptionalParserPackLifecycleError::Supervisor(other)),
485    }
486}
487
488/// Construct the stable refusal used after cleanup uncertainty or lock poisoning.
489fn runtime_unavailable_error() -> CliError {
490    CliError::ParserPack(OptionalParserPackLifecycleError::InvalidData {
491        reason: "optional parser runtime is unavailable until process restart".to_owned(),
492    })
493}
494
495#[cfg(test)]
496mod tests {
497    use super::*;
498    use std::sync::atomic::AtomicUsize;
499
500    #[test]
501    fn optional_jobs_sort_by_canonical_grammar_then_path() {
502        let job = |path: &str, language: &str| SymbolParseJob {
503            path: path.to_owned(),
504            native_path: path.into(),
505            expected_content_hash: "a".repeat(64),
506            language: Some(language.to_owned()),
507            fallback_summary: None,
508            purpose_needs_suggestion: false,
509        };
510        let jobs = [
511            job("z.zig", "zig"),
512            job("z.awk", "awk"),
513            job("a.zig", "zig"),
514            job("a.awk", "awk"),
515        ];
516        let mut scheduled = [
517            ("zig", &jobs[0]),
518            ("awk", &jobs[1]),
519            ("zig", &jobs[2]),
520            ("awk", &jobs[3]),
521        ];
522        sort_optional_jobs(&mut scheduled);
523        assert_eq!(
524            scheduled
525                .iter()
526                .map(|(language, job)| (*language, job.path.as_str()))
527                .collect::<Vec<_>>(),
528            [
529                ("awk", "a.awk"),
530                ("awk", "z.awk"),
531                ("zig", "a.zig"),
532                ("zig", "z.zig"),
533            ]
534        );
535    }
536
537    #[test]
538    fn cleanup_failure_quarantines_the_concrete_runtime_state() {
539        let mut runtime = OptionalParserRuntime::new();
540        runtime.state =
541            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained: None };
542        assert!(runtime.deactivate().is_ok());
543        assert!(matches!(
544            runtime.state,
545            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { .. }
546        ));
547    }
548
549    #[test]
550    fn operation_and_cleanup_failures_remain_distinct_at_the_runtime_boundary() {
551        let operation = CliError::ParserPack(OptionalParserPackLifecycleError::InvalidData {
552            reason: "synthetic operation failure".to_owned(),
553        });
554        let cleanup = CliError::ParserPack(OptionalParserPackLifecycleError::Supervisor(
555            ParserSupervisorError::Cleanup {
556                message: "synthetic cleanup failure".to_owned(),
557            },
558        ));
559
560        let result = combine_optional_operation_and_cleanup::<()>(Err(operation), Err(cleanup));
561        assert!(matches!(
562            result,
563            Err(CliError::OptionalParserOperationAndCleanup { .. })
564        ));
565    }
566
567    #[test]
568    fn cleanup_failure_retains_execution_owner_until_runtime_state_drops() {
569        struct DropProbe<'a>(&'a AtomicUsize);
570
571        impl Drop for DropProbe<'_> {
572            fn drop(&mut self) {
573                self.0.fetch_add(1, Ordering::Relaxed);
574            }
575        }
576
577        let drops = AtomicUsize::new(0);
578        {
579            let mut state = OptionalParserRuntimeState::Resident {
580                verified: DropProbe(&drops),
581            };
582            state.retain_after_cleanup_failure();
583            assert!(matches!(
584                &state,
585                OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained: Some(_) }
586            ));
587            assert_eq!(drops.load(Ordering::Relaxed), 0);
588        }
589        assert_eq!(drops.load(Ordering::Relaxed), 1);
590    }
591
592    #[test]
593    fn combined_parser_failure_quarantines_and_retains_the_execution_owner() {
594        struct DropProbe<'a>(&'a AtomicUsize);
595
596        impl Drop for DropProbe<'_> {
597            fn drop(&mut self) {
598                self.0.fetch_add(1, Ordering::Relaxed);
599            }
600        }
601
602        let drops = AtomicUsize::new(0);
603        let mut state = OptionalParserRuntimeState::Resident {
604            verified: DropProbe(&drops),
605        };
606        let error = ParserSupervisorError::OperationAndCleanup {
607            operation: Box::new(ParserSupervisorError::Cancelled { phase: "test" }),
608            cleanup: Box::new(ParserSupervisorError::Cleanup {
609                message: "synthetic cleanup failure".to_owned(),
610            }),
611        };
612
613        retain_runtime_after_parser_failure(&mut state, &error);
614        assert!(matches!(
615            &state,
616            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained: Some(_) }
617        ));
618        assert_eq!(drops.load(Ordering::Relaxed), 0);
619        drop(state);
620        assert_eq!(drops.load(Ordering::Relaxed), 1);
621    }
622
623    #[test]
624    fn pending_deactivation_is_consumed_under_the_runtime_lease() {
625        OPTIONAL_PARSER_DEACTIVATION_REQUESTED.store(true, Ordering::Release);
626        let mut runtime = OptionalParserRuntime::new();
627        assert!(service_pending_deactivation(&mut runtime).is_ok());
628        assert!(!OPTIONAL_PARSER_DEACTIVATION_REQUESTED.load(Ordering::Acquire));
629        assert!(matches!(
630            runtime.state,
631            OptionalParserRuntimeState::Inactive
632        ));
633    }
634
635    #[test]
636    fn group_activity_is_cleared_when_the_lease_drops() {
637        OPTIONAL_PARSER_GROUP_ACTIVE.store(false, Ordering::Release);
638        {
639            let _lease = OptionalParserGroupLease::acquire();
640            assert!(OPTIONAL_PARSER_GROUP_ACTIVE.load(Ordering::Acquire));
641        }
642        assert!(!OPTIONAL_PARSER_GROUP_ACTIVE.load(Ordering::Acquire));
643    }
644
645    #[test]
646    fn cancellation_interrupts_global_admission_wait() {
647        let runtime = Mutex::new(OptionalParserRuntime::new());
648        let _holder = runtime
649            .lock()
650            .unwrap_or_else(std::sync::PoisonError::into_inner);
651        let control = IndexWorkControl::new(projectatlas_core::IndexCancellation::new(), None);
652        control.cancel();
653        let result = lock_runtime(&runtime, &control);
654        assert!(matches!(
655            result,
656            Err(CliError::IndexWork(IndexWorkFailure::Cancelled {
657                stage: IndexWorkStage::SymbolParsing,
658            }))
659        ));
660    }
661}
projectatlas/runtime/optional_parser_runtime.rs

projectatlas/runtime/
optional_parser_runtime.rs
