nodedb_types/sync/
compensation.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Typed compensation hints for rejected sync deltas.
4//!
5//! When the Origin rejects a CRDT delta (constraint violation, RLS, rate limit),
6//! it sends a `CompensationHint` back to the edge client. The edge uses this
7//! to roll back optimistic local state and notify the application with a
8//! typed, actionable error — not a generic string.
9
10use serde::{Deserialize, Serialize};
11
12/// Typed compensation hint sent from Origin to edge when a delta is rejected.
13///
14/// The edge's `CompensationHandler` receives this and can programmatically
15/// decide how to react (prompt user, auto-retry with suffix, silently merge).
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 CompensationHint {
32    /// UNIQUE constraint violated — another device wrote the same value first.
33    #[serde(rename = "unique_violation")]
34    UniqueViolation {
35        /// The field that has the UNIQUE constraint (e.g., "username").
36        field: String,
37        /// The conflicting value that was already taken.
38        conflicting_value: String,
39    },
40
41    /// Foreign key reference missing — the referenced entity doesn't exist.
42    #[serde(rename = "foreign_key_missing")]
43    ForeignKeyMissing {
44        /// The ID that was referenced but not found.
45        referenced_id: String,
46    },
47
48    /// Permission denied — the user doesn't have write access.
49    /// No details are leaked (security: the edge is untrusted).
50    #[serde(rename = "permission_denied")]
51    PermissionDenied,
52
53    /// Rate limit exceeded — try again later.
54    #[serde(rename = "rate_limited")]
55    RateLimited {
56        /// Suggested delay before retrying (milliseconds).
57        retry_after_ms: u64,
58    },
59
60    /// Schema violation — the delta doesn't conform to the collection schema.
61    #[serde(rename = "schema_violation")]
62    SchemaViolation {
63        /// Which field failed validation.
64        field: String,
65        /// Human-readable reason.
66        reason: String,
67    },
68
69    /// Custom application-defined constraint violation.
70    #[serde(rename = "custom")]
71    Custom {
72        /// Constraint name.
73        constraint: String,
74        /// Typed payload for the application to interpret.
75        detail: String,
76    },
77
78    /// Data integrity violation — CRC32C checksum mismatch on delta payload.
79    /// The client should re-send the delta.
80    #[serde(rename = "integrity_violation")]
81    IntegrityViolation,
82
83    /// Transient rejection — the delta was admitted against a constraint
84    /// version not yet installed on the accepting replica. Not a failure:
85    /// re-push the delta as a new sequence after the suggested delay.
86    #[serde(rename = "retry")]
87    Retry {
88        /// Suggested delay before re-pushing (milliseconds).
89        retry_after_ms: u64,
90    },
91}
92
93impl CompensationHint {
94    /// Constraint name carried by [`Self::Custom`] when a delta's Loro peer id
95    /// is already owned by another replica.
96    ///
97    /// Unlike every other refusal, nothing about the row is wrong and no
98    /// conflict policy can resolve it: the producer's identity is unusable, and
99    /// only the producer can replace it. Defined here so the server that emits
100    /// the refusal and the client that acts on it cannot drift apart on the
101    /// spelling of a string that is the difference between a replica healing
102    /// itself and one that is permanently refused.
103    pub const PEER_ID_COLLISION: &'static str = "peer_id_collision";
104
105    /// Whether this refusal says the producer's Loro peer id is unusable.
106    pub fn is_peer_id_collision(&self) -> bool {
107        matches!(
108            self,
109            Self::Custom { constraint, .. } if constraint == Self::PEER_ID_COLLISION
110        )
111    }
112
113    /// Returns a short, machine-readable code for the hint type.
114    pub fn code(&self) -> &'static str {
115        match self {
116            Self::UniqueViolation { .. } => "UNIQUE_VIOLATION",
117            Self::ForeignKeyMissing { .. } => "FK_MISSING",
118            Self::PermissionDenied => "PERMISSION_DENIED",
119            Self::RateLimited { .. } => "RATE_LIMITED",
120            Self::SchemaViolation { .. } => "SCHEMA_VIOLATION",
121            Self::Custom { .. } => "CUSTOM",
122            Self::IntegrityViolation => "INTEGRITY_VIOLATION",
123            Self::Retry { .. } => "RETRY",
124        }
125    }
126}
127
128impl std::fmt::Display for CompensationHint {
129    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
130        match self {
131            Self::UniqueViolation {
132                field,
133                conflicting_value,
134            } => write!(
135                f,
136                "UNIQUE({field}): value '{conflicting_value}' already exists"
137            ),
138            Self::ForeignKeyMissing { referenced_id } => {
139                write!(f, "FK_MISSING: referenced ID '{referenced_id}' not found")
140            }
141            Self::PermissionDenied => write!(f, "PERMISSION_DENIED"),
142            Self::RateLimited { retry_after_ms } => {
143                write!(f, "RATE_LIMITED: retry after {retry_after_ms}ms")
144            }
145            Self::SchemaViolation { field, reason } => {
146                write!(f, "SCHEMA({field}): {reason}")
147            }
148            Self::Custom {
149                constraint, detail, ..
150            } => write!(f, "CUSTOM({constraint}): {detail}"),
151            Self::IntegrityViolation => write!(f, "INTEGRITY_VIOLATION: CRC32C mismatch"),
152            Self::Retry { retry_after_ms } => {
153                write!(f, "RETRY: retry after {retry_after_ms}ms")
154            }
155        }
156    }
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162
163    #[test]
164    fn compensation_codes() {
165        assert_eq!(
166            CompensationHint::UniqueViolation {
167                field: "email".into(),
168                conflicting_value: "a@b.com".into()
169            }
170            .code(),
171            "UNIQUE_VIOLATION"
172        );
173        assert_eq!(
174            CompensationHint::PermissionDenied.code(),
175            "PERMISSION_DENIED"
176        );
177        assert_eq!(
178            CompensationHint::RateLimited {
179                retry_after_ms: 5000
180            }
181            .code(),
182            "RATE_LIMITED"
183        );
184    }
185
186    #[test]
187    fn compensation_display() {
188        let hint = CompensationHint::UniqueViolation {
189            field: "username".into(),
190            conflicting_value: "alice".into(),
191        };
192        assert!(hint.to_string().contains("alice"));
193        assert!(hint.to_string().contains("username"));
194    }
195
196    #[test]
197    fn retry_code_and_display() {
198        let hint = CompensationHint::Retry {
199            retry_after_ms: 1000,
200        };
201        assert_eq!(hint.code(), "RETRY");
202        assert!(hint.to_string().contains("1000"));
203    }
204
205    #[test]
206    fn msgpack_roundtrip() {
207        let hint = CompensationHint::ForeignKeyMissing {
208            referenced_id: "user-42".into(),
209        };
210        let bytes = zerompk::to_msgpack_vec(&hint).unwrap();
211        let decoded: CompensationHint = zerompk::from_msgpack(&bytes).unwrap();
212        assert_eq!(hint, decoded);
213    }
214}
nodedb_types/sync/compensation.rs

nodedb_types/sync/
compensation.rs
