batpak 0.8.2

Event sourcing with causal graphs and caller-defined gates. Sync API, no async runtime.
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

//! Red-path tests keep the `unified_*_red` names for cross-surface edge cases
//! that should fail fast or prove defensive behavior across the unified store.
// justifies: INV-TEST-PANIC-AS-ASSERTION, INV-MACRO-BOUNDED-CAST; unified red-path reader tests in tests/unified_reader_red.rs use unwrap/panic as assertion style and narrow bounded test counters that fit within u32.
#![allow(clippy::unwrap_used, clippy::cast_possible_truncation, clippy::panic)]

#[path = "support/red_kinds.rs"]
mod red_kinds;
#[path = "support/red_test_coord.rs"]
mod red_test_coord;

use red_kinds::*;
use red_test_coord::*;

mod support;
use std::sync::Arc;
use support::prelude::*;
use tempfile::TempDir;

#[test]
fn sealed_segment_reads_via_mmap() {
    let dir = TempDir::new().expect("temp dir");
    let config = StoreConfig::new(dir.path()).with_segment_max_bytes(512);
    let store = Store::open(config).expect("open");
    let coord = test_coord();
    for i in 0u32..50 {
        store.append(&coord, kind_a(), &payload(i)).expect("append");
    }
    store.sync().expect("sync");
    let entries = store.by_entity("entity:test");
    let first = &entries[0];
    let event = store
        .get(batpak::id::EventId::from(first.event_id()))
        .expect("get from sealed segment");
    assert_eq!(
        event.coordinate.entity(),
        "entity:test",
        "PROPERTY: mmap read from sealed segment must return correct event.\n\
         Investigate: src/store/segment/scan.rs sealed_maps path."
    );
    store.close().expect("close");
}

#[test]
fn concurrent_sealed_reads_no_lock_contention() {
    let dir = TempDir::new().expect("temp dir");
    let config = StoreConfig::new(dir.path()).with_segment_max_bytes(512);
    let store = Arc::new(Store::open(config).expect("open"));
    let coord = test_coord();
    let mut ids = Vec::new();
    for i in 0u32..20 {
        let receipt = store.append(&coord, kind_a(), &payload(i)).expect("append");
        ids.push(receipt.event_id);
    }
    store.sync().expect("sync");

    let handles: Vec<_> = ids
        .iter()
        .map(|&id| {
            let store = Arc::clone(&store);
            std::thread::Builder::new()
                .name(format!("reader-{id}"))
                .spawn(move || {
                    store.get(id).expect("concurrent get");
                })
                .expect("spawn")
        })
        .collect();
    for handle in handles {
        handle.join().expect("reader thread");
    }
}

#[test]
fn evict_mmap_before_compaction_delete() {
    let dir = TempDir::new().expect("temp dir");
    let config = StoreConfig::new(dir.path()).with_segment_max_bytes(512);
    let store = Store::open(config).expect("open");
    let coord = test_coord();
    for i in 0u32..50 {
        store.append(&coord, kind_a(), &payload(i)).expect("append");
    }
    store.sync().expect("sync");
    let result = store
        .compact(&CompactionConfig::default())
        .map(|(result, _report)| result);
    assert!(
        result.is_ok(),
        "PROPERTY: compaction must succeed even with mmap'd segments.\n\
         Investigate: src/store/segment/scan.rs evict_segment drops Mmap before delete.\n\
         Got: {result:?}"
    );
    store.close().expect("close");
}