fdu_core/query/
query_status.rs

1//! Completeness and provenance for one coherent report snapshot.
2
3use std::path::PathBuf;
4use std::time::{Duration, SystemTime};
5
6use crate::{Coverage, CoverageReason, Freshness, Index, Issue, Source};
7
8use super::{IgnoredEntries, ReportSource, Request};
9
10/// Completeness of the facts used to answer one report.
11#[derive(Clone, Debug)]
12pub struct TreeStatus {
13    /// Whether every requested structural and content fact is represented.
14    pub complete: bool,
15    /// Coverage of the least complete requested tier.
16    pub coverage: Coverage,
17    /// Bounded, path-ordered operational failure details.
18    pub errors: Vec<Issue>,
19    /// Further failure details omitted by the shared retention bound.
20    pub errors_omitted: u64,
21}
22
23impl TreeStatus {
24    /// Derive status from the same immutable index snapshot used to build report rows.
25    pub fn of(index: &Index, request: &Request) -> Self {
26        let state = index.state();
27        let mut details: Vec<(PathBuf, Issue)> = Vec::with_capacity(crate::MAX_RETAINED_ISSUES);
28        let mut detail_count = 0_u64;
29        for issue in index.issues() {
30            detail_count = detail_count.saturating_add(1);
31            retain_first_detail(
32                &mut details,
33                (issue.path.clone().unwrap_or_default(), issue.clone()),
34            );
35        }
36        let unknown_population = request.query.selection.ignored != IgnoredEntries::Include
37            && index.control_table().refused_len() > 0;
38        if unknown_population {
39            for refusal in index.control_table().refusals() {
40                detail_count = detail_count.saturating_add(1);
41                let issue = Issue::provider_failure(
42                    Some(&refusal.path),
43                    format!(
44                        "ignored population is unknown below this refused control ({})",
45                        refusal.reason.label()
46                    ),
47                );
48                retain_first_detail(&mut details, (refusal.path, issue));
49            }
50        }
51        let wanted = index.content_identity(request.basis.content);
52        let admitted = index.content().and_then(|content| content.admit(&wanted));
53        let mut content_failures = 0_u64;
54        if request.basis.content.is_enabled() {
55            if let Some(content) = admitted {
56                for (path, analysis) in content.records() {
57                    let Some(reason) = analysis.operational_failure() else {
58                        continue;
59                    };
60                    content_failures = content_failures.saturating_add(1);
61                    detail_count = detail_count.saturating_add(1);
62                    let detail = analysis.error.clone().unwrap_or_else(|| match reason {
63                        crate::content::CoverageReason::IoError => {
64                            "content analysis could not read the file".to_string()
65                        }
66                        crate::content::CoverageReason::ChangedDuringRead => {
67                            "file changed during content analysis".to_string()
68                        }
69                        _ => unreachable!("operational_failure returns only operational reasons"),
70                    });
71                    retain_first_detail(
72                        &mut details,
73                        (path.to_path_buf(), Issue::provider_failure(Some(path), detail)),
74                    );
75                }
76            }
77        }
78        let content_tier_partial = request.basis.content.is_enabled()
79            && admitted
80                .and_then(crate::stored_state::ContentProjection::state)
81                .is_some_and(|tier| tier.freshness == Freshness::Partial);
82        let content_pending = index.content_has_pending(request.basis.content);
83        if content_tier_partial && content_failures == 0 {
84            detail_count = detail_count.saturating_add(1);
85            retain_first_detail(
86                &mut details,
87                (
88                    PathBuf::new(),
89                    Issue::provider_failure(
90                        None,
91                        "content analysis results became stale before they could be retained"
92                            .to_string(),
93                    ),
94                ),
95            );
96        }
97        let retained = u64::try_from(details.len()).unwrap_or(u64::MAX);
98        let errors = details.into_iter().map(|(_, issue)| issue).collect();
99        let complete = state.coverage == Coverage::Complete
100            && content_failures == 0
101            && !content_tier_partial
102            && !content_pending
103            && !unknown_population;
104        Self {
105            complete,
106            coverage: if content_failures == 0
107                && !content_tier_partial
108                && !content_pending
109                && !unknown_population
110            {
111                state.coverage
112            } else {
113                Coverage::Partial(CoverageReason::Failed)
114            },
115            errors,
116            errors_omitted: state
117                .issues
118                .omitted
119                .saturating_add(detail_count.saturating_sub(retained)),
120        }
121    }
122
123    /// Derive status from a one-shot walk, collapsing repeated causes first.
124    ///
125    /// Workers can meet one unreadable path more than once; the index retains each cause
126    /// once, so this path must too or the two routes disagree about `errors_omitted`.
127    pub(crate) fn of_walk(root: &std::path::Path, scan: &mut crate::ScanReport) -> Self {
128        crate::scan::normalize_walk_errors(root, &mut scan.errors);
129        let complete = scan.is_complete();
130        let mut details = Vec::with_capacity(crate::MAX_RETAINED_ISSUES);
131        let mut count = 0_u64;
132        for error in &scan.errors {
133            count = count.saturating_add(1);
134            let issue = crate::Issue::from_error_under(root, error);
135            retain_first_detail(&mut details, (issue.path.clone().unwrap_or_default(), issue));
136        }
137        let retained = u64::try_from(details.len()).unwrap_or(u64::MAX);
138        let errors = details.into_iter().map(|(_, issue)| issue).collect();
139        Self {
140            complete,
141            coverage: if complete {
142                Coverage::Complete
143            } else {
144                Coverage::Partial(CoverageReason::Inaccessible)
145            },
146            errors,
147            errors_omitted: count.saturating_sub(retained),
148        }
149    }
150}
151
152fn retain_first_detail(details: &mut Vec<(PathBuf, Issue)>, detail: (PathBuf, Issue)) {
153    let position = details
154        .binary_search_by(|current| {
155            current.0.cmp(&detail.0).then_with(|| current.1.message.cmp(&detail.1.message))
156        })
157        .unwrap_or_else(|position| position);
158    if position >= crate::MAX_RETAINED_ISSUES {
159        return;
160    }
161    details.insert(position, detail);
162    if details.len() > crate::MAX_RETAINED_ISSUES {
163        details.pop();
164    }
165}
166
167/// Source and currency of one retained tier.
168#[derive(Clone, Copy, PartialEq, Eq, Debug)]
169pub struct TierState {
170    /// Weakest source represented by this tier.
171    pub source: Source,
172    /// Current trust state of this tier.
173    pub freshness: Freshness,
174    /// When this tier's facts were observed, or `None` when the stored format cannot say.
175    pub observed_at_ns: Option<i64>,
176}
177
178/// Provenance of each tier contributing to a report.
179#[derive(Clone, Copy, PartialEq, Eq, Debug)]
180pub struct TierProvenance {
181    /// Retained filesystem-entry tier.
182    pub entries: TierState,
183    /// Sparse content-analysis tier when requested and present.
184    pub content: Option<TierState>,
185}
186
187/// How and when the coherent answer was produced.
188#[derive(Clone, Debug)]
189pub struct ReportProvenance {
190    /// Weakest source among tiers contributing to the report.
191    pub source: ReportSource,
192    /// Least fresh tier contributing to the report.
193    pub freshness: Freshness,
194    /// Conservative start watermark of the entry verification pass.
195    pub scan_started_at: Option<SystemTime>,
196    /// Caller-supplied instant at which this answer was generated.
197    pub generated_at: SystemTime,
198    /// Per-tier source, currency, and observation time.
199    pub tiers: TierProvenance,
200}
201
202impl ReportProvenance {
203    /// Derive report provenance for the requested tiers from one immutable index snapshot.
204    pub fn of(
205        index: &Index,
206        content_requested: crate::content::AnalysisSet,
207        generated_at: SystemTime,
208    ) -> Self {
209        let state = index.state();
210        let entries = TierState {
211            source: state.source,
212            freshness: state.freshness,
213            observed_at_ns: Some(index.writing_pass_started_at_ns()),
214        };
215        let scan_started_at = u64::try_from(index.writing_pass_started_at_ns())
216            .ok()
217            .map(|nanos| SystemTime::UNIX_EPOCH + Duration::from_nanos(nanos));
218        let content_pending = index.content_has_pending(content_requested);
219        let wanted = index.content_identity(content_requested);
220        let content = if content_requested.is_enabled() {
221            index.content().and_then(|content| content.admit(&wanted)).and_then(|content| {
222                content.state().map(|state| TierState {
223                    source: state.source,
224                    freshness: if content_pending { Freshness::Partial } else { state.freshness },
225                    observed_at_ns: state.observed_at_ns,
226                })
227            })
228        } else {
229            None
230        };
231        let source = content.map_or(entries.source, |state| entries.source.max(state.source));
232        let freshness = content
233            .map_or(entries.freshness, |state| least_fresh(entries.freshness, state.freshness));
234        Self {
235            source: report_source(source),
236            freshness,
237            scan_started_at,
238            generated_at,
239            tiers: TierProvenance { entries, content },
240        }
241    }
242
243    pub(crate) fn of_walk(started: SystemTime, generated_at: SystemTime, complete: bool) -> Self {
244        let observed_at_ns = crate::query::system_time_to_nanos(started);
245        let freshness = if complete { Freshness::Fresh } else { Freshness::Partial };
246        let entries = TierState { source: Source::Scanned, freshness, observed_at_ns };
247        Self {
248            source: ReportSource::ColdScan,
249            freshness,
250            scan_started_at: Some(started),
251            generated_at,
252            tiers: TierProvenance { entries, content: None },
253        }
254    }
255}
256
257fn least_fresh(left: Freshness, right: Freshness) -> Freshness {
258    let rank = |freshness| match freshness {
259        Freshness::Fresh => 0,
260        Freshness::Reconciling => 1,
261        Freshness::Stale => 2,
262        Freshness::Partial => 3,
263    };
264    if rank(left) >= rank(right) { left } else { right }
265}
266
267fn report_source(source: Source) -> ReportSource {
268    match source {
269        Source::Scanned => ReportSource::ColdScan,
270        Source::Revalidated | Source::JournalScoped => ReportSource::WarmRevalidate,
271        Source::Cached => ReportSource::CacheOnly,
272    }
273}
274
275#[cfg(test)]
276mod tests {
277    use std::fs;
278    use std::time::UNIX_EPOCH;
279
280    use super::{ReportProvenance, TreeStatus};
281    use crate::content::{AnalysisRequest, AnalysisSet, analyze_index};
282    use crate::query::{Basis, Query, Request};
283    use crate::scan::ScanConfig;
284    use crate::{Freshness, Index, Source};
285
286    #[test]
287    fn partial_content_tier_keeps_status_incomplete_without_a_failure_record() {
288        let mut index = Index::new("/unused");
289        let analysis = AnalysisSet::NONE.with_lines();
290        index.prepare_content_analysis(AnalysisRequest {
291            profile: analysis,
292            ..AnalysisRequest::default()
293        });
294        index.set_content_tier_state(Source::Scanned, Freshness::Partial, Some(1));
295        let mut basis = Basis::held_by(&index);
296        basis.content = analysis;
297        let request = Request::new(basis, Query::default(), UNIX_EPOCH);
298
299        let status = TreeStatus::of(&index, &request);
300
301        assert!(!status.complete);
302        assert_eq!(status.coverage, crate::Coverage::Partial(crate::CoverageReason::Failed));
303        assert_eq!(
304            status.errors[0].message,
305            "content analysis results became stale before they could be retained"
306        );
307    }
308
309    #[test]
310    fn metadata_only_provenance_excludes_an_unrequested_partial_content_tier() {
311        let mut index = Index::new("/unused");
312        index.prepare_content_analysis(AnalysisRequest {
313            profile: AnalysisSet::NONE.with_lines(),
314            ..AnalysisRequest::default()
315        });
316        index.set_content_tier_state(Source::Cached, Freshness::Partial, None);
317
318        let provenance = ReportProvenance::of(&index, AnalysisSet::NONE, UNIX_EPOCH);
319
320        assert_eq!(provenance.tiers.content, None);
321        assert_eq!(provenance.freshness, Freshness::Fresh);
322        assert_eq!(provenance.source, crate::query::ReportSource::ColdScan);
323    }
324
325    #[test]
326    fn repeated_content_read_failure_stays_incomplete_until_a_verified_recovery() {
327        let root = tempfile::tempdir().expect("root");
328        let path = root.path().join("failed.txt");
329        fs::write(&path, b"hello").expect("fixture");
330        let scan = ScanConfig::default();
331        let (mut index, report) = crate::scan::scan_into_index(root.path(), &scan).expect("scan");
332        assert!(report.is_complete());
333        fs::remove_file(&path).expect("make the retained candidate unreadable");
334        let analysis = AnalysisRequest { profile: AnalysisSet::NONE.with_lines(), workers: 1 };
335        let first = analyze_index(&mut index, analysis);
336        assert_eq!(first.lines.io_errors, 1);
337        let request = Request::new(Basis::held_by(&index), Query::default(), UNIX_EPOCH);
338
339        let first_status = TreeStatus::of(&index, &request);
340        assert!(!first_status.complete);
341        assert_eq!(first_status.errors.len(), 1);
342        assert_eq!(
343            first_status.errors[0].path.as_deref(),
344            Some(std::path::Path::new("failed.txt"))
345        );
346
347        let repeated = analyze_index(&mut index, analysis);
348        assert_eq!(repeated.lines.io_errors, 1, "a failed record must be retried");
349        assert!(!TreeStatus::of(&index, &request).complete);
350
351        fs::write(&path, b"recovered").expect("recover file");
352        crate::scan::reconcile(&mut index, &scan, &mut |_| {}).expect("reconcile recovery");
353        let recovered = analyze_index(&mut index, analysis);
354        assert_eq!(recovered.lines.analyzed, 1);
355        let status = TreeStatus::of(&index, &request);
356        assert!(status.complete, "a successful reread clears the operational failure");
357        assert!(status.errors.is_empty());
358    }
359
360    /// A walk can meet one unreadable directory from several workers. The summary fast
361    /// path must collapse those repeats as the index does, or `--view summary` and
362    /// `--view tree` disagree about how many errors were omitted (R113-4).
363    #[test]
364    fn walk_status_counts_each_unreadable_path_once() {
365        let root = std::path::Path::new("/root");
366        let denied = |number: usize| crate::Error::Io {
367            path: root.join(format!("denied-{number:02}")),
368            source: std::io::Error::from(std::io::ErrorKind::PermissionDenied),
369        };
370        let mut scan = crate::ScanReport::default();
371        for number in (0..40).rev() {
372            scan.errors.push(denied(number));
373            scan.errors.push(denied(number));
374        }
375
376        let status = TreeStatus::of_walk(root, &mut scan);
377
378        assert!(!status.complete);
379        assert_eq!(status.errors.len(), 40);
380        assert_eq!(status.errors_omitted, 0);
381        assert_eq!(status.errors[0].path.as_deref(), Some(std::path::Path::new("denied-00")));
382    }
383
384    #[test]
385    fn content_failures_retain_the_first_paths_and_count_every_omission() {
386        let root = tempfile::tempdir().expect("root");
387        for number in (0..66).rev() {
388            fs::write(root.path().join(format!("file-{number:02}.txt")), b"x").expect("fixture");
389        }
390        let (mut index, report) =
391            crate::scan::scan_into_index(root.path(), &ScanConfig::default()).expect("scan");
392        assert!(report.is_complete());
393        for number in 0..66 {
394            fs::remove_file(root.path().join(format!("file-{number:02}.txt")))
395                .expect("make candidate unreadable");
396        }
397        analyze_index(
398            &mut index,
399            AnalysisRequest { profile: AnalysisSet::NONE.with_lines(), workers: 2 },
400        );
401        let request = Request::new(Basis::held_by(&index), Query::default(), UNIX_EPOCH);
402
403        let status = TreeStatus::of(&index, &request);
404
405        assert!(!status.complete);
406        assert_eq!(status.errors.len(), crate::MAX_RETAINED_ISSUES);
407        assert_eq!(status.errors_omitted, 2);
408        let retained_paths: Vec<_> =
409            status.errors.iter().map(|issue| issue.path.as_deref().expect("path")).collect();
410        assert_eq!(retained_paths[0], std::path::Path::new("file-00.txt"));
411        assert_eq!(retained_paths[63], std::path::Path::new("file-63.txt"));
412        assert!(retained_paths.windows(2).all(|pair| pair[0] < pair[1]));
413        assert!(!retained_paths.contains(&std::path::Path::new("file-64.txt")));
414        assert!(!retained_paths.contains(&std::path::Path::new("file-65.txt")));
415        assert_eq!(
416            index
417                .content()
418                .expect("content")
419                .records()
420                .filter(|(_, record)| { record.operational_failure().is_some() })
421                .count(),
422            66,
423            "the diagnostic bound must not discard retained failure state"
424        );
425    }
426}
fdu_core/query/query_status.rs

fdu_core/query/
query_status.rs
