tailtriage_core/
run_builder.rs

1use crate::request_ids::add_duplicate_completed_request_id_warning;
2use crate::{
3    collector::generate_run_id, unix_time_ms, BuildError, CaptureLimits, CaptureMode,
4    EffectiveCoreConfig, EffectiveTokioSamplerConfig, InFlightSnapshot, QueueEvent, RequestEvent,
5    Run, RunEndReason, RunMetadata, RuntimeSnapshot, StageEvent, UnfinishedRequests,
6};
7use core::fmt;
8
9/// Options for assembling a completed [`Run`] artifact.
10///
11/// This API is for completed evidence assembly (for example, import/conversion
12/// paths), not live request instrumentation. Normal live instrumentation should
13/// use [`crate::Tailtriage::builder`].
14///
15/// When omitted:
16/// - mode defaults to [`CaptureMode::Light`]
17/// - capture limits default to `mode.core_defaults()`
18/// - host and pid remain `None`
19/// - run id uses the same core run-id generator as live capture
20/// - timestamps are filled from one captured current unix-ms value
21///
22/// `finalized_at_unix_ms` is always `Some(...)` for [`RunBuilder`] output.
23#[derive(Debug, Clone)]
24pub struct RunBuilderOptions {
25    service_name: String,
26    service_version: Option<String>,
27    run_id: Option<String>,
28    mode: Option<CaptureMode>,
29    capture_limits: Option<CaptureLimits>,
30    strict_lifecycle: bool,
31    started_at_unix_ms: Option<u64>,
32    finished_at_unix_ms: Option<u64>,
33    finalized_at_unix_ms: Option<u64>,
34    host: Option<String>,
35    pid: Option<u32>,
36    run_end_reason: Option<RunEndReason>,
37}
38
39impl RunBuilderOptions {
40    /// Creates options with a required service name.
41    #[must_use]
42    pub fn new(service_name: impl Into<String>) -> Self {
43        Self {
44            service_name: service_name.into(),
45            service_version: None,
46            run_id: None,
47            mode: None,
48            capture_limits: None,
49            strict_lifecycle: false,
50            started_at_unix_ms: None,
51            finished_at_unix_ms: None,
52            finalized_at_unix_ms: None,
53            host: None,
54            pid: None,
55            run_end_reason: None,
56        }
57    }
58
59    /// Sets optional service version metadata.
60    #[must_use]
61    pub fn service_version(mut self, service_version: impl Into<String>) -> Self {
62        self.service_version = Some(service_version.into());
63        self
64    }
65    /// Sets a caller-provided run identifier.
66    #[must_use]
67    pub fn run_id(mut self, run_id: impl Into<String>) -> Self {
68        self.run_id = Some(run_id.into());
69        self
70    }
71    /// Sets capture mode.
72    #[must_use]
73    pub fn mode(mut self, mode: CaptureMode) -> Self {
74        self.mode = Some(mode);
75        self
76    }
77    /// Sets effective capture limits for this completed run artifact.
78    #[must_use]
79    pub fn capture_limits(mut self, capture_limits: CaptureLimits) -> Self {
80        self.capture_limits = Some(capture_limits);
81        self
82    }
83    /// Sets strict lifecycle flag recorded in effective core config.
84    #[must_use]
85    pub const fn strict_lifecycle(mut self, strict_lifecycle: bool) -> Self {
86        self.strict_lifecycle = strict_lifecycle;
87        self
88    }
89    /// Sets start timestamp in unix milliseconds.
90    #[must_use]
91    pub const fn started_at_unix_ms(mut self, started_at_unix_ms: u64) -> Self {
92        self.started_at_unix_ms = Some(started_at_unix_ms);
93        self
94    }
95    /// Sets finish timestamp in unix milliseconds.
96    #[must_use]
97    pub const fn finished_at_unix_ms(mut self, finished_at_unix_ms: u64) -> Self {
98        self.finished_at_unix_ms = Some(finished_at_unix_ms);
99        self
100    }
101    /// Sets finalization timestamp in unix milliseconds.
102    #[must_use]
103    pub const fn finalized_at_unix_ms(mut self, finalized_at_unix_ms: u64) -> Self {
104        self.finalized_at_unix_ms = Some(finalized_at_unix_ms);
105        self
106    }
107    /// Sets optional host metadata.
108    #[must_use]
109    pub fn host(mut self, host: impl Into<String>) -> Self {
110        self.host = Some(host.into());
111        self
112    }
113    /// Sets optional process identifier metadata.
114    #[must_use]
115    pub const fn pid(mut self, pid: u32) -> Self {
116        self.pid = Some(pid);
117        self
118    }
119    /// Sets optional run-end reason metadata.
120    #[must_use]
121    pub const fn run_end_reason(mut self, run_end_reason: RunEndReason) -> Self {
122        self.run_end_reason = Some(run_end_reason);
123        self
124    }
125}
126
127/// Validation error returned when a pushed event or snapshot has invalid shape.
128#[derive(Debug, Clone, PartialEq, Eq)]
129pub enum RunBuilderEventError {
130    /// A field value on an event/snapshot failed validation.
131    InvalidEvent {
132        /// Event/snapshot type name.
133        event: &'static str,
134        /// Invalid field name.
135        field: &'static str,
136        /// Human-readable validation reason.
137        reason: String,
138    },
139}
140
141impl fmt::Display for RunBuilderEventError {
142    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
143        match self {
144            Self::InvalidEvent {
145                event,
146                field,
147                reason,
148            } => {
149                write!(f, "invalid {event}.{field}: {reason}")
150            }
151        }
152    }
153}
154
155impl std::error::Error for RunBuilderEventError {}
156
157/// Advanced/import API for completed-run artifact assembly.
158///
159/// [`RunBuilder`] assembles a finalized [`Run`] from already measured evidence.
160/// It validates event shape and timestamp ordering, but does not perform live
161/// request lifecycle tracking.
162///
163/// Elapsed duration fields such as request latency, stage latency, and queue
164/// wait are accepted as authoritative completed evidence. [`RunBuilder`] does
165/// not synthesize or repair durations from wall-clock timestamps.
166///
167/// Push methods use first-N retention through the same bounded
168/// retention/truncation helper used by the live collector. Overflow items are
169/// dropped and reflected in [`Run::truncation`].
170#[derive(Debug)]
171pub struct RunBuilder {
172    run: Run,
173    capture_limits: CaptureLimits,
174}
175
176impl RunBuilder {
177    /// Creates a new completed-run builder from [`RunBuilderOptions`].
178    ///
179    /// # Errors
180    ///
181    /// Returns [`BuildError::EmptyServiceName`] when the service name is blank.
182    ///
183    /// Returns [`BuildError::InvalidRunTimeBounds`] when finished timestamp is
184    /// earlier than start timestamp.
185    ///
186    /// Returns [`BuildError::InvalidFinalizationTime`] when finalization
187    /// timestamp is earlier than finished timestamp.
188    pub fn new(options: RunBuilderOptions) -> Result<Self, BuildError> {
189        if options.service_name.trim().is_empty() {
190            return Err(BuildError::EmptyServiceName);
191        }
192
193        let mode = options.mode.unwrap_or(CaptureMode::Light);
194        let capture_limits = options
195            .capture_limits
196            .unwrap_or_else(|| mode.core_defaults());
197        let ts = unix_time_ms();
198        let started_at_unix_ms = options.started_at_unix_ms.unwrap_or(ts);
199        let finished_at_unix_ms = options.finished_at_unix_ms.unwrap_or(ts);
200        let finalized_at_unix_ms_value =
201            options.finalized_at_unix_ms.unwrap_or(finished_at_unix_ms);
202        let finalized_at_unix_ms = Some(finalized_at_unix_ms_value);
203
204        if finished_at_unix_ms < started_at_unix_ms {
205            return Err(BuildError::InvalidRunTimeBounds {
206                started_at_unix_ms,
207                finished_at_unix_ms,
208            });
209        }
210
211        if finalized_at_unix_ms_value < finished_at_unix_ms {
212            return Err(BuildError::InvalidFinalizationTime {
213                finished_at_unix_ms,
214                finalized_at_unix_ms: finalized_at_unix_ms_value,
215            });
216        }
217
218        Ok(Self {
219            run: Run::new(RunMetadata {
220                run_id: options.run_id.unwrap_or_else(generate_run_id),
221                service_name: options.service_name,
222                service_version: options.service_version,
223                started_at_unix_ms,
224                finished_at_unix_ms,
225                finalized_at_unix_ms,
226                mode,
227                effective_core_config: Some(EffectiveCoreConfig {
228                    mode,
229                    capture_limits,
230                    strict_lifecycle: options.strict_lifecycle,
231                }),
232                effective_tokio_sampler_config: None,
233                host: options.host,
234                pid: options.pid,
235                lifecycle_warnings: Vec::new(),
236                unfinished_requests: UnfinishedRequests::default(),
237                run_end_reason: options.run_end_reason,
238            }),
239            capture_limits,
240        })
241    }
242
243    /// Appends a request event.
244    ///
245    /// The event is retained only while request capture-limit capacity remains;
246    /// otherwise it is dropped and `truncation.dropped_requests` is updated.
247    ///
248    /// # Errors
249    ///
250    /// Returns [`RunBuilderEventError`] when the event has invalid shape.
251    pub fn push_request(&mut self, event: RequestEvent) -> Result<(), RunBuilderEventError> {
252        validate_request_event(&event)?;
253        let _ = crate::retention::push_request_bounded(&mut self.run, self.capture_limits, event);
254        Ok(())
255    }
256    /// Appends a stage event.
257    ///
258    /// The event is retained only while stage capture-limit capacity remains;
259    /// otherwise it is dropped and `truncation.dropped_stages` is updated.
260    ///
261    /// # Errors
262    ///
263    /// Returns [`RunBuilderEventError`] when the event has invalid shape.
264    pub fn push_stage(&mut self, event: StageEvent) -> Result<(), RunBuilderEventError> {
265        validate_stage_event(&event)?;
266        let _ = crate::retention::push_stage_bounded(&mut self.run, self.capture_limits, event);
267        Ok(())
268    }
269    /// Appends a queue event.
270    ///
271    /// The event is retained only while queue capture-limit capacity remains;
272    /// otherwise it is dropped and `truncation.dropped_queues` is updated.
273    ///
274    /// # Errors
275    ///
276    /// Returns [`RunBuilderEventError`] when the event has invalid shape.
277    pub fn push_queue(&mut self, event: QueueEvent) -> Result<(), RunBuilderEventError> {
278        validate_queue_event(&event)?;
279        let _ = crate::retention::push_queue_bounded(&mut self.run, self.capture_limits, event);
280        Ok(())
281    }
282    /// Appends an in-flight snapshot.
283    ///
284    /// The snapshot is retained only while in-flight snapshot capture-limit
285    /// capacity remains; otherwise it is dropped and
286    /// `truncation.dropped_inflight_snapshots` is updated.
287    ///
288    /// # Errors
289    ///
290    /// Returns [`RunBuilderEventError`] when the snapshot has invalid shape.
291    pub fn push_inflight_snapshot(
292        &mut self,
293        snapshot: InFlightSnapshot,
294    ) -> Result<(), RunBuilderEventError> {
295        validate_inflight_snapshot(&snapshot)?;
296        let _ = crate::retention::push_inflight_snapshot_bounded(
297            &mut self.run,
298            self.capture_limits,
299            snapshot,
300        );
301        Ok(())
302    }
303    /// Appends a runtime snapshot.
304    ///
305    /// The snapshot is retained only while runtime snapshot capture-limit
306    /// capacity remains; otherwise it is dropped and
307    /// `truncation.dropped_runtime_snapshots` is updated.
308    ///
309    /// # Errors
310    ///
311    /// Currently returns `Ok(())` for all runtime snapshots. The `Result`
312    /// keeps this API consistent with other [`RunBuilder`] push methods and
313    /// leaves room for future runtime snapshot validation without changing
314    /// the method shape.
315    pub fn push_runtime_snapshot(
316        &mut self,
317        snapshot: RuntimeSnapshot,
318    ) -> Result<(), RunBuilderEventError> {
319        let _ = crate::retention::push_runtime_snapshot_bounded(
320            &mut self.run,
321            self.capture_limits,
322            snapshot,
323        );
324        Ok(())
325    }
326    /// Adds one lifecycle warning string.
327    pub fn add_lifecycle_warning(&mut self, warning: impl Into<String>) {
328        self.run.metadata.lifecycle_warnings.push(warning.into());
329    }
330    /// Sets unfinished-request metadata.
331    pub fn set_unfinished_requests(&mut self, unfinished: UnfinishedRequests) {
332        self.run.metadata.unfinished_requests = unfinished;
333    }
334    /// Sets effective Tokio sampler configuration metadata.
335    pub fn set_effective_tokio_sampler_config(&mut self, config: EffectiveTokioSamplerConfig) {
336        self.run.metadata.effective_tokio_sampler_config = Some(config);
337    }
338    /// Sets run-end reason only when absent.
339    pub fn set_run_end_reason_if_absent(&mut self, reason: RunEndReason) {
340        if self.run.metadata.run_end_reason.is_none() {
341            self.run.metadata.run_end_reason = Some(reason);
342        }
343    }
344    /// Consumes the builder and returns the assembled finalized [`Run`].
345    ///
346    /// This does not perform lifecycle validation or synthesize missing
347    /// completions.
348    #[must_use]
349    pub fn finish(mut self) -> Run {
350        add_duplicate_completed_request_id_warning(&mut self.run);
351        self.run
352    }
353}
354
355fn invalid_event(
356    event: &'static str,
357    field: &'static str,
358    reason: impl Into<String>,
359) -> RunBuilderEventError {
360    RunBuilderEventError::InvalidEvent {
361        event,
362        field,
363        reason: reason.into(),
364    }
365}
366
367fn validate_request_event(event: &RequestEvent) -> Result<(), RunBuilderEventError> {
368    if event.request_id.trim().is_empty() {
369        return Err(invalid_event(
370            "RequestEvent",
371            "request_id",
372            "must not be empty",
373        ));
374    }
375    if event.route.trim().is_empty() {
376        return Err(invalid_event("RequestEvent", "route", "must not be empty"));
377    }
378    if event.finished_at_unix_ms < event.started_at_unix_ms {
379        return Err(invalid_event(
380            "RequestEvent",
381            "finished_at_unix_ms",
382            "must be >= started_at_unix_ms",
383        ));
384    }
385    if let (Some(started_at_run_us), Some(finished_at_run_us)) =
386        (event.started_at_run_us, event.finished_at_run_us)
387    {
388        if finished_at_run_us < started_at_run_us {
389            return Err(invalid_event(
390                "RequestEvent",
391                "finished_at_run_us",
392                "must be >= started_at_run_us",
393            ));
394        }
395    }
396    if event.outcome.trim().is_empty() {
397        return Err(invalid_event(
398            "RequestEvent",
399            "outcome",
400            "must not be empty",
401        ));
402    }
403    Ok(())
404}
405fn validate_stage_event(event: &StageEvent) -> Result<(), RunBuilderEventError> {
406    if event.request_id.trim().is_empty() {
407        return Err(invalid_event(
408            "StageEvent",
409            "request_id",
410            "must not be empty",
411        ));
412    }
413    if event.stage.trim().is_empty() {
414        return Err(invalid_event("StageEvent", "stage", "must not be empty"));
415    }
416    if event.finished_at_unix_ms < event.started_at_unix_ms {
417        return Err(invalid_event(
418            "StageEvent",
419            "finished_at_unix_ms",
420            "must be >= started_at_unix_ms",
421        ));
422    }
423    if let (Some(started_at_run_us), Some(finished_at_run_us)) =
424        (event.started_at_run_us, event.finished_at_run_us)
425    {
426        if finished_at_run_us < started_at_run_us {
427            return Err(invalid_event(
428                "StageEvent",
429                "finished_at_run_us",
430                "must be >= started_at_run_us",
431            ));
432        }
433    }
434    Ok(())
435}
436fn validate_queue_event(event: &QueueEvent) -> Result<(), RunBuilderEventError> {
437    if event.request_id.trim().is_empty() {
438        return Err(invalid_event(
439            "QueueEvent",
440            "request_id",
441            "must not be empty",
442        ));
443    }
444    if event.queue.trim().is_empty() {
445        return Err(invalid_event("QueueEvent", "queue", "must not be empty"));
446    }
447    if event.waited_until_unix_ms < event.waited_from_unix_ms {
448        return Err(invalid_event(
449            "QueueEvent",
450            "waited_until_unix_ms",
451            "must be >= waited_from_unix_ms",
452        ));
453    }
454    if let (Some(waited_from_run_us), Some(waited_until_run_us)) =
455        (event.waited_from_run_us, event.waited_until_run_us)
456    {
457        if waited_until_run_us < waited_from_run_us {
458            return Err(invalid_event(
459                "QueueEvent",
460                "waited_until_run_us",
461                "must be >= waited_from_run_us",
462            ));
463        }
464    }
465    Ok(())
466}
467fn validate_inflight_snapshot(snapshot: &InFlightSnapshot) -> Result<(), RunBuilderEventError> {
468    if snapshot.gauge.trim().is_empty() {
469        return Err(invalid_event(
470            "InFlightSnapshot",
471            "gauge",
472            "must not be empty",
473        ));
474    }
475    Ok(())
476}
tailtriage_core/run_builder.rs

tailtriage_core/
run_builder.rs
