kevy_replicate/
lib.rs

1//! kevy-replicate — primary-to-replica streaming replication.
2//!
3//! One primary streams every applied
4//! mutation to N read replicas over a long-lived TCP connection, using a
5//! RESP3-extended frame format with an offset envelope. New replicas join
6//! via an inline snapshot ship, then catch up from the live frame stream.
7//!
8//! - [`wire`] — RESP-based frame format (see `docs/wire.md`).
9//! - `wire_snapshot` (internal) — snapshot-ship framing for the joining
10//!   replica.
11//! - [`source`] — primary-side bounded backlog indexed by offset.
12//! - [`handshake`] — `REPLICATE FROM <offset> ID <id>` parse + `+ACK` format.
13//! - [`slot`] — per-replica state + reconnect-window expiry.
14//! - [`replica`] — replica-side blocking TCP client (handshake +
15//!   frame-decoding iterator).
16//!
17//! # Applying replicated frames
18//!
19//! `ReplicaClient` yields decoded `(offset, Argv)` tuples; *applying*
20//! them to a local store is the caller's responsibility — the right
21//! dispatcher depends on where the replica's data lives. The wire
22//! format intentionally carries the exact RESP argv the primary
23//! applied, so any dispatcher that hands `Argv` through Redis-verb
24//! routing produces a byte-equivalent local store.
25//!
26//! A replica against a primary built from this crate's primary-side
27//! pieces — the backlog ([`source`]), the handshake parser and `+ACK`
28//! ([`handshake`]) — applying each frame to a map:
29//!
30//! ```
31//! use std::collections::HashMap;
32//! use std::io::{Read, Write};
33//! use kevy_replicate::feed::FeedPosition;
34//! use kevy_replicate::handshake::{HandshakeReq, encode_ack};
35//! use kevy_replicate::replica::ReplicaClient;
36//! use kevy_replicate::source::ReplicationSource;
37//!
38//! // primary: two applied writes sit in the backlog
39//! let mut backlog = ReplicationSource::new(1 << 20);
40//! for cmd in [["SET", "a", "1"], ["SET", "b", "2"]] {
41//!     backlog.push_mutation(&kevy_resp::Argv::from(cmd.map(|w| w.as_bytes().to_vec()).to_vec()));
42//! }
43//! let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
44//! let addr = listener.local_addr()?;
45//! let primary = std::thread::spawn(move || -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
46//!     let (mut sock, _) = listener.accept()?;
47//!     let (mut buf, mut argv) = (Vec::new(), kevy_resp::Argv::default());
48//!     while kevy_resp::parse_command_into(&buf, &mut argv)?.is_none() {
49//!         let mut chunk = [0u8; 256];
50//!         let n = sock.read(&mut chunk)?;
51//!         buf.extend_from_slice(&chunk[..n]);
52//!     }
53//!     let req = HandshakeReq::parse(&argv)?;
54//!     sock.write_all(&encode_ack(FeedPosition::new(1, req.from.offset)))?;
55//!     for frame in backlog.frames_from(req.from.offset).map_err(|e| format!("{e:?}"))? {
56//!         sock.write_all(&frame.bytes)?;
57//!     }
58//!     Ok(()) // closing the socket ends the stream
59//! });
60//!
61//! // replica: apply every frame in offset order
62//! let mut store = HashMap::new();
63//! for frame in ReplicaClient::connect(addr, "replica-a", 0)? {
64//!     let frame = frame?;
65//!     if let [b"SET", key, value] = frame.argv.iter().collect::<Vec<_>>()[..] {
66//!         store.insert(key.to_vec(), value.to_vec());
67//!     }
68//! }
69//! primary.join().expect("primary thread").map_err(|e| e.to_string())?;
70//! assert_eq!(store.get(&b"b"[..]).map(Vec::as_slice), Some(&b"2"[..]));
71//! assert_eq!(store.len(), 2);
72//! # Ok::<(), Box<dyn std::error::Error>>(())
73//! ```
74//!
75//! In kevy itself the dispatcher is `kevy::KevyCommands::dispatch`
76//! over a `kevy::KeyspaceStore`: the frames carry the exact argv the
77//! primary applied, so the replica's store ends up byte-equivalent.
78//!
79//! # Features
80//!
81//! - `secure` (default) — Noise IK encrypted, mutually authenticated
82//!   links: `replica::ReplicaSecurity` and
83//!   `replica::ConnectOptions::with_security`. Off, a replica connects
84//!   in plaintext only and the handshake code is not linked.
85//!
86//! See the `replica_apply_dispatch_mirrors_primary_store` integration
87//! test in `crates/kevy/tests/replication.rs` for the pattern under
88//! the full primary+replica end-to-end harness.
89//!
90//! The kevy binary also ships full **server-as-replica** mode (it
91//! auto-spawns a `ReplicaClient` when `[replication] role = "replica"`,
92//! routing frames into the reactor with re-replication suppression);
93//! the in-process recipe above is for any user that wants to drive
94//! replication themselves.
95#![forbid(unsafe_code)]
96#![warn(missing_docs)]
97
98pub mod feed;
99pub mod handshake;
100pub mod replica;
101mod replica_connect;
102mod replica_decode;
103mod replica_error;
104mod replica_event;
105#[cfg(not(feature = "secure"))]
106#[path = "replica_plain.rs"]
107mod replica_secure;
108#[cfg(feature = "secure")]
109mod replica_secure;
110pub mod slot;
111pub mod source;
112pub mod wire;
113mod wire_snapshot;
114
115const _: () = {
116    const fn send_sync<T: Send + Sync>() {}
117    send_sync::<feed::FeedPosition>();
118    send_sync::<feed::FeedRead>();
119    send_sync::<feed::FeedFrame<'static>>();
120    send_sync::<feed::FeedSource>();
121    send_sync::<handshake::HandshakeReq>();
122    send_sync::<handshake::HandshakeError>();
123    send_sync::<replica::DecodedFrame>();
124    send_sync::<replica::ReplicaEvent>();
125    send_sync::<replica::ReplicaClient>();
126    send_sync::<replica::ReplicaError>();
127    #[cfg(feature = "secure")]
128    send_sync::<replica::ReplicaSecurity>();
129    send_sync::<replica::ConnectOptions>();
130    send_sync::<slot::ReplicaSlot>();
131    send_sync::<slot::SlotTable>();
132    send_sync::<source::Frame>();
133    send_sync::<source::FromOffset>();
134    send_sync::<source::ReplicationSource>();
135    send_sync::<source::FramesIter<'static>>();
136    send_sync::<wire::WireError>();
137    send_sync::<wire::SnapshotMarker>();
138};
kevy_replicate/lib.rs

kevy_replicate/
lib.rs
