lace_sans_io/runtime_support/compaction/
mod.rs

1//! In-place native arena compaction shared by the native filesystem and LMDB
2//! stores.
3//!
4//! [`Compaction`] owns steps A (plan), B (state), C (batched move), E (tail
5//! release), and F (finish) of `docs/architecture/store-compaction.md`, and
6//! their recovery. Each store owns D, which publishes its own admission
7//! metadata from the plan rows, and the removal of its indexes. All IO goes
8//! through [`CompactionFiles`], so tests can drop writes made after the last
9//! sync. The kept set arrives as rows; the store and this module remain
10//! policy-neutral.
11
12mod format;
13mod root_files;
14
15#[cfg(test)]
16mod tests;
17
18use std::collections::BTreeMap;
19use std::io::{self, ErrorKind, Write};
20
21use crate::core::{ErrorCategory, MachineError, MachineErrorSource, StoreError, StorePosition};
22use crate::records::{MAX_H3_RECORD_BYTES, RecordHash, blake3_hash};
23use crate::runtime_support::record_log_row::RecordBytesRange;
24
25pub use format::{CompactionStage, PlanRow};
26use format::{
27    JOURNAL_HEADER_BYTES, PLAN_HEADER_BYTES, PLAN_ROW_BYTES, PlanHeader, corrupt, journal_entries,
28    journal_entry_header, journal_header,
29};
30pub use root_files::RootCompactionFiles;
31
32pub const COMPACTION_PLAN_FILE: &str = "compaction-plan";
33pub const COMPACTION_STATE_FILE: &str = "compaction-state";
34pub const COMPACTION_JOURNAL_FILE: &str = "compaction-journal";
35/// Suffix of the temporary file a [`CompactionFiles::replace`] renames.
36pub const TEMPORARY_SUFFIX: &str = ".lace-tmp";
37/// Default `J`: the most plan-row bytes one move batch covers.
38pub const DEFAULT_COMPACTION_BATCH_BYTES: u64 = 64 << 20;
39const _: () = assert!(DEFAULT_COMPACTION_BATCH_BYTES >= MAX_H3_RECORD_BYTES as u64);
40
41/// Plan rows read per plan IO when walking the plan.
42const PLAN_READ_ROWS: u64 = 4096;
43
44/// Whether `name` is a root entry that only compaction creates.
45pub fn is_compaction_entry(name: &str) -> bool {
46    let name = name.strip_suffix(TEMPORARY_SUFFIX).unwrap_or(name);
47    [
48        COMPACTION_PLAN_FILE,
49        COMPACTION_STATE_FILE,
50        COMPACTION_JOURNAL_FILE,
51    ]
52    .contains(&name)
53}
54
55/// Root IO used by compaction. Named files live beside the arena.
56pub trait CompactionFiles {
57    fn arena_capacity(&self) -> u64;
58    fn arena_read(&mut self, offset: u64, buf: &mut [u8]) -> io::Result<()>;
59    fn arena_write(&mut self, offset: u64, bytes: &[u8]) -> io::Result<()>;
60    fn arena_sync(&mut self) -> io::Result<()>;
61    /// Release `[offset, capacity)`; `ErrorKind::Unsupported` when the
62    /// filesystem cannot.
63    fn arena_release_tail(&mut self, offset: u64) -> io::Result<()>;
64    fn len(&mut self, name: &str) -> io::Result<Option<u64>>;
65    fn read_at(&mut self, name: &str, offset: u64, buf: &mut [u8]) -> io::Result<()>;
66    /// A writer whose `commit` atomically and durably replaces `name`
67    /// (temporary file, sync, rename, directory sync). Dropping it uncommitted
68    /// leaves `name` unchanged.
69    fn replace(&mut self, name: &str) -> io::Result<Box<dyn Replacement>>;
70    /// Durably remove `name` when present.
71    fn remove(&mut self, name: &str) -> io::Result<()>;
72}
73
74/// An uncommitted replacement of one named file.
75pub trait Replacement: Write {
76    fn commit(self: Box<Self>) -> io::Result<()>;
77}
78
79/// Whether the freed arena tail was released to the filesystem.
80#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub enum ArenaRelease {
82    Ok,
83    /// The filesystem cannot release blocks; the root is consistent anyway.
84    Unsupported,
85}
86
87#[derive(Debug, Clone, PartialEq, Eq)]
88pub struct CompactionReport {
89    /// High-water the sweep evaluated; every record at or below it was decided.
90    pub boundary: StorePosition,
91    pub retained: u64,
92    pub removed: u64,
93    /// Arena bytes between the new and the previous committed end.
94    pub bytes_released: u64,
95    pub release: ArenaRelease,
96}
97
98/// Outcome of looking for an interrupted compaction.
99pub enum Resumed<F> {
100    /// No compaction is pending; stray plan and journal files were removed.
101    Idle(F),
102    Pending(Compaction<F>),
103}
104
105/// One compaction of a root whose arena no mapping can observe.
106pub struct Compaction<F> {
107    files: F,
108    header: PlanHeader,
109    stage: CompactionStage,
110    new_end: u64,
111    /// Verify every source of the next batch: it may have been interrupted.
112    recovering: bool,
113}
114
115impl<F: CompactionFiles> Compaction<F> {
116    /// Steps A and B: durably write the plan for `kept` (old rows in id order,
117    /// `retained` of them) and the initial move state. The root must hold no
118    /// compaction files ([`Self::resume`] returned `Idle`).
119    pub fn begin(
120        mut files: F,
121        boundary: StorePosition,
122        old_end: u64,
123        retained: u64,
124        kept: impl IntoIterator<Item = (RecordHash, RecordBytesRange)>,
125        batch_bytes: u64,
126    ) -> Result<Self, StoreError> {
127        if batch_bytes == 0 {
128            return Err(StoreError::StoreFailed(
129                "compaction batch must hold at least one byte".into(),
130            ));
131        }
132        let header = PlanHeader {
133            boundary,
134            retained,
135            batch_bytes,
136            old_end,
137        };
138        let mut plan = files.replace(COMPACTION_PLAN_FILE).map_err(io_error)?;
139        plan.write_all(&header.encode()).map_err(io_error)?;
140        let (mut count, mut new_end) = (0u64, 0u64);
141        for (hash, range) in kept {
142            let row = PlanRow {
143                hash,
144                old_offset: range.offset,
145                new_offset: new_end,
146                len: range.len,
147            };
148            plan.write_all(&row.encode()).map_err(io_error)?;
149            new_end = new_end
150                .checked_add(range.len)
151                .filter(|end| *end <= old_end)
152                .ok_or_else(|| StoreError::StoreFailed("kept rows exceed the arena".into()))?;
153            count += 1;
154        }
155        if count != retained {
156            return Err(StoreError::StoreFailed(format!(
157                "kept rows count {count} differs from the retained count {retained}"
158            )));
159        }
160        plan.commit().map_err(io_error)?;
161        let stage = CompactionStage::Move { cursor: 0 };
162        write_state(&mut files, stage)?;
163        Ok(Self {
164            files,
165            header,
166            stage,
167            new_end,
168            recovering: false,
169        })
170    }
171
172    /// Continue an interrupted compaction from its durable files, or report
173    /// that none is pending.
174    pub fn resume(mut files: F) -> Result<Resumed<F>, StoreError> {
175        let state = read_small(&mut files, COMPACTION_STATE_FILE)?;
176        let Some(state) = state else {
177            // Before B or after F removed the state: the root is consistent.
178            files.remove(COMPACTION_JOURNAL_FILE).map_err(io_error)?;
179            files.remove(COMPACTION_PLAN_FILE).map_err(io_error)?;
180            return Ok(Resumed::Idle(files));
181        };
182        let stage = CompactionStage::decode(&state)?;
183        let plan_len = files
184            .len(COMPACTION_PLAN_FILE)
185            .map_err(io_error)?
186            .ok_or_else(|| corrupt("compaction-state without compaction-plan"))?;
187        let mut header = [0; PLAN_HEADER_BYTES as usize];
188        if plan_len < PLAN_HEADER_BYTES {
189            return Err(corrupt("compaction-plan is short"));
190        }
191        files
192            .read_at(COMPACTION_PLAN_FILE, 0, &mut header)
193            .map_err(io_error)?;
194        let header = PlanHeader::decode(&header)?;
195        if header.plan_len() != Some(plan_len) {
196            return Err(corrupt("compaction-plan length differs from its header"));
197        }
198        let cursor = match stage {
199            CompactionStage::Move { cursor } | CompactionStage::Reload { cursor } => cursor,
200        };
201        if cursor > header.retained {
202            return Err(corrupt("compaction-state cursor exceeds the plan"));
203        }
204        let mut compaction = Self {
205            files,
206            header,
207            stage,
208            new_end: 0,
209            recovering: true,
210        };
211        if header.retained > 0 {
212            let last = compaction.plan_rows(header.retained - 1, 1)?[0];
213            compaction.new_end = last.new_offset + last.len;
214        }
215        Ok(Resumed::Pending(compaction))
216    }
217
218    pub fn boundary(&self) -> StorePosition {
219        self.header.boundary
220    }
221
222    pub fn retained(&self) -> u64 {
223        self.header.retained
224    }
225
226    pub fn stage(&self) -> CompactionStage {
227        self.stage
228    }
229
230    /// The committed arena end after compaction.
231    pub fn new_end(&self) -> u64 {
232        self.new_end
233    }
234
235    pub fn files_mut(&mut self) -> &mut F {
236        &mut self.files
237    }
238
239    /// Up to `max` plan rows starting at row index `start`.
240    pub fn plan_rows(&mut self, start: u64, max: u64) -> Result<Vec<PlanRow>, StoreError> {
241        let count = max.min(self.header.retained.saturating_sub(start));
242        let len = usize::try_from(count * PLAN_ROW_BYTES)
243            .map_err(|_| StoreError::StoreFailed("plan read exceeds memory".into()))?;
244        let mut bytes = vec![0; len];
245        self.files
246            .read_at(
247                COMPACTION_PLAN_FILE,
248                PLAN_HEADER_BYTES + start * PLAN_ROW_BYTES,
249                &mut bytes,
250            )
251            .map_err(io_error)?;
252        let capacity = self.files.arena_capacity();
253        bytes
254            .as_chunks::<{ PLAN_ROW_BYTES as usize }>()
255            .0
256            .iter()
257            .map(|row| PlanRow::decode(row, capacity))
258            .collect()
259    }
260
261    /// Step C for one batch. Returns `true` once every plan row is moved.
262    pub fn move_batch(&mut self) -> Result<bool, StoreError> {
263        let CompactionStage::Move { cursor } = self.stage else {
264            return Ok(true);
265        };
266        if cursor == self.header.retained {
267            return Ok(true);
268        }
269        let rows = self.batch_rows(cursor)?;
270        let first = rows[0];
271        let last = rows[rows.len() - 1];
272        let region = (first.new_offset, last.new_offset + last.len);
273        let journal = if self.recovering {
274            self.read_journal()?
275        } else {
276            Vec::new()
277        };
278        let journal = journal_entries(&journal, cursor)
279            .into_iter()
280            .collect::<BTreeMap<_, _>>();
281
282        // Resolve every source before the first write of this batch.
283        let mut writes = Vec::new();
284        for (index, row) in (cursor..).zip(&rows) {
285            if row.new_offset == row.old_offset {
286                continue;
287            }
288            let bytes = if self.recovering {
289                match self.recovered_source(row, journal.get(&index).copied())? {
290                    Some(bytes) => bytes,
291                    None => continue,
292                }
293            } else {
294                self.read_arena(row.old_range())?
295            };
296            let overlaps = row.old_offset < region.1 && region.0 < row.old_offset + row.len;
297            writes.push((index, *row, overlaps, bytes));
298        }
299
300        // Sources this batch's destinations can overwrite are journaled first.
301        if writes.iter().any(|(_, _, overlaps, _)| *overlaps) {
302            let mut journal = self
303                .files
304                .replace(COMPACTION_JOURNAL_FILE)
305                .map_err(io_error)?;
306            journal
307                .write_all(&journal_header(cursor))
308                .map_err(io_error)?;
309            for (index, row, _, bytes) in writes.iter().filter(|write| write.2) {
310                journal
311                    .write_all(&journal_entry_header(*index, row.len))
312                    .map_err(io_error)?;
313                journal.write_all(bytes).map_err(io_error)?;
314            }
315            journal.commit().map_err(io_error)?;
316        }
317        for (_, row, _, bytes) in &writes {
318            self.files
319                .arena_write(row.new_offset, bytes)
320                .map_err(io_error)?;
321        }
322        self.files.arena_sync().map_err(io_error)?;
323        let next = CompactionStage::Move {
324            cursor: cursor + rows.len() as u64,
325        };
326        write_state(&mut self.files, next)?;
327        self.stage = next;
328        self.recovering = false;
329        Ok(cursor + rows.len() as u64 == self.header.retained)
330    }
331
332    /// Record LMDB reload progress; the arena move must be complete.
333    pub fn set_reload_cursor(&mut self, cursor: u64) -> Result<(), StoreError> {
334        let moved = match self.stage {
335            CompactionStage::Move { cursor } => cursor == self.header.retained,
336            CompactionStage::Reload { .. } => true,
337        };
338        if !moved || cursor > self.header.retained {
339            return Err(StoreError::StoreFailed(
340                "compaction reload precedes the complete move or exceeds the plan".into(),
341            ));
342        }
343        let next = CompactionStage::Reload { cursor };
344        write_state(&mut self.files, next)?;
345        self.stage = next;
346        Ok(())
347    }
348
349    /// Steps E and F, after the store published its metadata (D).
350    pub fn finish(mut self) -> Result<CompactionReport, StoreError> {
351        let complete = match self.stage {
352            CompactionStage::Move { cursor } => cursor == self.header.retained,
353            CompactionStage::Reload { cursor } => cursor == self.header.retained,
354        };
355        if !complete {
356            return Err(StoreError::StoreFailed(
357                "compaction finish precedes its complete move".into(),
358            ));
359        }
360        let release = match self.files.arena_release_tail(self.new_end) {
361            Ok(()) => ArenaRelease::Ok,
362            Err(error) if error.kind() == ErrorKind::Unsupported => ArenaRelease::Unsupported,
363            Err(error) => return Err(io_error(error)),
364        };
365        for name in [
366            COMPACTION_JOURNAL_FILE,
367            COMPACTION_STATE_FILE,
368            COMPACTION_PLAN_FILE,
369        ] {
370            self.files.remove(name).map_err(io_error)?;
371        }
372        Ok(CompactionReport {
373            boundary: self.header.boundary,
374            retained: self.header.retained,
375            removed: self.header.boundary.raw() - self.header.retained,
376            bytes_released: self.header.old_end.saturating_sub(self.new_end),
377            release,
378        })
379    }
380
381    /// Consecutive rows from `cursor` whose total length is at most `J`; the
382    /// first row always belongs to the batch.
383    fn batch_rows(&mut self, cursor: u64) -> Result<Vec<PlanRow>, StoreError> {
384        let mut rows = Vec::new();
385        let mut total = 0u64;
386        let mut next = cursor;
387        while next < self.header.retained {
388            for row in self.plan_rows(next, PLAN_READ_ROWS)? {
389                if !rows.is_empty() && total + row.len > self.header.batch_bytes {
390                    return Ok(rows);
391                }
392                total += row.len;
393                rows.push(row);
394            }
395            next = cursor + rows.len() as u64;
396        }
397        Ok(rows)
398    }
399
400    /// Verified bytes for `row` of an interrupted batch, or `None` when its
401    /// destination already holds them.
402    fn recovered_source(
403        &mut self,
404        row: &PlanRow,
405        journaled: Option<&[u8]>,
406    ) -> Result<Option<Vec<u8>>, StoreError> {
407        let destination = self.read_arena(row.new_range())?;
408        if record_bytes_match(&row.hash, &destination) {
409            return Ok(None);
410        }
411        if let Some(bytes) = journaled.filter(|bytes| record_bytes_match(&row.hash, bytes)) {
412            return Ok(Some(bytes.to_vec()));
413        }
414        let old = self.read_arena(row.old_range())?;
415        if record_bytes_match(&row.hash, &old) {
416            return Ok(Some(old));
417        }
418        Err(StoreError::CorruptStore(format!(
419            "no verified copy of compacted record {} remains",
420            row.hash
421        )))
422    }
423
424    fn read_journal(&mut self) -> Result<Vec<u8>, StoreError> {
425        let Some(len) = self.files.len(COMPACTION_JOURNAL_FILE).map_err(io_error)? else {
426            return Ok(Vec::new());
427        };
428        // A batch journal holds at most `J` record bytes plus a 16-byte header
429        // per row; anything longer cannot be this batch's journal.
430        let bound = self
431            .header
432            .batch_bytes
433            .max(MAX_H3_RECORD_BYTES as u64)
434            .saturating_mul(2)
435            .saturating_add(JOURNAL_HEADER_BYTES as u64);
436        if len > bound {
437            return Ok(Vec::new());
438        }
439        let mut bytes = vec![0; len as usize];
440        self.files
441            .read_at(COMPACTION_JOURNAL_FILE, 0, &mut bytes)
442            .map_err(io_error)?;
443        Ok(bytes)
444    }
445
446    fn read_arena(&mut self, range: RecordBytesRange) -> Result<Vec<u8>, StoreError> {
447        let len = usize::try_from(range.len)
448            .map_err(|_| StoreError::StoreFailed("record length exceeds memory".into()))?;
449        let mut bytes = vec![0; len];
450        self.files
451            .arena_read(range.offset, &mut bytes)
452            .map_err(io_error)?;
453        Ok(bytes)
454    }
455}
456
457/// Whether `bytes` is exactly the record `hash` names: its glyph line and the
458/// BLAKE3 digest of every byte after it (Lace-010).
459pub fn record_bytes_match(hash: &RecordHash, bytes: &[u8]) -> bool {
460    const GLYPH_PREFIX: &[u8] = "🖧: ".as_bytes();
461    let glyph_len = GLYPH_PREFIX.len() + hash.as_bytes().len() + 1;
462    bytes.len() >= glyph_len
463        && bytes.starts_with(GLYPH_PREFIX)
464        && &bytes[GLYPH_PREFIX.len()..glyph_len - 1] == hash.as_bytes()
465        && bytes[glyph_len - 1] == b'\n'
466        && blake3_hash(&bytes[glyph_len..]) == hash.digest()
467}
468
469fn write_state<F: CompactionFiles>(
470    files: &mut F,
471    stage: CompactionStage,
472) -> Result<(), StoreError> {
473    let mut state = files.replace(COMPACTION_STATE_FILE).map_err(io_error)?;
474    state.write_all(&stage.encode()).map_err(io_error)?;
475    state.commit().map_err(io_error)
476}
477
478/// A whole small file, bounded so a corrupt one is not read into memory.
479fn read_small<F: CompactionFiles>(
480    files: &mut F,
481    name: &str,
482) -> Result<Option<Vec<u8>>, StoreError> {
483    let Some(len) = files.len(name).map_err(io_error)? else {
484        return Ok(None);
485    };
486    if len > 4096 {
487        return Err(corrupt("compaction file is oversized"));
488    }
489    let mut bytes = vec![0; len as usize];
490    files.read_at(name, 0, &mut bytes).map_err(io_error)?;
491    Ok(Some(bytes))
492}
493
494/// The `Store` machine error a compaction driver reports for `error`.
495pub fn store_error(error: StoreError) -> MachineError {
496    MachineError {
497        category: ErrorCategory::Store,
498        side: None,
499        record_hash: None,
500        detail: format!("{error:?}"),
501        source: Some(MachineErrorSource::Store(error)),
502    }
503}
504
505fn io_error(error: io::Error) -> StoreError {
506    match error.kind() {
507        ErrorKind::InvalidData | ErrorKind::UnexpectedEof => {
508            StoreError::CorruptStore(error.to_string())
509        }
510        ErrorKind::StorageFull | ErrorKind::QuotaExceeded => StoreError::CapacityExceeded,
511        ErrorKind::WouldBlock => StoreError::Unavailable(error.to_string()),
512        _ => StoreError::StoreFailed(format!("compaction IO failed: {error}")),
513    }
514}
lace_sans_io/runtime_support/compaction/mod.rs

lace_sans_io/runtime_support/compaction/
mod.rs
