nodedb_lite/sync/array/inbound/
reject.rs

1//! Reject-message handler: Origin refused an op this replica emitted.
2//!
3//! [`reject_disposition`] decides the op's fate. Only a `Drop` disposition
4//! removes the op from the pending queue. The push loop handles schema
5//! re-sends and parking. Every reject is logged.
6
7use nodedb_array::sync::hlc::Hlc;
8use nodedb_types::sync::wire::array::ArrayRejectMsg;
9
10use crate::error::LiteError;
11use crate::storage::engine::StorageEngine;
12use crate::sync::array::reject_policy::{RejectDisposition, reject_disposition};
13
14use super::dispatcher::ArrayInbound;
15use super::outcome::InboundOutcome;
16
17impl<S: StorageEngine> ArrayInbound<S> {
18    /// Process a reject message from Origin.
19    ///
20    /// - `Drop`: removes the op from the pending queue and logs the loss at
21    ///   error level.
22    /// - `CatchUp`: persists the array's catch-up flag, so the next catch-up
23    ///   request starts at `Hlc::ZERO`.
24    /// - `ResendSchema` and `Park`: keep the op queued and log the reject.
25    ///
26    /// Returns [`InboundOutcome::RejectAcknowledged`] in every case, including
27    /// a `Drop` for an op no longer queued.
28    pub async fn handle_reject(&self, msg: &ArrayRejectMsg) -> Result<InboundOutcome, LiteError> {
29        let op_hlc = Hlc::from_bytes(&msg.op_hlc_bytes);
30        match reject_disposition(msg.reason) {
31            RejectDisposition::Drop => {
32                let was_present = self.pending.remove(op_hlc).await?;
33                tracing::error!(
34                    array = %msg.array,
35                    reason = ?msg.reason,
36                    detail = %msg.detail,
37                    op_hlc_physical_ms = op_hlc.physical_ms,
38                    was_in_pending = was_present,
39                    "array op permanently refused by Origin; op dropped, the write never reaches Origin"
40                );
41            }
42            RejectDisposition::CatchUp => {
43                tracing::warn!(
44                    array = %msg.array,
45                    detail = %msg.detail,
46                    "array catch-up cursor is below Origin's retention floor; requesting a snapshot"
47                );
48                if let Err(e) = self.catchup.record_reject_retention_floor(&msg.array).await {
49                    tracing::warn!(
50                        array = %msg.array,
51                        error = %e,
52                        "array_inbound: failed to persist catchup_needed flag"
53                    );
54                }
55            }
56            disposition @ (RejectDisposition::ResendSchema | RejectDisposition::Park) => {
57                tracing::warn!(
58                    array = %msg.array,
59                    reason = ?msg.reason,
60                    detail = %msg.detail,
61                    op_hlc_physical_ms = op_hlc.physical_ms,
62                    ?disposition,
63                    "array op refused by Origin; op kept for retry"
64                );
65            }
66        }
67
68        Ok(InboundOutcome::RejectAcknowledged)
69    }
70}
71
72#[cfg(test)]
73mod tests {
74    use nodedb_types::sync::wire::array::{ArrayRejectMsg, ArrayRejectReason};
75
76    use super::super::fixtures::{hlc, make_inbound, put_op};
77    use super::super::outcome::InboundOutcome;
78
79    fn reject(reason: ArrayRejectReason) -> ArrayRejectMsg {
80        ArrayRejectMsg {
81            array: "any".into(),
82            op_hlc_bytes: hlc(10).to_bytes(),
83            reason,
84            detail: "refused".into(),
85        }
86    }
87
88    #[tokio::test(flavor = "multi_thread")]
89    async fn shape_invalid_reject_drops_op_from_pending() {
90        let (inbound, _schemas, pending, _storage) = make_inbound().await;
91        pending.enqueue(&put_op("any", 10, 1)).await.unwrap();
92
93        let outcome = inbound
94            .handle_reject(&reject(ArrayRejectReason::ShapeInvalid))
95            .await
96            .unwrap();
97        assert_eq!(outcome, InboundOutcome::RejectAcknowledged);
98        assert_eq!(pending.len().await.unwrap(), 0, "op must be removed");
99    }
100
101    #[tokio::test(flavor = "multi_thread")]
102    async fn retryable_rejects_keep_op_in_pending() {
103        let (inbound, _schemas, pending, _storage) = make_inbound().await;
104        pending.enqueue(&put_op("any", 10, 1)).await.unwrap();
105
106        for reason in [
107            ArrayRejectReason::ArrayUnknown,
108            ArrayRejectReason::SchemaTooNew,
109            ArrayRejectReason::EngineUnavailable,
110            ArrayRejectReason::SnapshotCapacity,
111            ArrayRejectReason::SnapshotBufferLost,
112            ArrayRejectReason::Unauthenticated,
113            ArrayRejectReason::RetentionFloor,
114        ] {
115            let outcome = inbound.handle_reject(&reject(reason)).await.unwrap();
116            assert_eq!(outcome, InboundOutcome::RejectAcknowledged);
117            assert_eq!(
118                pending.len().await.unwrap(),
119                1,
120                "{reason:?} must keep the op queued"
121            );
122        }
123    }
124}
nodedb_lite/sync/array/inbound/reject.rs

nodedb_lite/sync/array/inbound/
reject.rs
