Provide unique ids across Coordinator instances

Previously we tried to do this from inside the `Coordinator` /
`CoordinatorFactory` types, but that didn't handle the case when two
`CoordinatorFactory` instances were created (almost) simultaneously.

Instead, push the job of determining a unique ID outside of these
types, allowing it to be somewhere higher in the call stack (ideally
nigh-global).

Author: shepmaster

compiler/base/orchestrator/src/coordinator.rs

@@ -820,25 +820,33 @@ enum DemultiplexCommand {
     ListenOnce(JobId, oneshot::Sender<WorkerMessage>),
 }
 
-#[derive(Debug, Copy, Clone)]
-pub struct CoordinatorId {
-    start: u64,
-    id: u64,
+/// The [`Coordinator`][] usually represents a mostly-global resource,
+/// such as a Docker container. To avoid conflicts, each container
+/// must have a unique name, but that uniqueness can only be
+/// guaranteed by whoever is creating [`Coordinator`][]s via the
+/// [`CoordinatorFactory`][].
+pub trait IdProvider: Send + Sync + fmt::Debug + 'static {
+    fn next(&self) -> String;
 }
 
-/// Enforces a limited number of concurrent `Coordinator`s.
+/// A reasonable choice when there's a single [`IdProvider`][] in the
+/// entire process.
+///
+/// This represents uniqueness via a combination of
+///
+/// 1. **process start time** — this helps avoid conflicts from other
+///    processes, assuming they were started at least one second apart.
+///
+/// 2. **instance counter** — this avoids conflicts from other
+///    [`Coordinator`][]s started inside this process.
 #[derive(Debug)]
-pub struct CoordinatorFactory {
-    semaphore: Arc<Semaphore>,
-
+pub struct GlobalIdProvider {
     start: u64,
     id: AtomicU64,
 }
 
-impl CoordinatorFactory {
-    pub fn new(maximum: usize) -> Self {
-        let semaphore = Arc::new(Semaphore::new(maximum));
-
+impl GlobalIdProvider {
+    pub fn new() -> Self {
         let now = std::time::SystemTime::now();
         let start = now
             .duration_since(std::time::UNIX_EPOCH)
@@ -847,28 +855,40 @@ impl CoordinatorFactory {
 
         let id = AtomicU64::new(0);
 
-        Self {
-            semaphore,
-            start,
-            id,
-        }
+        Self { start, id }
     }
+}
 
-    fn next_id(&self) -> CoordinatorId {
+impl IdProvider for GlobalIdProvider {
+    fn next(&self) -> String {
         let start = self.start;
         let id = self.id.fetch_add(1, Ordering::SeqCst);
 
-        CoordinatorId { start, id }
+        format!("{start}-{id}")
+    }
+}
+
+/// Enforces a limited number of concurrent `Coordinator`s.
+#[derive(Debug)]
+pub struct CoordinatorFactory {
+    semaphore: Arc<Semaphore>,
+    ids: Arc<dyn IdProvider>,
+}
+
+impl CoordinatorFactory {
+    pub fn new(ids: Arc<dyn IdProvider>, maximum: usize) -> Self {
+        let semaphore = Arc::new(Semaphore::new(maximum));
+
+        Self { semaphore, ids }
     }
 
     pub fn build<B>(&self) -> Coordinator<B>
     where
-        B: Backend + From<CoordinatorId>,
+        B: Backend + From<Arc<dyn IdProvider>>,
     {
         let semaphore = self.semaphore.clone();
 
-        let id = self.next_id();
-        let backend = B::from(id);
+        let backend = B::from(self.ids.clone());
 
         Coordinator::new(semaphore, backend)
     }
@@ -2586,25 +2606,20 @@ fn basic_secure_docker_command() -> Command {
 }
 
 pub struct DockerBackend {
-    id: CoordinatorId,
-    instance: AtomicU64,
+    ids: Arc<dyn IdProvider>,
 }
 
-impl From<CoordinatorId> for DockerBackend {
-    fn from(id: CoordinatorId) -> Self {
-        Self {
-            id,
-            instance: Default::default(),
-        }
+impl From<Arc<dyn IdProvider>> for DockerBackend {
+    fn from(ids: Arc<dyn IdProvider>) -> Self {
+        Self { ids }
     }
 }
 
 impl DockerBackend {
     fn next_name(&self) -> String {
-        let CoordinatorId { start, id } = self.id;
-        let instance = self.instance.fetch_add(1, Ordering::SeqCst);
+        let id = self.ids.next();
 
-        format!("playground-{start}-{id}-{instance}")
+        format!("playground-{id}")
     }
 }
 
@@ -2794,8 +2809,8 @@ mod tests {
         project_dir: TempDir,
     }
 
-    impl From<CoordinatorId> for TestBackend {
-        fn from(_id: CoordinatorId) -> Self {
+    impl From<Arc<dyn IdProvider>> for TestBackend {
+        fn from(_ids: Arc<dyn IdProvider>) -> Self {
             static COMPILE_WORKER_ONCE: Once = Once::new();
 
             COMPILE_WORKER_ONCE.call_once(|| {
@@ -2849,8 +2864,12 @@ mod tests {
             .unwrap_or(5)
     });
 
-    static TEST_COORDINATOR_FACTORY: Lazy<CoordinatorFactory> =
-        Lazy::new(|| CoordinatorFactory::new(*MAX_CONCURRENT_TESTS));
+    static TEST_COORDINATOR_ID_PROVIDER: Lazy<Arc<GlobalIdProvider>> =
+        Lazy::new(|| Arc::new(GlobalIdProvider::new()));
+
+    static TEST_COORDINATOR_FACTORY: Lazy<CoordinatorFactory> = Lazy::new(|| {
+        CoordinatorFactory::new(TEST_COORDINATOR_ID_PROVIDER.clone(), *MAX_CONCURRENT_TESTS)
+    });
 
     fn new_coordinator_test() -> Coordinator<TestBackend> {
         TEST_COORDINATOR_FACTORY.build()

ui/src/main.rs

@@ -1,6 +1,6 @@
 #![deny(rust_2018_idioms)]
 
-use orchestrator::coordinator::CoordinatorFactory;
+use orchestrator::coordinator::{CoordinatorFactory, GlobalIdProvider, IdProvider};
 use std::{
     net::SocketAddr,
     path::{Path, PathBuf},
@@ -47,6 +47,7 @@ struct Config {
     metrics_token: Option<String>,
     feature_flags: FeatureFlags,
     request_db_path: Option<PathBuf>,
+    id_provider: Arc<dyn IdProvider>,
     coordinators_one_off_limit: usize,
     coordinators_websocket_limit: usize,
     port: u16,
@@ -106,6 +107,8 @@ impl Config {
 
         let request_db_path = env::var_os("PLAYGROUND_REQUEST_DATABASE").map(Into::into);
 
+        let id_provider = Arc::new(GlobalIdProvider::new());
+
         let coordinators_one_off_limit = env::var("PLAYGROUND_COORDINATORS_ONE_OFF_LIMIT")
             .ok()
             .and_then(|l| l.parse().ok())
@@ -123,6 +126,7 @@ impl Config {
             metrics_token,
             feature_flags,
             request_db_path,
+            id_provider,
             coordinators_one_off_limit,
             coordinators_websocket_limit,
             port,
@@ -162,11 +166,11 @@ impl Config {
     }
 
     fn coordinator_one_off_factory(&self) -> CoordinatorFactory {
-        CoordinatorFactory::new(self.coordinators_one_off_limit)
+        CoordinatorFactory::new(self.id_provider.clone(), self.coordinators_one_off_limit)
     }
 
     fn coordinator_websocket_factory(&self) -> CoordinatorFactory {
-        CoordinatorFactory::new(self.coordinators_websocket_limit)
+        CoordinatorFactory::new(self.id_provider.clone(), self.coordinators_websocket_limit)
     }
 
     fn server_socket_addr(&self) -> SocketAddr {