kevy_replicate/
replica_event.rs

1//! The events a [`crate::replica::ReplicaClient`] yields — split from
2//! `replica.rs` so each file stays under the 500-LOC house rule.
3//! Re-exported from [`crate::replica`], so caller paths are unchanged.
4
5use crate::feed::FeedPosition;
6use crate::replica::DecodedFrame;
7
8/// Event yielded by [`ReplicaClient::next_event`](crate::replica::ReplicaClient::next_event). A driver loop
9/// pattern-matches and applies each:
10/// - [`Self::Frame`] → run through the local dispatcher.
11/// - [`Self::SnapshotBegin`] → caller should reset / prepare the
12///   local store for a fresh-from-snapshot fill.
13/// - [`Self::SnapshotChunk`] → append the bytes to the caller's
14///   accumulating snapshot buffer.
15/// - [`Self::SnapshotEnd`] → caller hands the accumulated buffer to
16///   `kevy_persist::load_snapshot`; [`ReplicaClient`](crate::replica::ReplicaClient) has already
17///   advanced `expected_offset` to `ack_offset`, so the next
18///   [`Self::Frame`] arrives at `ack_offset` with no gap.
19///
20/// A replica joining through a snapshot sees the whole sequence:
21///
22/// ```
23/// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
24/// use kevy_replicate::replica::{ReplicaClient, ReplicaEvent};
25/// use kevy_replicate::wire;
26///
27/// # let mut stream = wire::encode_snapshot_begin();
28/// # stream.extend(wire::encode_snapshot_chunk(b"dump"));
29/// # stream.extend(wire::encode_snapshot_end(9));
30/// # stream.extend(wire::encode_frame(9, &argv(&["SET", "k", "v"])));
31/// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", stream);
32/// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
33/// let mut snapshot = Vec::new();
34/// while let Some(event) = client.next_event() {
35///     match event? {
36///         ReplicaEvent::SnapshotBegin => snapshot.clear(),
37///         ReplicaEvent::SnapshotChunk(bytes) => snapshot.extend(bytes),
38///         ReplicaEvent::SnapshotEnd { ack_offset } => assert_eq!(ack_offset, 9),
39///         ReplicaEvent::Frame(frame) => assert_eq!(frame.offset, 9),
40///         _ => {}
41///     }
42/// }
43/// assert_eq!(snapshot, b"dump");
44/// # Ok::<(), Box<dyn std::error::Error>>(())
45/// ```
46#[derive(Debug, Clone, PartialEq, Eq, Hash)]
47#[non_exhaustive]
48pub enum ReplicaEvent {
49    /// A live mutation frame.
50    ///
51    /// ```
52    /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
53    /// use kevy_replicate::replica::{DecodedFrame, ReplicaClient, ReplicaEvent};
54    ///
55    /// # let stream = kevy_replicate::wire::encode_frame(0, &argv(&["SET", "k", "v"]));
56    /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", stream);
57    /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
58    /// let expected = DecodedFrame::new(0, argv(&["SET", "k", "v"]));
59    /// assert_eq!(client.next_event().transpose()?, Some(ReplicaEvent::Frame(expected)));
60    /// # Ok::<(), Box<dyn std::error::Error>>(())
61    /// ```
62    Frame(DecodedFrame),
63    /// In-stream heartbeat: the primary's tail at send time — its feed
64    /// generation (the REPL.TOKEN / REPL.WAIT gen truth; `0` = the
65    /// primary spoke the legacy one-number heartbeat, "unknown") and its
66    /// `next_offset`. Lets the replica compute lag (applied vs primary)
67    /// and judge link liveness. Occupies no offset space.
68    ///
69    /// ```
70    /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
71    /// use kevy_replicate::feed::FeedPosition;
72    /// use kevy_replicate::replica::{ReplicaClient, ReplicaEvent};
73    ///
74    /// # let stream = kevy_replicate::wire::encode_ping(FeedPosition::new(1, 120));
75    /// # let (addr, _primary) = fake_primary(b"+ACK 1 100\r\n", stream);
76    /// let mut client = ReplicaClient::connect(addr, "replica-a", 100)?;
77    /// let Some(ReplicaEvent::Ping(tail)) = client.next_event().transpose()? else { panic!() };
78    /// assert_eq!(tail.offset - client.expected_offset(), 20); // frames behind the primary
79    /// # Ok::<(), Box<dyn std::error::Error>>(())
80    /// ```
81    Ping(FeedPosition),
82    /// Snapshot ship begin marker (`+SNAPSHOT\r\n`).
83    ///
84    /// ```
85    /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
86    /// use kevy_replicate::replica::{ReplicaClient, ReplicaEvent};
87    ///
88    /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", kevy_replicate::wire::encode_snapshot_begin());
89    /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
90    /// assert_eq!(client.next_event().transpose()?, Some(ReplicaEvent::SnapshotBegin));
91    /// # Ok::<(), Box<dyn std::error::Error>>(())
92    /// ```
93    SnapshotBegin,
94    /// One snapshot chunk's payload bytes (RESP bulk string body).
95    ///
96    /// ```
97    /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
98    /// use kevy_replicate::replica::{ReplicaClient, ReplicaEvent};
99    /// use kevy_replicate::wire;
100    ///
101    /// # let mut stream = wire::encode_snapshot_begin();
102    /// # stream.extend(wire::encode_snapshot_chunk(b"part-1"));
103    /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", stream);
104    /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
105    /// client.next_event(); // SnapshotBegin
106    /// let chunk = ReplicaEvent::SnapshotChunk(b"part-1".to_vec());
107    /// assert_eq!(client.next_event().transpose()?, Some(chunk));
108    /// # Ok::<(), Box<dyn std::error::Error>>(())
109    /// ```
110    SnapshotChunk(Vec<u8>),
111    /// Snapshot ship end marker carrying the offset the next live
112    /// frame will have.
113    ///
114    /// ```
115    /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
116    /// use kevy_replicate::replica::{ReplicaClient, ReplicaEvent};
117    /// use kevy_replicate::wire;
118    ///
119    /// # let mut stream = wire::encode_snapshot_begin();
120    /// # stream.extend(wire::encode_snapshot_end(57));
121    /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", stream);
122    /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
123    /// client.next_event(); // SnapshotBegin
124    /// let end = client.next_event().transpose()?;
125    /// assert_eq!(end, Some(ReplicaEvent::SnapshotEnd { ack_offset: 57 }));
126    /// # Ok::<(), Box<dyn std::error::Error>>(())
127    /// ```
128    SnapshotEnd {
129        /// The offset the primary's `next_offset` was at when the
130        /// snapshot started. After this event, [`ReplicaClient::expected_offset`](crate::replica::ReplicaClient::expected_offset)
131        /// equals this value.
132        ///
133        /// ```
134        /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
135        /// use kevy_replicate::replica::{ReplicaClient, ReplicaEvent};
136        /// use kevy_replicate::wire;
137        ///
138        /// # let mut stream = wire::encode_snapshot_begin();
139        /// # stream.extend(wire::encode_snapshot_end(57));
140        /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", stream);
141        /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
142        /// client.next_event(); // SnapshotBegin
143        /// let Some(ReplicaEvent::SnapshotEnd { ack_offset }) = client.next_event().transpose()? else {
144        ///     panic!("expected the end marker")
145        /// };
146        /// assert_eq!(client.expected_offset(), ack_offset); // the next live frame is 57
147        /// # Ok::<(), Box<dyn std::error::Error>>(())
148        /// ```
149        ack_offset: u64,
150    },
151}
kevy_replicate/replica_event.rs

kevy_replicate/
replica_event.rs
