appcore_sync/sync/
conflict.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: conflict.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/09/25 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/09/25 00:00:00 by dnettoRaw
8//      ###########      S: 1.0.3-rc
9// =============================================================================
10
11//! Bounded UI-ready sync conflicts and idempotent resolution records.
12
13use super::error::{SyncError, SyncResult};
14use parking_lot::Mutex;
15use serde::{Deserialize, Serialize};
16use std::collections::BTreeMap;
17use std::sync::Arc;
18
19/// Maximum UTF-8 bytes accepted for a conflict identity field.
20pub const MAX_CONFLICT_ID_BYTES: usize = 128;
21/// Maximum UTF-8 bytes accepted for a conflict reason.
22pub const MAX_CONFLICT_REASON_BYTES: usize = 512;
23/// Maximum conflicts retained by the in-memory resolution registry.
24pub const MAX_SYNC_CONFLICTS: usize = 4_096;
25
26/// Generic replication conflict category, independent of application schema.
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(rename_all = "snake_case")]
29pub enum SyncConflictKind {
30    /// One sequence contains different bytes for the same position.
31    SequencePayload,
32    /// A checkpoint hash disagrees with the observed leader chain.
33    CheckpointHash,
34    /// A batch identity was reused with different metadata or events.
35    BatchIdentity,
36}
37
38/// Payload-free conflict observation suitable for a UI or diagnostics API.
39#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
40pub struct SyncConflict {
41    /// Stable conflict identity used by resolution requests.
42    pub conflict_id: String,
43    /// Peer that produced the conflicting observation.
44    pub peer_id: String,
45    /// Generic conflict category.
46    pub kind: SyncConflictKind,
47    /// Conflicting replication sequence, when applicable.
48    pub sequence: Option<u64>,
49    /// Local content digest, never the content payload.
50    pub local_hash: String,
51    /// Remote content digest, never the content payload.
52    pub remote_hash: String,
53    /// Bounded operator-visible explanation.
54    pub reason: String,
55    /// Detection timestamp supplied by the caller.
56    pub detected_at_ms: u64,
57}
58
59impl SyncConflict {
60    /// Validates the bounded, payload-free conflict contract.
61    pub fn validate(&self) -> SyncResult<()> {
62        for (name, value, max) in [
63            (
64                "conflict_id",
65                self.conflict_id.as_str(),
66                MAX_CONFLICT_ID_BYTES,
67            ),
68            ("peer_id", self.peer_id.as_str(), MAX_CONFLICT_ID_BYTES),
69            ("reason", self.reason.as_str(), MAX_CONFLICT_REASON_BYTES),
70        ] {
71            validate_text(name, value, max)?;
72        }
73        validate_hash(&self.local_hash)?;
74        validate_hash(&self.remote_hash)?;
75        Ok(())
76    }
77}
78
79/// Operator or application choice recorded for one conflict.
80#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
81#[serde(rename_all = "snake_case")]
82pub enum SyncConflictResolution {
83    /// Preserve the local replication state.
84    KeepLocal,
85    /// Accept the remote observation through an owner-controlled operation.
86    AcceptRemote,
87    /// Defer the decision for manual review.
88    Defer,
89}
90
91/// Idempotent resolution command for one conflict.
92#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
93pub struct SyncConflictResolutionRequest {
94    /// Conflict being resolved.
95    pub conflict_id: String,
96    /// Stable caller-generated idempotency key.
97    pub idempotency_key: String,
98    /// Resolution selected by the owner.
99    pub resolution: SyncConflictResolution,
100    /// Resolution timestamp supplied by the caller.
101    pub resolved_at_ms: u64,
102}
103
104impl SyncConflictResolutionRequest {
105    /// Validates the bounded idempotency and conflict identity fields.
106    pub fn validate(&self) -> SyncResult<()> {
107        validate_text("conflict_id", &self.conflict_id, MAX_CONFLICT_ID_BYTES)?;
108        validate_text(
109            "idempotency_key",
110            &self.idempotency_key,
111            MAX_CONFLICT_ID_BYTES,
112        )
113    }
114}
115
116/// Result of applying or replaying a conflict resolution.
117#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
118pub struct SyncConflictResolutionReceipt {
119    /// Conflict resolved by this receipt.
120    pub conflict_id: String,
121    /// Idempotency key accepted for the resolution.
122    pub idempotency_key: String,
123    /// Recorded resolution.
124    pub resolution: SyncConflictResolution,
125    /// Whether this call replayed an already-recorded identical resolution.
126    pub idempotent: bool,
127    /// Resolution timestamp.
128    pub resolved_at_ms: u64,
129}
130
131#[derive(Debug, Clone)]
132struct ConflictRecord {
133    conflict: SyncConflict,
134    resolution: Option<SyncConflictResolutionReceipt>,
135}
136
137/// Conflict registry contract with idempotent recording and resolution.
138pub trait SyncConflictStore: Send + Sync {
139    /// Records one conflict, accepting an identical replay.
140    fn record(&self, conflict: SyncConflict) -> SyncResult<()>;
141    /// Applies or replays one resolution without mutating replication payloads.
142    fn resolve(
143        &self,
144        request: SyncConflictResolutionRequest,
145    ) -> SyncResult<SyncConflictResolutionReceipt>;
146    /// Lists payload-free conflicts in stable identity order.
147    fn list(&self) -> SyncResult<Vec<SyncConflict>>;
148}
149
150/// Bounded process-local conflict store for Runtime/UI coordination.
151#[derive(Debug, Clone)]
152pub struct InMemorySyncConflictStore {
153    records: Arc<Mutex<BTreeMap<String, ConflictRecord>>>,
154}
155
156impl Default for InMemorySyncConflictStore {
157    fn default() -> Self {
158        Self::new()
159    }
160}
161
162impl InMemorySyncConflictStore {
163    /// Creates an empty bounded conflict store.
164    pub fn new() -> Self {
165        Self {
166            records: Arc::new(Mutex::new(BTreeMap::new())),
167        }
168    }
169}
170
171impl SyncConflictStore for InMemorySyncConflictStore {
172    fn record(&self, conflict: SyncConflict) -> SyncResult<()> {
173        conflict.validate()?;
174        let mut records = self.records.lock();
175        if let Some(existing) = records.get(&conflict.conflict_id) {
176            if existing.conflict == conflict {
177                return Ok(());
178            }
179            return Err(SyncError::InvalidConflict(
180                "conflict identity was reused with different evidence".to_string(),
181            ));
182        }
183        if records.len() >= MAX_SYNC_CONFLICTS {
184            return Err(SyncError::ConflictStoreFull);
185        }
186        records.insert(
187            conflict.conflict_id.clone(),
188            ConflictRecord {
189                conflict,
190                resolution: None,
191            },
192        );
193        Ok(())
194    }
195
196    fn resolve(
197        &self,
198        request: SyncConflictResolutionRequest,
199    ) -> SyncResult<SyncConflictResolutionReceipt> {
200        request.validate()?;
201        let mut records = self.records.lock();
202        let record = records
203            .get_mut(&request.conflict_id)
204            .ok_or_else(|| SyncError::ConflictNotFound(request.conflict_id.clone()))?;
205        if let Some(existing) = &record.resolution {
206            if existing.idempotency_key == request.idempotency_key
207                && existing.resolution == request.resolution
208            {
209                return Ok(SyncConflictResolutionReceipt {
210                    idempotent: true,
211                    ..existing.clone()
212                });
213            }
214            return Err(SyncError::ConflictAlreadyResolved(
215                request.conflict_id.clone(),
216            ));
217        }
218        let receipt = SyncConflictResolutionReceipt {
219            conflict_id: request.conflict_id,
220            idempotency_key: request.idempotency_key,
221            resolution: request.resolution,
222            idempotent: false,
223            resolved_at_ms: request.resolved_at_ms,
224        };
225        record.resolution = Some(receipt.clone());
226        Ok(receipt)
227    }
228
229    fn list(&self) -> SyncResult<Vec<SyncConflict>> {
230        Ok(self
231            .records
232            .lock()
233            .values()
234            .map(|record| record.conflict.clone())
235            .collect())
236    }
237}
238
239fn validate_text(name: &str, value: &str, max: usize) -> SyncResult<()> {
240    if value.trim().is_empty() || value.len() > max || value.chars().any(char::is_control) {
241        return Err(SyncError::InvalidConflict(format!(
242            "{name} is empty, too long or contains control characters"
243        )));
244    }
245    Ok(())
246}
247
248fn validate_hash(value: &str) -> SyncResult<()> {
249    if value.len() != 64
250        || !value
251            .bytes()
252            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
253    {
254        return Err(SyncError::InvalidConflict(
255            "conflict hash must be lowercase SHA-256".to_string(),
256        ));
257    }
258    Ok(())
259}
260
261#[cfg(test)]
262mod tests {
263    use super::*;
264
265    fn conflict() -> SyncConflict {
266        SyncConflict {
267            conflict_id: "conflict-a".to_string(),
268            peer_id: "peer-a".to_string(),
269            kind: SyncConflictKind::SequencePayload,
270            sequence: Some(42),
271            local_hash: "11".repeat(32),
272            remote_hash: "22".repeat(32),
273            reason: "same sequence has different bytes".to_string(),
274            detected_at_ms: 100,
275        }
276    }
277
278    #[test]
279    fn resolution_replay_is_idempotent_and_payload_free() {
280        let store = InMemorySyncConflictStore::new();
281        store.record(conflict()).unwrap();
282        let request = SyncConflictResolutionRequest {
283            conflict_id: "conflict-a".to_string(),
284            idempotency_key: "resolve-a".to_string(),
285            resolution: SyncConflictResolution::KeepLocal,
286            resolved_at_ms: 200,
287        };
288        let first = store.resolve(request.clone()).unwrap();
289        let replay = store.resolve(request).unwrap();
290        assert!(!first.idempotent);
291        assert!(replay.idempotent);
292        assert_eq!(store.list().unwrap()[0].local_hash, "11".repeat(32));
293    }
294
295    #[test]
296    fn conflicting_resolution_is_rejected_without_overwrite() {
297        let store = InMemorySyncConflictStore::new();
298        store.record(conflict()).unwrap();
299        store
300            .resolve(SyncConflictResolutionRequest {
301                conflict_id: "conflict-a".to_string(),
302                idempotency_key: "resolve-a".to_string(),
303                resolution: SyncConflictResolution::KeepLocal,
304                resolved_at_ms: 200,
305            })
306            .unwrap();
307        assert!(matches!(
308            store.resolve(SyncConflictResolutionRequest {
309                conflict_id: "conflict-a".to_string(),
310                idempotency_key: "resolve-b".to_string(),
311                resolution: SyncConflictResolution::AcceptRemote,
312                resolved_at_ms: 300,
313            }),
314            Err(SyncError::ConflictAlreadyResolved(id)) if id == "conflict-a"
315        ));
316    }
317}
appcore_sync/sync/conflict.rs

appcore_sync/sync/
conflict.rs
