hawdb-storage 0.16.0

An embedded Rust graph database for local data-plane workloads
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

use super::*;
use crate::background::{CheckpointOperationError, CheckpointWorkError};
use crate::projection::ProjectedRelationshipPredicate;
use hawdb_core::{RuntimeMemoryError, RuntimeTaskContext, Value};
use hawdb_qos::{
    IoConcurrencyBudget, RuntimeGovernor, RuntimeGovernorConfig, RuntimeMemorySnapshot,
    RuntimeResourceBudget, RuntimeResourceSnapshot, RuntimeWorkRequest,
};
use std::num::NonZeroUsize;

fn governor(bytes: u64) -> RuntimeGovernor {
    RuntimeGovernor::new(
        RuntimeGovernorConfig {
            memory_budget_bytes: Some(bytes),
            background_task_limit: Some(NonZeroUsize::MIN),
            ..RuntimeGovernorConfig::shared_host()
        },
        RuntimeResourceSnapshot::from_parts(
            RuntimeResourceBudget::from_limits(NonZeroUsize::MIN, None, None),
            RuntimeMemorySnapshot::from_limits(Some(1 << 30), Some(1 << 30), None, None, None),
        ),
        IoConcurrencyBudget::new(2, 1),
    )
}

fn source() -> (String, ProjectedGraphDefinition) {
    let definition = ProjectedGraphDefinition {
        node_labels: vec!["Memory".into()],
        rel_types: vec!["LINKS".into()],
        relationship_predicates: BTreeMap::from([(
            "LINKS".into(),
            ProjectedRelationshipPredicate::Eq {
                property: "body".into(),
                value: Value::Binary((0..257 * 1024 + 3).map(|i| (i % 256) as u8).collect()),
            },
        )]),
    };
    let data =
        ProjectedGraphArtifactData::new(Vec::new(), vec![0], Vec::new(), vec![0], Vec::new())
            .unwrap();
    (
        encode_projected_graph_artifacts(19, 23, [("wide-predicate", &definition, data)]),
        definition,
    )
}

#[test]
fn checkpoint_units_projection_artifact_memory_admits_before_allocating_payload() {
    let (source, _) = source();
    let governor = governor(1);
    let permit = governor
        .try_admit(RuntimeWorkRequest::background_maintenance(1))
        .unwrap();
    let work = CheckpointWorkContext::new(permit.bind_task_context(RuntimeTaskContext::default()));
    let observation = crate::test_allocator::AllocationObservation::start();
    let result =
        work.classify(|work| decode_projected_graph_artifacts_with_work_context(&source, work));
    let allocations = observation.finish();
    assert!(matches!(
        result,
        Err(CheckpointOperationError::Work(CheckpointWorkError::Memory(
            RuntimeMemoryError::ReservationExceeded { .. }
        )))
    ));
    assert_eq!(
        allocations, 0,
        "artifact denial must precede any unadmitted payload allocation over 64 KiB"
    );
    drop(work);
    drop(permit);
    assert_eq!(governor.snapshot().admitted_memory_bytes, 0);
}

#[test]
fn checkpoint_units_projection_artifact_memory_retains_complete_predicate_and_arrays_after_execution(
) {
    let (source, definition) = source();
    let ceiling = 32 * 1024 * 1024;
    let governor = governor(ceiling);
    let permit = governor
        .try_admit(RuntimeWorkRequest::background_maintenance(ceiling))
        .unwrap();
    let work = CheckpointWorkContext::new(permit.bind_task_context(RuntimeTaskContext::default()));
    let (epoch, artifacts) =
        decode_projected_graph_artifacts_with_work_context(&source, &work).unwrap();
    assert_eq!(epoch, 23);
    assert_eq!(artifacts.len(), 1);
    let artifact = artifacts.get("wide-predicate").unwrap();
    assert_eq!(artifact.definition, definition);
    assert_eq!(artifact.projection_epoch, 19);
    assert_eq!(artifact.commit_epoch, 23);
    assert!(artifact.data.nodes.is_empty());
    assert_eq!(artifact.data.csr_offsets, [0]);
    assert_eq!(artifact.data.csc_offsets, [0]);
    assert!(artifact.data.csr_targets.is_empty());
    assert!(artifact.data.csc_sources.is_empty());
    drop(source);
    drop(definition);
    drop(work);
    drop(permit);
    assert_eq!(governor.snapshot().active_background_tasks, 0);
    assert_eq!(governor.snapshot().active_cpu_slots, 0);
    assert_eq!(governor.snapshot().admitted_memory_bytes, ceiling);
    let artifact = artifacts.get("wide-predicate").unwrap();
    let ProjectedRelationshipPredicate::Eq { property, value } = artifact
        .definition
        .relationship_predicates
        .get("LINKS")
        .unwrap()
    else {
        unreachable!()
    };
    assert_eq!(property, "body");
    let Value::Binary(bytes) = value else {
        unreachable!()
    };
    assert_eq!(bytes.len(), 257 * 1024 + 3);
    assert!(bytes
        .iter()
        .enumerate()
        .all(|(i, byte)| *byte == (i % 256) as u8));
    assert_eq!(artifact.data.csr_offsets, [0]);
    assert_eq!(artifact.data.csc_offsets, [0]);
    drop(artifacts);
    assert_eq!(governor.snapshot().admitted_memory_bytes, 0);
}