armdb 0.12.0

sharded bitcask key-value storage optimized for NVMe
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

//! FixedStore participation in `db.tx()`: mixed-backend commit and rollback.

use std::sync::Arc;

use armdb::armour::Db;
use armdb::{CollectionMeta, Config, FixedConfig, NoHook, RapiraCodec, TtlHook, TypedMap, ZeroMap};
use armour_core::GetType;
use rapira::Rapira;
use zerocopy::{FromBytes, Immutable, IntoBytes};

type Id = [u8; 8];

fn id(n: u64) -> Id {
    n.to_be_bytes()
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct Message {
    len: u32,
}
impl CollectionMeta for Message {
    type SelfId = Id;
    const NAME: &'static str = "messages";
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, GetType, IntoBytes, FromBytes, Immutable)]
#[repr(C)]
struct SeenMarker {
    at: u64,
}
impl CollectionMeta for SeenMarker {
    type SelfId = Id;
    const NAME: &'static str = "seen";
}

type Messages = TypedMap<Id, Message, RapiraCodec, TtlHook<Message, NoHook>>;
type Seen = ZeroMap<Id, 8, SeenMarker, TtlHook<SeenMarker, NoHook>, armdb::durability::Fixed>;

fn open_pair(db: &Arc<Db>) -> (Arc<Messages>, Arc<Seen>) {
    let messages = db
        .open_typed_map::<Message, RapiraCodec, _>(Config::test(), NoHook, &[])
        .unwrap();
    let seen = db
        .open_zero_map_fixed::<SeenMarker, 8, _>(FixedConfig::test(), NoHook, &[])
        .unwrap();
    (messages, seen)
}

#[test]
fn tx_mixed_backends_commit_together() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let (messages, seen) = open_pair(&db);

    db.tx()
        .put(&messages, &id(1), &Message { len: 42 })
        .unwrap()
        .put(&seen, &id(1), &SeenMarker { at: 7 })
        .unwrap()
        .durable()
        .commit()
        .unwrap();

    assert_eq!(messages.get(&id(1)).map(|m| m.len), Some(42));
    assert_eq!(seen.get(&id(1)).map(|s| s.at), Some(7));
}

#[test]
fn tx_mixed_backends_roll_back_together() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let (messages, seen) = open_pair(&db);

    let err = db
        .tx()
        .put(&messages, &id(2), &Message { len: 1 })
        .unwrap()
        .put(&seen, &id(2), &SeenMarker { at: 1 })
        .unwrap()
        .guard(&messages, &id(999), |cur: Option<&Message>| {
            if cur.is_none() {
                Err(armdb::DbError::TxConflict)
            } else {
                Ok(())
            }
        })
        .unwrap()
        .commit()
        .unwrap_err();

    assert!(matches!(err, armdb::DbError::TxConflict), "{err}");
    assert!(messages.get(&id(2)).is_none());
    assert!(
        seen.get(&id(2)).is_none(),
        "the Fixed half must not be written either"
    );
    assert_eq!(
        db.pending_transactions().len(),
        0,
        "nothing reached the log"
    );
}

/// `insert` works on a FixedStore participant, both outcomes, mixed with a
/// Bitcask collection in the same transaction.
#[test]
fn tx_insert_on_a_fixed_collection() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let (messages, seen) = open_pair(&db);

    db.tx()
        .insert(&seen, &id(1), &SeenMarker { at: 7 })
        .unwrap()
        .insert(&messages, &id(1), &Message { len: 42 })
        .unwrap()
        .commit()
        .unwrap();
    assert_eq!(seen.get(&id(1)).map(|s| s.at), Some(7));
    assert_eq!(messages.get(&id(1)).map(|m| m.len), Some(42));

    let pending_before_refused_insert = db.pending_transactions().len();
    let err = db
        .tx()
        .insert(&seen, &id(1), &SeenMarker { at: 9 })
        .unwrap()
        .put(&messages, &id(2), &Message { len: 1 })
        .unwrap()
        .commit()
        .unwrap_err();
    assert!(matches!(err, armdb::DbError::KeyExists), "{err}");
    assert_eq!(
        seen.get(&id(1)).map(|s| s.at),
        Some(7),
        "the Fixed row keeps its value"
    );
    assert!(
        messages.get(&id(2)).is_none(),
        "the Bitcask half of the refused transaction is not written either"
    );
    assert_eq!(
        db.pending_transactions().len(),
        pending_before_refused_insert,
        "the refused transaction adds nothing to the log"
    );
}