nodedb_lite/sync/client/
row_push_reject.rs

1//! Row push refusals waiting to go back to Origin.
2//!
3//! When this replica refuses a `RowPush`, the dispatch loop queues a
4//! `RowPushReject` here. The push loop drains the queue each tick and sends
5//! every refusal, so Origin records the row this replica does not hold.
6
7use nodedb_types::error::{ErrorDetails, NodeDbError};
8use nodedb_types::sync::wire::{RowPushMsg, RowPushRefusal, RowPushRejectMsg};
9
10use super::state::SyncClient;
11
12/// Most refusals held for the next tick. The queue drains every tick, so
13/// the cap is reached only when refusals arrive faster than one tick sends
14/// them. A refusal past the cap is logged and not sent.
15pub(super) const ROW_PUSH_REJECT_CAP: usize = 1024;
16
17impl SyncClient {
18    /// Queue the refusal of `msg`, which failed to apply with `error`.
19    pub async fn queue_row_push_reject(&self, msg: &RowPushMsg, error: &NodeDbError) {
20        let reject = row_push_reject(msg, error);
21        let mut queue = self.pending_row_push_rejects.lock().await;
22        if queue.len() >= ROW_PUSH_REJECT_CAP {
23            tracing::error!(
24                collection = %reject.collection,
25                document_id = %reject.document_id,
26                sequence = reject.sequence,
27                cap = ROW_PUSH_REJECT_CAP,
28                "RowPushReject queue full; Origin is not told about this refused row"
29            );
30            return;
31        }
32        queue.push(reject);
33    }
34
35    /// Take every queued refusal.
36    pub async fn drain_row_push_rejects(&self) -> Vec<RowPushRejectMsg> {
37        std::mem::take(&mut *self.pending_row_push_rejects.lock().await)
38    }
39
40    /// Put back `unsent`, drained refusals a failed send did not deliver,
41    /// ahead of any queued since. Refusals past the cap are logged and
42    /// not sent.
43    pub async fn requeue_row_push_rejects(&self, unsent: Vec<RowPushRejectMsg>) {
44        let mut queue = self.pending_row_push_rejects.lock().await;
45        let newer = std::mem::replace(&mut *queue, unsent);
46        queue.extend(newer);
47        if queue.len() > ROW_PUSH_REJECT_CAP {
48            for dropped in queue.drain(ROW_PUSH_REJECT_CAP..) {
49                tracing::error!(
50                    collection = %dropped.collection,
51                    document_id = %dropped.document_id,
52                    sequence = dropped.sequence,
53                    cap = ROW_PUSH_REJECT_CAP,
54                    "RowPushReject queue full; Origin is not told about this refused row"
55                );
56            }
57        }
58    }
59}
60
61/// The refusal of `msg`. A payload that does not decode is `Malformed`.
62/// Any other error is `ApplyFailed`.
63fn row_push_reject(msg: &RowPushMsg, error: &NodeDbError) -> RowPushRejectMsg {
64    let detail = error.to_string();
65    let refusal = if matches!(error.details(), ErrorDetails::Serialization { .. }) {
66        RowPushRefusal::Malformed { detail }
67    } else {
68        RowPushRefusal::ApplyFailed { detail }
69    };
70    RowPushRejectMsg {
71        collection: msg.collection.clone(),
72        document_id: msg.document_id.clone(),
73        sequence: msg.sequence,
74        peer_id: msg.peer_id,
75        refusal,
76    }
77}
78
79#[cfg(test)]
80mod tests {
81    use nodedb_types::sync::wire::RowOp;
82
83    use super::*;
84    use crate::sync::client::SyncConfig;
85
86    fn row_push() -> RowPushMsg {
87        RowPushMsg {
88            collection: "cfg".into(),
89            document_id: "k1".into(),
90            payload: vec![0xc0],
91            op: RowOp::Upsert,
92            lsn: 9,
93            peer_id: 4,
94            sequence: 12,
95        }
96    }
97
98    #[tokio::test]
99    async fn a_decode_error_is_queued_as_malformed() {
100        let client = SyncClient::new(SyncConfig::new("wss://localhost:9090/sync", "t"));
101        let error = NodeDbError::serialization("msgpack", "not a row map");
102        client.queue_row_push_reject(&row_push(), &error).await;
103
104        let drained = client.drain_row_push_rejects().await;
105        assert_eq!(drained.len(), 1);
106        assert_eq!(drained[0].collection, "cfg");
107        assert_eq!(drained[0].document_id, "k1");
108        assert_eq!(drained[0].sequence, 12);
109        assert_eq!(drained[0].peer_id, 4);
110        assert!(matches!(
111            drained[0].refusal,
112            RowPushRefusal::Malformed { .. }
113        ));
114        assert!(client.drain_row_push_rejects().await.is_empty());
115    }
116
117    #[tokio::test]
118    async fn a_storage_error_is_queued_as_apply_failed() {
119        let client = SyncClient::new(SyncConfig::new("wss://localhost:9090/sync", "t"));
120        let error = NodeDbError::storage("disk full");
121        client.queue_row_push_reject(&row_push(), &error).await;
122        let drained = client.drain_row_push_rejects().await;
123        assert!(matches!(
124            drained[0].refusal,
125            RowPushRefusal::ApplyFailed { .. }
126        ));
127    }
128
129    #[tokio::test]
130    async fn requeued_refusals_go_ahead_of_newer_ones() {
131        let client = SyncClient::new(SyncConfig::new("wss://localhost:9090/sync", "t"));
132        let error = NodeDbError::storage("disk full");
133        let mut first = row_push();
134        first.sequence = 1;
135        client.queue_row_push_reject(&first, &error).await;
136        let unsent = client.drain_row_push_rejects().await;
137
138        let mut second = row_push();
139        second.sequence = 2;
140        client.queue_row_push_reject(&second, &error).await;
141        client.requeue_row_push_rejects(unsent).await;
142
143        let order: Vec<u64> = client
144            .drain_row_push_rejects()
145            .await
146            .iter()
147            .map(|r| r.sequence)
148            .collect();
149        assert_eq!(order, vec![1, 2]);
150    }
151}
nodedb_lite/sync/client/row_push_reject.rs

nodedb_lite/sync/client/
row_push_reject.rs
