nodedb_lite/sync/array/inbound/
delta.rs

1//! Delta and delta-batch handlers — single-op and batched op application.
2
3use nodedb_array::sync::op_codec;
4use nodedb_types::sync::wire::array::{ArrayDeltaBatchMsg, ArrayDeltaMsg};
5
6use crate::error::LiteError;
7use crate::storage::engine::StorageEngine;
8
9use super::dispatcher::{ArrayInbound, map_apply_outcome};
10use super::outcome::InboundOutcome;
11
12impl<S: StorageEngine> ArrayInbound<S> {
13    /// Apply a single delta message from Origin.
14    ///
15    /// Decodes the op payload, observes the HLC, then delegates to
16    /// [`nodedb_array::sync::apply::apply_op`].
17    pub fn handle_delta(&self, msg: &ArrayDeltaMsg) -> Result<InboundOutcome, LiteError> {
18        let op = op_codec::decode_op(&msg.op_payload).map_err(|e| LiteError::Storage {
19            detail: format!("handle_delta decode: {e}"),
20        })?;
21
22        self.replica.observe(op.header.hlc)?;
23
24        let outcome = self.apply_single_op(&op)?;
25        Ok(map_apply_outcome(outcome))
26    }
27
28    /// Apply a batch of delta messages from Origin.
29    ///
30    /// Returns one [`InboundOutcome`] per op payload in the batch.
31    pub fn handle_delta_batch(
32        &self,
33        msg: &ArrayDeltaBatchMsg,
34    ) -> Result<Vec<InboundOutcome>, LiteError> {
35        let mut outcomes = Vec::with_capacity(msg.op_payloads.len());
36        for payload in &msg.op_payloads {
37            let op = op_codec::decode_op(payload).map_err(|e| LiteError::Storage {
38                detail: format!("handle_delta_batch decode: {e}"),
39            })?;
40            self.replica.observe(op.header.hlc)?;
41            let outcome = self.apply_single_op(&op)?;
42            outcomes.push(map_apply_outcome(outcome));
43        }
44        Ok(outcomes)
45    }
46}
47
48#[cfg(test)]
49mod tests {
50    use nodedb_array::sync::apply::ApplyRejection;
51    use nodedb_array::sync::hlc::Hlc;
52    use nodedb_array::sync::op::{ArrayOp, ArrayOpHeader, ArrayOpKind};
53    use nodedb_array::sync::op_codec;
54    use nodedb_array::sync::replica_id::ReplicaId;
55    use nodedb_array::types::cell_value::value::CellValue;
56    use nodedb_array::types::coord::value::CoordValue;
57    use nodedb_types::sync::wire::array::{ArrayDeltaBatchMsg, ArrayDeltaMsg};
58
59    use super::super::fixtures::{hlc, make_inbound, put_op, simple_schema};
60    use super::super::outcome::InboundOutcome;
61
62    #[tokio::test(flavor = "multi_thread")]
63    async fn handle_delta_applies_put() {
64        let (inbound, schemas, _pending, storage) = make_inbound().await;
65
66        // Register schema in both the schemas registry AND the engine's catalog.
67        schemas
68            .put_schema("arr", &simple_schema("arr"))
69            .await
70            .unwrap();
71        {
72            let mut state = inbound.engine.array_state.lock().await;
73            state
74                .create_array(&storage, "arr", simple_schema("arr"), None, None)
75                .await
76                .unwrap();
77        }
78        let schema_hlc = schemas.schema_hlc("arr").unwrap();
79
80        let op = ArrayOp {
81            header: ArrayOpHeader {
82                array: "arr".into(),
83                hlc: hlc(100),
84                schema_hlc,
85                valid_from_ms: 0,
86                valid_until_ms: -1,
87                system_from_ms: 100,
88            },
89            kind: ArrayOpKind::Put,
90            coord: vec![CoordValue::Int64(3)],
91            attrs: Some(vec![CellValue::Float64(42.0)]),
92        };
93
94        let payload = op_codec::encode_op(&op).unwrap();
95        let msg = ArrayDeltaMsg {
96            array: "arr".into(),
97            op_payload: payload,
98            producer_id: 0,
99            epoch: 0,
100            seq: 0,
101        };
102        let outcome = inbound.handle_delta(&msg).unwrap();
103        assert_eq!(outcome, InboundOutcome::Applied);
104    }
105
106    #[tokio::test(flavor = "multi_thread")]
107    async fn handle_delta_idempotent() {
108        let (inbound, schemas, _pending, storage) = make_inbound().await;
109        schemas
110            .put_schema("arr", &simple_schema("arr"))
111            .await
112            .unwrap();
113        {
114            let mut state = inbound.engine.array_state.lock().await;
115            state
116                .create_array(&storage, "arr", simple_schema("arr"), None, None)
117                .await
118                .unwrap();
119        }
120        let schema_hlc = schemas.schema_hlc("arr").unwrap();
121
122        let op = ArrayOp {
123            header: ArrayOpHeader {
124                array: "arr".into(),
125                hlc: hlc(200),
126                schema_hlc,
127                valid_from_ms: 0,
128                valid_until_ms: -1,
129                system_from_ms: 200,
130            },
131            kind: ArrayOpKind::Put,
132            coord: vec![CoordValue::Int64(1)],
133            attrs: Some(vec![CellValue::Float64(1.0)]),
134        };
135        let payload = op_codec::encode_op(&op).unwrap();
136        let msg = ArrayDeltaMsg {
137            array: "arr".into(),
138            op_payload: payload.clone(),
139            producer_id: 0,
140            epoch: 0,
141            seq: 0,
142        };
143
144        // First application — should be Applied.
145        let o1 = inbound.handle_delta(&msg).unwrap();
146        assert_eq!(o1, InboundOutcome::Applied);
147
148        // Second application of identical op — must be Idempotent.
149        let msg2 = ArrayDeltaMsg {
150            array: "arr".into(),
151            op_payload: payload,
152            producer_id: 0,
153            epoch: 0,
154            seq: 0,
155        };
156        let o2 = inbound.handle_delta(&msg2).unwrap();
157        assert_eq!(o2, InboundOutcome::Idempotent);
158    }
159
160    #[tokio::test(flavor = "multi_thread")]
161    async fn handle_delta_unknown_array_returns_rejected() {
162        let (inbound, _schemas, _pending, _storage) = make_inbound().await;
163        // No array registered → ApplyRejection::ArrayUnknown.
164        let op = put_op("unknown_arr", 50, 50);
165        let payload = op_codec::encode_op(&op).unwrap();
166        let msg = ArrayDeltaMsg {
167            array: "unknown_arr".into(),
168            op_payload: payload,
169            producer_id: 0,
170            epoch: 0,
171            seq: 0,
172        };
173        let outcome = inbound.handle_delta(&msg).unwrap();
174        assert!(
175            matches!(
176                outcome,
177                InboundOutcome::Rejected(ApplyRejection::ArrayUnknown { .. })
178            ),
179            "expected ArrayUnknown rejection, got: {outcome:?}"
180        );
181    }
182
183    #[tokio::test(flavor = "multi_thread")]
184    async fn handle_delta_schema_too_new_returns_rejected() {
185        let (inbound, schemas, _pending, storage) = make_inbound().await;
186        schemas
187            .put_schema("arr", &simple_schema("arr"))
188            .await
189            .unwrap();
190        {
191            let mut state = inbound.engine.array_state.lock().await;
192            state
193                .create_array(&storage, "arr", simple_schema("arr"), None, None)
194                .await
195                .unwrap();
196        }
197        // Local schema_hlc is hlc(X); op carries schema_hlc far in the future.
198        let schema_hlc_future = Hlc::new(u64::MAX >> 16, 0, ReplicaId::new(99)).unwrap();
199        let op = ArrayOp {
200            header: ArrayOpHeader {
201                array: "arr".into(),
202                hlc: hlc(10),
203                schema_hlc: schema_hlc_future,
204                valid_from_ms: 0,
205                valid_until_ms: -1,
206                system_from_ms: 10,
207            },
208            kind: ArrayOpKind::Put,
209            coord: vec![CoordValue::Int64(0)],
210            attrs: Some(vec![CellValue::Float64(0.0)]),
211        };
212        let payload = op_codec::encode_op(&op).unwrap();
213        let msg = ArrayDeltaMsg {
214            array: "arr".into(),
215            op_payload: payload,
216            producer_id: 0,
217            epoch: 0,
218            seq: 0,
219        };
220        let outcome = inbound.handle_delta(&msg).unwrap();
221        assert!(
222            matches!(
223                outcome,
224                InboundOutcome::Rejected(ApplyRejection::SchemaTooNew { .. })
225            ),
226            "expected SchemaTooNew, got: {outcome:?}"
227        );
228    }
229
230    #[tokio::test(flavor = "multi_thread")]
231    async fn handle_delta_batch_processes_all() {
232        let (inbound, schemas, _pending, storage) = make_inbound().await;
233        schemas.put_schema("b", &simple_schema("b")).await.unwrap();
234        {
235            let mut state = inbound.engine.array_state.lock().await;
236            state
237                .create_array(&storage, "b", simple_schema("b"), None, None)
238                .await
239                .unwrap();
240        }
241        let schema_hlc = schemas.schema_hlc("b").unwrap();
242
243        let ops: Vec<ArrayOp> = (1u64..=4)
244            .map(|i| ArrayOp {
245                header: ArrayOpHeader {
246                    array: "b".into(),
247                    hlc: hlc(i * 100),
248                    schema_hlc,
249                    valid_from_ms: 0,
250                    valid_until_ms: -1,
251                    system_from_ms: (i * 100) as i64,
252                },
253                kind: ArrayOpKind::Put,
254                coord: vec![CoordValue::Int64(i as i64)],
255                attrs: Some(vec![CellValue::Float64(i as f64)]),
256            })
257            .collect();
258
259        let op_payloads: Vec<Vec<u8>> = ops
260            .iter()
261            .map(|op| op_codec::encode_op(op).unwrap())
262            .collect();
263
264        let msg = ArrayDeltaBatchMsg {
265            array: "b".into(),
266            op_payloads,
267        };
268        let outcomes = inbound.handle_delta_batch(&msg).unwrap();
269        assert_eq!(outcomes.len(), 4);
270        assert!(outcomes.iter().all(|o| *o == InboundOutcome::Applied));
271    }
272}
nodedb_lite/sync/array/inbound/delta.rs

nodedb_lite/sync/array/inbound/
delta.rs
