kevy 6.4.0

kevy — a pure-Rust, zero-dependency, Redis-compatible KV server.
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

//! Unit tests for [`super`] — the replication state instance.

use super::*;

#[test]
fn fresh_state_defaults() {
    let r = ReplicationState::new(1, false, 0);
    assert!(!r.is_replica());
    assert!(r.read_only());
    assert!(r.current_upstream().is_none());
    assert_eq!(r.applied_offset_sum(), 0);
    assert!(r.write_denied_reply(|| 0).is_none());
    assert!(r.read_denied_reply().is_none());
    assert!(!r.write_possibly_gated());
    assert!(!r.read_possibly_gated());
}

#[test]
fn applied_offset_sum_is_per_runner_sum_not_max() {
    let r = ReplicationState::new(3, false, 0);
    assert_eq!(r.applied_offset_sum(), 0, "no runners → 0");
    // Simulate a 3-runner fleet's registry (start_runners sizes
    // this in production).
    r.progress.size_runner_slots(3);
    r.progress.record_ping(0, 1, 100, 40);
    r.progress.record_applied(1, 25);
    r.progress.record_applied(2, 35);
    assert_eq!(r.applied_offset_sum(), 100, "sum across runners, not max");
    // Plain store semantics: a resync rewind must show through.
    r.progress.record_applied(1, 5);
    assert_eq!(r.applied_offset_sum(), 80);
    // Out-of-range slot is ignored (registry resize race guard).
    r.progress.record_applied(9, 1_000);
    assert_eq!(r.applied_offset_sum(), 80);
    // stop_runners clears the registry.
    r.stop_runners();
    assert_eq!(r.applied_offset_sum(), 0);
}

#[test]
fn start_runners_before_runtime_wiring_errors() {
    let r = ReplicationState::new(1, false, 0);
    let result = r.start_runners((IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 6400));
    assert!(result.is_err(), "inboxes never taken → no runtime drains them");
    assert!(!r.is_replica());
}

#[test]
fn promote_bumps_epoch_only_from_replica() {
    let r = ReplicationState::new(1, false, 0);
    r.promote_stop_runners();
    assert_eq!(r.promotion_epoch(), 0, "primary → primary is not a promotion");
    r.force_replica_flag();
    r.promote_stop_runners();
    assert_eq!(r.promotion_epoch(), 1);
    assert!(!r.is_replica());
}

#[test]
fn write_denied_orders_fences() {
    let r = ReplicationState::new(1, false, 0);
    r.set_quiesce(Some("10.0.0.9:7000".into()));
    assert!(r.write_possibly_gated(), "quiesce raises the gate bit");
    let reply = r.write_denied_reply(|| 0).expect("quiesced");
    assert!(reply.starts_with(b"-QUIESCED"), "{}", String::from_utf8_lossy(&reply));
    assert!(r.set_quorum_fence(true), "flag changed");
    let reply = r.write_denied_reply(|| 0).expect("fenced");
    assert!(reply.starts_with(b"-NOREPLICAS"), "fence outranks quiesce");
    assert!(!r.set_quorum_fence(true), "unchanged flag reports false");
    r.set_quorum_fence(false);
    r.set_quiesce(None);
    r.force_replica_flag();
    let reply = r.write_denied_reply(|| 0).expect("read-only replica");
    assert!(reply.starts_with(b"-READONLY"));
    r.set_read_only(false);
    assert!(r.write_denied_reply(|| 0).is_none(), "writable replica");
    assert!(!r.write_possibly_gated(), "writable replica clears the bit");
}

#[test]
fn every_gate_writer_bumps_the_control_epoch() {
    let r = ReplicationState::new(1, false, 0);
    let epoch = r.control_epoch_handle();
    let mut last = epoch.load(Ordering::Acquire);
    let assert_bumped = |last: &mut u64, what: &str| {
        let now = epoch.load(Ordering::Acquire);
        assert!(now > *last, "{what} must bump the control epoch");
        *last = now;
    };
    r.set_read_only(false);
    assert_bumped(&mut last, "set_read_only");
    r.set_min_replicas(2);
    assert_bumped(&mut last, "set_min_replicas");
    r.set_max_staleness_ms(500);
    assert_bumped(&mut last, "set_max_staleness_ms");
    r.set_quiesce(Some("x:1".into()));
    assert_bumped(&mut last, "set_quiesce");
    r.set_quorum_fence(true);
    assert_bumped(&mut last, "set_quorum_fence(changed)");
    assert!(!r.set_quorum_fence(true));
    assert_eq!(epoch.load(Ordering::Acquire), last, "unchanged fence: no bump");
    r.force_replica_flag();
    assert_bumped(&mut last, "force_replica_flag");
    r.stop_runners();
    assert_bumped(&mut last, "stop_runners");
}

#[test]
fn min_replicas_on_primary_keeps_write_gate_raised() {
    let r = ReplicationState::new(1, false, 0);
    r.set_min_replicas(1);
    assert!(r.write_possibly_gated(), "count is dynamic → always re-judge");
    let denied = r.write_denied_reply(|| 0).expect("no healthy replicas");
    assert!(denied.starts_with(b"-NOREPLICAS"));
    assert!(r.write_denied_reply(|| 1).is_none(), "count satisfied");
}