armdb 0.11.2

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

#![cfg(feature = "rpc")]

use std::sync::Arc;
use std::time::Duration;

use armdb::armour::Db;
use armdb::{QueueConfig, QueueSpec, queue};
use armour_rpc::RpcClient;
use rapira::Rapira;

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    armour_core::GetType,
    zerocopy::FromBytes,
    zerocopy::IntoBytes,
    zerocopy::Immutable,
    zerocopy::KnownLayout,
)]
#[repr(transparent)]
pub struct FixedJob(u64);

#[derive(Clone, Debug, PartialEq, Eq, armour_core::GetType, Rapira)]
pub struct VarJob(String);

queue!(
    RpcFixedJobs,
    payload = FixedJob,
    name = "queue_rpc_fixed_jobs"
);
queue!(
    RpcVarJobs,
    payload = VarJob,
    name = "queue_rpc_var_jobs",
    var
);

fn wait_for_socket(socket: &std::path::Path) {
    for _ in 0..100 {
        if socket.exists() {
            return;
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    panic!("UDS socket never appeared");
}

#[test]
fn queue_physical_collections_are_not_generic_rpc_routes() {
    let db_dir = tempfile::tempdir().expect("db tempdir");
    let socket_dir = tempfile::tempdir().expect("socket tempdir");
    let socket = socket_dir.path().join("armdb.sock");
    let db = Arc::new(Db::open_test(db_dir.path()).expect("open db"));
    let fixed = db
        .open_queue::<RpcFixedJobs>(QueueConfig::balanced().build().expect("fixed config"))
        .expect("open fixed queue");
    let var = db
        .open_queue::<RpcVarJobs>(QueueConfig::balanced().build().expect("var config"))
        .expect("open var queue");

    // The regression is a capability boundary: registering any of these
    // physical trees exposes generic Write/Remove/Take/ApplyBatch operations
    // that bypass queue lease, counter, and retention bookkeeping.
    for name in [
        RpcFixedJobs::NAME,
        RpcVarJobs::NAME,
        RpcVarJobs::PAYLOAD_NAME,
    ] {
        assert!(
            db.rpc_handler_for(name).is_none(),
            "queue physical collection `{name}` must not have a generic RPC handler"
        );
    }

    db.listen_uds(socket.clone());
    wait_for_socket(&socket);
    compio::runtime::RuntimeBuilder::new()
        .build()
        .expect("runtime")
        .block_on(async {
            for name in [
                RpcFixedJobs::NAME,
                RpcVarJobs::NAME,
                RpcVarJobs::PAYLOAD_NAME,
            ] {
                let mut client = RpcClient::connect_uds_by_name(&socket, name)
                    .await
                    .expect("connect unknown collection client");
                assert!(
                    client.get_schema().await.is_err(),
                    "RPC dispatch unexpectedly exposed queue physical collection `{name}`"
                );
            }
        });

    // Suppressing the generic adapter does not suppress the opaque queue API.
    fixed.push(FixedJob(7), None).expect("fixed queue push");
    let fixed_row = fixed
        .lease(1, Some(Duration::from_secs(1)))
        .expect("fixed queue lease")
        .pop()
        .expect("fixed lease row");
    assert_eq!(*fixed_row.payload(), FixedJob(7));
    fixed.ack(fixed_row.token()).expect("fixed queue ack");

    var.push(VarJob("body".to_owned()), None)
        .expect("var queue push");
    let var_row = var
        .lease(1, Some(Duration::from_secs(1)))
        .expect("var queue lease")
        .pop()
        .expect("var lease row");
    assert_eq!(var_row.payload(), &VarJob("body".to_owned()));
    var.ack(var_row.token()).expect("var queue ack");
}