Skip to main content

player_plugin/
scope.rs

1//! Metadata-only plugin runtime scope lifecycle.
2
3mod playback;
4
5use std::num::NonZeroU64;
6use std::panic::{AssertUnwindSafe, catch_unwind};
7use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
8use std::sync::{Arc, Mutex, mpsc};
9use std::time::{Duration, Instant};
10
11use thiserror::Error;
12
13use crate::PluginPlan;
14
15pub use playback::{
16    MAX_PLUGIN_CORRELATION_ID_BYTES, PluginActivePlaybackCorrelation, PluginNextPrewarmCorrelation,
17    PluginPlaybackAttachment, PluginPlaybackAttachmentToken, PluginPlaybackAuthority,
18    PluginPlaybackError, PluginPlaybackRole, PluginPlaybackTransitionReport,
19    PluginSessionCorrelation,
20};
21
22/// Maximum diagnostic bytes retained for a scope failure reason.
23pub const MAX_PLUGIN_SCOPE_REASON_BYTES: usize = 512;
24/// Maximum direct children retained by one scope.
25pub const MAX_PLUGIN_SCOPE_CHILDREN: usize = 64;
26/// Maximum owner disposers retained by one scope.
27pub const MAX_PLUGIN_SCOPE_OWNERS: usize = 64;
28/// Maximum scope nesting below the runtime root.
29pub const MAX_PLUGIN_SCOPE_DEPTH: usize = 16;
30/// Maximum scope registrations during one runtime lifetime.
31pub const MAX_PLUGIN_RUNTIME_SCOPE_REGISTRATIONS: usize = 1_024;
32/// Maximum owner registrations during one runtime lifetime.
33pub const MAX_PLUGIN_RUNTIME_OWNER_REGISTRATIONS: usize = 1_024;
34/// Maximum detailed quarantine records retained in one close report.
35pub const MAX_PLUGIN_SCOPE_QUARANTINE_RECORDS: usize = 128;
36/// Default total deadline for direct scope settlement.
37pub const DEFAULT_PLUGIN_SCOPE_CLOSE_TIMEOUT: Duration = Duration::from_millis(250);
38/// Default total deadline used when a runtime is dropped.
39pub const DEFAULT_PLUGIN_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(500);
40
41#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
42pub enum PluginScopeKind {
43    Root,
44    Player,
45    Playback,
46    NextPrewarm,
47    Operation,
48    Worker,
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
52pub enum PluginScopeState {
53    Created,
54    Starting,
55    Running,
56    Draining,
57    Closed,
58    Failed,
59    Cancelled,
60    Quarantined,
61}
62
63/// Non-zero disposer identity that is unique within one `PluginRuntime`.
64#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
65pub struct PluginOwnerToken(NonZeroU64);
66
67impl PluginOwnerToken {
68    pub fn get(self) -> u64 {
69        self.0.get()
70    }
71}
72
73#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
74pub enum PluginScopeQuarantineReason {
75    Failed,
76    Panicked,
77    TimedOut,
78    WorkerUnavailable,
79}
80
81/// Signals that an owner cleanup completed but could not release its resource.
82///
83/// The scope records the typed failure without retaining an arbitrary external
84/// error payload. Boundary adapters remain responsible for their own detailed
85/// diagnostics.
86#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
87#[error("plugin owner cleanup failed")]
88pub struct PluginOwnerDisposalError;
89
90#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
91pub struct PluginScopeQuarantine {
92    pub owner_token: PluginOwnerToken,
93    pub reason: PluginScopeQuarantineReason,
94}
95
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
97pub enum PluginScopeResource {
98    Children,
99    Owners,
100    Depth,
101    ScopeRegistrations,
102    OwnerRegistrations,
103    ActivePlaybackSlot,
104    NextPrewarmSlot,
105}
106
107/// Bounded aggregate from one settlement attempt.
108///
109/// A `Quarantined` final state means cleanup was isolated after a panic,
110/// timeout, worker failure, or concurrent close. It does not prove that the
111/// quarantined resource was released.
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub struct PluginScopeCloseReport {
114    pub children_closed: usize,
115    pub children_quarantined: usize,
116    pub busy_scopes_quarantined: usize,
117    pub disposers_run: usize,
118    pub owners_settled: usize,
119    pub owners_quarantined: usize,
120    pub disposer_failures: usize,
121    pub disposer_panics: usize,
122    pub disposer_timeouts: usize,
123    pub disposer_worker_failures: usize,
124    pub quarantined_owners: Vec<PluginScopeQuarantine>,
125    pub quarantine_records_dropped: usize,
126    pub final_state: PluginScopeState,
127}
128
129impl Default for PluginScopeCloseReport {
130    fn default() -> Self {
131        Self {
132            children_closed: 0,
133            children_quarantined: 0,
134            busy_scopes_quarantined: 0,
135            disposers_run: 0,
136            owners_settled: 0,
137            owners_quarantined: 0,
138            disposer_failures: 0,
139            disposer_panics: 0,
140            disposer_timeouts: 0,
141            disposer_worker_failures: 0,
142            quarantined_owners: Vec::new(),
143            quarantine_records_dropped: 0,
144            final_state: PluginScopeState::Closed,
145        }
146    }
147}
148
149#[derive(Debug, Error, Clone, PartialEq, Eq)]
150pub enum PluginScopeError {
151    #[error("plugin scope `{kind:?}` cannot transition from {state:?}")]
152    InvalidTransition {
153        kind: PluginScopeKind,
154        state: PluginScopeState,
155    },
156    #[error("plugin scope `{kind:?}` is already terminal in {state:?}")]
157    Terminal {
158        kind: PluginScopeKind,
159        state: PluginScopeState,
160    },
161    #[error("plugin scope close is already in progress")]
162    Busy,
163    #[error("plugin scope failure reason must contain 1 to {limit} UTF-8 bytes")]
164    InvalidFailureReason { limit: usize },
165    #[error("plugin scope {resource:?} capacity exceeds {limit}")]
166    CapacityExceeded {
167        resource: PluginScopeResource,
168        limit: usize,
169    },
170    #[error("plugin scope owner token space is exhausted")]
171    OwnerTokenExhausted,
172    #[error("plugin scope close timeout cannot be represented by the monotonic clock")]
173    InvalidCloseTimeout,
174}
175
176/// Runtime owner for one immutable plan and its root scope.
177pub struct PluginRuntime {
178    plan: PluginPlan,
179    root: PluginScope,
180    playback: Mutex<playback::PluginPlaybackSlots>,
181}
182
183impl PluginRuntime {
184    pub fn new(plan: PluginPlan) -> Self {
185        Self {
186            plan,
187            root: PluginScope::new_root(),
188            playback: Mutex::new(playback::PluginPlaybackSlots::default()),
189        }
190    }
191
192    pub fn plan(&self) -> &PluginPlan {
193        &self.plan
194    }
195
196    pub fn root_scope(&self) -> PluginScope {
197        self.root.clone()
198    }
199
200    /// Closes the root with one total deadline shared by the complete scope tree.
201    ///
202    /// An already draining root is marked `Quarantined` so shutdown never waits
203    /// on an unbounded concurrent close.
204    pub fn shutdown(&self, timeout: Duration) -> Result<PluginScopeCloseReport, PluginScopeError> {
205        self.begin_playback_shutdown();
206        self.root.close_with_timeout(timeout)
207    }
208}
209
210impl Drop for PluginRuntime {
211    fn drop(&mut self) {
212        let _ = self.shutdown(DEFAULT_PLUGIN_RUNTIME_SHUTDOWN_TIMEOUT);
213    }
214}
215
216/// A hierarchical lifecycle coordinator that never carries media data.
217#[derive(Clone)]
218pub struct PluginScope {
219    context: Arc<ScopeContext>,
220    inner: Arc<Mutex<ScopeInner>>,
221}
222
223struct ScopeContext {
224    next_owner_token: AtomicU64,
225    registered_scopes: AtomicUsize,
226    registered_owners: AtomicUsize,
227}
228
229struct ScopeInner {
230    kind: PluginScopeKind,
231    depth: usize,
232    state: PluginScopeState,
233    failure_reason: Option<String>,
234    settlement_report: Option<PluginScopeCloseReport>,
235    children: Vec<PluginScope>,
236    owners: Vec<PluginOwnerDisposer>,
237}
238
239struct PluginOwnerDisposer {
240    token: PluginOwnerToken,
241    disposer: Box<dyn FnOnce() -> Result<(), PluginOwnerDisposalError> + Send + 'static>,
242}
243
244#[derive(Debug, Clone, Copy, PartialEq, Eq)]
245enum BusyChildPolicy {
246    Restore,
247    Quarantine,
248}
249
250#[derive(Debug, Clone, Copy, PartialEq, Eq)]
251enum DisposerOutcome {
252    Completed,
253    Failed,
254    Panicked,
255    WorkerUnavailable,
256}
257
258impl PluginScope {
259    fn new_root() -> Self {
260        let context = Arc::new(ScopeContext {
261            next_owner_token: AtomicU64::new(1),
262            registered_scopes: AtomicUsize::new(1),
263            registered_owners: AtomicUsize::new(0),
264        });
265        Self::new(PluginScopeKind::Root, 0, context)
266    }
267
268    fn new(kind: PluginScopeKind, depth: usize, context: Arc<ScopeContext>) -> Self {
269        Self {
270            context,
271            inner: Arc::new(Mutex::new(ScopeInner {
272                kind,
273                depth,
274                state: PluginScopeState::Created,
275                failure_reason: None,
276                settlement_report: None,
277                children: Vec::new(),
278                owners: Vec::new(),
279            })),
280        }
281    }
282
283    pub fn kind(&self) -> PluginScopeKind {
284        self.lock().kind
285    }
286
287    pub fn state(&self) -> PluginScopeState {
288        self.lock().state
289    }
290
291    pub fn failure_reason(&self) -> Option<String> {
292        self.lock().failure_reason.clone()
293    }
294
295    /// Returns the most recent bounded terminal settlement aggregate.
296    pub fn last_close_report(&self) -> Option<PluginScopeCloseReport> {
297        self.lock().settlement_report.clone()
298    }
299
300    pub fn start(&self) -> Result<(), PluginScopeError> {
301        let mut inner = self.lock();
302        if inner.state != PluginScopeState::Created {
303            return Err(PluginScopeError::InvalidTransition {
304                kind: inner.kind,
305                state: inner.state,
306            });
307        }
308        inner.state = PluginScopeState::Starting;
309        inner.state = PluginScopeState::Running;
310        Ok(())
311    }
312
313    pub fn create_child(&self, kind: PluginScopeKind) -> Result<PluginScope, PluginScopeError> {
314        let mut inner = self.lock();
315        Self::ensure_mutable(&inner)?;
316        inner.children.retain(|child| {
317            !matches!(
318                child.state(),
319                PluginScopeState::Closed
320                    | PluginScopeState::Failed
321                    | PluginScopeState::Cancelled
322                    | PluginScopeState::Quarantined
323            )
324        });
325        let finite_slot = match kind {
326            PluginScopeKind::Playback => Some(PluginScopeResource::ActivePlaybackSlot),
327            PluginScopeKind::NextPrewarm => Some(PluginScopeResource::NextPrewarmSlot),
328            PluginScopeKind::Root
329            | PluginScopeKind::Player
330            | PluginScopeKind::Operation
331            | PluginScopeKind::Worker => None,
332        };
333        if let Some(resource) = finite_slot
334            && inner.children.iter().any(|child| child.kind() == kind)
335        {
336            return Err(PluginScopeError::CapacityExceeded { resource, limit: 1 });
337        }
338        if inner.children.len() >= MAX_PLUGIN_SCOPE_CHILDREN {
339            return Err(PluginScopeError::CapacityExceeded {
340                resource: PluginScopeResource::Children,
341                limit: MAX_PLUGIN_SCOPE_CHILDREN,
342            });
343        }
344        if inner.depth >= MAX_PLUGIN_SCOPE_DEPTH {
345            return Err(PluginScopeError::CapacityExceeded {
346                resource: PluginScopeResource::Depth,
347                limit: MAX_PLUGIN_SCOPE_DEPTH,
348            });
349        }
350        if !reserve_bounded(
351            &self.context.registered_scopes,
352            MAX_PLUGIN_RUNTIME_SCOPE_REGISTRATIONS,
353        ) {
354            return Err(PluginScopeError::CapacityExceeded {
355                resource: PluginScopeResource::ScopeRegistrations,
356                limit: MAX_PLUGIN_RUNTIME_SCOPE_REGISTRATIONS,
357            });
358        }
359        let child = Self::new(kind, inner.depth + 1, self.context.clone());
360        inner.children.push(child.clone());
361        Ok(child)
362    }
363
364    pub fn add_disposer<F>(&self, disposer: F) -> Result<(), PluginScopeError>
365    where
366        F: FnOnce() + Send + 'static,
367    {
368        self.add_owner_disposer(disposer).map(|_| ())
369    }
370
371    /// Registers one owner cleanup and returns its runtime-local quarantine token.
372    pub fn add_owner_disposer<F>(&self, disposer: F) -> Result<PluginOwnerToken, PluginScopeError>
373    where
374        F: FnOnce() + Send + 'static,
375    {
376        self.add_fallible_owner_disposer(move || {
377            disposer();
378            Ok(())
379        })
380    }
381
382    /// Registers one owner cleanup that can report a typed settlement failure.
383    pub fn add_fallible_owner_disposer<F>(
384        &self,
385        disposer: F,
386    ) -> Result<PluginOwnerToken, PluginScopeError>
387    where
388        F: FnOnce() -> Result<(), PluginOwnerDisposalError> + Send + 'static,
389    {
390        let mut inner = self.lock();
391        Self::ensure_mutable(&inner)?;
392        if inner.owners.len() >= MAX_PLUGIN_SCOPE_OWNERS {
393            return Err(PluginScopeError::CapacityExceeded {
394                resource: PluginScopeResource::Owners,
395                limit: MAX_PLUGIN_SCOPE_OWNERS,
396            });
397        }
398        if !reserve_bounded(
399            &self.context.registered_owners,
400            MAX_PLUGIN_RUNTIME_OWNER_REGISTRATIONS,
401        ) {
402            return Err(PluginScopeError::CapacityExceeded {
403                resource: PluginScopeResource::OwnerRegistrations,
404                limit: MAX_PLUGIN_RUNTIME_OWNER_REGISTRATIONS,
405            });
406        }
407        let token = match self.next_owner_token() {
408            Ok(token) => token,
409            Err(error) => {
410                self.context
411                    .registered_owners
412                    .fetch_sub(1, Ordering::Relaxed);
413                return Err(error);
414            }
415        };
416        inner.owners.push(PluginOwnerDisposer {
417            token,
418            disposer: Box::new(disposer),
419        });
420        Ok(token)
421    }
422
423    pub fn fail(
424        &self,
425        reason: impl Into<String>,
426    ) -> Result<PluginScopeCloseReport, PluginScopeError> {
427        self.fail_with_timeout(reason, DEFAULT_PLUGIN_SCOPE_CLOSE_TIMEOUT)
428    }
429
430    pub fn fail_with_timeout(
431        &self,
432        reason: impl Into<String>,
433        timeout: Duration,
434    ) -> Result<PluginScopeCloseReport, PluginScopeError> {
435        let reason = validate_failure_reason(reason.into())?;
436        self.settle_with_timeout(
437            PluginScopeState::Failed,
438            Some(reason),
439            timeout,
440            BusyChildPolicy::Restore,
441        )
442    }
443
444    pub fn cancel(&self) -> Result<PluginScopeCloseReport, PluginScopeError> {
445        self.cancel_with_timeout(DEFAULT_PLUGIN_SCOPE_CLOSE_TIMEOUT)
446    }
447
448    pub fn cancel_with_timeout(
449        &self,
450        timeout: Duration,
451    ) -> Result<PluginScopeCloseReport, PluginScopeError> {
452        self.settle_with_timeout(
453            PluginScopeState::Cancelled,
454            None,
455            timeout,
456            BusyChildPolicy::Restore,
457        )
458    }
459
460    pub fn close(&self) -> Result<PluginScopeCloseReport, PluginScopeError> {
461        self.close_with_timeout_and_policy(
462            DEFAULT_PLUGIN_SCOPE_CLOSE_TIMEOUT,
463            BusyChildPolicy::Restore,
464        )
465    }
466
467    /// Closes the scope with one total deadline shared by descendants and owners.
468    ///
469    /// Remaining time is divided across pending cleanup so one stalled owner
470    /// cannot consume every sibling's budget. Concurrent draining work is
471    /// isolated as `Quarantined`; `close` retains the retryable `Busy` behavior.
472    pub fn close_with_timeout(
473        &self,
474        timeout: Duration,
475    ) -> Result<PluginScopeCloseReport, PluginScopeError> {
476        self.close_with_timeout_and_policy(timeout, BusyChildPolicy::Quarantine)
477    }
478
479    fn close_with_timeout_and_policy(
480        &self,
481        timeout: Duration,
482        busy_child_policy: BusyChildPolicy,
483    ) -> Result<PluginScopeCloseReport, PluginScopeError> {
484        match self.settle_with_timeout(PluginScopeState::Closed, None, timeout, busy_child_policy) {
485            Err(PluginScopeError::Busy) if busy_child_policy == BusyChildPolicy::Quarantine => {
486                self.quarantine_busy_or_settle_now()
487            }
488            result => result,
489        }
490    }
491
492    fn settle_with_timeout(
493        &self,
494        terminal: PluginScopeState,
495        failure_reason: Option<String>,
496        timeout: Duration,
497        busy_child_policy: BusyChildPolicy,
498    ) -> Result<PluginScopeCloseReport, PluginScopeError> {
499        let deadline = Instant::now()
500            .checked_add(timeout)
501            .ok_or(PluginScopeError::InvalidCloseTimeout)?;
502        self.settle_until(terminal, failure_reason, deadline, busy_child_policy)
503    }
504
505    fn settle_until(
506        &self,
507        terminal: PluginScopeState,
508        failure_reason: Option<String>,
509        deadline: Instant,
510        busy_child_policy: BusyChildPolicy,
511    ) -> Result<PluginScopeCloseReport, PluginScopeError> {
512        let (previous_state, previous_failure_reason, children, owners) = {
513            let mut inner = self.lock();
514            match inner.state {
515                PluginScopeState::Closed
516                | PluginScopeState::Failed
517                | PluginScopeState::Cancelled
518                | PluginScopeState::Quarantined => {
519                    return Ok(PluginScopeCloseReport {
520                        final_state: inner.state,
521                        ..PluginScopeCloseReport::default()
522                    });
523                }
524                PluginScopeState::Draining => return Err(PluginScopeError::Busy),
525                PluginScopeState::Created
526                | PluginScopeState::Starting
527                | PluginScopeState::Running => {}
528            }
529            let previous_state = inner.state;
530            let previous_failure_reason = inner.failure_reason.clone();
531            inner.state = PluginScopeState::Draining;
532            inner.failure_reason = failure_reason;
533            (
534                previous_state,
535                previous_failure_reason,
536                std::mem::take(&mut inner.children),
537                std::mem::take(&mut inner.owners),
538            )
539        };
540
541        let mut report = PluginScopeCloseReport::default();
542        let mut pending_children = children;
543        while let Some(child) = pending_children.pop() {
544            let remaining_items = pending_children.len() + owners.len() + 1;
545            let child_deadline = fair_share_deadline(deadline, remaining_items);
546            match child.settle_until(
547                PluginScopeState::Closed,
548                None,
549                child_deadline,
550                busy_child_policy,
551            ) {
552                Ok(child_report) => report.merge_child(child_report),
553                Err(PluginScopeError::Busy) if busy_child_policy == BusyChildPolicy::Quarantine => {
554                    match child.quarantine_busy_or_settle_now() {
555                        Ok(child_report) => report.merge_child(child_report),
556                        Err(error) => {
557                            self.restore_after_child_error(
558                                previous_state,
559                                previous_failure_reason,
560                                pending_children,
561                                child,
562                                owners,
563                            );
564                            return Err(error);
565                        }
566                    }
567                }
568                Err(error) => {
569                    let inner = self.lock();
570                    if inner.state == PluginScopeState::Quarantined {
571                        drop(inner);
572                        match child.quarantine_busy_or_settle_now() {
573                            Ok(child_report) => {
574                                report.merge_child(child_report);
575                                continue;
576                            }
577                            Err(quarantine_error) => {
578                                self.restore_after_child_error(
579                                    previous_state,
580                                    previous_failure_reason,
581                                    pending_children,
582                                    child,
583                                    owners,
584                                );
585                                return Err(quarantine_error);
586                            }
587                        }
588                    }
589                    drop(inner);
590                    self.restore_after_child_error(
591                        previous_state,
592                        previous_failure_reason,
593                        pending_children,
594                        child,
595                        owners,
596                    );
597                    return Err(error);
598                }
599            }
600        }
601
602        let mut pending_owners = owners;
603        while let Some(owner) = pending_owners.pop() {
604            let owner_deadline = fair_share_deadline(deadline, pending_owners.len() + 1);
605            report.run_owner(owner, owner_deadline);
606        }
607
608        let mut inner = self.lock();
609        if inner.state == PluginScopeState::Draining {
610            inner.state = if report.requires_quarantine() {
611                PluginScopeState::Quarantined
612            } else {
613                terminal
614            };
615        }
616        report.final_state = inner.state;
617        if let Some(existing) = inner.settlement_report.take() {
618            report.merge_aggregate(existing);
619        }
620        inner.settlement_report = Some(report.clone());
621        Ok(report)
622    }
623
624    fn next_owner_token(&self) -> Result<PluginOwnerToken, PluginScopeError> {
625        let raw = self
626            .context
627            .next_owner_token
628            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
629                current.checked_add(1)
630            })
631            .map_err(|_| PluginScopeError::OwnerTokenExhausted)?;
632        NonZeroU64::new(raw)
633            .map(PluginOwnerToken)
634            .ok_or(PluginScopeError::OwnerTokenExhausted)
635    }
636
637    fn restore_after_child_error(
638        &self,
639        previous_state: PluginScopeState,
640        previous_failure_reason: Option<String>,
641        pending_children: Vec<PluginScope>,
642        child: PluginScope,
643        owners: Vec<PluginOwnerDisposer>,
644    ) {
645        let mut inner = self.lock();
646        inner.children.extend(pending_children);
647        inner.children.push(child);
648        inner.owners.extend(owners);
649        if inner.state == PluginScopeState::Draining {
650            inner.state = previous_state;
651            inner.failure_reason = previous_failure_reason;
652        }
653    }
654
655    fn lock(&self) -> std::sync::MutexGuard<'_, ScopeInner> {
656        self.inner.lock().unwrap_or_else(|error| error.into_inner())
657    }
658
659    fn same_identity(&self, other: &PluginScope) -> bool {
660        Arc::ptr_eq(&self.inner, &other.inner)
661    }
662
663    fn quarantine_busy_scope(&self) -> Result<PluginScopeCloseReport, PluginScopeError> {
664        let mut inner = self.lock();
665        match inner.state {
666            PluginScopeState::Draining => {
667                inner.state = PluginScopeState::Quarantined;
668                let report = PluginScopeCloseReport {
669                    busy_scopes_quarantined: 1,
670                    final_state: PluginScopeState::Quarantined,
671                    ..PluginScopeCloseReport::default()
672                };
673                inner.settlement_report = Some(report.clone());
674                Ok(report)
675            }
676            PluginScopeState::Closed
677            | PluginScopeState::Failed
678            | PluginScopeState::Cancelled
679            | PluginScopeState::Quarantined => Ok(PluginScopeCloseReport {
680                final_state: inner.state,
681                ..PluginScopeCloseReport::default()
682            }),
683            PluginScopeState::Created | PluginScopeState::Starting | PluginScopeState::Running => {
684                Err(PluginScopeError::InvalidTransition {
685                    kind: inner.kind,
686                    state: inner.state,
687                })
688            }
689        }
690    }
691
692    fn quarantine_busy_or_settle_now(&self) -> Result<PluginScopeCloseReport, PluginScopeError> {
693        for _ in 0..2 {
694            match self.quarantine_busy_scope() {
695                Ok(report) => return Ok(report),
696                Err(PluginScopeError::InvalidTransition { .. }) => {
697                    match self.settle_with_timeout(
698                        PluginScopeState::Closed,
699                        None,
700                        Duration::ZERO,
701                        BusyChildPolicy::Quarantine,
702                    ) {
703                        Err(PluginScopeError::Busy) => continue,
704                        result => return result,
705                    }
706                }
707                Err(error) => return Err(error),
708            }
709        }
710        self.quarantine_busy_scope()
711    }
712
713    fn ensure_mutable(inner: &ScopeInner) -> Result<(), PluginScopeError> {
714        match inner.state {
715            PluginScopeState::Created | PluginScopeState::Starting | PluginScopeState::Running => {
716                Ok(())
717            }
718            PluginScopeState::Draining => Err(PluginScopeError::Busy),
719            PluginScopeState::Closed
720            | PluginScopeState::Failed
721            | PluginScopeState::Cancelled
722            | PluginScopeState::Quarantined => Err(PluginScopeError::Terminal {
723                kind: inner.kind,
724                state: inner.state,
725            }),
726        }
727    }
728
729    fn transition_kind(
730        &self,
731        expected: PluginScopeKind,
732        target: PluginScopeKind,
733    ) -> Result<(), PluginScopeError> {
734        let mut inner = self.lock();
735        Self::ensure_mutable(&inner)?;
736        if inner.kind != expected {
737            return Err(PluginScopeError::InvalidTransition {
738                kind: inner.kind,
739                state: inner.state,
740            });
741        }
742        inner.kind = target;
743        Ok(())
744    }
745
746    fn transition_child_kind(
747        &self,
748        child: &PluginScope,
749        expected: PluginScopeKind,
750        target: PluginScopeKind,
751    ) -> Result<(), PluginScopeError> {
752        let mut inner = self.lock();
753        Self::ensure_mutable(&inner)?;
754        inner.children.retain(|candidate| {
755            !matches!(
756                candidate.state(),
757                PluginScopeState::Closed
758                    | PluginScopeState::Failed
759                    | PluginScopeState::Cancelled
760                    | PluginScopeState::Quarantined
761            )
762        });
763        if !inner
764            .children
765            .iter()
766            .any(|candidate| Arc::ptr_eq(&candidate.inner, &child.inner))
767        {
768            return Err(PluginScopeError::InvalidTransition {
769                kind: child.kind(),
770                state: child.state(),
771            });
772        }
773        let finite_slot = match target {
774            PluginScopeKind::Playback => Some(PluginScopeResource::ActivePlaybackSlot),
775            PluginScopeKind::NextPrewarm => Some(PluginScopeResource::NextPrewarmSlot),
776            PluginScopeKind::Root
777            | PluginScopeKind::Player
778            | PluginScopeKind::Operation
779            | PluginScopeKind::Worker => None,
780        };
781        if let Some(resource) = finite_slot
782            && inner.children.iter().any(|candidate| {
783                !Arc::ptr_eq(&candidate.inner, &child.inner) && candidate.kind() == target
784            })
785        {
786            return Err(PluginScopeError::CapacityExceeded { resource, limit: 1 });
787        }
788        child.transition_kind(expected, target)
789    }
790}
791
792impl PluginScopeCloseReport {
793    fn merge_child(&mut self, child: Self) {
794        if child.final_state == PluginScopeState::Quarantined {
795            self.children_quarantined += 1;
796        } else {
797            self.children_closed += 1;
798        }
799        self.merge_aggregate(child);
800    }
801
802    fn merge_aggregate(&mut self, other: Self) {
803        self.children_closed += other.children_closed;
804        self.children_quarantined += other.children_quarantined;
805        self.busy_scopes_quarantined += other.busy_scopes_quarantined;
806        self.disposers_run += other.disposers_run;
807        self.owners_settled += other.owners_settled;
808        self.owners_quarantined += other.owners_quarantined;
809        self.disposer_panics += other.disposer_panics;
810        self.disposer_timeouts += other.disposer_timeouts;
811        self.disposer_worker_failures += other.disposer_worker_failures;
812        self.disposer_failures += other.disposer_failures;
813        self.quarantine_records_dropped += other.quarantine_records_dropped;
814        for quarantine in other.quarantined_owners {
815            self.push_quarantine(quarantine);
816        }
817    }
818
819    fn run_owner(&mut self, owner: PluginOwnerDisposer, deadline: Instant) {
820        let token = owner.token;
821        if Instant::now() >= deadline {
822            // Quarantine retains the captured owner so its Drop cannot run on the
823            // caller after the close budget has already expired.
824            std::mem::forget(owner);
825            self.disposer_timeouts += 1;
826            self.quarantine_owner(token, PluginScopeQuarantineReason::TimedOut);
827            return;
828        }
829        let owner_holder = Arc::new(Mutex::new(Some(owner)));
830        let worker_owner_holder = owner_holder.clone();
831        let (sender, receiver) = mpsc::sync_channel(1);
832        let thread_name = format!("vesper-plugin-dispose-{}", token.get());
833        let spawn_result = std::thread::Builder::new()
834            .name(thread_name)
835            .spawn(move || {
836                let owner = worker_owner_holder
837                    .lock()
838                    .unwrap_or_else(|error| error.into_inner())
839                    .take();
840                let outcome = match owner {
841                    Some(owner) => match catch_unwind(AssertUnwindSafe(owner.disposer)) {
842                        Ok(Ok(())) => DisposerOutcome::Completed,
843                        Ok(Err(_)) => DisposerOutcome::Failed,
844                        Err(_) => DisposerOutcome::Panicked,
845                    },
846                    None => DisposerOutcome::WorkerUnavailable,
847                };
848                let _ = sender.send(outcome);
849            });
850
851        if spawn_result.is_err() {
852            // The failed worker drops only its Arc clone. Retain the original
853            // holder so a captured native owner is not destroyed on this thread.
854            std::mem::forget(owner_holder);
855            self.disposer_worker_failures += 1;
856            self.quarantine_owner(token, PluginScopeQuarantineReason::WorkerUnavailable);
857            return;
858        }
859        drop(owner_holder);
860
861        self.disposers_run += 1;
862        let remaining = deadline.saturating_duration_since(Instant::now());
863        match receiver.recv_timeout(remaining) {
864            Ok(DisposerOutcome::Completed) => self.owners_settled += 1,
865            Ok(DisposerOutcome::Failed) => {
866                self.disposer_failures += 1;
867                self.quarantine_owner(token, PluginScopeQuarantineReason::Failed);
868            }
869            Ok(DisposerOutcome::Panicked) => {
870                self.disposer_panics += 1;
871                self.quarantine_owner(token, PluginScopeQuarantineReason::Panicked);
872            }
873            Ok(DisposerOutcome::WorkerUnavailable) => {
874                self.disposer_worker_failures += 1;
875                self.quarantine_owner(token, PluginScopeQuarantineReason::WorkerUnavailable);
876            }
877            Err(mpsc::RecvTimeoutError::Timeout) => {
878                self.disposer_timeouts += 1;
879                self.quarantine_owner(token, PluginScopeQuarantineReason::TimedOut);
880            }
881            Err(mpsc::RecvTimeoutError::Disconnected) => {
882                self.disposer_worker_failures += 1;
883                self.quarantine_owner(token, PluginScopeQuarantineReason::WorkerUnavailable);
884            }
885        }
886    }
887
888    fn quarantine_owner(
889        &mut self,
890        owner_token: PluginOwnerToken,
891        reason: PluginScopeQuarantineReason,
892    ) {
893        self.owners_quarantined += 1;
894        self.push_quarantine(PluginScopeQuarantine {
895            owner_token,
896            reason,
897        });
898    }
899
900    fn push_quarantine(&mut self, quarantine: PluginScopeQuarantine) {
901        if self.quarantined_owners.len() < MAX_PLUGIN_SCOPE_QUARANTINE_RECORDS {
902            self.quarantined_owners.push(quarantine);
903        } else {
904            self.quarantine_records_dropped += 1;
905        }
906    }
907
908    fn requires_quarantine(&self) -> bool {
909        self.children_quarantined > 0
910            || self.busy_scopes_quarantined > 0
911            || self.owners_quarantined > 0
912    }
913}
914
915fn validate_failure_reason(reason: String) -> Result<String, PluginScopeError> {
916    if reason.is_empty() || reason.len() > MAX_PLUGIN_SCOPE_REASON_BYTES {
917        return Err(PluginScopeError::InvalidFailureReason {
918            limit: MAX_PLUGIN_SCOPE_REASON_BYTES,
919        });
920    }
921    Ok(reason)
922}
923
924fn reserve_bounded(counter: &AtomicUsize, limit: usize) -> bool {
925    counter
926        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
927            (current < limit).then_some(current + 1)
928        })
929        .is_ok()
930}
931
932fn fair_share_deadline(deadline: Instant, remaining_items: usize) -> Instant {
933    let now = Instant::now();
934    let divisor = u32::try_from(remaining_items.max(1)).unwrap_or(u32::MAX);
935    now.checked_add(deadline.saturating_duration_since(now) / divisor)
936        .unwrap_or(deadline)
937}
938
939#[cfg(test)]
940mod tests {
941    use super::*;
942
943    #[test]
944    fn failed_child_restoration_preserves_concurrent_quarantine() {
945        let parent = PluginScope::new_root();
946        let child = match parent.create_child(PluginScopeKind::Worker) {
947            Ok(child) => child,
948            Err(error) => panic!("child fixture failed: {error}"),
949        };
950        if let Err(error) = parent.add_owner_disposer(|| {}) {
951            panic!("owner fixture failed: {error}");
952        }
953        let owners = {
954            let mut inner = parent.lock();
955            inner.children.clear();
956            inner.state = PluginScopeState::Quarantined;
957            inner.failure_reason = Some("concurrent shutdown".to_owned());
958            std::mem::take(&mut inner.owners)
959        };
960
961        parent.restore_after_child_error(
962            PluginScopeState::Running,
963            Some("previous failure".to_owned()),
964            Vec::new(),
965            child,
966            owners,
967        );
968
969        let inner = parent.lock();
970        assert_eq!(inner.state, PluginScopeState::Quarantined);
971        assert_eq!(inner.failure_reason.as_deref(), Some("concurrent shutdown"));
972        assert_eq!(inner.children.len(), 1);
973        assert_eq!(inner.owners.len(), 1);
974    }
975
976    #[test]
977    fn fallible_owner_cleanup_is_quarantined_without_panicking() {
978        let scope = PluginScope::new_root();
979        let token = scope
980            .add_fallible_owner_disposer(|| Err(PluginOwnerDisposalError))
981            .expect("fallible owner");
982
983        let report = scope
984            .close_with_timeout(Duration::from_secs(1))
985            .expect("scope close");
986
987        assert_eq!(report.final_state, PluginScopeState::Quarantined);
988        assert_eq!(report.owners_settled, 0);
989        assert_eq!(report.owners_quarantined, 1);
990        assert_eq!(report.disposer_failures, 1);
991        assert_eq!(report.disposer_panics, 0);
992        assert!(report.quarantined_owners.iter().any(|entry| {
993            entry.owner_token == token && entry.reason == PluginScopeQuarantineReason::Failed
994        }));
995    }
996}