Delay every data update poll by a random amount, drawn afresh each time

pipeline-engines/src/data_update_service.rs

@@ -55,7 +55,7 @@
 //! and a `429` response surfaces the server's `Retry-After` in the error but the
 //! next poll still uses the configured interval rather than that delay.
 
-use std::collections::hash_map::DefaultHasher;
+use std::collections::hash_map::RandomState;
 use std::hash::{Hash, Hasher};
 use std::io::Read;
 use std::path::Path;
@@ -279,30 +279,32 @@ impl DataUpdateService {
                     .unwrap_or_else(|_| Duration::from_secs(0));
                 return Instant::now()
                     + wait
-                    + self.randomisation(data_file.configuration().max_randomisation_seconds);
+                    + Self::randomisation(data_file.configuration().max_randomisation_seconds);
             }
         }
-        Instant::now() + self.poll_interval(data_file)
+        Instant::now() + Self::poll_interval(data_file)
     }
 
-    /// The base polling interval for a file plus randomisation.
-    fn poll_interval(&self, data_file: &AspectEngineDataFile) -> Duration {
+    /// The polling interval for a file plus a random delay of up to its
+    /// maximum randomisation, as each poll waits.
+    fn poll_interval(data_file: &AspectEngineDataFile) -> Duration {
         let config = data_file.configuration();
         Duration::from_secs(config.polling_interval_seconds)
-            + self.randomisation(config.max_randomisation_seconds)
+            + Self::randomisation(config.max_randomisation_seconds)
     }
 
-    /// A pseudo-random extra delay between zero and `max_seconds`.
+    /// A random delay between zero and `max_seconds`, so instances that start
+    /// together poll at different times.
     ///
-    /// A hash of the current time seeds the choice, which is enough to stagger
-    /// many instances without pulling in a random-number-generator dependency.
-    fn randomisation(&self, max_seconds: u64) -> Duration {
+    /// Each `RandomState` has its own keys, which the standard library draws
+    /// from the operating system, so the delay differs between calls and
+    /// between processes without a random number generator dependency.
+    fn randomisation(max_seconds: u64) -> Duration {
         if max_seconds == 0 {
             return Duration::from_secs(0);
         }
-        let mut hasher = DefaultHasher::new();
-        Instant::now().elapsed().as_nanos().hash(&mut hasher);
-        std::thread::current().id().hash(&mut hasher);
+        let mut hasher = std::hash::BuildHasher::build_hasher(&RandomState::new());
+        std::time::SystemTime::now().hash(&mut hasher);
         Duration::from_secs(hasher.finish() % max_seconds)
     }
 
@@ -513,12 +515,7 @@ impl ServiceHandle {
                 // Poll now. A failure is logged, and the next interval retries.
                 let result = self.check_for_update_from_url(&registration);
                 log_poll(&registration, result);
-                let interval = Duration::from_secs(
-                    registration
-                        .data_file
-                        .configuration()
-                        .polling_interval_seconds,
-                );
+                let interval = DataUpdateService::poll_interval(&registration.data_file);
                 let next = now + interval;
                 *registration
                     .next_poll
@@ -819,4 +816,140 @@ mod tests {
         let service = DataUpdateService::new();
         assert_eq!(service.registered_count(), 0);
     }
+
+    /// An engine with one data file and no update URL, so a poll of it makes
+    /// no request.
+    struct IdleEngine {
+        filter: fiftyone_pipeline_core::EvidenceKeyFilterWhitelist,
+        data_files: Vec<Arc<AspectEngineDataFile>>,
+    }
+
+    impl fiftyone_pipeline_core::FlowElement for IdleEngine {
+        fn process(
+            &self,
+            _data: &mut fiftyone_pipeline_core::FlowData,
+        ) -> fiftyone_pipeline_core::Result<()> {
+            Ok(())
+        }
+        fn data_key(&self) -> &str {
+            "idle"
+        }
+        fn evidence_key_filter(&self) -> &dyn fiftyone_pipeline_core::EvidenceKeyFilter {
+            &self.filter
+        }
+        fn properties(&self) -> &[fiftyone_pipeline_core::PropertyMetaData] {
+            &[]
+        }
+    }
+
+    impl crate::AspectEngine for IdleEngine {
+        fn data_source_tier(&self) -> &str {
+            "Lite"
+        }
+        fn aspect_properties(&self) -> &[crate::AspectPropertyMetaData] {
+            &[]
+        }
+    }
+
+    impl OnPremiseAspectEngine for IdleEngine {
+        fn data_files(&self) -> &[Arc<AspectEngineDataFile>] {
+            &self.data_files
+        }
+        fn refresh(&self, _identifier: Option<&str>) -> fiftyone_pipeline_core::Result<()> {
+            Ok(())
+        }
+    }
+
+    /// A data file with no update URL that polls every `interval_seconds`
+    /// plus up to `max_randomisation_seconds`.
+    fn polled_file(interval_seconds: u64, max_randomisation_seconds: u64) -> AspectEngineDataFile {
+        AspectEngineDataFile::new(
+            crate::DataFileConfiguration::memory_builder()
+                .polling_interval_seconds(interval_seconds)
+                .max_randomisation_seconds(max_randomisation_seconds)
+                .build(),
+        )
+    }
+
+    /// The delay before the first poll is drawn at random, so instances that
+    /// start together poll at different times. 200 draws from 600 seconds
+    /// give about 170 different delays.
+    #[test]
+    fn the_first_poll_is_delayed_at_random() {
+        let service = DataUpdateService::new();
+        let data_file = polled_file(0, 600);
+        let delays: std::collections::BTreeSet<u64> = (0..200)
+            .map(|_| {
+                let start = Instant::now();
+                service
+                    .initial_poll_due(&data_file)
+                    .duration_since(start)
+                    .as_secs()
+            })
+            .collect();
+        eprintln!("{} different delays in 200", delays.len());
+        assert!(
+            delays.len() > 100,
+            "{} different delays in 200: {delays:?}",
+            delays.len()
+        );
+    }
+
+    /// Each poll after the first also waits the polling interval plus a
+    /// random delay, as the specification says.
+    #[test]
+    fn each_later_poll_adds_a_random_delay() {
+        const INTERVAL_SECONDS: u64 = 1000;
+        let handle = ServiceHandle {
+            client: reqwest::blocking::Client::new(),
+            inner: Arc::new(Inner {
+                registrations: Mutex::new(Vec::new()),
+                wake: Condvar::new(),
+                shutting_down: Mutex::new(false),
+            }),
+        };
+        let due = Instant::now();
+        for _ in 0..5 {
+            let data_file = Arc::new(polled_file(INTERVAL_SECONDS, INTERVAL_SECONDS));
+            let engine = Arc::new(IdleEngine {
+                filter: fiftyone_pipeline_core::EvidenceKeyFilterWhitelist::new(
+                    Vec::<String>::new(),
+                ),
+                data_files: vec![Arc::clone(&data_file)],
+            });
+            handle
+                .inner
+                .registrations
+                .lock()
+                .unwrap()
+                .push(Arc::new(Registration {
+                    engine,
+                    data_file,
+                    next_poll: Mutex::new(Some(due)),
+                    watcher: Mutex::new(None),
+                }));
+        }
+
+        handle.run_due_polls();
+
+        // Without a delay every next poll is due at most one interval after
+        // the polls ran. Five delays of under two seconds out of a thousand
+        // happen once in about 3 * 10^13 runs.
+        let undelayed = Instant::now() + Duration::from_secs(INTERVAL_SECONDS + 1);
+        let delayed = handle
+            .inner
+            .registrations
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|registration| {
+                registration
+                    .next_poll
+                    .lock()
+                    .unwrap()
+                    .is_some_and(|next| next > undelayed)
+            })
+            .count();
+        assert!(delayed > 0, "no poll after the first added a delay");
+    }
 }