Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_datime/src/time/ntp.rs

13.2 KiB, 80 runs

created by r1870400018:8568, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1//! Network Time Protocol (NTP) implementation for fe2o3_datime
2//!
3//! This module provides functionality to synchronise time with NTP servers,
4//! query network time, and calculate clock offset and drift.
5//!
6//! NTP Protocol Reference: RFC 5905
7//!
8//! # Examples
9//!
10//! ```ignore
11//! use oxedyne_fe2o3_datime::time::ntp::NtpClient;
12//!
13//! let client = NtpClient::new("pool.ntp.org", 123)?;
14//! let ntp_time = client.query_time()?;
15//! println!("Network time: {}", ntp_time.network_time);
16//! println!("Offset: {} ms", ntp_time.offset_millis);
17//! ```
18//!
19//! [Written with AI entirely](https://need2know.ai/entirely-ai/code)\
20//! Anthropic Claude
21
22use oxedyne_fe2o3_core::prelude::*;
23use std::{
24 net::{UdpSocket, ToSocketAddrs},
25 time::{Duration, SystemTime, UNIX_EPOCH},
26 thread,
27 sync::mpsc,
28};
29
30/// The RFC 5905 packet layout.
31#[repr(C)]
32#[derive(Debug, Clone, Copy)]
33struct NtpPacket {
34 li_vn_mode: u8, // 2-bit leap indicator, 3-bit version, 3-bit mode
35 stratum: u8,
36 poll: i8,
37 precision: i8,
38 root_delay: u32,
39 root_dispersion: u32,
40 ref_id: u32,
41 ref_timestamp: u64,
42 origin_timestamp: u64, // T1
43 receive_timestamp: u64, // T2
44 transmit_timestamp: u64, // T3
45}
46
47impl NtpPacket {
48 fn new_request() -> Self {
49 NtpPacket {
50 li_vn_mode: 0x1B, // Version 3, Mode 3 (client)
51 stratum: 0,
52 poll: 6, // 64 seconds
53 precision: -6, // ~15.6 ms precision
54 root_delay: 0,
55 root_dispersion: 0,
56 ref_id: 0,
57 ref_timestamp: 0,
58 origin_timestamp: 0,
59 receive_timestamp: 0,
60 transmit_timestamp: system_time_to_ntp_timestamp(SystemTime::now()),
61 }
62 }
63
64 fn to_bytes(&self) -> [u8; 48] {
65 let mut bytes = [0u8; 48];
66 bytes[0] = self.li_vn_mode;
67 bytes[1] = self.stratum;
68 bytes[2] = self.poll as u8;
69 bytes[3] = self.precision as u8;
70 bytes[4..8].copy_from_slice(&self.root_delay.to_be_bytes());
71 bytes[8..12].copy_from_slice(&self.root_dispersion.to_be_bytes());
72 bytes[12..16].copy_from_slice(&self.ref_id.to_be_bytes());
73 bytes[16..24].copy_from_slice(&self.ref_timestamp.to_be_bytes());
74 bytes[24..32].copy_from_slice(&self.origin_timestamp.to_be_bytes());
75 bytes[32..40].copy_from_slice(&self.receive_timestamp.to_be_bytes());
76 bytes[40..48].copy_from_slice(&self.transmit_timestamp.to_be_bytes());
77 bytes
78 }
79
80 fn from_bytes(bytes: &[u8; 48]) -> Self {
81 NtpPacket {
82 li_vn_mode: bytes[0],
83 stratum: bytes[1],
84 poll: bytes[2] as i8,
85 precision: bytes[3] as i8,
86 root_delay: u32::from_be_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]),
87 root_dispersion: u32::from_be_bytes([bytes[8], bytes[9], bytes[10], bytes[11]]),
88 ref_id: u32::from_be_bytes([bytes[12], bytes[13], bytes[14], bytes[15]]),
89 ref_timestamp: u64::from_be_bytes([
90 bytes[16], bytes[17], bytes[18], bytes[19],
91 bytes[20], bytes[21], bytes[22], bytes[23]
92 ]),
93 origin_timestamp: u64::from_be_bytes([
94 bytes[24], bytes[25], bytes[26], bytes[27],
95 bytes[28], bytes[29], bytes[30], bytes[31]
96 ]),
97 receive_timestamp: u64::from_be_bytes([
98 bytes[32], bytes[33], bytes[34], bytes[35],
99 bytes[36], bytes[37], bytes[38], bytes[39]
100 ]),
101 transmit_timestamp: u64::from_be_bytes([
102 bytes[40], bytes[41], bytes[42], bytes[43],
103 bytes[44], bytes[45], bytes[46], bytes[47]
104 ]),
105 }
106 }
107}
108
109#[derive(Debug, Clone)]
110pub struct NtpTimeResult {
111 pub network_time: SystemTime,
112 pub local_time: SystemTime,
113 pub offset_millis: i64, // positive when the local clock is ahead
114 pub delay_millis: u64,
115 pub stratum: u8,
116 pub reference_id: u32,
117}
118
119#[derive(Debug)]
120pub struct NtpClient {
121 server_addr: String,
122 port: u16,
123 timeout: Duration,
124}
125
126impl NtpClient {
127 pub fn new(server: &str, port: u16) -> Self {
128 NtpClient {
129 server_addr: server.to_string(),
130 port,
131 timeout: Duration::from_secs(5),
132 }
133 }
134
135 pub fn default(server: &str) -> Self {
136 Self::new(server, 123)
137 }
138
139 pub fn with_timeout(mut self, timeout: Duration) -> Self {
140 self.timeout = timeout;
141 self
142 }
143
144 pub fn query_time(&self) -> Outcome<NtpTimeResult> {
145 // Resolve server address
146 let server_addr = format!("{}:{}", self.server_addr, self.port);
147 let mut addrs = res!(server_addr.to_socket_addrs()
148 .map_err(|e| err!("Failed to resolve NTP server '{}': {}", server_addr, e; Invalid, Network)));
149
150 let addr = res!(addrs.next()
151 .ok_or_else(|| err!("No addresses found for NTP server '{}'", server_addr; Invalid, Network)));
152
153 // Create UDP socket
154 let socket = res!(UdpSocket::bind("0.0.0.0:0")
155 .map_err(|e| err!("Failed to create UDP socket: {}", e; Invalid, Network)));
156
157 res!(socket.set_read_timeout(Some(self.timeout))
158 .map_err(|e| err!("Failed to set socket timeout: {}", e; Invalid, Network)));
159
160 // Create and send NTP request
161 let request = NtpPacket::new_request();
162 let request_bytes = request.to_bytes();
163 let send_time = SystemTime::now();
164
165 res!(socket.send_to(&request_bytes, addr)
166 .map_err(|e| err!("Failed to send NTP request: {}", e; Invalid, Network)));
167
168 // Receive response
169 let mut response_bytes = [0u8; 48];
170 let (bytes_received, _) = res!(socket.recv_from(&mut response_bytes)
171 .map_err(|e| err!("Failed to receive NTP response: {}", e; Invalid, Network, Timeout)));
172
173 if bytes_received != 48 {
174 return Err(err!("Invalid NTP response size: {} bytes", bytes_received; Invalid, Network));
175 }
176
177 let receive_time = SystemTime::now();
178 let response = NtpPacket::from_bytes(&response_bytes);
179
180 // Calculate timestamps and offset
181 self.calculate_time_result(request, response, send_time, receive_time)
182 }
183
184 pub fn query_multiple_servers(servers: &[&str], max_concurrent: usize) -> Outcome<NtpTimeResult> {
185 if servers.is_empty() {
186 return Err(err!("No NTP servers provided"; Invalid, Input));
187 }
188
189 let max_queries = std::cmp::min(servers.len(), max_concurrent);
190 let (tx, rx) = mpsc::channel();
191
192 // Launch concurrent queries
193 let mut handles = Vec::new();
194 for &server in servers.iter().take(max_queries) {
195 let tx = tx.clone();
196 let server_owned = server.to_string(); // Convert to owned string
197 let handle = thread::spawn(move || {
198 let client = NtpClient::default(&server_owned);
199 let result = client.query_time();
200 let _ = tx.send(result); // Ignore send errors
201 });
202 handles.push(handle);
203 }
204
205 // Drop the original sender to signal no more messages
206 drop(tx);
207
208 // Collect results with timeout
209 let mut results = Vec::new();
210 let timeout = Duration::from_secs(5); // 5 second timeout total
211 let start_time = SystemTime::now();
212
213 while let Ok(result) = rx.recv() {
214 if let Ok(ntp_result) = result {
215 results.push(ntp_result);
216 }
217
218 // Check if we have enough results or timeout
219 if results.len() >= 3 || start_time.elapsed().unwrap_or(timeout) >= timeout {
220 break;
221 }
222 }
223
224 // Wait for all threads to complete (with timeout)
225 for handle in handles {
226 let _ = handle.join(); // Ignore join errors
227 }
228
229 if results.is_empty() {
230 return Err(err!("All NTP servers failed to respond"; Invalid, Network));
231 }
232
233 // Select the best result (lowest stratum, then lowest delay)
234 results.sort_by(|a, b| {
235 a.stratum.cmp(&b.stratum)
236 .then(a.delay_millis.cmp(&b.delay_millis))
237 });
238
239 match results.into_iter().next() {
240 Some(result) => Ok(result),
241 None => Err(err!("Internal error: no NTP results after filtering"; Invalid, Bug)),
242 }
243 }
244
245 fn calculate_time_result(
246 &self,
247 _request: NtpPacket,
248 response: NtpPacket,
249 send_time: SystemTime,
250 receive_time: SystemTime
251 ) -> Outcome<NtpTimeResult> {
252 // Convert NTP timestamps to SystemTime
253 let t1 = system_time_to_ntp_timestamp(send_time); // Origin time
254 let t2 = response.receive_timestamp; // Receive time at server
255 let t3 = response.transmit_timestamp; // Transmit time at server
256 let t4 = system_time_to_ntp_timestamp(receive_time); // Destination time
257
258 // Calculate offset and delay using NTP algorithm
259 // Offset = ((T2 - T1) + (T3 - T4)) / 2
260 // Delay = (T4 - T1) - (T3 - T2)
261
262 let offset_ntp = (((t2 as i64) - (t1 as i64)) + ((t3 as i64) - (t4 as i64))) / 2;
263 let delay_ntp = ((t4 as i64) - (t1 as i64)) - ((t3 as i64) - (t2 as i64));
264
265 // Convert to milliseconds
266 let offset_millis = ntp_timestamp_to_millis(offset_ntp as u64) as i64;
267 let delay_millis = std::cmp::max(0, ntp_timestamp_to_millis(delay_ntp as u64)) as u64;
268
269 // Calculate network time
270 let network_time = ntp_timestamp_to_system_time(t3);
271
272 Ok(NtpTimeResult {
273 network_time,
274 local_time: receive_time,
275 offset_millis,
276 delay_millis,
277 stratum: response.stratum,
278 reference_id: response.ref_id,
279 })
280 }
281}
282
283pub struct NtpPool;
284
285impl NtpPool {
286 pub const PUBLIC: &'static [&'static str] = &[
287 "pool.ntp.org",
288 "time.nist.gov",
289 "time.google.com",
290 "time.cloudflare.com",
291 ];
292
293 pub fn query_time() -> Outcome<NtpTimeResult> {
294 NtpClient::query_multiple_servers(Self::PUBLIC, 4)
295 }
296
297 pub fn query_reliable() -> Outcome<NtpTimeResult> {
298 let reliable_servers = &[
299 "time.google.com",
300 "time.cloudflare.com",
301 "pool.ntp.org",
302 ];
303 NtpClient::query_multiple_servers(reliable_servers, 3)
304 }
305}
306
307fn system_time_to_ntp_timestamp(time: SystemTime) -> u64 {
308 const NTP_EPOCH_OFFSET: u64 = 2_208_988_800; // Seconds from 1900 to 1970
309
310 match time.duration_since(UNIX_EPOCH) {
311 Ok(duration) => {
312 let seconds = duration.as_secs() + NTP_EPOCH_OFFSET;
313 let nanos = duration.subsec_nanos() as u64;
314 let fraction = (nanos * (1u64 << 32)) / 1_000_000_000;
315 (seconds << 32) | fraction
316 },
317 Err(_) => 0, // Handle times before Unix epoch
318 }
319}
320
321fn ntp_timestamp_to_system_time(ntp_timestamp: u64) -> SystemTime {
322 const NTP_EPOCH_OFFSET: u64 = 2_208_988_800;
323
324 let seconds = (ntp_timestamp >> 32) - NTP_EPOCH_OFFSET;
325 let fraction = ntp_timestamp & 0xFFFFFFFF;
326 let nanos = (fraction * 1_000_000_000) >> 32;
327
328 UNIX_EPOCH + Duration::new(seconds, nanos as u32)
329}
330
331fn ntp_timestamp_to_millis(ntp_timestamp: u64) -> u64 {
332 let seconds = ntp_timestamp >> 32;
333 let fraction = ntp_timestamp & 0xFFFFFFFF;
334 let millis = (fraction * 1000) >> 32;
335 seconds * 1000 + millis
336}
337
338#[cfg(test)]
339mod tests {
340 use super::*;
341
342 #[test]
343 fn test_ntp_packet_serialization() {
344 let packet = NtpPacket::new_request();
345 let bytes = packet.to_bytes();
346 let deserialised = NtpPacket::from_bytes(&bytes);
347
348 assert_eq!(packet.li_vn_mode, deserialised.li_vn_mode);
349 assert_eq!(packet.stratum, deserialised.stratum);
350 assert_eq!(packet.transmit_timestamp, deserialised.transmit_timestamp);
351 }
352
353 #[test]
354 fn test_timestamp_conversion() {
355 let now = SystemTime::now();
356 let ntp_ts = system_time_to_ntp_timestamp(now);
357 let converted_back = ntp_timestamp_to_system_time(ntp_ts);
358
359 // Should be within 1 second due to precision
360 let diff = now.duration_since(converted_back).unwrap_or_else(|_| {
361 converted_back.duration_since(now).unwrap_or(Duration::from_secs(0))
362 });
363
364 assert!(diff.as_secs() <= 1);
365 }
366
367 #[ignore] // Network test - only run manually
368 #[test]
369 fn test_ntp_query() -> Outcome<()> {
370 let client = NtpClient::default("pool.ntp.org");
371 let result = res!(client.query_time());
372
373 println!("Network time: {:?}", result.network_time);
374 println!("Offset: {} ms", result.offset_millis);
375 println!("Delay: {} ms", result.delay_millis);
376
377 // Sanity checks
378 assert!(result.delay_millis < 10000); // Less than 10 seconds delay
379 assert!(result.stratum > 0 && result.stratum < 16); // Valid stratum
380 Ok(())
381 }
382
383 #[ignore] // Network test - only run manually
384 #[test]
385 fn test_ntp_pool_query() -> Outcome<()> {
386 let result = res!(NtpPool::query_reliable());
387
388 println!("Pool result - Offset: {} ms, Delay: {} ms, Stratum: {}",
389 result.offset_millis, result.delay_millis, result.stratum);
390
391 assert!(result.stratum > 0);
392 Ok(())
393 }
394}