artifactize_core/monitor/
loader.rs

1//! One in-flight snapshot job, isolated from UI navigation. Generation checks discard stale
2//! scopes/pages; state hints coalesce in the driver, and a read reads again what they named.
3//!
4//! [`Loader`] is the UI-agnostic seam: the terminal monitor and the desktop app both read
5//! their catalog, Runs page and selected Run through it.
6use super::{Scope, catalog, model};
7use crate::{
8    changes::Changed,
9    store::{self, RunSummary, reader::Reader},
10    types::RunId,
11};
12use std::{path::PathBuf, sync::Arc};
13use time::OffsetDateTime;
14
15/// Which Runs page of which scope a snapshot reads, and the Run it opens.
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct Page {
18    pub scope: Scope,
19    /// The selected Run; `None` opens the first (or, with `tail`, the last) Run of the page.
20    pub open: Option<RunId>,
21    pub offset: u32,
22    pub tail: bool,
23    pub limit: u32,
24}
25
26/// Where the page of older Runs starts, when the page shown (`shown` Runs of `limit`) is full.
27pub fn older_offset(offset: u32, limit: u32, shown: u32) -> Option<u32> {
28    (shown >= limit).then(|| offset.saturating_add(limit))
29}
30
31/// Where the page of newer Runs starts, when this one is not the first.
32pub fn newer_offset(offset: u32, limit: u32) -> Option<u32> {
33    (offset > 0).then(|| offset.saturating_sub(limit))
34}
35
36/// The monitor's page plus what tells an obsolete job from a current one.
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub struct Selection {
39    pub state: PathBuf,
40    pub page: Page,
41    pub revision: u64,
42}
43
44/// Everything one snapshot read needs; the caller keeps the catalog between reads.
45pub struct Input {
46    pub state: PathBuf,
47    /// cwd, `--repo` or `--all` (`None`): only the first selection.
48    pub initial: Option<PathBuf>,
49    pub catalog: catalog::Catalog,
50    /// False before the first accepted snapshot, which picks the initial scope.
51    pub initialized: bool,
52    pub page: Page,
53    /// Read the database catalog and the shown Run afresh; Git discovery stays cached until
54    /// [`Loader::rescan`].
55    pub full: bool,
56    /// The Run the caller already shows; it is read again only when a hint named it.
57    pub displayed: Option<RunId>,
58    /// What the change hints since the last read named.
59    pub changed: Changed,
60    /// The page the caller already shows, reused while the database is unchanged.
61    pub runs: Option<Vec<RunSummary>>,
62}
63
64/// Where a snapshot was read from: the state identity and the database file.
65#[derive(Debug, Clone, Default, PartialEq, Eq)]
66pub struct Origin {
67    identity: Option<crate::types::StateId>,
68    database: Option<crate::platform::FileIdentity>,
69}
70impl Origin {
71    /// The origin of a snapshot read from the state `identity` in the database file `database`.
72    pub(crate) fn new(
73        identity: Option<crate::types::StateId>,
74        database: Option<crate::platform::FileIdentity>,
75    ) -> Self {
76        Self { identity, database }
77    }
78    /// The state was replaced since `previous` was read: Details, drafts and sessions
79    /// of the old state no longer apply.
80    pub fn replaces(&self, previous: &Self) -> bool {
81        previous.database.is_some() && previous.database != self.database
82            || previous.identity.is_some() && previous.identity != self.identity
83    }
84    pub fn identity(&self) -> Option<&crate::types::StateId> {
85        self.identity.as_ref()
86    }
87    /// The state's origin as it stands, read without a snapshot: its database file, then its
88    /// identity. A UI that keeps work outside snapshots (open reviews, drafts, claims taken)
89    /// compares it with [`Origin::replaces`] as the monitor compares snapshots.
90    pub fn read(state: &std::path::Path) -> Result<Self, String> {
91        let path = state.join(store::DATABASE);
92        let database =
93            match crate::platform::open_no_follow(std::fs::File::options().read(true), &path) {
94                Ok(file) => Some(crate::platform::file_identity(&file).map_err(|e| e.to_string())?),
95                Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
96                Err(error) => return Err(error.to_string()),
97            };
98        Ok(Self {
99            identity: store::read_state_id(state)?,
100            database,
101        })
102    }
103}
104
105/// One consistent read: the catalog, the scope's Runs page and the selected Run with its tree.
106pub struct Snapshot {
107    pub origin: Origin,
108    pub catalog: catalog::Catalog,
109    pub scope: Scope,
110    pub runs: Vec<RunSummary>,
111    pub selected_id: Option<RunId>,
112    /// The selected Run when it was read again: its requests, projected tree and a share in its
113    /// memory, kept after the rest so the memory goes back once nothing shows the Run.
114    pub selected: Option<super::live::Opened>,
115}
116
117/// A persistent, read-only snapshot worker with its caches. Call [`Loader::load`] on a
118/// blocking thread; it never writes.
119#[derive(Default)]
120pub struct Loader {
121    reader: Reader,
122    /// The catalog is supplied by the caller, so reset invalidates its Git caches on load.
123    reset_git: bool,
124}
125impl Loader {
126    /// The SELECT statements read since the last call, for tests that count what a read costs.
127    #[cfg(any(test, feature = "test-support"))]
128    pub fn traced_statements(&self) -> Vec<String> {
129        self.reader.traced_statements()
130    }
131
132    pub fn load(&mut self, mut input: Input) -> Result<Snapshot, String> {
133        if self.reset_git {
134            input.catalog.reset_git();
135        }
136        let result = load(&mut self.reader, input);
137        if result.is_ok() {
138            self.reset_git = false;
139        }
140        result
141    }
142    /// Ask Git again on the next load. The state is read by queries, which keep nothing that a
143    /// change could make wrong.
144    pub fn rescan(&mut self) {
145        self.reset_git = true;
146    }
147}
148
149pub struct Completed {
150    pub selection: Selection,
151    pub loader: Loader,
152    pub result: Result<Snapshot, String>,
153    /// The SELECT statements the read ran.
154    #[cfg(any(test, feature = "test-support"))]
155    pub queries: Vec<String>,
156}
157impl Completed {
158    /// Read the snapshot of `input` with `loader`, keeping what the caller needs to apply it.
159    pub fn run(selection: Selection, mut loader: Loader, input: Input) -> Self {
160        #[cfg(any(test, feature = "test-support"))]
161        loader.traced_statements();
162        let result = loader.load(input);
163        #[cfg(any(test, feature = "test-support"))]
164        let queries = loader.traced_statements();
165        Self {
166            selection,
167            loader,
168            result,
169            #[cfg(any(test, feature = "test-support"))]
170            queries,
171        }
172    }
173}
174
175fn load(reader: &mut Reader, input: Input) -> Result<Snapshot, String> {
176    let mut catalog = input.catalog;
177    let mut scope = input.page.scope.clone();
178    // Empty state is a normal baseline. File safety checks still run on the worker.
179    if !input
180        .state
181        .join(store::DATABASE)
182        .try_exists()
183        .map_err(|error| error.to_string())?
184    {
185        store::check_probe_files(&input.state)?;
186        catalog.update(&[], input.initial.as_deref());
187        if !input.initialized {
188            scope = catalog.initial(input.initial.as_deref());
189        }
190        reader.reset();
191        return Ok(Snapshot {
192            origin: Origin::default(),
193            catalog,
194            scope,
195            runs: Vec::new(),
196            selected_id: None,
197            selected: None,
198        });
199    }
200    let saved = reader.snapshot_optional(&input.state, |db, cache, state| {
201        let reuse_page = input.runs.is_some()
202            && !input.full
203            && input.initialized
204            && cache.has_catalog()
205            && !cache.changed();
206        if input.initialized {
207            catalog.selected = Some(scope.clone());
208        }
209        if input.full || !input.initialized || !cache.has_catalog() || cache.changed() {
210            let facts = cache.catalog(db, state)?;
211            if input.initialized {
212                catalog.selected = Some(scope.clone());
213            }
214            catalog.update(&facts, input.initial.as_deref());
215            if !input.initialized {
216                scope = catalog.initial(input.initial.as_deref());
217                catalog.selected = Some(scope.clone());
218                catalog.update(&facts, input.initial.as_deref());
219            }
220        }
221        if !input.full && input.initialized {
222            catalog.rebuild();
223        }
224        let paths = catalog.paths(&scope);
225        let runs = if reuse_page {
226            input.runs.clone().expect("accepted same-database page")
227        } else {
228            cache.page(
229                db,
230                state,
231                paths.as_deref(),
232                input.page.limit,
233                input.page.offset,
234                input.page.open.as_ref(),
235            )?
236        };
237        let selected = input
238            .page
239            .open
240            .as_ref()
241            .and_then(|id| runs.iter().find(|run| &run.id == id))
242            .or_else(|| {
243                if input.page.tail {
244                    runs.iter().take(input.page.limit as usize).next_back()
245                } else {
246                    runs.first()
247                }
248            });
249        let selected_id = selected.map(|run| run.id.clone());
250        if selected_id.is_none() {
251            // Nothing shows a Run: the one held goes, and with its last share its memory.
252            cache.close();
253        }
254        let mut runs = runs;
255        // The shown Run is read again when a hint named it; another Run is read in full.
256        let shown = |id: &RunId| input.displayed.as_ref() == Some(id);
257        let changed = input.changed.or_written(cache.changed());
258        let read = selected_id
259            .as_ref()
260            .filter(|id| !shown(id) || input.full || changed.run(id) || !cache.holds(id));
261        let selected = match read {
262            Some(id) => match cache.selected(db, state, id, !shown(id)) {
263                Ok(selected) => selected.map(|(run, requests)| {
264                    let footprint = cache.footprint().expect("the Run just read is held");
265                    (run, requests, footprint)
266                }),
267                // The Run's own record does not decode: it is listed with why, and not opened.
268                Err(store::receipts::Error::Unreadable(record)) => {
269                    let run = runs.iter_mut().find(|run| &run.id == id);
270                    let run = run.expect("the selected Run is listed");
271                    if !run.unreadable.contains(&record) {
272                        run.unreadable.push(record);
273                    }
274                    None
275                }
276                Err(error) => return Err(error),
277            },
278            None => None,
279        };
280        Ok((cache.identity(db)?, runs, selected_id, selected))
281    })?;
282    let Some((identity, runs, selected_id, selected)) = saved else {
283        catalog.update(&[], input.initial.as_deref());
284        if !input.initialized {
285            scope = catalog.initial(input.initial.as_deref());
286        }
287        return Ok(Snapshot {
288            origin: Origin {
289                identity: None,
290                database: reader.database(),
291            },
292            catalog,
293            scope,
294            runs: Vec::new(),
295            selected_id: None,
296            selected: None,
297        });
298    };
299    // No read transaction survives heavy projection or live-file reads.
300    let selected = selected.map(|(run, requests, footprint)| {
301        let nodes = Arc::new(model::tree(&run, &requests, OffsetDateTime::now_utc()));
302        (run, requests, nodes, footprint)
303    });
304    reader.require_current(&input.state)?;
305    Ok(Snapshot {
306        origin: Origin {
307            identity,
308            database: reader.database(),
309        },
310        catalog,
311        scope,
312        runs,
313        selected_id,
314        selected,
315    })
316}
artifactize_core/monitor/loader.rs

artifactize_core/monitor/
loader.rs
