nodedb_lite/sync/client/
handshake.rs

1//! Handshake: build outgoing `HandshakeMsg`, process incoming `HandshakeAckMsg`.
2
3use nodedb_types::sync::wire::{HandshakeAckMsg, HandshakeMsg};
4use nodedb_types::wire_version::WIRE_FORMAT_VERSION;
5
6use super::config::SyncState;
7use super::state::SyncClient;
8
9impl SyncClient {
10    /// Build a handshake message from current state and the instance identity.
11    ///
12    /// The identity is a parameter rather than client state because it changes
13    /// while the client is alive: a reported fork replaces it wholesale, and a
14    /// peer-id collision rotates part of it. A handshake built from a cached
15    /// copy would re-present the identity Origin just rejected.
16    pub async fn build_handshake(&self, identity: &crate::identity::LiteIdentity) -> HandshakeMsg {
17        let clock = self.clock.lock().await;
18        let shapes = self.shapes.lock().await;
19
20        let wire_clock = clock.to_wire();
21        let mut vector_clock = std::collections::HashMap::new();
22        // Origin expects: { collection: { doc_id: lamport_ts } }
23        // We send a simplified version: { "_global": { peer_hex: counter } }
24        vector_clock.insert("_global".to_string(), wire_clock);
25
26        HandshakeMsg {
27            jwt_token: self.config.jwt_token.clone(),
28            vector_clock,
29            subscribed_shapes: shapes.active_shape_ids(),
30            client_version: self.config.client_version.clone(),
31            lite_id: identity.lite_id.clone(),
32            epoch: identity.epoch,
33            wire_version: WIRE_FORMAT_VERSION,
34        }
35    }
36
37    /// Process a handshake acknowledgment from Origin.
38    pub async fn handle_handshake_ack(&self, ack: &HandshakeAckMsg) -> bool {
39        if !ack.success {
40            tracing::warn!(
41                error = ack.error.as_deref().unwrap_or("unknown"),
42                "handshake rejected by Origin"
43            );
44            return false;
45        }
46
47        *self.session_id.lock().await = Some(ack.session_id.clone());
48        // New session — every collection must be re-announced before its
49        // first delta so Origin materializes it, even if it was already
50        // announced in a prior session.
51        self.announced_collections.lock().await.clear();
52
53        let mut clock = self.clock.lock().await;
54        for (peer_hex, &counter) in &ack.server_clock {
55            if let Ok(peer_id) = u64::from_str_radix(peer_hex, 16) {
56                clock.advance(peer_id, counter);
57            }
58        }
59        drop(clock);
60
61        self.set_producer_id(ack.producer_id).await;
62        self.set_accepted_epoch(ack.accepted_epoch).await;
63
64        *self.state.lock().await = SyncState::Connected;
65        tracing::info!(
66            session = %ack.session_id,
67            producer_id = ack.producer_id,
68            accepted_epoch = ack.accepted_epoch,
69            "sync handshake accepted"
70        );
71        true
72    }
73}
74
75#[cfg(test)]
76mod tests {
77    use super::*;
78    use crate::sync::client::SyncConfig;
79
80    fn make_config() -> SyncConfig {
81        SyncConfig::new("wss://localhost:9090/sync", "test.jwt.token")
82    }
83
84    #[tokio::test]
85    async fn build_handshake() {
86        let client = SyncClient::new(make_config());
87
88        {
89            let mut shapes = client.shapes().lock().await;
90            shapes.subscribe(nodedb_types::sync::shape::ShapeDefinition {
91                shape_id: "s1".into(),
92                tenant_id: 1,
93                shape_type: nodedb_types::sync::shape::ShapeType::Document {
94                    collection: "orders".into(),
95                    predicate: Vec::new(),
96                },
97                description: "test".into(),
98                field_filter: vec![],
99            });
100        }
101
102        let identity = crate::identity::LiteIdentity {
103            lite_id: "lite-1".into(),
104            epoch: 3,
105            peer_id: 42,
106        };
107        let hs = client.build_handshake(&identity).await;
108        assert_eq!(hs.jwt_token, "test.jwt.token");
109        assert_eq!(hs.subscribed_shapes, vec!["s1"]);
110    }
111
112    #[tokio::test]
113    async fn handle_handshake_ack_success() {
114        let client = SyncClient::new(make_config());
115        let ack = HandshakeAckMsg {
116            success: true,
117            session_id: "sess-123".into(),
118            server_clock: std::collections::HashMap::new(),
119            error: None,
120            fork_detected: false,
121            server_wire_version: 1,
122            producer_id: 0,
123            accepted_epoch: 0,
124            // All zeros = signing unavailable for this session.
125            delta_signing_key: [0u8; 32],
126        };
127
128        assert!(client.handle_handshake_ack(&ack).await);
129        assert_eq!(client.state().await, SyncState::Connected);
130    }
131
132    #[tokio::test]
133    async fn handle_handshake_ack_failure() {
134        let client = SyncClient::new(make_config());
135        let ack = HandshakeAckMsg {
136            success: false,
137            session_id: String::new(),
138            server_clock: std::collections::HashMap::new(),
139            error: Some("invalid token".into()),
140            fork_detected: false,
141            server_wire_version: 1,
142            producer_id: 0,
143            accepted_epoch: 0,
144            // All zeros = signing unavailable for this session.
145            delta_signing_key: [0u8; 32],
146        };
147
148        assert!(!client.handle_handshake_ack(&ack).await);
149        assert_eq!(client.state().await, SyncState::Disconnected);
150    }
151}
nodedb_lite/sync/client/handshake.rs

nodedb_lite/sync/client/
handshake.rs
