soma_ops/
progress.rs

1use std::convert::Infallible;
2
3use serde::{Deserialize, Serialize};
4
5use crate::{OperationId, OperationName, Timestamp};
6
7const MAX_PHASE_CHARS: usize = 128;
8const MAX_UNIT_CHARS: usize = 64;
9const MAX_MESSAGE_CHARS: usize = 1_024;
10
11/// One bounded monotonic progress update for an operation.
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
13#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
14pub struct ProgressEvent {
15    operation_id: OperationId,
16    operation: OperationName,
17    sequence: u64,
18    occurred_at: Timestamp,
19    phase: String,
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    current: Option<u64>,
22    #[serde(default, skip_serializing_if = "Option::is_none")]
23    total: Option<u64>,
24    #[serde(default, skip_serializing_if = "Option::is_none")]
25    unit: Option<String>,
26    #[serde(default, skip_serializing_if = "Option::is_none")]
27    message: Option<String>,
28}
29
30impl ProgressEvent {
31    /// Creates a progress event without a known quantity.
32    pub fn new(
33        operation_id: OperationId,
34        operation: OperationName,
35        sequence: u64,
36        occurred_at: Timestamp,
37        phase: impl Into<String>,
38    ) -> Result<Self, ProgressError> {
39        if sequence == 0 {
40            return Err(ProgressError::ZeroSequence);
41        }
42        let phase = phase.into();
43        validate_text("phase", &phase, MAX_PHASE_CHARS)?;
44        Ok(Self {
45            operation_id,
46            operation,
47            sequence,
48            occurred_at,
49            phase,
50            current: None,
51            total: None,
52            unit: None,
53            message: None,
54        })
55    }
56
57    /// Adds a current quantity, optional total, and optional unit.
58    pub fn with_amount(
59        mut self,
60        current: u64,
61        total: Option<u64>,
62        unit: Option<impl Into<String>>,
63    ) -> Result<Self, ProgressError> {
64        if total.is_some_and(|total| total == 0 || current > total) {
65            return Err(ProgressError::InvalidAmount { current, total });
66        }
67        let unit = unit.map(Into::into);
68        if let Some(unit) = &unit {
69            validate_text("unit", unit, MAX_UNIT_CHARS)?;
70        }
71        self.current = Some(current);
72        self.total = total;
73        self.unit = unit;
74        Ok(self)
75    }
76
77    /// Adds a bounded human-readable progress message.
78    pub fn with_message(mut self, message: impl Into<String>) -> Result<Self, ProgressError> {
79        let message = message.into();
80        validate_text("message", &message, MAX_MESSAGE_CHARS)?;
81        self.message = Some(message);
82        Ok(self)
83    }
84
85    /// Returns the operation identity.
86    #[must_use]
87    pub fn operation_id(&self) -> &OperationId {
88        &self.operation_id
89    }
90
91    /// Returns the canonical operation name.
92    #[must_use]
93    pub fn operation(&self) -> &OperationName {
94        &self.operation
95    }
96
97    /// Returns the monotonic, one-based event sequence.
98    #[must_use]
99    pub const fn sequence(&self) -> u64 {
100        self.sequence
101    }
102
103    /// Returns the event timestamp.
104    #[must_use]
105    pub const fn occurred_at(&self) -> Timestamp {
106        self.occurred_at
107    }
108
109    /// Returns the current operation phase.
110    #[must_use]
111    pub fn phase(&self) -> &str {
112        &self.phase
113    }
114
115    /// Returns the current quantity when known.
116    #[must_use]
117    pub const fn current(&self) -> Option<u64> {
118        self.current
119    }
120
121    /// Returns the total quantity when known.
122    #[must_use]
123    pub const fn total(&self) -> Option<u64> {
124        self.total
125    }
126
127    /// Returns the quantity unit when present.
128    #[must_use]
129    pub fn unit(&self) -> Option<&str> {
130        self.unit.as_deref()
131    }
132
133    /// Returns the bounded progress message when present.
134    #[must_use]
135    pub fn message(&self) -> Option<&str> {
136        self.message.as_deref()
137    }
138}
139
140/// Consumer of operation progress events.
141pub trait ProgressSink: Send + Sync {
142    /// Sink-specific error.
143    type Error;
144
145    /// Delivers one progress event.
146    fn report(&self, event: &ProgressEvent) -> Result<(), Self::Error>;
147}
148
149/// Progress sink that intentionally discards events.
150#[derive(Debug, Clone, Copy, Default)]
151pub struct NoopProgressSink;
152
153impl ProgressSink for NoopProgressSink {
154    type Error = Infallible;
155
156    fn report(&self, _event: &ProgressEvent) -> Result<(), Self::Error> {
157        Ok(())
158    }
159}
160
161/// Invalid progress event.
162#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
163#[non_exhaustive]
164pub enum ProgressError {
165    /// Sequence numbers are one-based.
166    #[error("progress sequence must be greater than zero")]
167    ZeroSequence,
168    /// Current progress exceeded total or total was zero.
169    #[error("invalid progress amount: current={current}, total={total:?}")]
170    InvalidAmount {
171        /// Current quantity.
172        current: u64,
173        /// Total quantity when known.
174        total: Option<u64>,
175    },
176    /// Text was empty, oversized, or contained control characters.
177    #[error("invalid progress {field}: expected 1..={max_chars} non-control characters")]
178    InvalidText {
179        /// Progress field.
180        field: &'static str,
181        /// Maximum accepted character count.
182        max_chars: usize,
183    },
184}
185
186fn validate_text(field: &'static str, value: &str, max_chars: usize) -> Result<(), ProgressError> {
187    let chars = value.chars().count();
188    if chars == 0 || chars > max_chars || value.chars().any(char::is_control) {
189        return Err(ProgressError::InvalidText { field, max_chars });
190    }
191    Ok(())
192}
193
194#[cfg(test)]
195#[path = "progress_tests.rs"]
196mod tests;
soma_ops/progress.rs

soma_ops/
progress.rs
