nodedb 0.3.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal 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

// SPDX-License-Identifier: BUSL-1.1

//! `mirror_simulated_cross_region_latency`: inject 200 ms RTT into the
//! QUIC link; bootstrap completes, lag stays > 200 ms but < 5 s (Following),
//! BoundedStaleness(1000 ms) reads pass.
//!
//! This test simulates the observable lag from a cross-region deployment by
//! injecting a synthetic lag record with a 300 ms offset and verifying that:
//! 1. The status remains `Following` (lag < 5 s).
//! 2. A `BoundedStaleness(1000 ms)` read is accepted.
//! 3. A `BoundedStaleness(100 ms)` read is rejected (lag > 100 ms).
//!
//! The QUIC transport RTT is reflected in the lag record's `last_apply_ms`
//! value: a 200 ms round-trip means entries arrive ~200 ms late. The test
//! models this by writing a lag record with `last_apply_ms = now - 300`.

use std::time::Duration;

use nodedb_types::{DatabaseId, Lsn, MirrorMode, MirrorOrigin, MirrorStatus};
use tempfile::TempDir;

use nodedb::control::server::pgwire::ddl::database::{
    MirrorReadOutcome, check_mirror_read_consistency,
};
use nodedb::types::ReadConsistency;

use super::helpers::{TEST_SOURCE_CLUSTER, inject_lag_record_for_id, now_ms, open_tmp_catalog};

#[test]
fn mirror_simulated_cross_region_latency() {
    let dir = TempDir::new().unwrap();
    let catalog = open_tmp_catalog(&dir);
    let db_id = DatabaseId::new(5001);

    // Simulate 200 ms cross-region RTT: last entry applied 300 ms ago
    // (200 ms transit + ~100 ms apply queue drain).
    let simulated_lag_ms = 300u64;
    inject_lag_record_for_id(&catalog, db_id, simulated_lag_ms, 1);

    let origin = MirrorOrigin {
        source_cluster: TEST_SOURCE_CLUSTER.to_string(),
        source_database: DatabaseId::new(0),
        mode: MirrorMode::Async,
        last_applied: Lsn::new(1),
        status: MirrorStatus::Following,
    };

    // BoundedStaleness(1000 ms) must pass: 300 ms lag < 1000 ms bound.
    match check_mirror_read_consistency(
        &catalog,
        db_id,
        &origin,
        ReadConsistency::BoundedStaleness(Duration::from_millis(1_000)),
    ) {
        MirrorReadOutcome::ServeLocally => {}
        MirrorReadOutcome::Reject { message, .. } => {
            panic!("BoundedStaleness(1000ms) must pass with 300ms lag: {message}");
        }
    }

    // BoundedStaleness(100 ms) must fail: 300 ms lag > 100 ms bound.
    match check_mirror_read_consistency(
        &catalog,
        db_id,
        &origin,
        ReadConsistency::BoundedStaleness(Duration::from_millis(100)),
    ) {
        MirrorReadOutcome::Reject { sqlstate_code, .. } => {
            assert_eq!(
                sqlstate_code,
                nodedb_types::error::sqlstate::STALE_READ_NOT_LEADER
            );
        }
        MirrorReadOutcome::ServeLocally => {
            panic!("BoundedStaleness(100ms) must reject with 300ms lag");
        }
    }

    // Status must remain Following (lag < LAG_DEGRADED_MS = 5 s).
    use nodedb::control::mirror::{LAG_DEGRADED_MS, LagTransition, compute_lag_transition};
    let t = compute_lag_transition(&MirrorStatus::Following, simulated_lag_ms, now_ms(), false);
    assert_eq!(
        t,
        LagTransition::Unchanged,
        "300ms lag must not trigger Degraded (threshold is {}ms)",
        LAG_DEGRADED_MS
    );
}

#[test]
fn mirror_simulated_latency_at_threshold_boundary() {
    let dir = TempDir::new().unwrap();
    let catalog = open_tmp_catalog(&dir);
    let db_id = DatabaseId::new(5002);

    // Simulate RTT exactly at the BoundedStaleness window: 199 ms lag,
    // 200 ms bound — must pass.
    inject_lag_record_for_id(&catalog, db_id, 199, 1);

    let origin = MirrorOrigin {
        source_cluster: TEST_SOURCE_CLUSTER.to_string(),
        source_database: DatabaseId::new(0),
        mode: MirrorMode::Async,
        last_applied: Lsn::new(1),
        status: MirrorStatus::Following,
    };

    match check_mirror_read_consistency(
        &catalog,
        db_id,
        &origin,
        ReadConsistency::BoundedStaleness(Duration::from_millis(200)),
    ) {
        MirrorReadOutcome::ServeLocally => {}
        MirrorReadOutcome::Reject { message, .. } => {
            panic!("199ms lag should pass 200ms bound: {message}");
        }
    }
}