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_bytes,
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_bytes(job, options, control) {
256                Ok(bytes) => {
257                    if let Ok(content) = String::from_utf8(bytes) {
258                        content
259                    } else {
260                        outcomes.push(SymbolParseOutcome::BinaryOrNonUtf8 {
261                            path: job.path.clone(),
262                        });
263                        continue;
264                    }
265                }
266                Err(outcome) => {
267                    outcomes.push(*outcome);
268                    return Ok(outcomes);
269                }
270            };
271            ensure_selection_current(&lifecycle, &expected_selection)?;
272            let deadline = optional_parse_deadline(control)?;
273            let result = match &mut runtime.state {
274                OptionalParserRuntimeState::Resident { verified } => {
275                    verified.supervisor_mut().parse(
276                        language,
277                        content.as_bytes(),
278                        limits,
279                        deadline,
280                        OPTIONAL_PARSE_NO_PROGRESS_TIMEOUT,
281                        control.cancellation(),
282                    )
283                }
284                OptionalParserRuntimeState::Inactive
285                | OptionalParserRuntimeState::UnavailableAfterCleanupFailure { .. } => {
286                    return Err(runtime_unavailable_error());
287                }
288            };
289            if let Err(error) = result {
290                retain_runtime_after_parser_failure(&mut runtime.state, &error);
291                return Err(supervisor_error(error));
292            }
293            let outcome = parse_admitted_symbol_job(
294                job,
295                &content,
296                Some(projectatlas_core::symbols::ParserKind::TreeSitter),
297                options,
298                control,
299            );
300            let terminal = terminal_parse_outcome(&outcome);
301            outcomes.push(outcome);
302            if terminal {
303                break;
304            }
305        }
306        Ok(outcomes)
307    })();
308    group_lease.release();
309    let cleanup = service_pending_deactivation(&mut runtime);
310    combine_optional_operation_and_cleanup(operation, cleanup)
311}
312
313/// Quarantine one concrete execution owner only when the supervisor reports uncertain cleanup.
314fn retain_runtime_after_parser_failure<T>(
315    state: &mut OptionalParserRuntimeState<T>,
316    error: &ParserSupervisorError,
317) {
318    if error.has_mandatory_cleanup_failure() {
319        state.retain_after_cleanup_failure();
320    }
321}
322
323/// Return the canonical optional grammar identity for one registry-owned job.
324fn optional_language_id(job: &SymbolParseJob) -> Option<&str> {
325    let capability = language_capability(job.language.as_deref()?)?;
326    (capability.optional_pack == Some(BROAD_PARSER_PACK_ID)).then_some(capability.id)
327}
328
329/// Keep each grammar contiguous and paths stable inside its worker session.
330fn sort_optional_jobs(jobs: &mut [(&str, &SymbolParseJob)]) {
331    jobs.sort_by(|(left_language, left_job), (right_language, right_job)| {
332        left_language
333            .cmp(right_language)
334            .then_with(|| left_job.path.cmp(&right_job.path))
335    });
336}
337
338/// Parse non-optional jobs without changing the existing Rayon ownership.
339fn parse_built_in_jobs(
340    pool: &rayon::ThreadPool,
341    jobs: &[&SymbolParseJob],
342    options: &SymbolBuildOptions,
343    control: &IndexWorkControl,
344) -> Result<Vec<SymbolParseOutcome>, CliError> {
345    let mut outcomes = Vec::with_capacity(jobs.len());
346    for batch in jobs.chunks(BUILT_IN_PARSE_BATCH_SIZE) {
347        control.check(IndexWorkStage::SymbolParsing)?;
348        outcomes.extend(pool.install(|| {
349            batch
350                .par_iter()
351                .map(|job| super::parse_symbol_job_controlled(job, options, control))
352                .collect::<Vec<_>>()
353        }));
354    }
355    Ok(outcomes)
356}
357
358/// Return whether later jobs must not start after this outcome.
359fn terminal_parse_outcome(outcome: &SymbolParseOutcome) -> bool {
360    matches!(
361        outcome,
362        SymbolParseOutcome::SourceChanged { .. }
363            | SymbolParseOutcome::Io { .. }
364            | SymbolParseOutcome::InvalidInput { .. }
365            | SymbolParseOutcome::IndexWork(_)
366    )
367}
368
369/// Acquire the one process-global owner without hiding cancellation behind a blocking lock.
370fn lock_optional_runtime(
371    control: &IndexWorkControl,
372) -> Result<MutexGuard<'static, OptionalParserRuntime>, CliError> {
373    let runtime = OPTIONAL_PARSER_RUNTIME.get_or_init(|| Mutex::new(OptionalParserRuntime::new()));
374    lock_runtime(runtime, control)
375}
376
377/// Poll one concrete coordinator lock under the caller's cancellation and deadline.
378fn lock_runtime<'a>(
379    runtime: &'a Mutex<OptionalParserRuntime>,
380    control: &IndexWorkControl,
381) -> Result<MutexGuard<'a, OptionalParserRuntime>, CliError> {
382    loop {
383        control.check(IndexWorkStage::SymbolParsing)?;
384        match runtime.try_lock() {
385            Ok(guard) => return Ok(guard),
386            Err(TryLockError::WouldBlock) => thread::park_timeout(OPTIONAL_RUNTIME_ADMISSION_POLL),
387            Err(TryLockError::Poisoned(poisoned)) => {
388                let mut guard = poisoned.into_inner();
389                if let Err(error) = guard.quarantine() {
390                    return Err(supervisor_error(error));
391                }
392                return Err(runtime_unavailable_error());
393            }
394        }
395    }
396}
397
398/// Request cleanup without making default-core wait behind an active optional group.
399fn deactivate_if_initialized(control: &IndexWorkControl) -> Result<(), CliError> {
400    let Some(runtime) = OPTIONAL_PARSER_RUNTIME.get() else {
401        return Ok(());
402    };
403    OPTIONAL_PARSER_DEACTIVATION_REQUESTED.store(true, Ordering::Release);
404    loop {
405        if OPTIONAL_PARSER_GROUP_ACTIVE.load(Ordering::Acquire) {
406            return Ok(());
407        }
408        control.check(IndexWorkStage::SymbolParsing)?;
409        match runtime.try_lock() {
410            Ok(mut guard) => return service_pending_deactivation(&mut guard),
411            // A selected group that begins while this caller polls owns the pending request and
412            // will service it before releasing its lease. Otherwise the lock release is imminent.
413            Err(TryLockError::WouldBlock) => thread::park_timeout(OPTIONAL_RUNTIME_ADMISSION_POLL),
414            Err(TryLockError::Poisoned(poisoned)) => {
415                let mut guard = poisoned.into_inner();
416                OPTIONAL_PARSER_DEACTIVATION_REQUESTED.store(false, Ordering::Release);
417                return guard.quarantine().map_err(supervisor_error);
418            }
419        }
420    }
421}
422
423/// Apply one pending cleanup transition while the caller owns the process-global state.
424fn service_pending_deactivation(runtime: &mut OptionalParserRuntime) -> Result<(), CliError> {
425    if !OPTIONAL_PARSER_DEACTIVATION_REQUESTED.swap(false, Ordering::AcqRel) {
426        return Ok(());
427    }
428    runtime.deactivate().map_err(supervisor_error)
429}
430
431/// Revalidate the project-local selection immediately before optional source admission.
432fn ensure_selection_current(
433    lifecycle: &OptionalParserPackLifecycle,
434    expected: &OptionalParserPackSelectionKey,
435) -> Result<(), CliError> {
436    let current = lifecycle.derive_project_selection()?;
437    if current.selection_key() == Some(expected) {
438        return Ok(());
439    }
440    Err(CliError::ParserPack(
441        OptionalParserPackLifecycleError::InvalidData {
442            reason: "optional parser selection changed before source transfer".to_owned(),
443        },
444    ))
445}
446
447/// Preserve both an operation failure and a mandatory process-cleanup failure.
448fn combine_optional_operation_and_cleanup<T>(
449    operation: Result<T, CliError>,
450    cleanup: Result<(), CliError>,
451) -> Result<T, CliError> {
452    match (operation, cleanup) {
453        (Ok(value), Ok(())) => Ok(value),
454        (Err(operation), Ok(())) => Err(operation),
455        (Ok(_), Err(cleanup)) => Err(cleanup),
456        (Err(operation), Err(cleanup)) => Err(CliError::OptionalParserOperationAndCleanup {
457            operation: Box::new(operation),
458            cleanup: Box::new(cleanup),
459        }),
460    }
461}
462
463/// Bound one worker request by both the operation deadline and the proven file ceiling.
464fn optional_parse_deadline(control: &IndexWorkControl) -> Result<Instant, CliError> {
465    control.check(IndexWorkStage::SymbolParsing)?;
466    let now = Instant::now();
467    let per_file = now.checked_add(OPTIONAL_PARSE_TIMEOUT).ok_or_else(|| {
468        CliError::IndexWork(IndexWorkFailure::DeadlineExceeded {
469            stage: IndexWorkStage::SymbolParsing,
470        })
471    })?;
472    Ok(control
473        .deadline()
474        .map_or(per_file, |deadline| deadline.min(per_file)))
475}
476
477/// Preserve task cancellation and deadline state across the supervisor boundary.
478fn supervisor_error(error: ParserSupervisorError) -> CliError {
479    match error {
480        ParserSupervisorError::Cancelled { .. } => {
481            CliError::IndexWork(IndexWorkFailure::Cancelled {
482                stage: IndexWorkStage::SymbolParsing,
483            })
484        }
485        ParserSupervisorError::DeadlineExceeded { .. } => {
486            CliError::IndexWork(IndexWorkFailure::DeadlineExceeded {
487                stage: IndexWorkStage::SymbolParsing,
488            })
489        }
490        other => CliError::ParserPack(OptionalParserPackLifecycleError::Supervisor(other)),
491    }
492}
493
494/// Construct the stable refusal used after cleanup uncertainty or lock poisoning.
495fn runtime_unavailable_error() -> CliError {
496    CliError::ParserPack(OptionalParserPackLifecycleError::InvalidData {
497        reason: "optional parser runtime is unavailable until process restart".to_owned(),
498    })
499}
500
501#[cfg(test)]
502mod tests {
503    use super::*;
504    use std::sync::atomic::AtomicUsize;
505
506    #[test]
507    fn optional_jobs_sort_by_canonical_grammar_then_path() {
508        let job = |path: &str, language: &str| SymbolParseJob {
509            path: path.to_owned(),
510            native_path: path.into(),
511            expected_content_hash: "a".repeat(64),
512            language: Some(language.to_owned()),
513            fallback_summary: None,
514            purpose_needs_suggestion: false,
515        };
516        let jobs = [
517            job("z.zig", "zig"),
518            job("z.awk", "awk"),
519            job("a.zig", "zig"),
520            job("a.awk", "awk"),
521        ];
522        let mut scheduled = [
523            ("zig", &jobs[0]),
524            ("awk", &jobs[1]),
525            ("zig", &jobs[2]),
526            ("awk", &jobs[3]),
527        ];
528        sort_optional_jobs(&mut scheduled);
529        assert_eq!(
530            scheduled
531                .iter()
532                .map(|(language, job)| (*language, job.path.as_str()))
533                .collect::<Vec<_>>(),
534            [
535                ("awk", "a.awk"),
536                ("awk", "z.awk"),
537                ("zig", "a.zig"),
538                ("zig", "z.zig"),
539            ]
540        );
541    }
542
543    #[test]
544    fn cleanup_failure_quarantines_the_concrete_runtime_state() {
545        let mut runtime = OptionalParserRuntime::new();
546        runtime.state =
547            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained: None };
548        assert!(runtime.deactivate().is_ok());
549        assert!(matches!(
550            runtime.state,
551            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { .. }
552        ));
553    }
554
555    #[test]
556    fn operation_and_cleanup_failures_remain_distinct_at_the_runtime_boundary() {
557        let operation = CliError::ParserPack(OptionalParserPackLifecycleError::InvalidData {
558            reason: "synthetic operation failure".to_owned(),
559        });
560        let cleanup = CliError::ParserPack(OptionalParserPackLifecycleError::Supervisor(
561            ParserSupervisorError::Cleanup {
562                message: "synthetic cleanup failure".to_owned(),
563            },
564        ));
565
566        let result = combine_optional_operation_and_cleanup::<()>(Err(operation), Err(cleanup));
567        assert!(matches!(
568            result,
569            Err(CliError::OptionalParserOperationAndCleanup { .. })
570        ));
571    }
572
573    #[test]
574    fn cleanup_failure_retains_execution_owner_until_runtime_state_drops() {
575        struct DropProbe<'a>(&'a AtomicUsize);
576
577        impl Drop for DropProbe<'_> {
578            fn drop(&mut self) {
579                self.0.fetch_add(1, Ordering::Relaxed);
580            }
581        }
582
583        let drops = AtomicUsize::new(0);
584        {
585            let mut state = OptionalParserRuntimeState::Resident {
586                verified: DropProbe(&drops),
587            };
588            state.retain_after_cleanup_failure();
589            assert!(matches!(
590                &state,
591                OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained: Some(_) }
592            ));
593            assert_eq!(drops.load(Ordering::Relaxed), 0);
594        }
595        assert_eq!(drops.load(Ordering::Relaxed), 1);
596    }
597
598    #[test]
599    fn combined_parser_failure_quarantines_and_retains_the_execution_owner() {
600        struct DropProbe<'a>(&'a AtomicUsize);
601
602        impl Drop for DropProbe<'_> {
603            fn drop(&mut self) {
604                self.0.fetch_add(1, Ordering::Relaxed);
605            }
606        }
607
608        let drops = AtomicUsize::new(0);
609        let mut state = OptionalParserRuntimeState::Resident {
610            verified: DropProbe(&drops),
611        };
612        let error = ParserSupervisorError::OperationAndCleanup {
613            operation: Box::new(ParserSupervisorError::Cancelled { phase: "test" }),
614            cleanup: Box::new(ParserSupervisorError::Cleanup {
615                message: "synthetic cleanup failure".to_owned(),
616            }),
617        };
618
619        retain_runtime_after_parser_failure(&mut state, &error);
620        assert!(matches!(
621            &state,
622            OptionalParserRuntimeState::UnavailableAfterCleanupFailure { retained: Some(_) }
623        ));
624        assert_eq!(drops.load(Ordering::Relaxed), 0);
625        drop(state);
626        assert_eq!(drops.load(Ordering::Relaxed), 1);
627    }
628
629    #[test]
630    fn pending_deactivation_is_consumed_under_the_runtime_lease() {
631        OPTIONAL_PARSER_DEACTIVATION_REQUESTED.store(true, Ordering::Release);
632        let mut runtime = OptionalParserRuntime::new();
633        assert!(service_pending_deactivation(&mut runtime).is_ok());
634        assert!(!OPTIONAL_PARSER_DEACTIVATION_REQUESTED.load(Ordering::Acquire));
635        assert!(matches!(
636            runtime.state,
637            OptionalParserRuntimeState::Inactive
638        ));
639    }
640
641    #[test]
642    fn group_activity_is_cleared_when_the_lease_drops() {
643        OPTIONAL_PARSER_GROUP_ACTIVE.store(false, Ordering::Release);
644        {
645            let _lease = OptionalParserGroupLease::acquire();
646            assert!(OPTIONAL_PARSER_GROUP_ACTIVE.load(Ordering::Acquire));
647        }
648        assert!(!OPTIONAL_PARSER_GROUP_ACTIVE.load(Ordering::Acquire));
649    }
650
651    #[test]
652    fn cancellation_interrupts_global_admission_wait() {
653        let runtime = Mutex::new(OptionalParserRuntime::new());
654        let _holder = runtime
655            .lock()
656            .unwrap_or_else(std::sync::PoisonError::into_inner);
657        let control = IndexWorkControl::new(projectatlas_core::IndexCancellation::new(), None);
658        control.cancel();
659        let result = lock_runtime(&runtime, &control);
660        assert!(matches!(
661            result,
662            Err(CliError::IndexWork(IndexWorkFailure::Cancelled {
663                stage: IndexWorkStage::SymbolParsing,
664            }))
665        ));
666    }
667}
projectatlas/runtime/optional_parser_runtime.rs

projectatlas/runtime/
optional_parser_runtime.rs
