kevy_rt/
replica_inbox.rs

1//! Cross-thread inbox from an external replica runner into a
2//! [`Shard`]'s reactor thread. The replica runner lives on its own OS
3//! thread (it does blocking `TcpStream` reads from the upstream
4//! primary via `ReplicaClient`); applying mutations to the shard's
5//! `Store` must happen on the shard's reactor thread, so the runner
6//! drops events into this channel and the shard drains it once per
7//! tick.
8//!
9//! The kevy server (the embedder) creates one [`ReplicaInbox`] pair
10//! per shard before `Runtime::run`, hands the receivers to the
11//! runtime via `with_replica_inboxes`, and keeps the senders to wire
12//! into the runner threads. v1.18 spawns one runner per shard
13//! (matching the primary's per-shard listener layout), so the
14//! channels are 1:1.
15//!
16//! v1.18 cap: events are unbounded. Each [`ReplicaApply::Frame`]
17//! carries an owned [`Argv`] (snapshot path is `Vec<u8>` chunks); for
18//! a slow shard this can grow. Backpressure / capping is tracked as a
19//! follow-up — the v1.18 model assumes the shard's apply rate matches
20//! the upstream emit rate (single-machine cluster). The unbounded
21//! channel never blocks the runner thread, so a stuck shard never
22//! stalls the runner's TCP read (it just buffers).
23
24use std::sync::mpsc::{Receiver, SendError, Sender, channel};
25
26use crate::Argv;
27
28/// One event delivered from a replica runner to its target shard.
29/// Mirrors `kevy_replicate::replica::ReplicaEvent` except `Frame`
30/// carries an owned [`Argv`] (already decoded by the runner) instead
31/// of a `DecodedFrame { offset, argv }` — the offset is gap-checked
32/// by the runner on the way in, so the shard doesn't need it.
33#[derive(Debug)]
34pub enum ReplicaApply {
35    /// Upstream started shipping a full snapshot. The shard should
36    /// reset its accumulating snapshot buffer.
37    SnapshotBegin,
38    /// One chunk of snapshot bytes. The shard appends to its buffer.
39    SnapshotChunk(Vec<u8>),
40    /// Upstream finished the snapshot. The shard hands its buffered
41    /// bytes to `kevy_persist::load_snapshot_from` (replacing the
42    /// `Store` contents) and resumes at `ack_offset` for live frames.
43    /// `routed = true` (v3.2 single-source mode) means the payload is
44    /// the WHOLE upstream keyspace broadcast to every shard — each
45    /// shard loads only its own hash slice.
46    SnapshotEnd { ack_offset: u64, routed: bool },
47    /// One live mutation frame to be applied via `kevy::dispatch`
48    /// (inside a [`crate::ReplicatedApplyGuard`] scope so the apply
49    /// doesn't re-push into this shard's downstream
50    /// `ReplicationSource`).
51    Frame { offset: u64, argv: Argv },
52}
53
54/// Sender end of a per-shard replica inbox. `Send + Clone + Sync`
55/// (one std::sync::mpsc::Sender, no extra state) so the embedder can
56/// hand it freely to runner threads.
57#[derive(Clone)]
58pub struct ReplicaInboxSender {
59    inner: Sender<ReplicaApply>,
60}
61
62impl ReplicaInboxSender {
63    /// Send one event to the target shard. Fails only when the shard
64    /// has dropped its receiver (the runtime stopped or the shard
65    /// crashed) — the runner should treat that as "no more apply
66    /// possible" and exit.
67    pub fn send(&self, ev: ReplicaApply) -> Result<(), SendError<ReplicaApply>> {
68        self.inner.send(ev)
69    }
70}
71
72/// Receiver end. Lives inside the (private) `Shard`; drained once
73/// per reactor tick. Constructed by [`replica_inbox_pair`] and
74/// handed to the runtime via `Runtime::with_replica_inboxes`.
75pub struct ReplicaInboxReceiver {
76    pub(crate) inner: Receiver<ReplicaApply>,
77}
78
79/// Create a matched (sender, receiver) pair for one shard's replica
80/// inbox. The embedder calls this `nshards` times before
81/// `Runtime::run`.
82#[must_use]
83pub fn replica_inbox_pair() -> (ReplicaInboxSender, ReplicaInboxReceiver) {
84    let (tx, rx) = channel();
85    (
86        ReplicaInboxSender { inner: tx },
87        ReplicaInboxReceiver { inner: rx },
88    )
89}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94
95    #[test]
96    fn pair_round_trips_one_event() {
97        let (tx, rx) = replica_inbox_pair();
98        tx.send(ReplicaApply::SnapshotBegin).unwrap();
99        match rx.inner.recv().unwrap() {
100            ReplicaApply::SnapshotBegin => {}
101            other => panic!("expected SnapshotBegin, got {other:?}"),
102        }
103    }
104
105    #[test]
106    fn drop_receiver_makes_send_fail() {
107        let (tx, rx) = replica_inbox_pair();
108        drop(rx);
109        let err = tx.send(ReplicaApply::SnapshotBegin).unwrap_err();
110        match err.0 {
111            ReplicaApply::SnapshotBegin => {}
112            other => panic!("expected payload roundtrip, got {other:?}"),
113        }
114    }
115
116    #[test]
117    fn sender_is_clone_send_sync() {
118        fn assert_traits<T: Clone + Send + Sync>() {}
119        assert_traits::<ReplicaInboxSender>();
120    }
121}
kevy_rt/replica_inbox.rs

kevy_rt/
replica_inbox.rs
