ifc_lite_server/routes/parse/
stream_progress.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! The per-batch progress checkpoints a live parquet-stream parse reported,
6//! cached so a later cache hit can replay the SAME numbers.
7//!
8//! `progress` events are in units of geometry JOBS, not meshes: the pipeline
9//! calls its batch callback with `processed_jobs` / `total_jobs`
10//! (`processing/src/processor/mod.rs`), and a job can produce several meshes
11//! or none. A replay that counted the meshes it emitted would therefore
12//! report different units AND a different total from a miss on the same file,
13//! over the same public event. Nothing in the geometry blob records the job
14//! counts, so the live path writes them here and the replay reads them back.
15
16use crate::services::cache::DiskCache;
17use serde::{Deserialize, Serialize};
18
19/// Cache key for the checkpoint sidecar, keyed alongside the geometry and
20/// metadata entries for the same request.
21///
22/// `v1` is `{ total_jobs, after_batch }`. Bump on any change to what those
23/// numbers mean; an entry that fails to deserialize is treated as absent, so
24/// a bump degrades to the mesh-unit fallback rather than to garbage.
25pub(super) fn stream_progress_cache_key(cache_key: &str) -> String {
26    format!("{cache_key}-parquet-progress-v1")
27}
28
29/// What a live parse reported, in job units.
30#[derive(Debug, Default, Clone, Serialize, Deserialize)]
31pub(super) struct StreamProgress {
32    /// The `total` every `progress` event carried (the pipeline's `total_jobs`).
33    pub total_jobs: usize,
34    /// `processed_jobs` as of each `batch` event, in batch order. One entry
35    /// per batch, so `after_batch.len()` must equal the replay's batch count
36    /// for the checkpoints to line up.
37    pub after_batch: Vec<usize>,
38}
39
40/// Collects [`StreamProgress`] from the live event stream.
41///
42/// The pipeline emits `Batch` then the `Progress` that accounts for it, so a
43/// batch's checkpoint is the value of the NEXT progress event. Progress
44/// events with no batch before them (the leading `processed: 0`, and any
45/// chunk that produced no meshes) carry no batch to attach to and are only
46/// used for `total_jobs`.
47#[derive(Default)]
48pub(super) struct StreamProgressRecorder {
49    progress: StreamProgress,
50    batch_pending: bool,
51}
52
53impl StreamProgressRecorder {
54    pub(super) fn on_batch(&mut self) {
55        self.batch_pending = true;
56    }
57
58    pub(super) fn on_progress(&mut self, processed: usize, total: usize) {
59        self.progress.total_jobs = total;
60        if self.batch_pending {
61            self.progress.after_batch.push(processed);
62            self.batch_pending = false;
63        }
64    }
65
66    /// Take what was recorded, leaving the recorder empty.
67    pub(super) fn take(&mut self) -> StreamProgress {
68        std::mem::take(&mut self.progress)
69    }
70}
71
72/// Store the checkpoints beside the geometry. Best-effort: a failure here
73/// costs the next cache hit its job-unit progress, not its correctness.
74pub(super) async fn cache_stream_progress(
75    cache: &DiskCache,
76    cache_key: &str,
77    progress: &StreamProgress,
78) {
79    let Ok(bytes) = serde_json::to_vec(progress) else {
80        return;
81    };
82    if let Err(e) = cache
83        .set_bytes(&stream_progress_cache_key(cache_key), &bytes)
84        .await
85    {
86        tracing::warn!(error = %e, "Failed to cache stream progress checkpoints");
87    }
88}
89
90/// Read back the checkpoints for a cache hit. `None` for an entry written
91/// before this sidecar existed, or one that no longer deserializes.
92pub(super) async fn load_stream_progress(
93    cache: &DiskCache,
94    cache_key: &str,
95) -> Option<StreamProgress> {
96    let bytes = cache
97        .get_bytes(&stream_progress_cache_key(cache_key))
98        .await
99        .ok()??;
100    serde_json::from_slice(&bytes).ok()
101}
102
103#[cfg(test)]
104#[path = "stream_progress_tests.rs"]
105mod stream_progress_tests;
ifc_lite_server/routes/parse/stream_progress.rs

ifc_lite_server/routes/parse/
stream_progress.rs
