diff --git a/Cargo.toml b/Cargo.toml index b753544..f76426d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "endpoint-libs" -version = "3.3.1" +version = "3.3.2" edition = "2024" authors = ["Veon "] description = "Launch MCP services fast: describe endpoints once in RON, and endpoint-gen generates the Rust models, docs and MCP tool schemas that this crate serves over WebSocket RPC, with roles and typed errors built in." diff --git a/src/libs/utils.rs b/src/libs/utils.rs index 97f19c8..9420ad9 100644 --- a/src/libs/utils.rs +++ b/src/libs/utils.rs @@ -4,8 +4,44 @@ pub fn get_log_id() -> u64 { chrono::Utc::now().timestamp_micros() as _ } +/// A connection id unique within this process. +/// +/// It was the clock in microseconds, cut to 32 bits, which gives two +/// connections accepted in the same microsecond the same id. With one accept +/// loop that took two accepts inside a microsecond; with a shard per core it +/// is two shards accepting at once, and the second insert into the shared +/// connection table replaced the first. A counter cannot repeat until it wraps. +/// It starts from the clock so ids still differ across restarts in logs. pub fn get_conn_id() -> u32 { - chrono::Utc::now().timestamp_micros() as _ + use std::sync::OnceLock; + use std::sync::atomic::{AtomicU32, Ordering}; + static NEXT: OnceLock = OnceLock::new(); + NEXT.get_or_init(|| AtomicU32::new(chrono::Utc::now().timestamp_micros() as u32)) + .fetch_add(1, Ordering::Relaxed) +} + +#[cfg(test)] +mod conn_id_tests { + #[test] + fn ids_from_many_threads_never_repeat() { + let ids: Vec = std::thread::scope(|scope| { + let workers: Vec<_> = (0..8) + .map(|_| { + scope.spawn(|| { + (0..10_000) + .map(|_| super::get_conn_id()) + .collect::>() + }) + }) + .collect(); + workers + .into_iter() + .flat_map(|worker| worker.join().unwrap()) + .collect() + }); + let unique: std::collections::HashSet = ids.iter().copied().collect(); + assert_eq!(unique.len(), ids.len()); + } } pub fn get_time_milliseconds() -> i64 { diff --git a/tests/sharded_listen.rs b/tests/sharded_listen.rs index 96e41b8..c7ff0bf 100644 --- a/tests/sharded_listen.rs +++ b/tests/sharded_listen.rs @@ -5,7 +5,7 @@ //! must complete a handshake and a request whichever shard the kernel picks, //! and one stop must bring every shard down so `listen_until` returns. -#![cfg(feature = "ws-client")] +#![cfg(all(feature = "ws", feature = "ws-client"))] use std::sync::Arc; use std::time::Duration;