confium_coordinator/coordinator/
net_server.rs

1//! TCP coordinator server — wraps the in-memory Coordinator with a TCP server.
2//!
3//! Listens on a TCP port. Each client connection gets its own thread.
4//! Routes protocol messages to the Coordinator's API.
5
6use std::io;
7use std::net::TcpListener;
8use std::net::TcpStream;
9
10/// A byte-stream session: TCP directly, or any registry transport
11/// (e.g. noise) wrapped in [`confium_net::io::TransportIo`].
12pub trait SessionIo: std::io::Read + std::io::Write + Send {
13    /// Best-effort read timeout. Returns `false` when unsupported
14    /// (non-socket transports are message-framed and their `recv`
15    /// blocks only for the next message, which is the desired
16    /// behavior for a caller that previously set a socket timeout).
17    ///
18    /// Prefer the polling deadline adapter over this hook: under MRI
19    /// Ruby on windows-gnu the first recv on a socket carrying
20    /// SO_RCVTIMEO fails WSAENOTSOCK (audit ledger).
21    fn set_read_timeout(&mut self, _d: Option<std::time::Duration>) -> bool {
22        false
23    }
24
25    /// Best-effort non-blocking toggle for the polling deadline
26    /// adapter. Returns `false` when unsupported.
27    fn set_nonblocking(&mut self, _on: bool) -> bool {
28        false
29    }
30}
31
32impl SessionIo for TcpStream {
33    fn set_read_timeout(&mut self, d: Option<std::time::Duration>) -> bool {
34        std::net::TcpStream::set_read_timeout(self, d).is_ok()
35    }
36
37    fn set_nonblocking(&mut self, on: bool) -> bool {
38        std::net::TcpStream::set_nonblocking(self, on).is_ok()
39    }
40}
41
42impl SessionIo for confium_net::io::TransportIo {}
43use std::sync::{Arc, Mutex};
44use std::thread;
45
46use crate::coordinator::coordinator::Coordinator;
47use crate::coordinator::net::{ProtocolMessage, recv_message, send_message};
48use crate::coordinator::session::{Commitment, Share};
49use chrono::Utc;
50
51/// Thread-safe coordinator shared across connection handlers.
52pub type SharedCoordinator = Arc<Mutex<Coordinator>>;
53
54/// TCP coordinator server.
55pub struct CoordinatorServer {
56    addr: String,
57    coordinator: SharedCoordinator,
58    start_time: std::time::Instant,
59}
60
61impl CoordinatorServer {
62    /// Create a new server bound to `addr` (e.g., "127.0.0.1:0" for random port).
63    pub fn new(addr: &str) -> Self {
64        Self {
65            addr: addr.to_string(),
66            coordinator: Arc::new(Mutex::new(Coordinator::new())),
67            start_time: std::time::Instant::now(),
68        }
69    }
70
71    /// Get the shared coordinator handle.
72    pub fn shared_coordinator(&self) -> SharedCoordinator {
73        Arc::clone(&self.coordinator)
74    }
75
76    /// Start the server in a background thread. Returns the actual bound address.
77    pub fn start(&self) -> io::Result<String> {
78        let listener = TcpListener::bind(&self.addr)?;
79        let bound_addr = listener.local_addr()?.to_string();
80        let coordinator = Arc::clone(&self.coordinator);
81        let start_time = self.start_time;
82
83        thread::spawn(move || {
84            for stream_result in listener.incoming() {
85                match stream_result {
86                    Ok(stream) => {
87                        let coord = Arc::clone(&coordinator);
88                        thread::spawn(move || {
89                            let _ = handle_connection(
90                                Box::new(stream) as Box<dyn SessionIo>,
91                                coord,
92                                start_time,
93                            );
94                        });
95                    }
96                    Err(e) => {
97                        eprintln!("Coordinator: accept error: {e}");
98                    }
99                }
100            }
101        });
102
103        Ok(bound_addr)
104    }
105
106    /// Serve sessions over any registry transport URL (e.g.
107    /// `noise://0.0.0.0:18432?key=<hex>`). The scheme resolves at
108    /// link time; link `confium-net-noise` (or another transport
109    /// crate) into the binary to make it available.
110    pub fn start_url(&self, url: &str) -> io::Result<String> {
111        let mut listener =
112            confium_net::listen(url).map_err(|e| io::Error::other(format!("listen {url}: {e}")))?;
113        let coordinator = Arc::clone(&self.coordinator);
114        let start_time = self.start_time;
115
116        thread::spawn(move || {
117            loop {
118                match listener.accept() {
119                    Ok(transport) => {
120                        let coord = Arc::clone(&coordinator);
121                        thread::spawn(move || {
122                            let session = confium_net::io::TransportIo::new(transport);
123                            let _ = handle_connection(
124                                Box::new(session) as Box<dyn SessionIo>,
125                                coord,
126                                start_time,
127                            );
128                        });
129                    }
130                    Err(e) => {
131                        eprintln!("Coordinator: accept error: {e}");
132                    }
133                }
134            }
135        });
136
137        Ok(url.to_string())
138    }
139}
140
141fn handle_connection(
142    mut stream: Box<dyn SessionIo>,
143    coordinator: SharedCoordinator,
144    start_time: std::time::Instant,
145) -> io::Result<()> {
146    loop {
147        let msg = match recv_message(&mut stream) {
148            Ok(m) => m,
149            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
150            Err(_) => break,
151        };
152
153        let response = process_message(msg, &coordinator, start_time);
154        if let Some(resp) = response {
155            if send_message(&mut stream, &resp).is_err() {
156                break;
157            }
158        }
159    }
160    Ok(())
161}
162
163fn process_message(
164    msg: ProtocolMessage,
165    coordinator: &SharedCoordinator,
166    start_time: std::time::Instant,
167) -> Option<ProtocolMessage> {
168    match msg {
169        ProtocolMessage::Register {
170            signer_id,
171            quorum_id: _,
172        } => Some(ProtocolMessage::Registered {
173            signer_id: signer_id.clone(),
174        }),
175
176        ProtocolMessage::CreateSession {
177            quorum_id,
178            scheme,
179            message,
180            threshold,
181            num_parties,
182        } => {
183            let mut coord = coordinator.lock().unwrap();
184            let request = crate::coordinator::session::SessionRequest {
185                quorum_id,
186                scheme,
187                message,
188                threshold,
189                num_parties,
190                unlock_window_minutes: 240,
191                requested_by: "tcp-client".into(),
192            };
193            match coord.create_session(request) {
194                Ok(session_id) => Some(ProtocolMessage::SessionCreated { session_id }),
195                Err(e) => Some(ProtocolMessage::Error {
196                    message: format!("{e:?}"),
197                }),
198            }
199        }
200
201        ProtocolMessage::Commitment {
202            session_id,
203            signer_id,
204            bytes,
205            signature,
206        } => {
207            let mut coord = coordinator.lock().unwrap();
208            let commitment = Commitment {
209                signer_id: signer_id.clone(),
210                bytes,
211                signer_signature: signature,
212                submitted_at: Utc::now(),
213            };
214            match coord.submit_commitment(&session_id, commitment) {
215                Ok(()) => Some(ProtocolMessage::Ack { session_id }),
216                Err(e) => Some(ProtocolMessage::Error {
217                    message: format!("{e:?}"),
218                }),
219            }
220        }
221
222        ProtocolMessage::Share {
223            session_id,
224            signer_id,
225            bytes,
226            signature,
227        } => {
228            let mut coord = coordinator.lock().unwrap();
229            let share = Share {
230                signer_id: signer_id.clone(),
231                bytes,
232                signer_signature: signature,
233                submitted_at: Utc::now(),
234            };
235            match coord.submit_share(&session_id, share) {
236                Ok(()) => {
237                    let threshold = coord.session_threshold(&session_id).unwrap_or(0);
238                    let share_count = coord.session_share_count(&session_id).unwrap_or(0);
239                    if share_count >= threshold as usize {
240                        match coord.aggregate(&session_id) {
241                            Ok(sig) => Some(ProtocolMessage::Signature {
242                                session_id: session_id.clone(),
243                                bytes: sig.bytes,
244                                algorithm: sig.algorithm,
245                                contributing_signers: sig.contributing_signers,
246                            }),
247                            Err(e) => Some(ProtocolMessage::Error {
248                                message: format!("{e:?}"),
249                            }),
250                        }
251                    } else {
252                        Some(ProtocolMessage::Ack { session_id })
253                    }
254                }
255                Err(e) => Some(ProtocolMessage::Error {
256                    message: format!("{e:?}"),
257                }),
258            }
259        }
260
261        ProtocolMessage::GetStatus { session_id } => {
262            let coord = coordinator.lock().unwrap();
263            match session_id {
264                Some(sid) => {
265                    let state = coord.session_state(&sid);
266                    Some(ProtocolMessage::Status {
267                        session_id: sid,
268                        state: format!("{:?}", state),
269                    })
270                }
271                None => Some(ProtocolMessage::Error {
272                    message: "session_id required".into(),
273                }),
274            }
275        }
276
277        ProtocolMessage::HealthCheck => {
278            let coord = coordinator.lock().unwrap();
279            let session_count = coord.session_count();
280            Some(ProtocolMessage::HealthStatus {
281                alive: true,
282                ready: true,
283                session_count,
284                uptime_seconds: start_time.elapsed().as_secs(),
285            })
286        }
287
288        _ => Some(ProtocolMessage::Error {
289            message: "unexpected message type".into(),
290        }),
291    }
292}
293
294#[cfg(test)]
295mod tests {
296    use super::*;
297
298    #[test]
299    fn health_check_returns_status() {
300        let server = CoordinatorServer::new("127.0.0.1:0");
301        let coordinator = server.shared_coordinator();
302        let start_time = server.start_time;
303
304        let response = process_message(ProtocolMessage::HealthCheck, &coordinator, start_time);
305        match response {
306            Some(ProtocolMessage::HealthStatus {
307                alive,
308                ready,
309                session_count,
310                uptime_seconds,
311            }) => {
312                assert!(alive);
313                assert!(ready);
314                assert_eq!(session_count, 0);
315                assert!(uptime_seconds < 5);
316            }
317            _ => panic!("expected HealthStatus"),
318        }
319    }
320
321    #[test]
322    fn health_check_after_session_increments_count() {
323        let server = CoordinatorServer::new("127.0.0.1:0");
324        let coordinator = server.shared_coordinator();
325        let start_time = server.start_time;
326
327        let req = crate::coordinator::session::SessionRequest {
328            quorum_id: "q1".into(),
329            scheme: "CMP20".into(),
330            message: vec![0; 32],
331            threshold: 2,
332            num_parties: 3,
333            unlock_window_minutes: 60,
334            requested_by: "test".into(),
335        };
336        coordinator.lock().unwrap().create_session(req).unwrap();
337
338        let response = process_message(ProtocolMessage::HealthCheck, &coordinator, start_time);
339        match response {
340            Some(ProtocolMessage::HealthStatus { session_count, .. }) => {
341                assert_eq!(session_count, 1);
342            }
343            _ => panic!("expected HealthStatus"),
344        }
345    }
346}
confium_coordinator/coordinator/net_server.rs

confium_coordinator/coordinator/
net_server.rs
