batpak 0.2.0

Event sourcing with causal graphs and policy gates. Sync API, zero async.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

use crate::coordinate::Coordinate;
use crate::store::index::{DiskPos, IndexEntry, StoreIndex};
use crate::store::reader::Reader;
use crate::store::segment;
use crate::store::StoreError;
use std::path::Path;

/// Open the index using the fastest available path:
/// 1. Try loading a checkpoint file → if valid, restore from it + replay tail segments.
/// 2. Fall back to full segment scan if checkpoint is missing, corrupt, or stale.
pub(crate) fn open_index(
    index: &StoreIndex,
    reader: &Reader,
    data_dir: &Path,
    enable_checkpoint: bool,
) -> Result<(), StoreError> {
    if enable_checkpoint {
        if let Some((entries, interner_strings, watermark)) =
            crate::store::checkpoint::try_load_checkpoint(data_dir)
        {
            tracing::info!(
                "checkpoint v2 loaded: {} entries, {} interner strings, watermark segment {} offset {}",
                entries.len(),
                interner_strings.len(),
                watermark.watermark_segment_id,
                watermark.watermark_offset
            );
            crate::store::checkpoint::restore_from_checkpoint(index, entries, &interner_strings)?;
            // Replay segments newer than the watermark.
            replay_tail_segments(index, reader, data_dir, &watermark)?;
            return Ok(());
        }
        tracing::debug!("no valid checkpoint, performing full index rebuild");
    }
    rebuild_from_segments(index, reader, data_dir)
}

/// Replay only segments with ID > watermark, or frames at offset >= watermark_offset
/// within the watermark segment itself.
fn replay_tail_segments(
    index: &StoreIndex,
    reader: &Reader,
    data_dir: &Path,
    watermark: &crate::store::checkpoint::WatermarkInfo,
) -> Result<(), StoreError> {
    let mut entries: Vec<std::fs::DirEntry> = std::fs::read_dir(data_dir)?
        .filter_map(|e| e.ok())
        .filter(|e| {
            e.path()
                .extension()
                .map(|ext| ext == segment::SEGMENT_EXTENSION)
                .unwrap_or(false)
        })
        .collect();
    entries.sort_by_key(|e| e.file_name());

    for dir_entry in &entries {
        let seg_id = dir_entry
            .path()
            .file_stem()
            .and_then(|s| s.to_str())
            .and_then(|s| s.parse::<u64>().ok())
            .unwrap_or(0);

        if seg_id < watermark.watermark_segment_id {
            continue; // Already in checkpoint
        }

        let scanned = reader.scan_segment_index(&dir_entry.path())?;
        for se in scanned {
            // Skip frames already in the checkpoint
            if seg_id == watermark.watermark_segment_id && se.offset < watermark.watermark_offset {
                continue;
            }
            let coord = Coordinate::new(&se.entity, &se.scope)?;
            let entity_id = index.interner.intern(&se.entity);
            let scope_id = index.interner.intern(&se.scope);
            let clock = se.header.position.sequence;
            let entry = IndexEntry {
                event_id: se.header.event_id,
                correlation_id: se.header.correlation_id,
                causation_id: se.header.causation_id,
                coord,
                entity_id,
                scope_id,
                kind: se.header.event_kind,
                wall_ms: se.header.position.wall_ms,
                clock,
                hash_chain: se.hash_chain,
                disk_pos: DiskPos {
                    segment_id: se.segment_id,
                    offset: se.offset,
                    length: se.length,
                },
                global_sequence: index.global_sequence(),
            };
            index.insert(entry);
        }
    }
    Ok(())
}

/// Scan all segment files in `data_dir`, rebuild the in-memory index from their contents.
/// Used by both cold-start (`Store::open_with_cache`) and post-compaction index rebuild.
pub(crate) fn rebuild_from_segments(
    index: &StoreIndex,
    reader: &Reader,
    data_dir: &Path,
) -> Result<(), StoreError> {
    let mut entries: Vec<std::fs::DirEntry> = std::fs::read_dir(data_dir)?
        .filter_map(|e| e.ok())
        .filter(|e| {
            e.path()
                .extension()
                .map(|ext| ext == segment::SEGMENT_EXTENSION)
                .unwrap_or(false)
        })
        .collect();
    entries.sort_by_key(|e| e.file_name());

    for dir_entry in &entries {
        let scanned = reader.scan_segment_index(&dir_entry.path())?;
        for se in scanned {
            let coord = Coordinate::new(&se.entity, &se.scope)?;
            let entity_id = index.interner.intern(&se.entity);
            let scope_id = index.interner.intern(&se.scope);
            let clock = se.header.position.sequence;
            let entry = IndexEntry {
                event_id: se.header.event_id,
                correlation_id: se.header.correlation_id,
                causation_id: se.header.causation_id,
                coord,
                entity_id,
                scope_id,
                kind: se.header.event_kind,
                wall_ms: se.header.position.wall_ms,
                clock,
                hash_chain: se.hash_chain,
                disk_pos: DiskPos {
                    segment_id: se.segment_id,
                    offset: se.offset,
                    length: se.length,
                },
                global_sequence: index.global_sequence(),
            };
            index.insert(entry);
        }
    }

    Ok(())
}