hydracache 0.59.0

User-facing HydraCache runtime crate.
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

#![cfg(feature = "durable-value-store")]

use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};

use hydracache::{
    ClusterEpoch, DurableValueStore, PartitionId, ReplicatedValueRecord, ReplicatedValueStore,
    DURABLE_VALUE_FORMAT_VERSION,
};

#[test]
fn durable_value_store_reopen_recovers_records_and_tombstones() {
    let path = temp_store_path("reopen");
    let mut store = DurableValueStore::open_with_budget(&path, 1024).unwrap();
    let record = ReplicatedValueRecord::value(
        PartitionId::new(3),
        7,
        ClusterEpoch::new(2),
        b"sealed-user".to_vec(),
    );
    store.upsert("user:42", record.clone()).unwrap();
    store
        .tombstone("user:43", PartitionId::new(3), 8, ClusterEpoch::new(2))
        .unwrap();
    store.flush().unwrap();
    drop(store);

    let reopened = DurableValueStore::open_with_budget(&path, 1024).unwrap();

    assert_eq!(reopened.get("user:42").unwrap(), Some(record));
    assert!(reopened
        .get("user:43")
        .unwrap()
        .expect("tombstone")
        .is_tombstone());
}

#[test]
fn durable_value_store_unknown_future_format_refuses_to_open() {
    let path = temp_store_path("future-format");
    DurableValueStore::write_format_marker_for_test(&path, DURABLE_VALUE_FORMAT_VERSION + 1)
        .unwrap();

    let error = DurableValueStore::open(&path).unwrap_err();

    assert!(error
        .to_string()
        .contains("unsupported durable value-store format"));
}

#[test]
fn durable_value_store_over_budget_upsert_is_rejected_and_counted() {
    let path = temp_store_path("budget");
    let mut store = DurableValueStore::open_with_budget(&path, 4).unwrap();

    let error = store
        .upsert(
            "big",
            ReplicatedValueRecord::value(
                PartitionId::new(1),
                1,
                ClusterEpoch::new(1),
                b"too-large".to_vec(),
            ),
        )
        .unwrap_err();

    assert!(error.to_string().contains("budget exceeded"));
    assert_eq!(store.rejected_total(), 1);
    assert!(store.get("big").unwrap().is_none());
}

#[test]
fn durable_value_store_corrupt_record_is_detected_not_served() {
    let path = temp_store_path("corrupt");
    let mut store = DurableValueStore::open_with_budget(&path, 1024).unwrap();
    store
        .upsert(
            "user:42",
            ReplicatedValueRecord::value(
                PartitionId::new(1),
                1,
                ClusterEpoch::new(1),
                b"valid".to_vec(),
            ),
        )
        .unwrap();
    store
        .put_raw_record_for_test("user:42", b"not-a-valid-envelope")
        .unwrap();

    let error = store.get("user:42").unwrap_err();

    assert!(error.to_string().contains("durable value"));
}

fn temp_store_path(name: &str) -> PathBuf {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("clock after epoch")
        .as_nanos();
    std::env::temp_dir().join(format!(
        "hydracache-durable-value-store-{name}-{}-{nanos}",
        std::process::id()
    ))
}