qubit_progress/progress.rs
1/*******************************************************************************
2 *
3 * Copyright (c) 2025 - 2026 Haixing Hu.
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * Licensed under the Apache License, Version 2.0.
8 *
9 ******************************************************************************/
10use std::{
11 thread,
12 time::{
13 Duration,
14 Instant,
15 },
16};
17
18use crate::{
19 model::{
20 ProgressCounter,
21 ProgressEvent,
22 ProgressEventBuilder,
23 ProgressMetric,
24 ProgressPhase,
25 ProgressSchema,
26 ProgressStage,
27 },
28 reporter::ProgressReporter,
29 running::{
30 RunningProgressGuard,
31 RunningProgressLoop,
32 },
33};
34
35/// Tracks one logical progress-producing operation and reports events.
36///
37/// A `Progress` instance is scoped to one logical operation. It owns no domain
38/// state; callers keep their own counters and convert them into
39/// [`ProgressCounter`] values when reporting. The run manages elapsed time,
40/// periodic running-event throttling, optional stage metadata, a metric schema,
41/// and forwarding immutable events to one reporter.
42///
43/// # Examples
44///
45/// ```
46/// use std::time::Duration;
47///
48/// use qubit_progress::{
49/// Progress,
50/// ProgressMetric,
51/// ProgressSchema,
52/// WriterProgressReporter,
53/// };
54///
55/// let schema = ProgressSchema::new(vec![ProgressMetric::new("entries", "Entries")]);
56/// let reporter = WriterProgressReporter::from_writer(std::io::stdout());
57/// let mut progress = Progress::new(&reporter, Duration::from_secs(5), schema);
58///
59/// let started = progress.report_started(|event| event.counter("entries", |c| c.total(2)));
60/// assert!(started.elapsed().is_zero());
61///
62/// let _reported = progress.report_running_if_due(|event| {
63/// event.counter("entries", |counter| counter.total(2).completed(1).active(1))
64/// });
65///
66/// let finished = progress.report_finished(|event| {
67/// event.counter("entries", |counter| counter.total(2).completed(2).succeeded(2))
68/// });
69/// assert!(finished.elapsed() >= started.elapsed());
70/// ```
71pub struct Progress<'a> {
72 /// Reporter receiving lifecycle callbacks for this run.
73 reporter: &'a dyn ProgressReporter,
74 /// Metric schema carried by events emitted from this run.
75 schema: ProgressSchema,
76 /// Monotonic start time used to compute elapsed durations.
77 started_at: Instant,
78 /// Minimum interval between due-based running callbacks.
79 report_interval: Duration,
80 /// Next monotonic instant at which a due-based running callback may fire.
81 next_running_at: Instant,
82 /// Optional stage metadata attached to every event emitted by this run.
83 stage: Option<ProgressStage>,
84}
85
86impl<'a> Progress<'a> {
87 /// Creates a progress run starting at the current instant.
88 ///
89 /// # Parameters
90 ///
91 /// * `reporter` - Reporter receiving progress events.
92 /// * `report_interval` - Minimum delay between due-based running events.
93 /// * `schema` - Metric schema carried by emitted events.
94 ///
95 /// # Returns
96 ///
97 /// A progress run whose elapsed time is measured from now.
98 #[inline]
99 pub fn new(reporter: &'a dyn ProgressReporter, report_interval: Duration, schema: ProgressSchema) -> Self {
100 Self::from_start(reporter, report_interval, schema, Instant::now())
101 }
102
103 /// Creates a single-metric progress run starting at the current instant.
104 ///
105 /// # Parameters
106 ///
107 /// * `reporter` - Reporter receiving progress events.
108 /// * `report_interval` - Minimum delay between due-based running events.
109 /// * `metric_id` - Stable metric identifier.
110 /// * `metric_name` - Human-readable metric name.
111 ///
112 /// # Returns
113 ///
114 /// A progress run with a schema containing one metric.
115 #[inline]
116 pub fn single_metric(
117 reporter: &'a dyn ProgressReporter,
118 report_interval: Duration,
119 metric_id: &str,
120 metric_name: &str,
121 ) -> Self {
122 Self::new(
123 reporter,
124 report_interval,
125 ProgressSchema::new(vec![ProgressMetric::new(metric_id, metric_name)]),
126 )
127 }
128
129 /// Creates a progress run from an explicit start instant.
130 ///
131 /// # Parameters
132 ///
133 /// * `reporter` - Reporter receiving progress events.
134 /// * `report_interval` - Minimum delay between due-based running events.
135 /// * `schema` - Metric schema carried by emitted events.
136 /// * `started_at` - Monotonic instant representing operation start.
137 ///
138 /// # Returns
139 ///
140 /// A progress run using `started_at` for elapsed-time calculations.
141 #[inline]
142 fn from_start(
143 reporter: &'a dyn ProgressReporter,
144 report_interval: Duration,
145 schema: ProgressSchema,
146 started_at: Instant,
147 ) -> Self {
148 Self {
149 reporter,
150 schema,
151 started_at,
152 report_interval,
153 next_running_at: next_instant(started_at, report_interval),
154 stage: None,
155 }
156 }
157
158 /// Returns a copy configured with stage metadata.
159 ///
160 /// # Parameters
161 ///
162 /// * `stage` - Stage metadata attached to subsequently reported events.
163 ///
164 /// # Returns
165 ///
166 /// This progress run with `stage` recorded.
167 #[inline]
168 #[must_use]
169 pub fn with_stage(mut self, stage: ProgressStage) -> Self {
170 self.stage = Some(stage);
171 self
172 }
173
174 /// Returns a copy with stage metadata removed.
175 ///
176 /// # Returns
177 ///
178 /// This progress run without stage metadata.
179 #[inline]
180 #[must_use]
181 pub fn without_stage(mut self) -> Self {
182 self.stage = None;
183 self
184 }
185
186 /// Creates an event builder preconfigured with this run's schema, stage, and elapsed time.
187 ///
188 /// # Returns
189 ///
190 /// A progress event builder for this run.
191 #[inline]
192 pub fn event_builder(&self) -> ProgressEventBuilder {
193 self.event_builder_with_elapsed(self.elapsed())
194 }
195
196 /// Reports a started lifecycle event.
197 ///
198 /// # Parameters
199 ///
200 /// * `configure` - Closure that adds counters or stage overrides to the event builder.
201 ///
202 /// # Returns
203 ///
204 /// The event sent to the configured reporter.
205 ///
206 /// # Panics
207 ///
208 /// Propagates panics from the configured reporter.
209 #[inline]
210 pub fn report_started<F>(&self, configure: F) -> ProgressEvent
211 where
212 F: FnOnce(ProgressEventBuilder) -> ProgressEventBuilder,
213 {
214 self.report_with_elapsed(ProgressPhase::Started, Duration::ZERO, configure)
215 }
216
217 /// Reports a running lifecycle event immediately.
218 ///
219 /// # Parameters
220 ///
221 /// * `configure` - Closure that adds counters or stage overrides to the event builder.
222 ///
223 /// # Returns
224 ///
225 /// The event sent to the configured reporter.
226 ///
227 /// # Panics
228 ///
229 /// Propagates panics from the configured reporter.
230 pub fn report_running<F>(&mut self, configure: F) -> ProgressEvent
231 where
232 F: FnOnce(ProgressEventBuilder) -> ProgressEventBuilder,
233 {
234 let now = Instant::now();
235 let event = self.report_with_elapsed(
236 ProgressPhase::Running,
237 now.saturating_duration_since(self.started_at),
238 configure,
239 );
240 self.next_running_at = next_instant(now, self.report_interval);
241 event
242 }
243
244 /// Reports a running lifecycle event if the configured interval has passed.
245 ///
246 /// # Parameters
247 ///
248 /// * `configure` - Closure that adds counters or stage overrides when an event is due.
249 ///
250 /// # Returns
251 ///
252 /// `Some(event)` when a running event was emitted, or `None` when the next
253 /// running-event deadline has not been reached.
254 ///
255 /// This method does not call `configure` unless an event is due. It returns
256 /// immediately when not due, and when due it synchronously calls the
257 /// configured reporter. Any blocking behavior therefore comes from the
258 /// reporter implementation.
259 ///
260 /// # Panics
261 ///
262 /// Propagates panics from the configured reporter when an event is due.
263 pub fn report_running_if_due<F>(&mut self, configure: F) -> Option<ProgressEvent>
264 where
265 F: FnOnce(ProgressEventBuilder) -> ProgressEventBuilder,
266 {
267 let now = Instant::now();
268 if now < self.next_running_at {
269 return None;
270 }
271 let event = self.report_with_elapsed(
272 ProgressPhase::Running,
273 now.saturating_duration_since(self.started_at),
274 configure,
275 );
276 self.next_running_at = next_instant(now, self.report_interval);
277 Some(event)
278 }
279
280 /// Reports a finished lifecycle event.
281 ///
282 /// # Parameters
283 ///
284 /// * `configure` - Closure that adds final counters to the event builder.
285 ///
286 /// # Returns
287 ///
288 /// The event sent to the configured reporter.
289 ///
290 /// # Panics
291 ///
292 /// Propagates panics from the configured reporter.
293 #[inline]
294 pub fn report_finished<F>(&self, configure: F) -> ProgressEvent
295 where
296 F: FnOnce(ProgressEventBuilder) -> ProgressEventBuilder,
297 {
298 self.report_with_elapsed(ProgressPhase::Finished, self.elapsed(), configure)
299 }
300
301 /// Reports a failed lifecycle event.
302 ///
303 /// # Parameters
304 ///
305 /// * `configure` - Closure that adds final or current counters to the event builder.
306 ///
307 /// # Returns
308 ///
309 /// The event sent to the configured reporter.
310 ///
311 /// # Panics
312 ///
313 /// Propagates panics from the configured reporter.
314 #[inline]
315 pub fn report_failed<F>(&self, configure: F) -> ProgressEvent
316 where
317 F: FnOnce(ProgressEventBuilder) -> ProgressEventBuilder,
318 {
319 self.report_with_elapsed(ProgressPhase::Failed, self.elapsed(), configure)
320 }
321
322 /// Reports a canceled lifecycle event.
323 ///
324 /// # Parameters
325 ///
326 /// * `configure` - Closure that adds final or current counters to the event builder.
327 ///
328 /// # Returns
329 ///
330 /// The event sent to the configured reporter.
331 ///
332 /// # Panics
333 ///
334 /// Propagates panics from the configured reporter.
335 #[inline]
336 pub fn report_canceled<F>(&self, configure: F) -> ProgressEvent
337 where
338 F: FnOnce(ProgressEventBuilder) -> ProgressEventBuilder,
339 {
340 self.report_with_elapsed(ProgressPhase::Canceled, self.elapsed(), configure)
341 }
342
343 /// Spawns a scoped background reporter for periodic running events.
344 ///
345 /// The background reporter shares this progress run's reporter, schema,
346 /// start time, interval, and stage metadata. Worker threads should update
347 /// their own domain state first, then call
348 /// [`RunningProgressPointHandle::report`](crate::RunningProgressPointHandle::report)
349 /// on the handle returned by the guard. The guard must be stopped and
350 /// joined before the thread scope exits.
351 ///
352 /// # Parameters
353 ///
354 /// * `scope` - Thread scope that owns the background reporter thread.
355 /// * `snapshot` - Closure that builds fresh counters from caller-owned
356 /// domain state whenever a running event may be due.
357 ///
358 /// # Returns
359 ///
360 /// A guard that owns the reporter thread and can create worker-side point handles.
361 ///
362 /// # Panics
363 ///
364 /// Panics raised by the reporter thread are propagated by
365 /// [`RunningProgressGuard::stop_and_join`].
366 pub fn spawn_running_reporter<'scope, 'env, F>(
367 &self,
368 scope: &'scope thread::Scope<'scope, 'env>,
369 snapshot: F,
370 ) -> RunningProgressGuard<'scope>
371 where
372 'a: 'scope,
373 F: FnMut() -> Vec<ProgressCounter> + Send + 'scope,
374 {
375 RunningProgressLoop::spawn_scoped(scope, self.fork_for_running(), snapshot)
376 }
377
378 /// Reports a lifecycle event with an explicit elapsed duration.
379 ///
380 /// # Parameters
381 ///
382 /// * `phase` - Lifecycle phase to report.
383 /// * `elapsed` - Elapsed duration carried by the event.
384 /// * `configure` - Closure that adds counters or stage overrides.
385 ///
386 /// # Returns
387 ///
388 /// The event sent to the configured reporter.
389 ///
390 /// # Panics
391 ///
392 /// Propagates panics from the configured reporter.
393 fn report_with_elapsed<F>(&self, phase: ProgressPhase, elapsed: Duration, configure: F) -> ProgressEvent
394 where
395 F: FnOnce(ProgressEventBuilder) -> ProgressEventBuilder,
396 {
397 let event = configure(self.event_builder_with_elapsed(elapsed).phase(phase)).build();
398 self.reporter.report(&event);
399 event
400 }
401
402 /// Returns the metric schema for this progress run.
403 ///
404 /// # Returns
405 ///
406 /// The schema cloned into every event emitted by this run.
407 #[inline]
408 pub const fn schema(&self) -> &ProgressSchema {
409 &self.schema
410 }
411
412 /// Returns the elapsed duration since this run started.
413 ///
414 /// # Returns
415 ///
416 /// The monotonic elapsed duration for this progress run.
417 #[inline]
418 pub fn elapsed(&self) -> Duration {
419 self.started_at.elapsed()
420 }
421
422 /// Returns the start instant for this run.
423 ///
424 /// # Returns
425 ///
426 /// The monotonic instant used as this run's start time.
427 #[inline]
428 pub const fn started_at(&self) -> Instant {
429 self.started_at
430 }
431
432 /// Returns the configured running-event interval.
433 ///
434 /// # Returns
435 ///
436 /// The minimum delay between due-based running events.
437 #[inline]
438 pub const fn report_interval(&self) -> Duration {
439 self.report_interval
440 }
441
442 /// Returns the optional stage metadata attached to events.
443 ///
444 /// # Returns
445 ///
446 /// `Some(stage)` when stage metadata is configured, otherwise `None`.
447 #[inline]
448 pub const fn stage(&self) -> Option<&ProgressStage> {
449 self.stage.as_ref()
450 }
451
452 /// Creates a background-thread copy that reports running events for this run.
453 ///
454 /// # Returns
455 ///
456 /// A progress run with the same reporter, schema, start time, interval,
457 /// stage, and next running deadline as this run.
458 fn fork_for_running(&self) -> Self {
459 Self {
460 reporter: self.reporter,
461 schema: self.schema.clone(),
462 started_at: self.started_at,
463 report_interval: self.report_interval,
464 next_running_at: self.next_running_at,
465 stage: self.stage.clone(),
466 }
467 }
468
469 /// Creates an event builder with a specific elapsed duration.
470 ///
471 /// # Parameters
472 ///
473 /// * `elapsed` - Elapsed duration to attach to the event.
474 ///
475 /// # Returns
476 ///
477 /// A builder carrying this run's schema and optional stage.
478 fn event_builder_with_elapsed(&self, elapsed: Duration) -> ProgressEventBuilder {
479 let builder = ProgressEvent::builder(self.schema.clone()).elapsed(elapsed);
480 match self.stage.clone() {
481 Some(stage) => builder.stage(stage),
482 None => builder,
483 }
484 }
485}
486
487/// Computes the next reporting instant while avoiding overflow panics.
488///
489/// # Parameters
490///
491/// * `base` - Base instant for the deadline.
492/// * `interval` - Duration added to `base`.
493///
494/// # Returns
495///
496/// `base + interval`, or `base` when the addition overflows.
497fn next_instant(base: Instant, interval: Duration) -> Instant {
498 base.checked_add(interval).unwrap_or(base)
499}