clawft_kernel/cog_ingest/
store_log.rs

1//! Durable log behind a project's ingest store (ADR-100 decision 5).
2//!
3//! The owner daemon keeps each project's vectors in an in-memory index
4//! ([`super::store::VectorBackendStore`]); this log is what lets a restart
5//! rebuild it. One file per project under the daemon runtime dir.
6//!
7//! Layout: an 8-byte magic, then one frame per accepted batch:
8//! `u32 payload_len | payload | u32 crc`, where `crc` is the first four bytes
9//! of `blake3(payload)` and the payload is `u32 count` followed by records
10//! `u16 instance_len | instance | u16 node_len | node | u64 id | DIMS * f32`
11//! (all little endian). A batch is written with one `write_all` and synced
12//! before the ingest is acknowledged, so a crash leaves whole batches plus at
13//! most one torn frame at the tail; [`VectorLog::open`] drops a torn or
14//! corrupt tail and truncates the file back to the last good frame.
15//!
16//! The file is capped: a batch that would take it past the cap is refused up
17//! front (the store reports it full) and nothing is written. The log only
18//! grows; there is no compaction yet, so repeated upserts of one id consume
19//! space until the cap.
20
21use std::fs::{File, OpenOptions};
22use std::io::{self, Read, Seek, SeekFrom, Write};
23use std::path::{Path, PathBuf};
24
25use super::types::DIMS;
26
27/// File magic and format version.
28const MAGIC: &[u8; 8] = b"WCIVEC01";
29/// Default cap on one project's log file.
30pub const DEFAULT_MAX_LOG_BYTES: u64 = 64 * 1024 * 1024;
31/// Largest instance or node id a record may carry.
32const MAX_NAME: usize = 256;
33
34/// One accepted vector as persisted.
35#[derive(Debug, Clone, PartialEq)]
36pub struct LogRecord {
37    /// Instance that posted it.
38    pub instance: String,
39    /// Node whose bridge forwarded it.
40    pub node: String,
41    /// The id the instance gave it.
42    pub id: u64,
43    /// The vector.
44    pub values: [f32; DIMS],
45}
46
47impl LogRecord {
48    /// Encoded size of this record.
49    pub fn encoded_len(&self) -> u64 {
50        (2 + self.instance.len() + 2 + self.node.len() + 8 + DIMS * 4) as u64
51    }
52
53    /// Whether the ids fit the format (they are bounded by the bridge's
54    /// validation; this keeps the file reader's bound honest).
55    pub fn is_encodable(&self) -> bool {
56        self.instance.len() <= MAX_NAME && self.node.len() <= MAX_NAME
57    }
58
59    fn encode(&self, out: &mut Vec<u8>) {
60        out.extend_from_slice(&(self.instance.len() as u16).to_le_bytes());
61        out.extend_from_slice(self.instance.as_bytes());
62        out.extend_from_slice(&(self.node.len() as u16).to_le_bytes());
63        out.extend_from_slice(self.node.as_bytes());
64        out.extend_from_slice(&self.id.to_le_bytes());
65        for f in &self.values {
66            out.extend_from_slice(&f.to_le_bytes());
67        }
68    }
69}
70
71/// Bytes a batch of `records` adds to the file (frame overhead included).
72pub fn frame_len(records: &[LogRecord]) -> u64 {
73    // u32 len + u32 count + u32 crc
74    12 + records.iter().map(LogRecord::encoded_len).sum::<u64>()
75}
76
77struct Cursor<'a>(&'a [u8]);
78
79impl<'a> Cursor<'a> {
80    fn take(&mut self, n: usize) -> Option<&'a [u8]> {
81        (self.0.len() >= n).then(|| {
82            let (a, b) = self.0.split_at(n);
83            self.0 = b;
84            a
85        })
86    }
87    fn u16(&mut self) -> Option<u16> {
88        Some(u16::from_le_bytes(self.take(2)?.try_into().ok()?))
89    }
90    fn u32(&mut self) -> Option<u32> {
91        Some(u32::from_le_bytes(self.take(4)?.try_into().ok()?))
92    }
93    fn name(&mut self) -> Option<String> {
94        let n = self.u16()? as usize;
95        if n > MAX_NAME {
96            return None;
97        }
98        String::from_utf8(self.take(n)?.to_vec()).ok()
99    }
100    fn record(&mut self) -> Option<LogRecord> {
101        let instance = self.name()?;
102        let node = self.name()?;
103        let id = u64::from_le_bytes(self.take(8)?.try_into().ok()?);
104        let mut values = [0f32; DIMS];
105        for v in &mut values {
106            *v = f32::from_le_bytes(self.take(4)?.try_into().ok()?);
107        }
108        Some(LogRecord { instance, node, id, values })
109    }
110}
111
112fn crc(payload: &[u8]) -> [u8; 4] {
113    let h = blake3::hash(payload);
114    [h.as_bytes()[0], h.as_bytes()[1], h.as_bytes()[2], h.as_bytes()[3]]
115}
116
117/// What follows the last good frame.
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
119enum Tail {
120    /// Nothing: the file ends on a frame boundary.
121    Clean,
122    /// An incomplete frame at the end of the file: what a crash mid-append leaves.
123    Torn,
124    /// A complete frame that fails its checksum or does not parse. Anything
125    /// after it may still hold good data.
126    Corrupt,
127}
128
129/// Decode frames from `data` (after the magic). Returns the records, the
130/// byte offset (into `data`) of the end of the last good frame, and what
131/// follows it.
132fn decode_frames(data: &[u8]) -> (Vec<LogRecord>, usize, Tail) {
133    let mut out = Vec::new();
134    let mut good = 0usize;
135    let mut cur = Cursor(data);
136    let tail = loop {
137        if cur.0.is_empty() {
138            break Tail::Clean;
139        }
140        let Some(len) = cur.u32() else { break Tail::Torn };
141        let Some(payload) = cur.take(len as usize) else { break Tail::Torn };
142        let Some(sum) = cur.take(4) else { break Tail::Torn };
143        if sum != crc(payload) {
144            break Tail::Corrupt;
145        }
146        let mut p = Cursor(payload);
147        let Some(count) = p.u32() else { break Tail::Corrupt };
148        let mut batch = Vec::new();
149        let mut ok = true;
150        for _ in 0..count {
151            match p.record() {
152                Some(r) => batch.push(r),
153                None => {
154                    ok = false;
155                    break;
156                }
157            }
158        }
159        if !ok || !p.0.is_empty() {
160            break Tail::Corrupt;
161        }
162        out.extend(batch);
163        good = data.len() - cur.0.len();
164    };
165    (out, good, tail)
166}
167
168/// Why a log could not be opened.
169#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
170pub enum OpenError {
171    /// The file is not a usable log (wrong magic, over the cap). Safe to move aside.
172    #[error("{0}")]
173    Corrupt(String),
174    /// A filesystem error that may be transient; the file is not implicated.
175    #[error("{0}")]
176    Io(String),
177}
178
179/// Run blocking file work without stalling an async worker: on a
180/// multi-threaded tokio runtime the worker is handed off (`block_in_place`),
181/// anywhere else the closure just runs.
182pub fn run_blocking<T>(f: impl FnOnce() -> T) -> T {
183    match tokio::runtime::Handle::try_current() {
184        Ok(h) if h.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread => {
185            tokio::task::block_in_place(f)
186        }
187        _ => f(),
188    }
189}
190
191fn now_secs() -> u64 {
192    std::time::SystemTime::now()
193        .duration_since(std::time::UNIX_EPOCH)
194        .map(|d| d.as_secs())
195        .unwrap_or_default()
196}
197
198fn create_private_dir(dir: &Path) -> io::Result<()> {
199    let mut b = std::fs::DirBuilder::new();
200    b.recursive(true);
201    #[cfg(unix)]
202    {
203        use std::os::unix::fs::DirBuilderExt;
204        b.mode(0o700);
205    }
206    b.create(dir)
207}
208
209/// An open, append-only vector log.
210pub struct VectorLog {
211    path: PathBuf,
212    file: File,
213    size: u64,
214    cap: u64,
215}
216
217impl VectorLog {
218    /// Open (creating if absent) the log at `path` and return every record it
219    /// holds. A torn tail (a crash mid-append) is dropped and the file
220    /// truncated back to the last good frame. A complete frame that fails its
221    /// checksum is corruption: a warning with the dropped byte count is logged,
222    /// the dropped tail is first copied to `<file>.corrupt.<unix ts>`, and the
223    /// file is truncated so appends resume on a frame boundary. A file with
224    /// the wrong magic, or larger than `cap`, is [`OpenError::Corrupt`] and is
225    /// left untouched.
226    pub fn open(path: &Path, cap: u64) -> Result<(Self, Vec<LogRecord>), OpenError> {
227        let io = |e: io::Error| OpenError::Io(format!("{}: {e}", path.display()));
228        if let Some(parent) = path.parent() {
229            create_private_dir(parent).map_err(io)?;
230        }
231        let mut opts = OpenOptions::new();
232        opts.read(true).write(true).create(true);
233        #[cfg(unix)]
234        {
235            use std::os::unix::fs::OpenOptionsExt;
236            opts.mode(0o600);
237        }
238        let mut file = opts.open(path).map_err(io)?;
239        let len = file.metadata().map_err(io)?.len();
240        if len > cap {
241            return Err(OpenError::Corrupt(format!(
242                "{} is {len} bytes, over the {cap} byte cap",
243                path.display()
244            )));
245        }
246        let mut data = Vec::with_capacity(len as usize);
247        file.read_to_end(&mut data).map_err(io)?;
248        let (records, size) = if data.is_empty() {
249            file.write_all(MAGIC).map_err(io)?;
250            file.sync_data().map_err(io)?;
251            (Vec::new(), MAGIC.len() as u64)
252        } else if data.len() < MAGIC.len() || &data[..MAGIC.len()] != MAGIC {
253            return Err(OpenError::Corrupt(format!("{} is not a vector log", path.display())));
254        } else {
255            let body = &data[MAGIC.len()..];
256            let (records, good, tail) = decode_frames(body);
257            let size = (MAGIC.len() + good) as u64;
258            let dropped = body.len() - good;
259            // A length-prefix bit flip reads as "torn" yet can hide a lot of good data, so any
260            // dropped tail over this size is kept, whatever it was classified as.
261            const KEEP_OVER: usize = 512;
262            match tail {
263                Tail::Clean => {}
264                Tail::Torn => {
265                    tracing::warn!(path = %path.display(), dropped, "vector log ends in a torn frame; dropping it");
266                }
267                Tail::Corrupt => {
268                    tracing::warn!(
269                        path = %path.display(), dropped,
270                        "vector log has a corrupt frame; dropping it and everything after it"
271                    );
272                }
273            }
274            if tail == Tail::Corrupt || dropped > KEEP_OVER {
275                let aside = path.with_extension(format!(
276                    "{}.corrupt.{}",
277                    path.extension().and_then(|e| e.to_str()).unwrap_or("vec"),
278                    now_secs()
279                ));
280                tracing::warn!(path = %path.display(), dropped, kept = %aside.display(), "keeping a copy of the dropped vector log tail");
281                std::fs::write(&aside, &body[good..]).map_err(io)?;
282            }
283            if tail != Tail::Clean {
284                file.set_len(size).map_err(io)?;
285                file.sync_data().map_err(io)?;
286            }
287            (records, size)
288        };
289        file.seek(SeekFrom::Start(size)).map_err(io)?;
290        Ok((Self { path: path.to_path_buf(), file, size, cap }, records))
291    }
292
293    /// Whether a batch of `records` fits under the cap.
294    pub fn has_room_for(&self, records: &[LogRecord]) -> bool {
295        records.is_empty() || self.size + frame_len(records) <= self.cap
296    }
297
298    /// Append one batch as a single frame and sync it. Nothing is written
299    /// when the frame would pass the cap.
300    pub fn append(&mut self, records: &[LogRecord]) -> io::Result<()> {
301        if records.is_empty() {
302            return Ok(());
303        }
304        if !self.has_room_for(records) {
305            return Err(io::Error::other(format!(
306                "{} would pass its {} byte cap",
307                self.path.display(),
308                self.cap
309            )));
310        }
311        let mut payload = Vec::with_capacity(frame_len(records) as usize);
312        payload.extend_from_slice(&(records.len() as u32).to_le_bytes());
313        for r in records {
314            r.encode(&mut payload);
315        }
316        let mut frame = Vec::with_capacity(payload.len() + 8);
317        frame.extend_from_slice(&(payload.len() as u32).to_le_bytes());
318        frame.extend_from_slice(&payload);
319        frame.extend_from_slice(&crc(&payload));
320        self.file.write_all(&frame)?;
321        self.file.sync_data()?;
322        self.size += frame.len() as u64;
323        Ok(())
324    }
325
326    /// Bytes in the file.
327    pub fn size(&self) -> u64 {
328        self.size
329    }
330}
clawft_kernel/cog_ingest/store_log.rs

clawft_kernel/cog_ingest/
store_log.rs
