nodedb_types/sync/wire/
kv.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! KV row push messages, and the refusal of a row push.
4//!
5//! `KvPushMsg` carries one KV write from a Lite client to Origin. A put
6//! carries the `{key, value…}` row every KV read returns, the same shape
7//! Origin sends in a `RowPushMsg`. Origin answers each push with a
8//! `KvPushAckMsg`. A terminal refusal travels in that ack as
9//! `AckStatus::Rejected`, the way every engine push ack carries one.
10//!
11//! `RowPushRejectMsg` is Lite's refusal of an Origin `RowPushMsg` it could
12//! not apply.
13//!
14//! Wire opcodes:
15//! - `0x16` — `RowPushReject` (Lite → Origin)
16//! - `0xAE` — `KvPush`        (Lite → Origin)
17//! - `0xAF` — `KvPushAck`     (Origin → Lite)
18
19use serde::{Deserialize, Serialize};
20
21use crate::sync::wire::ack_status::AckStatus;
22
23/// The write a `KvPushMsg` carries.
24#[derive(
25    Debug,
26    Clone,
27    PartialEq,
28    Eq,
29    Serialize,
30    Deserialize,
31    zerompk::ToMessagePack,
32    zerompk::FromMessagePack,
33)]
34pub enum KvPushOp {
35    /// Store `row` at the entry's key.
36    Put {
37        /// The `{key, value…}` row as standard MessagePack.
38        row: Vec<u8>,
39        /// Absolute expiry in milliseconds since the Unix epoch. `0` means
40        /// the entry never expires.
41        expire_at_ms: u64,
42    },
43    /// Remove the entry's key.
44    Delete,
45}
46
47/// KV write push (Lite → Origin, 0xAE).
48#[derive(
49    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
50)]
51pub struct KvPushMsg {
52    /// Lite instance ID.
53    pub lite_id: String,
54    /// Target KV collection.
55    pub collection: String,
56    /// The entry's key bytes.
57    pub key: Vec<u8>,
58    /// The write.
59    pub op: KvPushOp,
60    /// Lite-assigned ID for ACK correlation.
61    pub batch_id: u64,
62    /// Stable identity of the originating producer.
63    pub producer_id: u64,
64    /// Producer epoch.
65    pub epoch: u64,
66    /// Per-stream monotonic sequence number within the epoch.
67    pub seq: u64,
68}
69
70/// KV write push acknowledgment (Origin → Lite, 0xAF).
71#[derive(
72    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
73)]
74pub struct KvPushAckMsg {
75    /// Collection acknowledged.
76    pub collection: String,
77    /// Key from the originating `KvPushMsg`.
78    pub key: Vec<u8>,
79    /// Batch ID from the originating `KvPushMsg`.
80    pub batch_id: u64,
81    /// `true` unless `status` is `AckStatus::Rejected`.
82    pub accepted: bool,
83    /// Refusal detail when `status` is `AckStatus::Rejected`.
84    pub reject_reason: Option<String>,
85    /// Highest sequence from this producer's stream that Origin applied.
86    pub applied_seq: u64,
87    /// Idempotency outcome of the push.
88    pub status: AckStatus,
89}
90
91/// Why Lite refused an Origin row push.
92#[derive(
93    Debug,
94    Clone,
95    PartialEq,
96    Eq,
97    Serialize,
98    Deserialize,
99    zerompk::ToMessagePack,
100    zerompk::FromMessagePack,
101)]
102pub enum RowPushRefusal {
103    /// The payload is not the row shape the collection's engine expects.
104    Malformed { detail: String },
105    /// The payload decoded, and the local write failed.
106    ApplyFailed { detail: String },
107}
108
109impl std::fmt::Display for RowPushRefusal {
110    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        match self {
112            Self::Malformed { detail } => write!(f, "malformed row push: {detail}"),
113            Self::ApplyFailed { detail } => write!(f, "row push apply failed: {detail}"),
114        }
115    }
116}
117
118/// Lite's refusal of an Origin `RowPushMsg` (Lite → Origin, 0x16).
119#[derive(
120    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
121)]
122pub struct RowPushRejectMsg {
123    /// Collection of the refused row.
124    pub collection: String,
125    /// Document ID of the refused row.
126    pub document_id: String,
127    /// `sequence` of the refused `RowPushMsg`.
128    pub sequence: u64,
129    /// `peer_id` of the refused `RowPushMsg`.
130    pub peer_id: u64,
131    /// Why Lite refused the row.
132    pub refusal: RowPushRefusal,
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138
139    #[test]
140    fn a_put_push_round_trips() {
141        let msg = KvPushMsg {
142            lite_id: "lite-1".into(),
143            collection: "cfg".into(),
144            key: b"k1".to_vec(),
145            op: KvPushOp::Put {
146                row: vec![0x81, 0xa1, b'k', 0x01],
147                expire_at_ms: 42,
148            },
149            batch_id: 7,
150            producer_id: 3,
151            epoch: 1,
152            seq: 9,
153        };
154        let bytes = zerompk::to_msgpack_vec(&msg).expect("encode");
155        let back: KvPushMsg = zerompk::from_msgpack(&bytes).expect("decode");
156        assert_eq!(back.op, msg.op);
157        assert_eq!(back.key, msg.key);
158        assert_eq!(back.seq, 9);
159    }
160
161    #[test]
162    fn a_row_push_reject_round_trips() {
163        let msg = RowPushRejectMsg {
164            collection: "cfg".into(),
165            document_id: "k1".into(),
166            sequence: 4,
167            peer_id: 2,
168            refusal: RowPushRefusal::Malformed {
169                detail: "not a row map".into(),
170            },
171        };
172        let bytes = zerompk::to_msgpack_vec(&msg).expect("encode");
173        let back: RowPushRejectMsg = zerompk::from_msgpack(&bytes).expect("decode");
174        assert_eq!(back.refusal, msg.refusal);
175        assert_eq!(back.sequence, 4);
176    }
177}
nodedb_types/sync/wire/kv.rs

nodedb_types/sync/wire/
kv.rs
