confium_tc/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, TcpStream};
8use std::sync::{Arc, Mutex};
9use std::thread;
10
11use crate::coordinator::coordinator::Coordinator;
12use crate::coordinator::net::{ProtocolMessage, recv_message, send_message};
13use crate::coordinator::session::{Commitment, Share};
14use chrono::Utc;
15
16/// Thread-safe coordinator shared across connection handlers.
17pub type SharedCoordinator = Arc<Mutex<Coordinator>>;
18
19/// TCP coordinator server.
20pub struct CoordinatorServer {
21    addr: String,
22    coordinator: SharedCoordinator,
23}
24
25impl CoordinatorServer {
26    /// Create a new server bound to `addr` (e.g., "127.0.0.1:0" for random port).
27    pub fn new(addr: &str) -> Self {
28        Self {
29            addr: addr.to_string(),
30            coordinator: Arc::new(Mutex::new(Coordinator::new())),
31        }
32    }
33
34    /// Get the shared coordinator handle.
35    pub fn shared_coordinator(&self) -> SharedCoordinator {
36        Arc::clone(&self.coordinator)
37    }
38
39    /// Start the server in a background thread. Returns the actual bound address.
40    pub fn start(&self) -> io::Result<String> {
41        let listener = TcpListener::bind(&self.addr)?;
42        let bound_addr = listener.local_addr()?.to_string();
43        let coordinator = Arc::clone(&self.coordinator);
44
45        thread::spawn(move || {
46            for stream_result in listener.incoming() {
47                match stream_result {
48                    Ok(stream) => {
49                        let coord = Arc::clone(&coordinator);
50                        thread::spawn(move || {
51                            let _ = handle_connection(stream, coord);
52                        });
53                    }
54                    Err(e) => {
55                        eprintln!("Coordinator: accept error: {e}");
56                    }
57                }
58            }
59        });
60
61        Ok(bound_addr)
62    }
63}
64
65fn handle_connection(mut stream: TcpStream, coordinator: SharedCoordinator) -> io::Result<()> {
66    loop {
67        let msg = match recv_message(&mut stream) {
68            Ok(m) => m,
69            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
70            Err(_) => break,
71        };
72
73        let response = process_message(msg, &coordinator);
74        if let Some(resp) = response {
75            if send_message(&mut stream, &resp).is_err() {
76                break;
77            }
78        }
79    }
80    Ok(())
81}
82
83fn process_message(
84    msg: ProtocolMessage,
85    coordinator: &SharedCoordinator,
86) -> Option<ProtocolMessage> {
87    match msg {
88        ProtocolMessage::Register {
89            signer_id,
90            quorum_id: _,
91        } => Some(ProtocolMessage::Registered {
92            signer_id: signer_id.clone(),
93        }),
94
95        ProtocolMessage::CreateSession {
96            quorum_id,
97            scheme,
98            message,
99            threshold,
100            num_parties,
101        } => {
102            let mut coord = coordinator.lock().unwrap();
103            let request = crate::coordinator::session::SessionRequest {
104                quorum_id,
105                scheme,
106                message,
107                threshold,
108                num_parties,
109                unlock_window_minutes: 240,
110                requested_by: "tcp-client".into(),
111            };
112            match coord.create_session(request) {
113                Ok(session_id) => Some(ProtocolMessage::SessionCreated { session_id }),
114                Err(e) => Some(ProtocolMessage::Error {
115                    message: format!("{e:?}"),
116                }),
117            }
118        }
119
120        ProtocolMessage::Commitment {
121            session_id,
122            signer_id,
123            bytes,
124            signature,
125        } => {
126            let mut coord = coordinator.lock().unwrap();
127            let commitment = Commitment {
128                signer_id: signer_id.clone(),
129                bytes,
130                signer_signature: signature,
131                submitted_at: Utc::now(),
132            };
133            match coord.submit_commitment(&session_id, commitment) {
134                Ok(()) => Some(ProtocolMessage::Ack { session_id }),
135                Err(e) => Some(ProtocolMessage::Error {
136                    message: format!("{e:?}"),
137                }),
138            }
139        }
140
141        ProtocolMessage::Share {
142            session_id,
143            signer_id,
144            bytes,
145            signature,
146        } => {
147            let mut coord = coordinator.lock().unwrap();
148            let share = Share {
149                signer_id: signer_id.clone(),
150                bytes,
151                signer_signature: signature,
152                submitted_at: Utc::now(),
153            };
154            match coord.submit_share(&session_id, share) {
155                Ok(()) => {
156                    // After each share, try to aggregate
157                    let threshold = coord.session_threshold(&session_id).unwrap_or(0);
158                    let share_count = coord.session_share_count(&session_id).unwrap_or(0);
159                    if share_count >= threshold as usize {
160                        match coord.aggregate(&session_id) {
161                            Ok(sig) => Some(ProtocolMessage::Signature {
162                                session_id: session_id.clone(),
163                                bytes: sig.bytes,
164                                algorithm: sig.algorithm,
165                                contributing_signers: sig.contributing_signers,
166                            }),
167                            Err(e) => Some(ProtocolMessage::Error {
168                                message: format!("{e:?}"),
169                            }),
170                        }
171                    } else {
172                        Some(ProtocolMessage::Ack { session_id })
173                    }
174                }
175                Err(e) => Some(ProtocolMessage::Error {
176                    message: format!("{e:?}"),
177                }),
178            }
179        }
180
181        ProtocolMessage::GetStatus { session_id } => {
182            let coord = coordinator.lock().unwrap();
183            match session_id {
184                Some(sid) => {
185                    let state = coord.session_state(&sid);
186                    Some(ProtocolMessage::Status {
187                        session_id: sid,
188                        state: format!("{:?}", state),
189                    })
190                }
191                None => Some(ProtocolMessage::Error {
192                    message: "session_id required".into(),
193                }),
194            }
195        }
196
197        ProtocolMessage::HealthCheck => Some(ProtocolMessage::HealthStatus {
198            alive: true,
199            ready: true,
200            session_count: coordinator.lock().unwrap().session_count(),
201            uptime_seconds: 0,
202        }),
203
204        _ => Some(ProtocolMessage::Error {
205            message: "unexpected message type".into(),
206        }),
207    }
208}
confium_tc/coordinator/net_server.rs

confium_tc/coordinator/
net_server.rs
