nodedb_types/sync/
violation.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Violation types for DLQ classification.
4//!
5//! When a sync delta is rejected, the `ViolationType` categorizes *why* it
6//! was rejected. This is stored in the DLQ on the Origin for forensic review
7//! and is separate from `CompensationHint` (which is what the edge sees).
8
9use serde::{Deserialize, Serialize};
10
11/// Why a sync delta was placed in the Dead-Letter Queue.
12///
13/// Used on the Origin side for audit/forensics. The edge never sees this
14/// directly — it only receives `CompensationHint` (which may be generic
15/// for security reasons).
16#[derive(
17    Debug,
18    Clone,
19    PartialEq,
20    Eq,
21    Serialize,
22    Deserialize,
23    rkyv::Archive,
24    rkyv::Serialize,
25    rkyv::Deserialize,
26    zerompk::ToMessagePack,
27    zerompk::FromMessagePack,
28)]
29#[serde(rename_all = "snake_case")]
30#[non_exhaustive]
31pub enum ViolationType {
32    /// RLS write policy rejected the delta.
33    #[serde(rename = "rls_policy_violation")]
34    RlsPolicyViolation { policy_name: String },
35    /// UNIQUE constraint violation.
36    #[serde(rename = "unique_violation")]
37    UniqueViolation { field: String, value: String },
38    /// Foreign key reference missing.
39    #[serde(rename = "foreign_key_missing")]
40    ForeignKeyMissing { referenced_id: String },
41    /// NOT NULL violation: a required field is absent or null.
42    #[serde(rename = "not_null_violation")]
43    NotNullViolation { field: String },
44    /// Permission denied (no write access to target resource).
45    #[serde(rename = "permission_denied")]
46    PermissionDenied,
47    /// Rate limit exceeded for this session.
48    #[serde(rename = "rate_limited")]
49    RateLimited,
50    /// JWT token expired during active session.
51    #[serde(rename = "token_expired")]
52    TokenExpired,
53    /// Schema validation failed.
54    #[serde(rename = "schema_violation")]
55    SchemaViolation { field: String, reason: String },
56    /// Generic constraint violation (catch-all).
57    #[serde(rename = "constraint_violation")]
58    ConstraintViolation { detail: String },
59    /// The delta bytes did not decode, so nothing was applied. Distinct from a
60    /// constraint violation: no rule was broken, the frame was unreadable.
61    #[serde(rename = "malformed_delta")]
62    MalformedDelta { detail: String },
63}
64
65impl std::fmt::Display for ViolationType {
66    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67        match self {
68            Self::RlsPolicyViolation { policy_name } => {
69                write!(f, "rls_policy:{policy_name}")
70            }
71            Self::UniqueViolation { field, value } => {
72                write!(f, "unique:{field}={value}")
73            }
74            Self::ForeignKeyMissing { referenced_id } => {
75                write!(f, "fk_missing:{referenced_id}")
76            }
77            Self::NotNullViolation { field } => write!(f, "not_null:{field}"),
78            Self::PermissionDenied => write!(f, "permission_denied"),
79            Self::RateLimited => write!(f, "rate_limited"),
80            Self::TokenExpired => write!(f, "token_expired"),
81            Self::SchemaViolation { field, reason } => {
82                write!(f, "schema:{field}={reason}")
83            }
84            Self::ConstraintViolation { detail } => write!(f, "constraint:{detail}"),
85            Self::MalformedDelta { detail } => write!(f, "malformed_delta:{detail}"),
86        }
87    }
88}
89
90impl ViolationType {
91    /// Convert a violation to the corresponding `CompensationHint` for the edge.
92    ///
93    /// Some violations map to a generic hint (e.g., RLS → PermissionDenied)
94    /// to avoid leaking security-sensitive information to untrusted edges.
95    pub fn to_compensation_hint(&self) -> super::compensation::CompensationHint {
96        use super::compensation::CompensationHint;
97        match self {
98            Self::UniqueViolation { field, value } => CompensationHint::UniqueViolation {
99                field: field.clone(),
100                conflicting_value: value.clone(),
101            },
102            Self::ForeignKeyMissing { referenced_id } => CompensationHint::ForeignKeyMissing {
103                referenced_id: referenced_id.clone(),
104            },
105            Self::RateLimited => CompensationHint::RateLimited {
106                retry_after_ms: 5000,
107            },
108            // Security-sensitive violations all map to generic PermissionDenied.
109            Self::RlsPolicyViolation { .. } | Self::PermissionDenied | Self::TokenExpired => {
110                CompensationHint::PermissionDenied
111            }
112            Self::SchemaViolation { field, reason } => CompensationHint::SchemaViolation {
113                field: field.clone(),
114                reason: reason.clone(),
115            },
116            Self::NotNullViolation { field } => CompensationHint::SchemaViolation {
117                field: field.clone(),
118                reason: "required field missing".into(),
119            },
120            Self::ConstraintViolation { detail } => CompensationHint::Custom {
121                constraint: "constraint".into(),
122                detail: detail.clone(),
123            },
124            Self::MalformedDelta { detail } => CompensationHint::Custom {
125                constraint: "malformed_delta".into(),
126                detail: detail.clone(),
127            },
128        }
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135
136    #[test]
137    fn violation_display() {
138        assert_eq!(
139            ViolationType::PermissionDenied.to_string(),
140            "permission_denied"
141        );
142        assert_eq!(ViolationType::RateLimited.to_string(), "rate_limited");
143        assert_eq!(
144            ViolationType::UniqueViolation {
145                field: "email".into(),
146                value: "x@y.com".into()
147            }
148            .to_string(),
149            "unique:email=x@y.com"
150        );
151        assert_eq!(
152            ViolationType::NotNullViolation {
153                field: "name".into()
154            }
155            .to_string(),
156            "not_null:name"
157        );
158    }
159
160    #[test]
161    fn rls_violation_maps_to_permission_denied() {
162        let v = ViolationType::RlsPolicyViolation {
163            policy_name: "user_write_own".into(),
164        };
165        let hint = v.to_compensation_hint();
166        // RLS details are NOT leaked to the edge.
167        assert!(matches!(
168            hint,
169            super::super::compensation::CompensationHint::PermissionDenied
170        ));
171    }
172
173    #[test]
174    fn unique_violation_preserves_details() {
175        let v = ViolationType::UniqueViolation {
176            field: "username".into(),
177            value: "alice".into(),
178        };
179        let hint = v.to_compensation_hint();
180        match hint {
181            super::super::compensation::CompensationHint::UniqueViolation {
182                field,
183                conflicting_value,
184            } => {
185                assert_eq!(field, "username");
186                assert_eq!(conflicting_value, "alice");
187            }
188            _ => panic!("expected UniqueViolation hint"),
189        }
190    }
191
192    #[test]
193    fn token_expired_maps_to_permission_denied() {
194        let hint = ViolationType::TokenExpired.to_compensation_hint();
195        assert!(matches!(
196            hint,
197            super::super::compensation::CompensationHint::PermissionDenied
198        ));
199    }
200
201    #[test]
202    fn malformed_delta_names_itself_rather_than_posing_as_a_constraint() {
203        let v = ViolationType::MalformedDelta {
204            detail: "orders/o1 could not be decoded".into(),
205        };
206        match v.to_compensation_hint() {
207            super::super::compensation::CompensationHint::Custom { constraint, detail } => {
208                assert_eq!(constraint, "malformed_delta");
209                assert!(detail.contains("orders/o1"));
210            }
211            other => panic!("expected a Custom hint naming the malformed delta, got {other:?}"),
212        }
213    }
214
215    #[test]
216    fn every_violation_is_terminal() {
217        // This type answers "why must the sender compensate?" — every variant
218        // is permanent. A *retryable* refusal is not a violation and must never
219        // be expressed as one: it travels as an `AckStatus::Gap` outcome, which
220        // holds the producer high-water-mark so the re-push is admitted. A
221        // retryable variant here would let a producer smuggle a retry through
222        // the terminal channel, where consumers correctly read it as final.
223        let hints = [
224            ViolationType::PermissionDenied,
225            ViolationType::RateLimited,
226            ViolationType::TokenExpired,
227            ViolationType::ConstraintViolation { detail: "d".into() },
228            ViolationType::MalformedDelta { detail: "d".into() },
229            ViolationType::SchemaViolation {
230                field: "f".into(),
231                reason: "r".into(),
232            },
233            ViolationType::UniqueViolation {
234                field: "f".into(),
235                value: "v".into(),
236            },
237            ViolationType::ForeignKeyMissing {
238                referenced_id: "r".into(),
239            },
240            ViolationType::NotNullViolation { field: "f".into() },
241            ViolationType::RlsPolicyViolation {
242                policy_name: "p".into(),
243            },
244        ]
245        .map(|v| v.to_compensation_hint());
246        for hint in hints {
247            assert!(
248                !matches!(
249                    hint,
250                    super::super::compensation::CompensationHint::Retry { .. }
251                ),
252                "a violation mapped to a retry hint: {hint:?}"
253            );
254        }
255    }
256}
nodedb_types/sync/violation.rs

nodedb_types/sync/
violation.rs
