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

pipedash_core/infrastructure/run_logs/
checkpoint.rs
