haematite 0.7.0

Content-addressed, branchable, actor-native storage engine
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

//! Internal helpers for [`crate::db::Database`]: shard-error mapping, single-
//! shard range collection, and event-key range bounds. Kept separate to keep
//! `db.rs` focused on the public surface.

use std::time::Duration;

use crate::shard::actor::{RangeItem, ShardError, ShardHandle};

use super::{DatabaseError, DbRange};

/// Map a runtime shard error into the public [`DatabaseError`], preserving the
/// optimistic-concurrency variants so callers can match them precisely.
pub fn map_shard_error(error: ShardError) -> DatabaseError {
    match error {
        ShardError::SequenceConflict { expected, actual } => {
            DatabaseError::SequenceConflict { expected, actual }
        }
        ShardError::CasMismatch { expected, actual } => {
            DatabaseError::CasMismatch { expected, actual }
        }
        ShardError::Spawn(message) => DatabaseError::ShardSpawn(message),
        ShardError::WriteDuringRootAdvanceEmission => DatabaseError::WriteDuringRootAdvanceEmission,
        other => DatabaseError::ShardError(other.to_string()),
    }
}

/// Run a `[from, to)` range against one shard handle and collect its entries.
pub fn range_on_handle(
    handle: &ShardHandle,
    from: &[u8],
    to: &[u8],
    timeout: Duration,
) -> Result<DbRange, DatabaseError> {
    let items = handle
        .range(from.to_vec(), to.to_vec(), timeout)
        .map_err(map_shard_error)?;
    collect_range_items(items)
}

/// Run a first-match liveness probe for one stream's event-entry range.
pub(super) fn has_live_events_on_handle(
    handle: &ShardHandle,
    key: &[u8],
    timeout: Duration,
) -> Result<bool, DatabaseError> {
    let from = event_range_start(key, 1);
    let to = event_range_end(key);
    handle
        .has_live_in_range(from, to, timeout)
        .map_err(map_shard_error)
}

/// Collect streamed range items up to the [`RangeItem::Done`] sentinel.
fn collect_range_items(items: Vec<RangeItem>) -> Result<DbRange, DatabaseError> {
    let mut entries = Vec::new();
    for item in items {
        match item {
            RangeItem::Entry { key, value } => entries.push((key, value)),
            RangeItem::Done => return Ok(entries),
        }
    }
    Err(DatabaseError::ShardError(
        "range result missing Done".to_owned(),
    ))
}

/// Inclusive lower bound for an event range: `key || 0x00 || seq.to_be_bytes()`.
pub(super) fn event_range_start(key: &[u8], seq: u64) -> Vec<u8> {
    let mut encoded = Vec::with_capacity(key.len().saturating_add(9));
    encoded.extend_from_slice(key);
    encoded.push(0);
    encoded.extend_from_slice(&seq.to_be_bytes());
    encoded
}

/// Exclusive upper bound for an event range: `key || 0x01`, which sorts after
/// every `key || 0x00 || ...` event key but before the `0xff` sequence key.
pub(super) fn event_range_end(key: &[u8]) -> Vec<u8> {
    let mut encoded = Vec::with_capacity(key.len().saturating_add(1));
    encoded.extend_from_slice(key);
    encoded.push(1);
    encoded
}

/// The per-stream sequence-counter key: `key || 0xff "seq"`.
///
/// This is byte-identical to the shard actor's `sequence_key` (its `SEQ_SUFFIX`)
/// and to the legacy `db::event_sequence_key`; it is the metadata key whose 8-byte
/// big-endian value is the stream's next-sequence. The single shared definition so
/// the local `append`, `read_stream_next_seq`, and the replicated
/// [`replicate_append`](crate::db::Database::replicate_append) all address the SAME
/// key (ยง event-store layout).
pub(super) fn event_sequence_key(key: &[u8]) -> Vec<u8> {
    let mut encoded = Vec::with_capacity(key.len().saturating_add(4));
    encoded.extend_from_slice(key);
    encoded.extend_from_slice(&[0xff, b's', b'e', b'q']);
    encoded
}