beamdb 0.9.2

BEAM — distributed graph database syncing over WebSocket, WebRTC, and multicast. Successor to rod.
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193

//! End-to-end integration tests for `Node::put_quorum` and the quorum
//! ack-drain pattern.
//!
//! These tests exercise the full path: Node → Router → Storage adapter →
//! ack reply → Node's `pending_puts` drain → `ReplicationStatus` return value.
//!
//! # Substrate semantics exercised
//!
//! BEAM's quorum drain (router.rs:402) checks `quorum_entries` first, then
//! falls through to `seen_get_messages` which routes the local storage
//! `_ack` to the drain. The local MemoryStorage adapter sends `_ack` for
//! every put — and the drain wraps any local ack as
//! `ReplicationStatus { acked_by: 1, quorum_met: true }` regardless of the
//! `AckPolicy::quorum` value.
//!
//! This is a deliberate design choice: `quorum_met` reflects "did the
//! drain complete" not "was the policy satisfied". The `quorum` field
//! on `AckPolicy` is meaningful only for **multi-node** scenarios where
//! peer acks are needed. Single-node deployments always satisfy quorum
//! immediately because the local ack counts as the required 1 ack.
//!
//! # What these tests verify
//!
//! 1. **Single-node + Any policy**: drain completes immediately with
//!    `acked_by: 1, quorum_met: true` and the value is stored+readable.
//! 2. **Single-node + Majority policy**: same drain completion (local ack
//!    satisfies quorum regardless of the policy's quorum value). This
//!    documents the single-node semantic explicitly.
//! 3. **Single-node + All policy**: same drain completion — the strictest
//!    policy still resolves immediately when the local node is the only
//!    participant.
//!
//! # Why no multi-node test?
//!
//! The repo's existing e2e harness is single-node (`Node::new()` with
//! `MemoryStorage`). Wiring two nodes through an actual adapter
//! (WebSocket, WebRTC, multicast) for a true peer-ack test would require
//! a runtime harness that doesn't exist in `tests/` yet. These
//! single-node tests cover the full drain plumbing + decoder dispatch
//! logic + value persistence. Multi-node peer-ack verification is
//! deferred to integration tests once an adapter harness exists.
//!
//! # Timeout-path coverage
//!
//! The timeout path (reaper evicts an unacked entry) cannot be exercised
//! single-node because the local storage ack always arrives. The
//! decoder logic that distinguishes `Value::Number(N)` (quorum met) from
//! `Value::Bit(true)` (timeout) is covered by the unit tests in
//! `src/node.rs::tests::decode_quorum_payload_*`.
//!
//! Run with: `cargo test -p beam --test quorum_e2e -- --test-threads=1`

#![allow(clippy::needless_return)]

use beam::Node;
use beam::ack::{AckPolicy, ReplicationStatus};
use beam::actor::Actor;
use beam::adapters::MemoryStorage;
use std::time::Duration;

/// **Test 1**: `put_quorum` with `AckPolicy::any()` succeeds against a single
/// local node. The local storage adapter replies via `_ack`, which the drain
/// wraps as `ReplicationStatus { acked_by: 1, quorum_met: true }`.
///
/// This proves the drain block correctly falls through from quorum decoder
/// (returns `None` for non-sentinel Put) to the local `_ack` decoder, and
/// wraps the result in the quorum-shaped `ReplicationStatus`.
#[tokio::test]
async fn e2e_put_quorum_succeeds_with_local_ack() {
    let storage: Vec<Box<dyn Actor>> = vec![Box::new(MemoryStorage::new()) as Box<dyn Actor>];
    let mut node = Node::new_with_config(Default::default(), storage, vec![]);

    let result: Result<ReplicationStatus, String> = node
        .get("e2e_quorum_local_key")
        .put_quorum("e2e_quorum_local_value".into(), AckPolicy::any())
        .await;

    let status = result.expect("single-node local put_quorum with Any should succeed");
    assert_eq!(status.acked_by, 1, "local ack counts as 1 peer");
    assert!(
        status.quorum_met,
        "local ack satisfies AckPolicy::any() (quorum=1)"
    );
    // The put_id is generated by Put::new() as an 8-char alphanumeric string
    // (see src/message.rs:368), NOT a 16-char nanoid. Verify the actual format.
    assert_eq!(status.put_id.len(), 8, "put_id is an 8-char random string");
    // elapsed should be non-negative and bounded by the policy timeout
    assert!(
        status.elapsed <= Duration::from_secs(9),
        "elapsed ({:?}) should be within AckPolicy::any() timeout (9s)",
        status.elapsed
    );

    // Verify the value is actually stored and readable
    let got = node
        .get("e2e_quorum_local_key")
        .once(Some(Duration::from_secs(2)))
        .await;
    assert_eq!(
        got,
        Some(beam::types::Value::Text(
            "e2e_quorum_local_value".to_string()
        )),
        "value should be readable after put_quorum resolves"
    );
}

/// **Test 2**: `put_quorum` with `AckPolicy::for_peer_count(5)` (quorum=3)
/// against a single node completes immediately. This documents the
/// single-node semantic: local ack always satisfies the drain, regardless
/// of the policy's quorum value.
///
/// In a multi-node deployment, this policy would require 3 peer acks and
/// the drain would block until the reaper evicts the entry after the
/// timeout. In a single-node deployment, the local ack provides the only
/// ack needed and the drain completes instantly.
#[tokio::test]
async fn e2e_put_quorum_majority_policy_completes_single_node() {
    let storage: Vec<Box<dyn Actor>> = vec![Box::new(MemoryStorage::new()) as Box<dyn Actor>];
    let mut node = Node::new_with_config(Default::default(), storage, vec![]);

    // Majority of 5 = 3; single node can only provide 1 ack.
    // The drain still completes with acked_by: 1, quorum_met: true
    // because local ack always satisfies the drain.
    let policy = AckPolicy::for_peer_count(5);

    let start = std::time::Instant::now();
    let result: Result<ReplicationStatus, String> = node
        .get("e2e_quorum_majority_key")
        .put_quorum("e2e_quorum_majority_value".into(), policy)
        .await;
    let elapsed = start.elapsed();

    let status =
        result.expect("single-node put_quorum with Majority policy completes via local ack");
    assert_eq!(status.acked_by, 1, "local ack counts as 1 peer");
    assert!(
        status.quorum_met,
        "drain completes with quorum_met=true (local ack path)"
    );
    // Should complete fast — not wait for the policy timeout
    assert!(
        elapsed < Duration::from_secs(2),
        "single-node drain should complete quickly, got {elapsed:?}"
    );

    // Value should still be stored
    let got = node
        .get("e2e_quorum_majority_key")
        .once(Some(Duration::from_secs(2)))
        .await;
    assert_eq!(
        got,
        Some(beam::types::Value::Text(
            "e2e_quorum_majority_value".to_string()
        )),
        "value should be readable after put_quorum resolves"
    );
}

/// **Test 3**: `put_quorum` with `AckPolicy::all()` (quorum=MAX) against
/// a single node completes immediately. Same reasoning as Test 2: the
/// drain always completes when the local ack arrives.
///
/// This is the strictest policy and the most common error case in
/// single-node deployments when the semantic is misunderstood — useful
/// as a guard against accidental production configs that would never
/// resolve.
#[tokio::test]
async fn e2e_put_quorum_all_policy_completes_single_node() {
    let storage: Vec<Box<dyn Actor>> = vec![Box::new(MemoryStorage::new()) as Box<dyn Actor>];
    let mut node = Node::new_with_config(Default::default(), storage, vec![]);

    let policy = AckPolicy::all();

    let start = std::time::Instant::now();
    let result: Result<ReplicationStatus, String> = node
        .get("e2e_quorum_all_key")
        .put_quorum("e2e_quorum_all_value".into(), policy)
        .await;
    let elapsed = start.elapsed();

    let status = result.expect("single-node put_quorum with All policy completes via local ack");
    assert_eq!(status.acked_by, 1, "local ack counts as 1 peer");
    assert!(
        status.quorum_met,
        "drain completes with quorum_met=true (local ack path)"
    );
    assert!(
        elapsed < Duration::from_secs(2),
        "single-node drain should complete quickly, got {elapsed:?}"
    );
}