kcode_k1_audio_classification_projection_cache/
lib.rs

1use kcode_k1_audio_classification_callback_proof::verify_callback;
2use kcode_k1_audio_classification_projection_cache_format::{
3    CacheCursor, CacheRecord, decode_cursor, decode_record, encode_cursor, encode_record,
4    fragment_filename, parse_fragment_filename,
5};
6use kcode_k1_txn_ordering::K1TxnOrdering;
7pub use kcode_k1_txn_ordering::TxId;
8use std::collections::BTreeSet;
9use std::fs::{self, File, OpenOptions};
10use std::io::Write;
11use std::path::{Path, PathBuf};
12
13const LOGS: &str = "audio-classification-fragments";
14const CURSOR: &str = "audio-classification-cursor.json";
15const TEMP: &str = "audio-classification-cursor.json.tmp";
16const SQLITE: &str = "audio-classification.sqlite3";
17
18#[derive(Clone, Debug, PartialEq, Eq)]
19pub struct CachedCallback {
20    pub callback_txid: TxId,
21    pub fragment_id: TxId,
22    pub payload: Vec<u8>,
23}
24
25fn cached(record: CacheRecord) -> CachedCallback {
26    CachedCallback {
27        callback_txid: record.callback_txid,
28        fragment_id: record.fragment_id,
29        payload: record.payload,
30    }
31}
32
33pub struct ProjectionCache {
34    root: PathBuf,
35    sequence: u64,
36    callbacks: BTreeSet<[u8; 12]>,
37    poisoned: bool,
38}
39
40type Loaded = (Vec<CachedCallback>, Option<TxId>);
41
42impl ProjectionCache {
43    pub fn open(
44        root: &Path,
45        ordering: &K1TxnOrdering,
46    ) -> Result<(Self, Vec<CachedCallback>, Option<TxId>), String> {
47        fs::create_dir_all(root).map_err(|error| context("create root", root, error))?;
48        for suffix in ["", "-wal", "-shm"] {
49            remove_file(&root.join(format!("{SQLITE}{suffix}")))?;
50        }
51        let (records, cursor) = match load(root, ordering)? {
52            Some(value) => value,
53            None => {
54                reset(root)?;
55                (Vec::new(), None)
56            }
57        };
58        let cache = Self {
59            root: root.into(),
60            sequence: records.len() as u64,
61            callbacks: records
62                .iter()
63                .map(|record| *record.callback_txid.as_bytes())
64                .collect(),
65            poisoned: false,
66        };
67        Ok((cache, records, cursor))
68    }
69
70    pub fn append(
71        &mut self,
72        callback_txid: TxId,
73        fragment_id: TxId,
74        payload: &[u8],
75    ) -> Result<(), String> {
76        self.ensure_healthy()?;
77        if self.callbacks.contains(callback_txid.as_bytes()) {
78            return Err("duplicate callback transaction".into());
79        }
80        let sequence = self
81            .sequence
82            .checked_add(1)
83            .ok_or("cache sequence overflow")?;
84        let record = encode_record(&CacheRecord {
85            sequence,
86            callback_txid,
87            fragment_id,
88            payload: payload.to_vec(),
89        })?;
90        let cursor = encode_cursor(CacheCursor {
91            last_sequence: sequence,
92            last_callback_txid: callback_txid,
93        })?;
94        if let Err(error) = persist(&self.root, fragment_id, &record, &cursor) {
95            self.poisoned = true;
96            return Err(error);
97        }
98        self.sequence = sequence;
99        self.callbacks.insert(*callback_txid.as_bytes());
100        Ok(())
101    }
102
103    pub fn clear(&mut self) -> Result<(), String> {
104        self.ensure_healthy()?;
105        if let Err(error) = reset(&self.root) {
106            self.poisoned = true;
107            return Err(error);
108        }
109        self.sequence = 0;
110        self.callbacks.clear();
111        Ok(())
112    }
113
114    pub fn ensure_healthy(&self) -> Result<(), String> {
115        (!self.poisoned)
116            .then_some(())
117            .ok_or_else(|| "cache is poisoned; reopen required".into())
118    }
119}
120
121fn load(root: &Path, ordering: &K1TxnOrdering) -> Result<Option<Loaded>, String> {
122    let logs = root.join(LOGS);
123    let cursor_path = root.join(CURSOR);
124    if metadata(&root.join(TEMP))?.is_some() {
125        return Ok(None);
126    }
127    if !metadata(&logs)?.is_some_and(|value| value.file_type().is_dir()) {
128        return Ok(None);
129    }
130    let Some(cursor_metadata) = metadata(&cursor_path)? else {
131        let mut entries = fs::read_dir(&logs).map_err(|error| context("read", &logs, error))?;
132        return match entries.next() {
133            None => Ok(Some((Vec::new(), None))),
134            Some(Ok(_)) => Ok(None),
135            Some(Err(error)) => Err(context("read", &logs, error)),
136        };
137    };
138    if !cursor_metadata.file_type().is_file() {
139        return Ok(None);
140    }
141    let Ok(cursor) = decode_cursor(&read(&cursor_path)?) else {
142        return Ok(None);
143    };
144    let Some(mut records) = read_records(&logs)? else {
145        return Ok(None);
146    };
147    records.sort_by_key(|record| record.0);
148    if u64::try_from(records.len()).ok() != Some(cursor.last_sequence) {
149        return Ok(None);
150    }
151    let mut callbacks = BTreeSet::new();
152    if records.iter().enumerate().any(|(index, record)| {
153        record.0 != index as u64 + 1 || !callbacks.insert(*record.1.callback_txid.as_bytes())
154    }) || records.last().map(|record| record.1.callback_txid) != Some(cursor.last_callback_txid)
155    {
156        return Ok(None);
157    }
158    for (_, record) in &records {
159        if !verify_callback(ordering, record.callback_txid, &record.payload)? {
160            return Ok(None);
161        }
162    }
163    Ok(Some((
164        records.into_iter().map(|record| record.1).collect(),
165        Some(cursor.last_callback_txid),
166    )))
167}
168
169fn read_records(logs: &Path) -> Result<Option<Vec<(u64, CachedCallback)>>, String> {
170    let mut records = Vec::new();
171    for entry in fs::read_dir(logs).map_err(|error| context("read", logs, error))? {
172        let entry = entry.map_err(|error| context("read", logs, error))?;
173        let kind = entry
174            .file_type()
175            .map_err(|error| context("inspect", &entry.path(), error))?;
176        if !kind.is_file() {
177            return Ok(None);
178        }
179        let name = entry.file_name();
180        let Some(name) = name.to_str() else {
181            return Ok(None);
182        };
183        let Ok(fragment) = parse_fragment_filename(name) else {
184            return Ok(None);
185        };
186        let bytes = read(&entry.path())?;
187        if bytes.is_empty() || bytes.last() != Some(&b'\n') {
188            return Ok(None);
189        }
190        let mut previous = 0;
191        for line in bytes.split_inclusive(|byte| *byte == b'\n') {
192            let Ok(record) = decode_record(line) else {
193                return Ok(None);
194            };
195            if record.sequence <= previous || record.fragment_id != fragment {
196                return Ok(None);
197            }
198            previous = record.sequence;
199            let sequence = record.sequence;
200            records.push((sequence, cached(record)));
201        }
202    }
203    Ok(Some(records))
204}
205
206fn persist(root: &Path, fragment: TxId, record: &[u8], cursor: &[u8]) -> Result<(), String> {
207    let logs = root.join(LOGS);
208    let path = logs.join(fragment_filename(fragment));
209    let mut file = OpenOptions::new()
210        .create(true)
211        .append(true)
212        .open(&path)
213        .map_err(|error| context("open", &path, error))?;
214    file.write_all(record)
215        .and_then(|_| file.flush())
216        .and_then(|_| file.sync_all())
217        .map_err(|error| context("sync", &path, error))?;
218    sync_dir(&logs)?;
219    let temporary = root.join(TEMP);
220    let mut file =
221        File::create(&temporary).map_err(|error| context("create", &temporary, error))?;
222    file.write_all(cursor)
223        .and_then(|_| file.flush())
224        .and_then(|_| file.sync_all())
225        .map_err(|error| context("sync", &temporary, error))?;
226    fs::rename(&temporary, root.join(CURSOR))
227        .map_err(|error| context("rename", &temporary, error))?;
228    sync_dir(root)
229}
230
231fn reset(root: &Path) -> Result<(), String> {
232    remove_path(&root.join(LOGS))?;
233    remove_file(&root.join(CURSOR))?;
234    remove_file(&root.join(TEMP))?;
235    let logs = root.join(LOGS);
236    fs::create_dir(&logs).map_err(|error| context("create", &logs, error))?;
237    sync_dir(&logs)?;
238    sync_dir(root)
239}
240
241fn metadata(path: &Path) -> Result<Option<fs::Metadata>, String> {
242    match fs::symlink_metadata(path) {
243        Ok(value) => Ok(Some(value)),
244        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
245        Err(error) => Err(context("inspect", path, error)),
246    }
247}
248
249fn read(path: &Path) -> Result<Vec<u8>, String> {
250    fs::read(path).map_err(|error| context("read", path, error))
251}
252
253fn remove_path(path: &Path) -> Result<(), String> {
254    match fs::symlink_metadata(path) {
255        Ok(value) if value.file_type().is_dir() => {
256            fs::remove_dir_all(path).map_err(|error| context("remove", path, error))
257        }
258        Ok(_) => fs::remove_file(path).map_err(|error| context("remove", path, error)),
259        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
260        Err(error) => Err(context("inspect", path, error)),
261    }
262}
263
264fn remove_file(path: &Path) -> Result<(), String> {
265    match fs::remove_file(path) {
266        Ok(()) => Ok(()),
267        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
268        Err(error) => Err(context("remove", path, error)),
269    }
270}
271
272fn sync_dir(path: &Path) -> Result<(), String> {
273    File::open(path)
274        .and_then(|file| file.sync_all())
275        .map_err(|error| context("sync", path, error))
276}
277
278fn context(action: &str, path: &Path, error: impl std::fmt::Display) -> String {
279    format!("{action} {}: {error}", path.display())
280}
281
282#[cfg(test)]
283mod tests;
kcode_k1_audio_classification_projection_cache/lib.rs

kcode_k1_audio_classification_projection_cache/
lib.rs
