pulsedb/sync/
types.rs

1//! Core types for the PulseDB sync protocol.
2//!
3//! This module defines the wire types used for synchronizing data between
4//! PulseDB instances: change payloads, cursors, handshake messages, and
5//! the instance identity type.
6
7use std::collections::BTreeMap;
8
9use serde::{Deserialize, Serialize};
10
11use crate::collective::Collective;
12use crate::experience::Experience;
13use crate::insight::DerivedInsight;
14use crate::relation::ExperienceRelation;
15use crate::types::{CollectiveId, ExperienceId, InsightId, RelationId, Timestamp};
16
17pub use crate::types::InstanceId;
18
19// ============================================================================
20// SyncCursor — Tracks sync position per peer
21// ============================================================================
22
23/// Tracks the sync position for a specific peer instance.
24///
25/// Each peer maintains a cursor recording the last WAL sequence number
26/// successfully synced. This enables incremental sync — only changes
27/// after the cursor position are transferred.
28#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
29pub struct SyncCursor {
30    /// The peer instance this cursor tracks.
31    pub instance_id: InstanceId,
32
33    /// The last WAL sequence number successfully synced from/to this peer.
34    pub last_sequence: u64,
35}
36
37impl SyncCursor {
38    /// Creates a new cursor at sequence 0 (beginning of WAL).
39    pub fn new(instance_id: InstanceId) -> Self {
40        Self {
41            instance_id,
42            last_sequence: 0,
43        }
44    }
45}
46
47// ============================================================================
48// SyncEntityType — What kind of entity changed
49// ============================================================================
50
51/// Discriminant for the type of entity in a sync change.
52#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
53#[repr(u8)]
54pub enum SyncEntityType {
55    /// An experience was created, updated, archived, or deleted.
56    Experience = 0,
57    /// A relation was created or deleted.
58    Relation = 1,
59    /// An insight was created or deleted.
60    Insight = 2,
61    /// A collective was created.
62    Collective = 3,
63}
64
65// ============================================================================
66// SerializableExperienceUpdate — Wire-safe mirror of ExperienceUpdate
67// ============================================================================
68
69/// Wire-safe version of [`crate::ExperienceUpdate`] for sync payloads.
70///
71/// The original `ExperienceUpdate` does not derive `Serialize`/`Deserialize`,
72/// so this struct mirrors its fields with full serde support. Use the `From`
73/// impls to convert between the two.
74#[derive(Clone, Debug, Default, Serialize, Deserialize)]
75pub struct SerializableExperienceUpdate {
76    /// New importance score (0.0–1.0).
77    pub importance: Option<f32>,
78
79    /// New confidence score (0.0–1.0).
80    pub confidence: Option<f32>,
81
82    /// Replace domain tags entirely.
83    pub domain: Option<Vec<String>>,
84
85    /// Replace related files entirely.
86    pub related_files: Option<Vec<String>>,
87
88    /// Set archived status.
89    pub archived: Option<bool>,
90
91    /// Full G-counter applications map for CRDT merge.
92    pub applications: Option<BTreeMap<InstanceId, u32>>,
93
94    /// Last reinforcement timestamp for max-timestamp merge.
95    pub last_reinforced: Option<Timestamp>,
96}
97
98impl From<crate::experience::ExperienceUpdate> for SerializableExperienceUpdate {
99    fn from(update: crate::experience::ExperienceUpdate) -> Self {
100        Self {
101            importance: update.importance,
102            confidence: update.confidence,
103            domain: update.domain,
104            related_files: update.related_files,
105            archived: update.archived,
106            applications: None,
107            last_reinforced: None,
108        }
109    }
110}
111
112impl From<SerializableExperienceUpdate> for crate::experience::ExperienceUpdate {
113    fn from(update: SerializableExperienceUpdate) -> Self {
114        Self {
115            importance: update.importance,
116            confidence: update.confidence,
117            domain: update.domain,
118            related_files: update.related_files,
119            archived: update.archived,
120        }
121    }
122}
123
124// ============================================================================
125// SyncPayload — Full data for each mutation type
126// ============================================================================
127
128/// The payload of a sync change, containing all data needed to apply
129/// the change on the receiving end.
130///
131/// Uses full payloads (not deltas) so the receiver has everything needed
132/// including embeddings for HNSW insertion.
133#[derive(Clone, Debug, Serialize, Deserialize)]
134pub enum SyncPayload {
135    /// A new experience was created.
136    ExperienceCreated(Experience),
137
138    /// An existing experience was updated.
139    ExperienceUpdated {
140        /// The experience that was updated.
141        id: ExperienceId,
142        /// The fields that changed.
143        update: SerializableExperienceUpdate,
144        /// When the update occurred.
145        timestamp: Timestamp,
146    },
147
148    /// An experience was soft-deleted (archived).
149    ExperienceArchived {
150        /// The archived experience.
151        id: ExperienceId,
152        /// When the archive occurred.
153        timestamp: Timestamp,
154    },
155
156    /// An experience was permanently deleted.
157    ExperienceDeleted {
158        /// The deleted experience.
159        id: ExperienceId,
160        /// When the deletion occurred.
161        timestamp: Timestamp,
162    },
163
164    /// A new relation was created.
165    RelationCreated(ExperienceRelation),
166
167    /// A relation was deleted.
168    RelationDeleted {
169        /// The deleted relation.
170        id: RelationId,
171        /// When the deletion occurred.
172        timestamp: Timestamp,
173    },
174
175    /// A new insight was created.
176    InsightCreated(DerivedInsight),
177
178    /// An insight was deleted.
179    InsightDeleted {
180        /// The deleted insight.
181        id: InsightId,
182        /// When the deletion occurred.
183        timestamp: Timestamp,
184    },
185
186    /// A new collective was created.
187    CollectiveCreated(Collective),
188}
189
190// ============================================================================
191// SyncChange — A single change to sync
192// ============================================================================
193
194/// A single change event to be synchronized between PulseDB instances.
195///
196/// Contains the full payload needed to apply the change, plus metadata
197/// about the source instance and WAL position.
198#[derive(Clone, Debug, Serialize, Deserialize)]
199pub struct SyncChange {
200    /// Source WAL sequence number.
201    pub sequence: u64,
202
203    /// The instance that produced this change.
204    pub source_instance: InstanceId,
205
206    /// Which collective this change belongs to.
207    pub collective_id: CollectiveId,
208
209    /// What kind of entity changed.
210    pub entity_type: SyncEntityType,
211
212    /// The full change data.
213    pub payload: SyncPayload,
214
215    /// When the change occurred.
216    pub timestamp: Timestamp,
217}
218
219// ============================================================================
220// SyncStatus — Current state of the sync system
221// ============================================================================
222
223/// Current operational status of the sync system.
224#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
225pub enum SyncStatus {
226    /// Sync is idle, waiting for the next poll interval.
227    Idle,
228    /// Sync is actively transferring data.
229    Syncing,
230    /// Sync encountered an error.
231    Error(String),
232    /// Disconnected from the remote peer.
233    Disconnected,
234}
235
236// ============================================================================
237// Handshake messages
238// ============================================================================
239
240/// Request sent during sync handshake to establish a connection.
241#[derive(Clone, Debug, Serialize, Deserialize)]
242pub struct HandshakeRequest {
243    /// The local instance ID.
244    pub instance_id: InstanceId,
245    /// The sync protocol version.
246    pub protocol_version: u32,
247    /// Capabilities advertised by this instance.
248    pub capabilities: Vec<String>,
249}
250
251/// Response to a handshake request.
252#[derive(Clone, Debug, Serialize, Deserialize)]
253pub struct HandshakeResponse {
254    /// The remote instance ID.
255    pub instance_id: InstanceId,
256    /// The remote's protocol version.
257    pub protocol_version: u32,
258    /// Whether the handshake was accepted.
259    pub accepted: bool,
260    /// Reason for rejection, if not accepted.
261    pub reason: Option<String>,
262}
263
264// ============================================================================
265// Pull request/response
266// ============================================================================
267
268/// Request to pull changes from a remote peer.
269#[derive(Clone, Debug, Serialize, Deserialize)]
270pub struct PullRequest {
271    /// The cursor position to pull changes from.
272    pub cursor: SyncCursor,
273    /// Maximum number of changes to return in this batch.
274    pub batch_size: usize,
275    /// Optional filter: only pull changes for these collectives.
276    pub collectives: Option<Vec<CollectiveId>>,
277}
278
279/// Response containing pulled changes.
280#[derive(Clone, Debug, Serialize, Deserialize)]
281pub struct PullResponse {
282    /// The changes since the cursor position.
283    pub changes: Vec<SyncChange>,
284    /// Whether there are more changes available.
285    pub has_more: bool,
286    /// The updated cursor position after this batch.
287    pub new_cursor: SyncCursor,
288}
289
290// ============================================================================
291// Push response
292// ============================================================================
293
294/// Response after pushing changes to a remote peer.
295#[derive(Clone, Debug, Serialize, Deserialize)]
296pub struct PushResponse {
297    /// Number of changes accepted by the remote.
298    pub accepted: usize,
299    /// Number of changes rejected by the remote.
300    pub rejected: usize,
301    /// The remote's updated cursor position.
302    pub new_cursor: SyncCursor,
303}
304
305#[cfg(test)]
306mod tests {
307    use super::*;
308
309    #[test]
310    fn test_instance_id_new_is_unique() {
311        let a = InstanceId::new();
312        let b = InstanceId::new();
313        assert_ne!(a, b);
314    }
315
316    #[test]
317    fn test_instance_id_nil() {
318        let id = InstanceId::nil();
319        assert_eq!(id, InstanceId::default());
320        assert_eq!(id, InstanceId::nil());
321    }
322
323    #[test]
324    fn test_instance_id_bytes_roundtrip() {
325        let id = InstanceId::new();
326        let bytes = *id.as_bytes();
327        let restored = InstanceId::from_bytes(bytes);
328        assert_eq!(id, restored);
329    }
330
331    #[test]
332    fn test_instance_id_display() {
333        let id = InstanceId::nil();
334        assert_eq!(id.to_string(), "00000000-0000-0000-0000-000000000000");
335    }
336
337    #[test]
338    fn test_instance_id_postcard_roundtrip() {
339        let id = InstanceId::new();
340        let bytes = postcard::to_allocvec(&id).unwrap();
341        let restored: InstanceId = postcard::from_bytes(&bytes).unwrap();
342        assert_eq!(id, restored);
343    }
344
345    #[test]
346    fn test_sync_cursor_new() {
347        let id = InstanceId::new();
348        let cursor = SyncCursor::new(id);
349        assert_eq!(cursor.instance_id, id);
350        assert_eq!(cursor.last_sequence, 0);
351    }
352
353    #[test]
354    fn test_sync_cursor_postcard_roundtrip() {
355        let cursor = SyncCursor {
356            instance_id: InstanceId::new(),
357            last_sequence: 42,
358        };
359        let bytes = postcard::to_allocvec(&cursor).unwrap();
360        let restored: SyncCursor = postcard::from_bytes(&bytes).unwrap();
361        assert_eq!(cursor, restored);
362    }
363
364    #[test]
365    fn test_sync_entity_type_repr() {
366        assert_eq!(SyncEntityType::Experience as u8, 0);
367        assert_eq!(SyncEntityType::Relation as u8, 1);
368        assert_eq!(SyncEntityType::Insight as u8, 2);
369        assert_eq!(SyncEntityType::Collective as u8, 3);
370    }
371
372    #[test]
373    fn test_serializable_experience_update_from_conversion() {
374        let update = crate::experience::ExperienceUpdate {
375            importance: Some(0.9),
376            confidence: None,
377            domain: Some(vec!["rust".to_string()]),
378            related_files: None,
379            archived: Some(false),
380        };
381        let serializable: SerializableExperienceUpdate = update.into();
382        assert_eq!(serializable.importance, Some(0.9));
383        assert_eq!(serializable.confidence, None);
384        assert_eq!(serializable.domain, Some(vec!["rust".to_string()]));
385        assert_eq!(serializable.archived, Some(false));
386    }
387
388    #[test]
389    fn test_serializable_experience_update_into_conversion() {
390        let serializable = SerializableExperienceUpdate {
391            importance: Some(0.5),
392            confidence: Some(0.8),
393            domain: None,
394            related_files: Some(vec!["main.rs".to_string()]),
395            archived: None,
396            applications: None,
397            last_reinforced: None,
398        };
399        let update: crate::experience::ExperienceUpdate = serializable.into();
400        assert_eq!(update.importance, Some(0.5));
401        assert_eq!(update.confidence, Some(0.8));
402        assert_eq!(update.related_files, Some(vec!["main.rs".to_string()]));
403    }
404
405    #[test]
406    fn test_serializable_experience_update_postcard_roundtrip() {
407        let update = SerializableExperienceUpdate {
408            importance: Some(0.7),
409            confidence: Some(0.9),
410            domain: Some(vec!["test".to_string()]),
411            related_files: None,
412            archived: Some(true),
413            applications: Some(std::collections::BTreeMap::from([(InstanceId::new(), 2)])),
414            last_reinforced: Some(Timestamp::now()),
415        };
416        let bytes = postcard::to_allocvec(&update).unwrap();
417        let restored: SerializableExperienceUpdate = postcard::from_bytes(&bytes).unwrap();
418        assert_eq!(update.importance, restored.importance);
419        assert_eq!(update.confidence, restored.confidence);
420        assert_eq!(update.domain, restored.domain);
421        assert_eq!(update.archived, restored.archived);
422        assert_eq!(update.applications, restored.applications);
423        assert_eq!(update.last_reinforced, restored.last_reinforced);
424    }
425
426    #[test]
427    fn test_sync_status_equality() {
428        assert_eq!(SyncStatus::Idle, SyncStatus::Idle);
429        assert_eq!(SyncStatus::Error("x".into()), SyncStatus::Error("x".into()));
430        assert_ne!(SyncStatus::Idle, SyncStatus::Syncing);
431    }
432
433    #[test]
434    fn test_handshake_request_postcard_roundtrip() {
435        let req = HandshakeRequest {
436            instance_id: InstanceId::new(),
437            protocol_version: crate::sync::SYNC_PROTOCOL_VERSION,
438            capabilities: vec!["push".to_string(), "pull".to_string()],
439        };
440        let bytes = postcard::to_allocvec(&req).unwrap();
441        let restored: HandshakeRequest = postcard::from_bytes(&bytes).unwrap();
442        assert_eq!(req.instance_id, restored.instance_id);
443        assert_eq!(req.protocol_version, restored.protocol_version);
444        assert_eq!(req.capabilities, restored.capabilities);
445    }
446
447    #[test]
448    fn test_pull_request_postcard_roundtrip() {
449        let req = PullRequest {
450            cursor: SyncCursor::new(InstanceId::new()),
451            batch_size: 500,
452            collectives: Some(vec![CollectiveId::new()]),
453        };
454        let bytes = postcard::to_allocvec(&req).unwrap();
455        let restored: PullRequest = postcard::from_bytes(&bytes).unwrap();
456        assert_eq!(req.cursor, restored.cursor);
457        assert_eq!(req.batch_size, restored.batch_size);
458    }
459
460    #[test]
461    fn test_push_response_postcard_roundtrip() {
462        let resp = PushResponse {
463            accepted: 10,
464            rejected: 2,
465            new_cursor: SyncCursor {
466                instance_id: InstanceId::new(),
467                last_sequence: 100,
468            },
469        };
470        let bytes = postcard::to_allocvec(&resp).unwrap();
471        let restored: PushResponse = postcard::from_bytes(&bytes).unwrap();
472        assert_eq!(resp.accepted, restored.accepted);
473        assert_eq!(resp.rejected, restored.rejected);
474        assert_eq!(resp.new_cursor, restored.new_cursor);
475    }
476}
pulsedb/sync/types.rs

pulsedb/sync/
types.rs
