nodedb_lite/sync/array/inbound/
dispatcher.rs

1//! [`ArrayInbound`] — dispatcher for inbound array CRDT wire messages.
2//!
3//! This file holds the struct, the constructor, the snapshot-assembly buffer,
4//! and the small shared helpers used by the per-message-family `impl` blocks
5//! in sibling modules ([`super::delta`], [`super::snapshot`], [`super::schema`],
6//! [`super::reject`]).
7
8use std::collections::{BTreeMap, HashMap};
9use std::sync::{Arc, Mutex};
10
11use nodedb_array::sync::apply::ApplyOutcome;
12use nodedb_array::sync::op::ArrayOp;
13use nodedb_array::sync::snapshot::{SnapshotChunk, SnapshotHeader};
14
15use crate::error::LiteError;
16use crate::storage::engine::StorageEngine;
17use crate::sync::array::catchup::CatchupTracker;
18use crate::sync::array::pending::PendingQueue;
19use crate::sync::array::replica_state::ReplicaState;
20use crate::sync::array::schema_registry::SchemaRegistry;
21
22use super::apply::LiteApplyEngine;
23use super::outcome::InboundOutcome;
24
25/// In-flight snapshot assembly state keyed by `(array_name, snapshot_hlc_bytes)`.
26///
27/// An assembly exists only after its header arrived; chunks buffer against it.
28pub(super) struct SnapshotAssembly {
29    pub(super) header: SnapshotHeader,
30    pub(super) chunks: BTreeMap<u32, SnapshotChunk>,
31}
32
33impl SnapshotAssembly {
34    pub(super) fn new(header: SnapshotHeader) -> Self {
35        Self {
36            header,
37            chunks: BTreeMap::new(),
38        }
39    }
40}
41
42/// Dispatcher for inbound array CRDT wire messages from Origin.
43///
44/// Each `handle_*` method (defined in sibling modules) is stateless from the
45/// transport layer's perspective — callers pass the already-decoded wire
46/// message and receive an [`InboundOutcome`].
47///
48/// Snapshot state is buffered internally in `snapshots` until all chunks for a
49/// given `(array, snapshot_hlc)` have arrived.
50pub struct ArrayInbound<S: StorageEngine> {
51    pub(super) engine: Arc<LiteApplyEngine<S>>,
52    pub(super) schemas: Arc<SchemaRegistry<S>>,
53    pub(super) replica: Arc<ReplicaState>,
54    pub(super) pending: Arc<PendingQueue<S>>,
55    /// Catchup tracker — the high-water mark of applied ops per array, and
56    /// the arrays a `RetentionFloor` reject flagged for a snapshot.
57    pub(super) catchup: Arc<CatchupTracker<S>>,
58    /// In-flight snapshot chunk buffers.
59    /// Key: `(array_name, snapshot_hlc_bytes)`.
60    pub(super) snapshots: Mutex<HashMap<(String, [u8; 18]), SnapshotAssembly>>,
61}
62
63impl<S: StorageEngine> ArrayInbound<S> {
64    /// Construct from the component parts shared with `NodeDbLite`.
65    pub fn new(
66        engine: Arc<LiteApplyEngine<S>>,
67        schemas: Arc<SchemaRegistry<S>>,
68        replica: Arc<ReplicaState>,
69        pending: Arc<PendingQueue<S>>,
70        catchup: Arc<CatchupTracker<S>>,
71    ) -> Self {
72        Self {
73            engine,
74            schemas,
75            replica,
76            pending,
77            catchup,
78            snapshots: Mutex::new(HashMap::new()),
79        }
80    }
81
82    /// The stable replica identity for this Lite peer.
83    ///
84    /// Used by the transport layer to construct `ArrayAckMsg` bodies.
85    pub fn replica_id(&self) -> u64 {
86        self.replica.replica_id().as_u64()
87    }
88
89    /// Drive [`nodedb_array::sync::apply::apply_op`] on a borrowed
90    /// [`LiteApplyEngine`].
91    pub(super) fn apply_single_op(&self, op: &ArrayOp) -> Result<ApplyOutcome, LiteError> {
92        let mut engine_ref: &LiteApplyEngine<S> = &self.engine;
93        nodedb_array::sync::apply::apply_op(&mut engine_ref, op).map_err(|e| LiteError::Storage {
94            detail: format!("apply: {e}"),
95        })
96    }
97}
98
99/// Convert an [`ApplyOutcome`] from `nodedb-array` into an [`InboundOutcome`].
100pub(super) fn map_apply_outcome(outcome: ApplyOutcome) -> InboundOutcome {
101    match outcome {
102        ApplyOutcome::Applied => InboundOutcome::Applied,
103        ApplyOutcome::Idempotent => InboundOutcome::Idempotent,
104        ApplyOutcome::Rejected(r) => InboundOutcome::Rejected(r),
105    }
106}
nodedb_lite/sync/array/inbound/dispatcher.rs

nodedb_lite/sync/array/inbound/
dispatcher.rs
