kevy 5.3.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
117
118
119
120
121
122
123
124
125
126

//! E2e: `MOVE-SCOPE` command round-trips against a mock
//! target. Validates the full wire path (TCP connect → ship request
//! → bulk parse → embedded command apply on target → `+OK <count>`
//! reply → source `migration_commit`) without requiring two real
//! `kevy_rt::Runtime` peers in the same test binary (the
//! cluster-port flake-prone multi-server harness can't yet
//! cleanly host two configs in one binary).

#![cfg(not(target_arch = "wasm32"))]

use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::Arc;
use std::time::{Duration, Instant};

use kevy_config::Config;
use kevy_resp::Argv;
use kevy_store::Store;

/// Spawn a TCP listener that accepts one connection, drains the
/// request bytes until EOF or a sensible cap, replies `+OK 1`, and
/// returns the request bytes via a channel. Single-shot — closes
/// after the one exchange.
fn spawn_mock_target() -> (u16, std::sync::mpsc::Receiver<Vec<u8>>) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        let (mut s, _) = listener.accept().expect("mock target accept");
        s.set_read_timeout(Some(Duration::from_secs(5))).ok();
        // Read until we have enough bytes to recognise a complete
        // MOVE-SCOPE-INGEST command. Simplest robust approach: read
        // a small chunk a few times.
        let mut buf = Vec::new();
        let mut chunk = [0u8; 8192];
        let deadline = Instant::now() + Duration::from_secs(2);
        loop {
            match s.read(&mut chunk) {
                Ok(0) => break,
                Ok(n) => buf.extend_from_slice(&chunk[..n]),
                Err(_) => break,
            }
            // Heuristic: once we've seen the closing CRLF after the
            // outer bulk, stop. Production target would use the
            // proper parser, but for the mock we just acknowledge.
            if buf.len() >= 32 && Instant::now() > deadline - Duration::from_secs(1) {
                break;
            }
            if Instant::now() > deadline {
                break;
            }
        }
        let _ = s.write_all(b"+OK 1\r\n");
        let _ = tx.send(buf);
    });
    // Wait a few ms for the listener thread to actually bind.
    for _ in 0..50 {
        if TcpStream::connect_timeout(
            &std::net::SocketAddr::from(([127, 0, 0, 1], port)),
            Duration::from_millis(20),
        )
        .is_ok()
        {
            break;
        }
        std::thread::sleep(Duration::from_millis(5));
    }
    (port, rx)
}

fn argv(parts: &[&[u8]]) -> Argv {
    let mut a = Argv::default();
    for p in parts {
        a.push(p);
    }
    a
}

#[test]
fn move_scope_ships_prefix_slice_to_mock_target_and_commits() {
    let (port, rx) = spawn_mock_target();

    // Install scope_integration with self_node_id = A, peers = A + B.
    let mut cfg = Config::default();
    cfg.cluster.node_id = "A".to_string();
    cfg.cluster.peers = kevy_config::PeerEntry::parse_list(&format!(
        "A@127.0.0.1:11000,B@127.0.0.1:{port}",
    ))
    .unwrap();
    // The state carries node identity + peer table straight from the
    // config, so the public dispatch path exercises the same routing
    // `kevy::serve` would.
    let kevy = kevy::KevyCommands::with_state(Arc::new(
        kevy::RuntimeState::new(Arc::new(cfg), std::path::PathBuf::new(), 1).unwrap(),
    ));

    // Pre-fill the source store with two keys under the prefix.
    let mut store = Store::new();
    store.set(b"test:a", b"1".to_vec(), None, false, false);
    store.set(b"test:b", b"2".to_vec(), None, false, false);

    // Issue the MOVE-SCOPE command via the public dispatch path.
    let args = argv(&[b"MOVE-SCOPE", b"test:", b"FROM", b"A", b"TO", b"B"]);
    let reply = kevy.dispatch(&mut store, &args);

    let reply_s = String::from_utf8_lossy(&reply).to_string();
    assert!(
        reply_s.starts_with('+') || reply_s.starts_with('-'),
        "reply should be a simple-string or error: {reply_s:?}"
    );

    if reply_s.starts_with('+') {
        // Success path — mock target should have received the
        // request.
        let request = rx
            .recv_timeout(Duration::from_secs(2))
            .expect("mock target should have received the request");
        let req_s = String::from_utf8_lossy(&request);
        assert!(
            req_s.contains("MOVE-SCOPE-INGEST"),
            "request shape: {req_s:?}",
        );
        assert!(req_s.contains("test:a"), "key 1 in request: {req_s:?}");
        assert!(req_s.contains("test:b"), "key 2 in request: {req_s:?}");
    }
}