pipedash_core/infrastructure/run_logs/
checkpoint.rs

1use std::fs::{
2    self,
3    File,
4    OpenOptions,
5};
6use std::io::{
7    self,
8    BufWriter,
9    Seek,
10    SeekFrom,
11    Write,
12};
13use std::path::Path;
14use std::sync::Arc;
15
16use serde::{
17    Deserialize,
18    Serialize,
19};
20use sha2::{
21    Digest,
22    Sha256,
23};
24
25use super::block::INDEX_RECORD_BYTES;
26use super::document::Document;
27use super::ingest::{
28    BLOCKS_FILE,
29    INDEX_FILE,
30};
31use super::outline::{
32    Outline,
33    StepBound,
34};
35use super::store::{
36    short_hex,
37    ReadLease,
38};
39use super::MAX_LINE_BYTES;
40use crate::domain::LogCursor;
41
42pub(super) const CHECKPOINT_FILE: &str = "checkpoint.json";
43pub(super) const TAIL_FILE: &str = "tail.bin";
44
45#[derive(Serialize, Deserialize)]
46pub(super) struct Resume<O = Outline> {
47    pub cursor: LogCursor,
48    pub line_count: u32,
49    pub text_bytes: u64,
50    pub first_ts: Option<i64>,
51    pub blocks_len: u64,
52    pub index_len: u64,
53    pub styles: usize,
54    pub outline: O,
55}
56
57#[derive(Serialize, Deserialize)]
58pub(super) struct Checkpoint<O = Outline> {
59    pub steps_hash: String,
60    pub raw_hash: Option<String>,
61    pub resume: Option<Resume<O>>,
62}
63
64pub(super) fn steps_hash(steps: &[StepBound]) -> String {
65    let json = serde_json::to_vec(steps).unwrap_or_default();
66    short_hex(&Sha256::digest(json))
67}
68
69pub(super) fn copy_prefix(from: &Path, to: &Path, len: u64) -> io::Result<File> {
70    fs::copy(from, to)?;
71    let mut file = OpenOptions::new().write(true).open(to)?;
72    file.set_len(len)?;
73    file.seek(SeekFrom::Start(len))?;
74    Ok(file)
75}
76
77pub(super) fn remove(dir: &Path) {
78    for name in [CHECKPOINT_FILE, TAIL_FILE] {
79        if let Err(e) = fs::remove_file(dir.join(name)) {
80            if e.kind() != io::ErrorKind::NotFound {
81                tracing::warn!("Couldn't remove a stale log checkpoint: {e}");
82            }
83        }
84    }
85}
86
87impl Checkpoint<&Outline> {
88    pub(super) fn write(&self, dir: &Path, tail: &[u8]) -> io::Result<()> {
89        let mut writer = BufWriter::new(File::create(dir.join(CHECKPOINT_FILE))?);
90        serde_json::to_writer(&mut writer, self).map_err(io::Error::from)?;
91        writer
92            .into_inner()
93            .map_err(|e| e.into_error())?
94            .sync_all()?;
95        if !tail.is_empty() {
96            let mut file = File::create(dir.join(TAIL_FILE))?;
97            file.write_all(tail)?;
98            file.sync_all()?;
99        }
100        Ok(())
101    }
102}
103
104impl Checkpoint {
105    pub(super) fn load(document: &Document) -> Option<Self> {
106        let dir = document.dir();
107        let checkpoint: Self =
108            serde_json::from_slice(&fs::read(dir.join(CHECKPOINT_FILE)).ok()?).ok()?;
109        let meta = document.meta();
110        let fits = checkpoint.resume.as_ref().is_none_or(|resume| {
111            let len = |name: &str| fs::metadata(dir.join(name)).map(|m| m.len()).ok();
112            resume.styles <= meta.styles.len()
113                && resume.line_count <= meta.line_count
114                && resume.index_len % INDEX_RECORD_BYTES as u64 == 0
115                && len(BLOCKS_FILE) >= Some(resume.blocks_len)
116                && len(INDEX_FILE) >= Some(resume.index_len)
117                && resume.outline.spool_fits(dir)
118                && len(TAIL_FILE).unwrap_or(0) <= MAX_LINE_BYTES as u64
119        });
120        fits.then_some(checkpoint)
121    }
122}
123
124pub struct Base {
125    pub revision: i64,
126    pub(super) document: Arc<Document>,
127    pub(super) _lease: ReadLease,
128    pub(super) checkpoint: Option<Checkpoint>,
129}
130
131impl Base {
132    pub fn cursor(&self, steps: &[StepBound]) -> Option<&LogCursor> {
133        let resume = self.checkpoint.as_ref()?.resume.as_ref()?;
134        resume.outline.accepts(steps).then_some(&resume.cursor)
135    }
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[test]
143    fn copied_prefix_can_be_truncated_and_extended_without_changing_the_source() {
144        let dir = tempfile::tempdir().unwrap();
145        let from = dir.path().join("base");
146        let to = dir.path().join("resumed");
147        fs::write(&from, b"prefix-stale-tail").unwrap();
148
149        let mut file = copy_prefix(&from, &to, 6).unwrap();
150        file.write_all(b"-new-tail").unwrap();
151        file.sync_all().unwrap();
152
153        assert_eq!(fs::read(&to).unwrap(), b"prefix-new-tail");
154        assert_eq!(fs::read(&from).unwrap(), b"prefix-stale-tail");
155    }
156}
pipedash_core/infrastructure/run_logs/checkpoint.rs

pipedash_core/infrastructure/run_logs/
checkpoint.rs
