qubit_progress/
progress.rs

1// =============================================================================
2//    Copyright (c) 2025 - 2026 Haixing Hu.
3//
4//    SPDX-License-Identifier: Apache-2.0
5//
6//    Licensed under the Apache License, Version 2.0.
7// =============================================================================
8//! Operation lifecycle, metric state and report scheduling.
9// qubit-style: allow multiple-public-types
10// qubit-style: allow coverage-cfg
11
12use std::{
13    sync::Arc,
14    sync::atomic::{
15        AtomicU64,
16        Ordering,
17    },
18    time::{
19        Duration,
20        Instant,
21    },
22};
23
24use crate::{
25    Event,
26    Metric,
27    MetricHandle,
28    MetricSnapshot,
29    OperationAttributes,
30    Phase,
31    Reporter,
32    Stage,
33    auto_reporter::{
34        self,
35        AutoReporter,
36    },
37    error::{
38        CompletionError,
39        ConfigurationError,
40        DeliveryError,
41        EmissionError,
42        FinishError,
43        RecoverableFinishError,
44        StartError,
45        TerminalError,
46    },
47    internal::OperationState,
48    validation::{
49        validate_attributes,
50        validate_metrics,
51        validate_stage,
52    },
53};
54
55#[cfg(coverage)]
56use crate::{
57    NoopReporter,
58    error::ReporterError,
59};
60
61/// Process-local source of nonzero operation identifiers.
62static NEXT_OPERATION_ID: AtomicU64 = AtomicU64::new(1);
63
64#[cfg(coverage)]
65/// Reporter that fails only after the Started event.
66struct CoverageTerminalReporter {
67    /// Number of delivery attempts observed by the reporter.
68    attempts: AtomicU64,
69}
70
71#[cfg(coverage)]
72impl Reporter for CoverageTerminalReporter {
73    /// Fails terminal delivery after allowing operation startup.
74    fn report(&self, _event: &Event) -> Result<(), ReporterError> {
75        if self.attempts.fetch_add(1, Ordering::Relaxed) == 0 {
76            Ok(())
77        } else {
78            Err(ReporterError::message("coverage terminal failure"))
79        }
80    }
81}
82
83/// Reporter retained by a progress operation.
84enum ReporterHandle<'reporter> {
85    Borrowed(&'reporter dyn Reporter),
86    Owned(Arc<dyn Reporter>),
87}
88
89impl ReporterHandle<'_> {
90    fn as_reporter(&self) -> &dyn Reporter {
91        match self {
92            Self::Borrowed(reporter) => *reporter,
93            Self::Owned(reporter) => reporter.as_ref(),
94        }
95    }
96}
97
98/// Configures one [`Progress`] operation before it starts.
99pub struct ProgressBuilder<'reporter> {
100    /// Reporter receiving complete events.
101    reporter: ReporterHandle<'reporter>,
102    /// Minimum interval between due-based running reports.
103    interval: Duration,
104    /// Stable operation metrics.
105    metrics: Vec<Metric>,
106    /// Optional initial stage.
107    stage: Option<Stage>,
108    /// Correlation attributes shared by all operation events.
109    attributes: OperationAttributes,
110}
111
112impl<'reporter> ProgressBuilder<'reporter> {
113    /// Sets the minimum interval between due-based running reports.
114    #[must_use]
115    pub const fn interval(mut self, interval: Duration) -> Self {
116        self.interval = interval;
117        self
118    }
119    /// Adds one stable metric to the operation.
120    #[must_use]
121    pub fn metric(mut self, metric: Metric) -> Self {
122        self.metrics.push(metric);
123        self
124    }
125    /// Adds stage metadata to the Started event and subsequent events.
126    #[must_use]
127    pub fn stage(mut self, stage: Stage) -> Self {
128        self.stage = Some(stage);
129        self
130    }
131    /// Adds or replaces one operation correlation attribute.
132    #[must_use]
133    pub fn attribute(mut self, key: &str, value: &str) -> Self {
134        self.attributes.insert(key, value);
135        self
136    }
137    /// Replaces all operation correlation attributes.
138    #[must_use]
139    pub fn attributes(mut self, attributes: OperationAttributes) -> Self {
140        self.attributes = attributes;
141        self
142    }
143    /// Validates configuration, samples enablement and emits Started when
144    /// enabled.
145    pub fn start(self) -> Result<Progress<'reporter>, StartError> {
146        validate_metrics(&self.metrics)?;
147        if let Some(stage) = &self.stage {
148            validate_stage(stage)?;
149        }
150        validate_attributes(&self.attributes)?;
151
152        let enabled = self.reporter.as_reporter().is_enabled();
153        let operation_state = OperationState::new();
154        let operation_id = enabled.then(allocate_operation_id).transpose()?;
155        let mut progress = Progress {
156            reporter: self.reporter,
157            enabled,
158            metrics: self
159                .metrics
160                .into_iter()
161                .map(|metric| {
162                    MetricHandle::new(metric, Arc::clone(&operation_state))
163                })
164                .collect(),
165            operation_state,
166            stage: self.stage,
167            attributes: Arc::new(self.attributes),
168            interval: self.interval,
169            started_at: Instant::now(),
170            next_due_elapsed: None,
171            operation_id,
172            next_sequence: 0,
173        };
174
175        if enabled {
176            let metrics = progress.metric_snapshots();
177            progress
178                .emit(Phase::Started, metrics, Duration::ZERO)
179                .map_err(StartError::from)?;
180            progress.next_due_elapsed = Some(progress.interval);
181            progress.started_at = Instant::now();
182        }
183        Ok(progress)
184    }
185}
186
187/// One started progress operation.
188///
189/// Terminal methods consume this value, preventing reports after a terminal
190/// phase and preventing duplicate terminal events in safe Rust.
191#[must_use]
192pub struct Progress<'reporter> {
193    /// Reporter selected by the builder.
194    reporter: ReporterHandle<'reporter>,
195    /// Stable enablement sampled once at start.
196    enabled: bool,
197    /// Live metrics carried by each event.
198    metrics: Vec<MetricHandle>,
199    /// Shared lifecycle and in-flight update gate.
200    operation_state: Arc<OperationState>,
201    /// Optional current stage.
202    stage: Option<Stage>,
203    /// Immutable correlation attributes shared by all events.
204    attributes: Arc<OperationAttributes>,
205    /// Minimum due-report spacing.
206    interval: Duration,
207    /// Monotonic operation start time.
208    started_at: Instant,
209    /// Next due elapsed deadline for a positive interval.
210    next_due_elapsed: Option<Duration>,
211    /// Nonzero identifier for enabled operations.
212    operation_id: Option<u64>,
213    /// Sequence reserved for the next event attempt.
214    next_sequence: u64,
215}
216
217impl<'reporter> Progress<'reporter> {
218    /// Creates a builder borrowing one reporter.
219    #[must_use]
220    pub fn builder(
221        reporter: &'reporter dyn Reporter,
222    ) -> ProgressBuilder<'reporter> {
223        ProgressBuilder {
224            reporter: ReporterHandle::Borrowed(reporter),
225            interval: Duration::ZERO,
226            metrics: Vec::new(),
227            stage: None,
228            attributes: OperationAttributes::new(),
229        }
230    }
231    /// Creates a builder that owns one shared reporter.
232    #[must_use]
233    pub fn builder_arc(
234        reporter: Arc<dyn Reporter>,
235    ) -> ProgressBuilder<'static> {
236        ProgressBuilder {
237            reporter: ReporterHandle::Owned(reporter),
238            interval: Duration::ZERO,
239            metrics: Vec::new(),
240            stage: None,
241            attributes: OperationAttributes::new(),
242        }
243    }
244    /// Returns enablement sampled when this operation started.
245    #[must_use]
246    pub const fn is_enabled(&self) -> bool {
247        self.enabled
248    }
249    /// Returns monotonic elapsed time since `start()`.
250    #[must_use]
251    pub fn elapsed(&self) -> Duration {
252        self.started_at.elapsed()
253    }
254    /// Returns a cloneable live metric selected by its stable ID.
255    pub fn metric(&self, metric_id: &str) -> Option<MetricHandle> {
256        self.metrics
257            .iter()
258            .find(|metric| metric.id() == metric_id)
259            .cloned()
260    }
261    /// Immediately emits a Running event from current metric state.
262    pub fn report(&mut self) -> Result<(), EmissionError> {
263        if !self.enabled {
264            return Ok(());
265        }
266        let metrics = self.metric_snapshots();
267        let elapsed = self.elapsed();
268        let result = self.emit(Phase::Running, metrics, elapsed);
269        self.reset_deadline();
270        result
271    }
272    /// Emits a Running event only when the configured interval is due.
273    pub fn report_if_due(&mut self) -> Result<(), EmissionError> {
274        if !self.enabled || !self.is_due() {
275            return Ok(());
276        }
277        self.report()
278    }
279    /// Replaces stage metadata attached to subsequent events.
280    pub fn set_stage(
281        &mut self,
282        stage: Stage,
283    ) -> Result<(), ConfigurationError> {
284        validate_stage(&stage)?;
285        self.stage = Some(stage);
286        Ok(())
287    }
288    /// Removes stage metadata from subsequent events.
289    pub fn clear_stage(&mut self) {
290        self.stage = None;
291    }
292    /// Consumes this operation and emits a successful terminal event without
293    /// checking whether metric work is complete.
294    pub fn finish_unchecked(self) -> Result<Duration, TerminalError> {
295        self.terminal(Phase::Succeeded)
296    }
297    /// Consumes this operation and emits a successful terminal event only when
298    /// no metric has active work and every known total has been completed.
299    #[allow(clippy::result_large_err)]
300    pub fn finish(mut self) -> Result<Duration, FinishError> {
301        let elapsed = self.elapsed();
302        let finish_guard = self.operation_state.begin_finish();
303        if let Err(source) = self.validate_finish() {
304            finish_guard.close();
305            return Err(FinishError::Incomplete { elapsed, source });
306        }
307        finish_guard.close();
308        if !self.enabled {
309            return Ok(elapsed);
310        }
311        self.emit(Phase::Succeeded, self.metric_snapshots(), elapsed)
312            .map(|()| elapsed)
313            .map_err(|source| {
314                FinishError::Terminal(TerminalError::new(elapsed, source))
315            })
316    }
317    /// Consumes this operation and emits a successful terminal event while
318    /// preserving the operation when completion validation fails.
319    #[allow(clippy::result_large_err)]
320    pub fn finish_recoverable(
321        mut self,
322    ) -> Result<Duration, RecoverableFinishError<'reporter>> {
323        let elapsed = self.elapsed();
324        let finish_guard = self.operation_state.begin_finish();
325        if let Err(source) = self.validate_finish() {
326            finish_guard.reopen();
327            return Err(RecoverableFinishError::Incomplete {
328                progress: self,
329                source,
330            });
331        }
332        finish_guard.close();
333        if !self.enabled {
334            return Ok(elapsed);
335        }
336        self.emit(Phase::Succeeded, self.metric_snapshots(), elapsed)
337            .map(|()| elapsed)
338            .map_err(|source| {
339                RecoverableFinishError::Terminal(TerminalError::new(
340                    elapsed, source,
341                ))
342            })
343    }
344    /// Consumes this operation and emits a failed terminal event.
345    pub fn fail(self) -> Result<Duration, TerminalError> {
346        self.terminal(Phase::Failed)
347    }
348    /// Consumes this operation and emits a cancelled terminal event.
349    pub fn cancel(self) -> Result<Duration, TerminalError> {
350        self.terminal(Phase::Cancelled)
351    }
352    /// Spawns a scoped automatic Running reporter that exclusively borrows this
353    /// operation.
354    pub fn spawn_auto_reporter<'scope, 'env>(
355        &'scope mut self,
356        scope: &'scope std::thread::Scope<'scope, 'env>,
357    ) -> AutoReporter<'scope, 'reporter>
358    where
359        'reporter: 'scope,
360    {
361        auto_reporter::spawn(self, scope)
362    }
363    /// Copies each metric into one independently consistent event snapshot.
364    fn metric_snapshots(&self) -> Vec<MetricSnapshot> {
365        self.metrics.iter().map(MetricHandle::snapshot).collect()
366    }
367    /// Validates the metric invariants required for successful finish.
368    fn validate_finish(&self) -> Result<(), CompletionError> {
369        for metric in &self.metrics {
370            let snapshot = metric.snapshot();
371            if snapshot.active() != 0 {
372                return Err(CompletionError::ActiveWork {
373                    metric_id: snapshot.id().to_owned(),
374                    active: snapshot.active(),
375                });
376            }
377            if let Some(total) = snapshot.total()
378                && snapshot.completed() != total
379            {
380                return Err(CompletionError::IncompleteTotal {
381                    metric_id: snapshot.id().to_owned(),
382                    completed: snapshot.completed(),
383                    total,
384                });
385            }
386        }
387        Ok(())
388    }
389    /// Delivers one complete event after reserving its delivery sequence.
390    fn emit(
391        &mut self,
392        phase: Phase,
393        metrics: Vec<MetricSnapshot>,
394        elapsed: Duration,
395    ) -> Result<(), EmissionError> {
396        let operation_id =
397            self.operation_id.ok_or(EmissionError::SequenceExhausted)?;
398        let sequence = self.next_sequence;
399        self.next_sequence = sequence
400            .checked_add(1)
401            .ok_or(EmissionError::SequenceExhausted)?;
402        let event = Event::new(
403            operation_id,
404            sequence,
405            phase,
406            self.stage.clone(),
407            Arc::clone(&self.attributes),
408            metrics,
409            elapsed,
410        );
411        match self.reporter.as_reporter().report(&event) {
412            Ok(()) => Ok(()),
413            Err(source) => {
414                Err(EmissionError::Delivery(DeliveryError::new(event, source)))
415            }
416        }
417    }
418    /// Tests whether a due-based running report can run now.
419    fn is_due(&self) -> bool {
420        self.interval.is_zero()
421            || self
422                .next_due_elapsed
423                .is_some_and(|deadline| self.elapsed() >= deadline)
424    }
425    /// Returns the configured interval to the crate-private background loop.
426    pub(crate) const fn report_interval(&self) -> Duration {
427        self.interval
428    }
429    /// Returns how long the background loop should wait for the next deadline.
430    pub(crate) fn time_until_due(&self) -> Duration {
431        self.next_due_elapsed
432            .map(|deadline| deadline.saturating_sub(self.elapsed()))
433            .unwrap_or(Duration::MAX)
434    }
435    /// Pushes the next positive-interval deadline after a running attempt.
436    fn reset_deadline(&mut self) {
437        self.next_due_elapsed = self.elapsed().checked_add(self.interval);
438    }
439    /// Emits one terminal phase while retaining elapsed time on failure.
440    #[inline(never)]
441    fn terminal(mut self, phase: Phase) -> Result<Duration, TerminalError> {
442        let elapsed = self.elapsed();
443        let finish_guard = self.operation_state.begin_finish();
444        finish_guard.close();
445        if !self.enabled {
446            return Ok(elapsed);
447        }
448        self.emit(phase, self.metric_snapshots(), elapsed)
449            .map(|()| elapsed)
450            .map_err(|source| TerminalError::new(elapsed, source))
451    }
452}
453
454impl Drop for Progress<'_> {
455    /// Closes live handles when a caller abandons an unfinished operation.
456    #[inline(never)]
457    fn drop(&mut self) {
458        self.operation_state.close();
459    }
460}
461
462/// Allocates a nonzero operation ID without wrapping or reuse.
463#[inline(never)]
464fn allocate_operation_id() -> Result<u64, StartError> {
465    loop {
466        let current = NEXT_OPERATION_ID.load(Ordering::Relaxed);
467        if current == 0 {
468            return Err(StartError::OperationIdExhausted);
469        }
470        let next = current.checked_add(1).unwrap_or(0);
471        if NEXT_OPERATION_ID
472            .compare_exchange_weak(
473                current,
474                next,
475                Ordering::Relaxed,
476                Ordering::Relaxed,
477            )
478            .is_ok()
479        {
480            return Ok(current);
481        }
482    }
483}
484
485/// Exercises progress-only edge paths from the instrumented library build.
486#[cfg(coverage)]
487#[doc(hidden)]
488pub fn __coverage_progress_edges() {
489    let previous = NEXT_OPERATION_ID.swap(0, Ordering::Relaxed);
490    assert!(matches!(
491        allocate_operation_id(),
492        Err(StartError::OperationIdExhausted)
493    ));
494    NEXT_OPERATION_ID.store(previous.max(1), Ordering::Relaxed);
495
496    let progress = Progress::builder(&NoopReporter)
497        .metric(Metric::new("coverage", "Coverage"))
498        .start()
499        .expect("coverage progress must start");
500    progress.cancel().expect("coverage progress must cancel");
501
502    let reporter = CoverageTerminalReporter {
503        attempts: AtomicU64::new(0),
504    };
505    let progress = Progress::builder_arc(Arc::new(reporter))
506        .metric(Metric::new("coverage-terminal", "Coverage terminal"))
507        .start()
508        .expect("coverage terminal progress must start");
509    assert!(progress.cancel().is_err());
510}
qubit_progress/progress.rs

qubit_progress/
progress.rs
