laminar_connectors/files/manifest/
mod.rs

1//! Exact file ingestion inventory.
2//!
3//! Only immutable file identities (paths) are correctness state. The exact set
4//! supports logarithmic membership/insertion, while an immutable serialized
5//! fragment log makes source-checkpoint snapshots constant-time.
6
7use std::collections::BTreeSet;
8use std::fmt;
9use std::sync::{Arc, RwLock};
10
11use crate::checkpoint::{PersistentOffset, SourceCheckpoint};
12
13/// Read-only exact membership view used by the discovery task.
14///
15/// The view shares the live path index, so newly completed files become known
16/// without rebuilding or copying the full inventory.
17#[derive(Clone)]
18pub(super) struct FileInventorySnapshot {
19    paths: Arc<RwLock<BTreeSet<String>>>,
20}
21
22impl FileInventorySnapshot {
23    /// Returns `true` only when the exact path has been ingested.
24    pub(super) fn contains(&self, path: &str) -> bool {
25        self.paths
26            .read()
27            .unwrap_or_else(std::sync::PoisonError::into_inner)
28            .contains(path)
29    }
30}
31
32impl fmt::Debug for FileInventorySnapshot {
33    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
34        f.debug_struct("FileInventorySnapshot")
35            .field(
36                "path_count",
37                &self
38                    .paths
39                    .read()
40                    .unwrap_or_else(std::sync::PoisonError::into_inner)
41                    .len(),
42            )
43            .finish()
44    }
45}
46
47/// Tracks every ingested file with exact membership.
48pub struct FileIngestionManifest {
49    paths: Arc<RwLock<BTreeSet<String>>>,
50    serialized_paths: PersistentOffset,
51}
52
53impl FileIngestionManifest {
54    /// Creates an empty manifest.
55    #[must_use]
56    pub fn new() -> Self {
57        Self {
58            paths: Arc::new(RwLock::new(BTreeSet::new())),
59            serialized_paths: PersistentOffset::new("[", ",", "]"),
60        }
61    }
62
63    /// Returns `true` only when the exact path has been ingested.
64    pub fn contains(&self, path: &str) -> bool {
65        self.paths
66            .read()
67            .unwrap_or_else(std::sync::PoisonError::into_inner)
68            .contains(path)
69    }
70
71    /// Inserts an immutable file identity in `O(log N)`.
72    pub fn insert(&mut self, path: String) {
73        let fragment = serde_json::Value::String(path.clone()).to_string();
74        let inserted = self
75            .paths
76            .write()
77            .unwrap_or_else(std::sync::PoisonError::into_inner)
78            .insert(path);
79        if inserted {
80            self.serialized_paths.push_fragment(fragment);
81        }
82    }
83
84    /// Returns the number of processed paths.
85    #[must_use]
86    pub fn processed_count(&self) -> usize {
87        self.paths
88            .read()
89            .unwrap_or_else(std::sync::PoisonError::into_inner)
90            .len()
91    }
92
93    /// Creates an `O(1)` exact membership view for the discovery engine.
94    #[must_use]
95    pub(super) fn snapshot_for_dedup(&self) -> FileInventorySnapshot {
96        FileInventorySnapshot {
97            paths: Arc::clone(&self.paths),
98        }
99    }
100
101    /// Adds the immutable serialized inventory snapshot to a source checkpoint
102    /// without materializing or copying the full path set.
103    pub fn to_checkpoint(&self, checkpoint: &mut SourceCheckpoint) {
104        checkpoint.set_persistent_offset("manifest", self.serialized_paths.clone());
105    }
106
107    /// Restores the exact inventory from a durable checkpoint.
108    ///
109    /// # Errors
110    ///
111    /// Returns an error if deserialization fails.
112    pub fn from_checkpoint(checkpoint: &SourceCheckpoint) -> Result<Self, serde_json::Error> {
113        let paths: BTreeSet<String> = match checkpoint.get_offset("manifest") {
114            Some(json) => serde_json::from_str(json)?,
115            None => BTreeSet::new(),
116        };
117        let mut manifest = Self::new();
118        for path in paths {
119            manifest.insert(path);
120        }
121        Ok(manifest)
122    }
123}
124
125impl Default for FileIngestionManifest {
126    fn default() -> Self {
127        Self::new()
128    }
129}
130
131impl fmt::Debug for FileIngestionManifest {
132    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
133        f.debug_struct("FileIngestionManifest")
134            .field("path_count", &self.processed_count())
135            .field(
136                "serialized_fragments",
137                &self.serialized_paths.fragment_count(),
138            )
139            .finish()
140    }
141}
142#[cfg(test)]
143mod tests;
laminar_connectors/files/manifest/mod.rs

laminar_connectors/files/manifest/
mod.rs
