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

projectatlas/runtime/
optional_parser_runtime.rs
