From e0522c82ff689861016741ad39495ea26d59ccab Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 26 Oct 2026 23:31:24 +0000 Subject: [PATCH 1/3] Test the file watcher's lifetime, master switch and file filter --- pipeline-engines/tests/file_watcher.rs | 201 +++++++++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 pipeline-engines/tests/file_watcher.rs diff --git a/pipeline-engines/tests/file_watcher.rs b/pipeline-engines/tests/file_watcher.rs new file mode 100644 index 0000000..32a5369 --- /dev/null +++ b/pipeline-engines/tests/file_watcher.rs @@ -0,0 +1,201 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +//! Tests of the data update service's file system watcher, which refreshes an +//! engine when its data file changes on disk. +#![cfg(feature = "data-update")] + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use fiftyone_pipeline_core::{ + EvidenceKeyFilter, EvidenceKeyFilterWhitelist, FlowData, FlowElement, PropertyMetaData, Result, +}; +use fiftyone_pipeline_engines::{ + AspectEngine, AspectEngineDataFile, AspectPropertyMetaData, DataFileConfiguration, + DataUpdateService, OnPremiseAspectEngine, +}; + +/// How long a test waits for a refresh it expects. +const REFRESH_WAIT: Duration = Duration::from_secs(10); + +/// How long a test waits before deciding a refresh it does not expect has +/// not happened. +const QUIET_WAIT: Duration = Duration::from_secs(2); + +/// How long the watcher is given to start before the file changes. +const WATCHER_START: Duration = Duration::from_millis(300); + +/// An engine that counts its refreshes. +struct CountingEngine { + filter: EvidenceKeyFilterWhitelist, + data_files: Vec>, + refreshes: Arc, +} + +impl FlowElement for CountingEngine { + fn process(&self, _data: &mut FlowData) -> Result<()> { + Ok(()) + } + fn data_key(&self) -> &str { + "counting" + } + fn evidence_key_filter(&self) -> &dyn EvidenceKeyFilter { + &self.filter + } + fn properties(&self) -> &[PropertyMetaData] { + &[] + } +} + +impl AspectEngine for CountingEngine { + fn data_source_tier(&self) -> &str { + "Lite" + } + fn aspect_properties(&self) -> &[AspectPropertyMetaData] { + &[] + } +} + +impl OnPremiseAspectEngine for CountingEngine { + fn data_files(&self) -> &[Arc] { + &self.data_files + } + fn refresh(&self, _identifier: Option<&str>) -> Result<()> { + self.refreshes.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +/// A new, empty folder for one test, holding a data file. +fn folder(test: &str) -> (PathBuf, PathBuf) { + let dir = std::env::temp_dir().join(format!( + "fiftyone-watcher-{test}-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("device.dat"); + std::fs::write(&path, b"version 1").unwrap(); + (dir, path) +} + +/// An engine whose data file at `path` is watched, with automatic updates on +/// or off. There is no update URL, so a poll makes no request. +fn watched_engine(path: &Path, automatic_updates: bool) -> Arc { + let config = DataFileConfiguration::builder(path) + .automatic_updates_enabled(automatic_updates) + .file_system_watcher_enabled(true) + .build(); + Arc::new(CountingEngine { + filter: EvidenceKeyFilterWhitelist::new(Vec::::new()), + data_files: vec![Arc::new(AspectEngineDataFile::new(config))], + refreshes: Arc::new(AtomicUsize::new(0)), + }) +} + +fn register(service: &DataUpdateService, engine: &Arc) { + let data_file = Arc::clone(&engine.data_files[0]); + let engine: Arc = engine.clone(); + service.register(engine, data_file).unwrap(); + std::thread::sleep(WATCHER_START); +} + +/// The refresh count once it reaches `at_least`, or after `limit`. +fn refreshes_after(count: &AtomicUsize, at_least: usize, limit: Duration) -> usize { + let start = Instant::now(); + while count.load(Ordering::SeqCst) < at_least && start.elapsed() < limit { + std::thread::sleep(Duration::from_millis(20)); + } + count.load(Ordering::SeqCst) +} + +/// Dropping the service releases the engines registered with it and stops +/// their watchers, so the data set is freed and a later change to the file +/// refreshes nothing. +#[test] +fn dropping_the_service_releases_the_engine_and_stops_the_watcher() { + let (dir, path) = folder("drop"); + let engine = watched_engine(&path, true); + let refreshes = Arc::clone(&engine.refreshes); + let service = DataUpdateService::new(); + register(&service, &engine); + let released = Arc::downgrade(&engine); + + drop(engine); + drop(service); + + assert!( + released.upgrade().is_none(), + "the engine is still alive after the service and its caller dropped it" + ); + std::fs::write(&path, b"version 2").unwrap(); + assert_eq!(refreshes_after(&refreshes, 1, QUIET_WAIT), 0); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Automatic updates are the master switch for the watcher, as the +/// specification says. With them off a change to the file refreshes nothing, +/// and with them on it refreshes the engine. +#[test] +fn the_watcher_runs_only_with_automatic_updates_on() { + let (dir, path) = folder("switch"); + let service = DataUpdateService::new(); + + let off = watched_engine(&path, false); + register(&service, &off); + std::fs::write(&path, b"version 2").unwrap(); + assert_eq!( + refreshes_after(&off.refreshes, 1, QUIET_WAIT), + 0, + "the watcher refreshed an engine with automatic updates off" + ); + + let on = watched_engine(&path, true); + register(&service, &on); + std::fs::write(&path, b"version 3").unwrap(); + assert!(refreshes_after(&on.refreshes, 1, REFRESH_WAIT) >= 1); + let _ = std::fs::remove_dir_all(&dir); +} + +/// The watcher refreshes the engine when its data file changes, and not when +/// another file in the same folder does. +#[test] +fn the_watcher_ignores_other_files_in_the_folder() { + let (dir, path) = folder("others"); + let service = DataUpdateService::new(); + let engine = watched_engine(&path, true); + register(&service, &engine); + + std::fs::write(dir.join("application.log"), b"a line").unwrap(); + assert_eq!( + refreshes_after(&engine.refreshes, 1, QUIET_WAIT), + 0, + "another file in the folder refreshed the engine" + ); + + std::fs::write(&path, b"version 2").unwrap(); + assert!(refreshes_after(&engine.refreshes, 1, REFRESH_WAIT) >= 1); + let _ = std::fs::remove_dir_all(&dir); +} From 427690accc712a6ef0a9a517947deef7b7326ec1 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 26 Oct 2026 23:32:09 +0000 Subject: [PATCH 2/3] Release a watched engine with its service, run the watcher only with automatic updates on, and watch only the data file The watcher's callback held the registration that owned the watcher, so the registration, its engine and the data set the engine held, the open data file and the watcher's thread outlived the service. The callback now holds the registration weakly. Automatic updates are the specification's master switch for the watcher as well as for polling, and the watcher now refreshes the engine only for events that name the data file, from the modified time the file had when it was registered. --- README.md | 10 ++-- device-detection/src/lib.rs | 16 +++--- ip-intelligence-onpremise/src/builder.rs | 8 +-- ip-intelligence/src/lib.rs | 9 ++-- ip-intelligence/src/on_premise.rs | 7 ++- pipeline-engines/src/data_file.rs | 11 ++-- pipeline-engines/src/data_update_service.rs | 59 +++++++++++++-------- 7 files changed, 77 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index 291428c..69ce600 100644 --- a/README.md +++ b/README.md @@ -115,10 +115,12 @@ path, so cloud-only users and most of CI build without a C compiler. `.data_update_license_keys(..)`, and that refreshes the engine when the data file changes on disk. The Distributor serves nothing without a licence key, so a build with polling on and no key fails with an error that says so, as - it does in .NET, and a deployment of the free Lite file turns polling off - with `.auto_update(false)`. `.file_system_watcher(false)` turns the watcher - off, and `pipeline.resource::()` reaches the service for - a check on demand. + it does in .NET, and a deployment of the free Lite file turns automatic + updates off with `.auto_update(false)`, which stops the polling and the file + watcher together, as the specification's master switch does. + `.file_system_watcher(false)` turns only the watcher off, and + `pipeline.resource::()` reaches the service for a check + on demand. - JavaScript minification in `fiftyone-javascript-builder` is opt-in behind the `minify` feature and off by default. See ARCHITECTURE.md for why. diff --git a/device-detection/src/lib.rs b/device-detection/src/lib.rs index 5a3fad4..7efd9f8 100644 --- a/device-detection/src/lib.rs +++ b/device-detection/src/lib.rs @@ -90,10 +90,11 @@ //! Distributor serves nothing without the licence keys set with //! `.data_update_license_keys(..)`, so a build with polling on and no key //! fails with an error that says so, as it does in .NET, and a deployment of -//! the free Lite file turns polling off with `.auto_update(false)`. -//! `.file_system_watcher(false)` turns the watcher off, and -//! `pipeline.resource::()` reaches the service for a check -//! on demand. +//! the free Lite file turns automatic updates off with `.auto_update(false)`, +//! which stops the polling and the file watcher together, as the +//! specification's master switch does. `.file_system_watcher(false)` turns +//! only the watcher off, and `pipeline.resource::()` +//! reaches the service for a check on demand. //! //! ## Examples //! @@ -314,14 +315,17 @@ impl DeviceDetectionOnPremisePipelineBuilder { /// with each new file. The 51Degrees Distributor serves nothing without a /// licence key, so [`Self::build`] refuses with this on and no key set, as /// .NET does, and a deployment of the Lite file turns this off or sets - /// [`Self::data_update_license_keys`]. + /// [`Self::data_update_license_keys`]. This is the master switch for + /// automatic updates, so with it off the file-system watcher does not run + /// either. pub fn auto_update(mut self, enabled: bool) -> Self { self.auto_update = enabled; self } /// Enable or disable the file-system watcher, which refreshes the engine - /// when the data file changes on disk. On by default. + /// when the data file changes on disk. On by default, and it runs only + /// while [`Self::auto_update`] is on. pub fn file_system_watcher(mut self, enabled: bool) -> Self { self.file_system_watcher = enabled; self diff --git a/ip-intelligence-onpremise/src/builder.rs b/ip-intelligence-onpremise/src/builder.rs index aa8512e..e4d54e7 100644 --- a/ip-intelligence-onpremise/src/builder.rs +++ b/ip-intelligence-onpremise/src/builder.rs @@ -156,15 +156,17 @@ impl IpIntelligenceOnPremiseEngineBuilder { /// /// When enabled, the [`DataUpdateService`] the engine is registered with /// polls the update URL on the polling interval and applies a downloaded - /// update by calling the engine's `refresh`. + /// update by calling the engine's `refresh`. This is the master switch for + /// automatic updates, so with it off the file-system watcher does not run + /// either. pub fn auto_update(mut self, enabled: bool) -> Self { self.auto_update_enabled = enabled; self } /// Enable or disable the file-system watcher for the data file. On by - /// default. When enabled, the update service refreshes the engine when the - /// file changes on disk. + /// default. When enabled, and [`Self::auto_update`] is on, the update + /// service refreshes the engine when the file changes on disk. pub fn file_system_watcher(mut self, enabled: bool) -> Self { self.file_system_watcher_enabled = enabled; self diff --git a/ip-intelligence/src/lib.rs b/ip-intelligence/src/lib.rs index 3b4f845..1ac4bd1 100644 --- a/ip-intelligence/src/lib.rs +++ b/ip-intelligence/src/lib.rs @@ -65,10 +65,11 @@ //! Distributor serves nothing without the licence keys set with //! `.data_update_license_keys(..)`, so a build with polling on and no key //! fails with an error that says so, as it does in .NET, and a deployment -//! without a key turns polling off with `.auto_update(false)`. -//! `.file_system_watcher(false)` turns the watcher off, and -//! `pipeline.resource::()` reaches the service for a check -//! on demand. +//! without a key turns automatic updates off with `.auto_update(false)`, which +//! stops the polling and the file watcher together, as the specification's +//! master switch does. `.file_system_watcher(false)` turns only the watcher +//! off, and `pipeline.resource::()` reaches the service for +//! a check on demand. //! //! # Features //! diff --git a/ip-intelligence/src/on_premise.rs b/ip-intelligence/src/on_premise.rs index 699ca27..4fd1019 100644 --- a/ip-intelligence/src/on_premise.rs +++ b/ip-intelligence/src/on_premise.rs @@ -128,14 +128,17 @@ impl OnPremiseIpIntelligencePipelineBuilder { /// with each new file. The 51Degrees Distributor serves nothing without a /// licence key, so [`Self::build`] refuses with this on and no key set, as /// .NET does, and a deployment without one turns this off or sets - /// [`Self::data_update_license_keys`]. + /// [`Self::data_update_license_keys`]. This is the master switch for + /// automatic updates, so with it off the file-system watcher does not run + /// either. pub fn auto_update(mut self, enabled: bool) -> Self { self.auto_update = enabled; self } /// Enable or disable the file-system watcher, which refreshes the engine - /// when the data file changes on disk. On by default. + /// when the data file changes on disk. On by default, and it runs only + /// while [`Self::auto_update`] is on. pub fn file_system_watcher(mut self, enabled: bool) -> Self { self.file_system_watcher = enabled; self diff --git a/pipeline-engines/src/data_file.rs b/pipeline-engines/src/data_file.rs index 087ec11..e325b21 100644 --- a/pipeline-engines/src/data_file.rs +++ b/pipeline-engines/src/data_file.rs @@ -104,9 +104,12 @@ pub struct DataFileConfiguration { /// polling. pub data_update_url: Option, /// Whether the service should periodically poll [`Self::data_update_url`]. + /// This is the master switch for automatic updates, so with it off there + /// is no file-system watcher either. pub automatic_updates_enabled: bool, /// Whether a file-system watcher should refresh the engine when the file at - /// [`Self::data_file_path`] changes on disk. + /// [`Self::data_file_path`] changes on disk. The watcher runs only when + /// [`Self::automatic_updates_enabled`] is on as well. pub file_system_watcher_enabled: bool, /// Whether the service should poll for an update once, synchronously, when /// the data file is first registered. @@ -231,13 +234,15 @@ impl DataFileConfigurationBuilder { self } - /// Enable or disable automatic remote polling. + /// Enable or disable automatic updates, the master switch for remote + /// polling and the file-system watcher. pub fn automatic_updates_enabled(mut self, enabled: bool) -> Self { self.config.automatic_updates_enabled = enabled; self } - /// Enable or disable the file-system watcher. + /// Enable or disable the file-system watcher, which runs only while + /// automatic updates are on. pub fn file_system_watcher_enabled(mut self, enabled: bool) -> Self { self.config.file_system_watcher_enabled = enabled; self diff --git a/pipeline-engines/src/data_update_service.rs b/pipeline-engines/src/data_update_service.rs index 6c6e83d..e862b48 100644 --- a/pipeline-engines/src/data_update_service.rs +++ b/pipeline-engines/src/data_update_service.rs @@ -34,7 +34,9 @@ //! stagger many instances), sending an `If-Modified-Since` header so the //! server can answer `304 Not Modified` cheaply. //! - **Automatic update from file.** A [`notify`] file-system watcher refreshes -//! the engine when the data file changes on disk. +//! the engine when the data file changes on disk. Like polling, it runs only +//! while automatic updates are on, which the specification makes the master +//! switch for both. //! - **Programmatic update.** [`DataUpdateService::check_for_update`] triggers an //! immediate check on demand. //! @@ -141,8 +143,10 @@ impl DataUpdateService { /// - If the configuration sets `update_on_startup`, a synchronous remote /// poll runs now and its result is returned. A failed startup update is /// returned as an error so the caller can decide whether to proceed. - /// - If `file_system_watcher_enabled` and the file has a path, a watcher is - /// installed. + /// - If `automatic_updates_enabled` and `file_system_watcher_enabled` are + /// both on and the file has a path, a watcher is installed. Automatic + /// updates are the master switch for the watcher, as the specification + /// says. /// - The file is added to the scheduler's list so it is polled on its /// interval (when `automatic_updates_enabled`). /// @@ -171,8 +175,12 @@ impl DataUpdateService { startup_status = self.check_for_update_from_url(®istration)?; } - // Install the file-system watcher if requested. - if config.file_system_watcher_enabled && data_file.data_file_path().is_some() { + // Install the file-system watcher if requested and automatic updates, + // its master switch, are on. + if config.automatic_updates_enabled + && config.file_system_watcher_enabled + && data_file.data_file_path().is_some() + { self.install_watcher(®istration); } @@ -331,6 +339,10 @@ impl DataUpdateService { /// Install a file-system watcher that refreshes the engine when the data /// file changes on disk. + /// + /// The registration owns the watcher, so the watcher's callback holds the + /// registration weakly. A strong reference would keep the registration, + /// its engine and the watcher's thread alive after the service is dropped. fn install_watcher(&self, registration: &Arc) { use notify::{RecursiveMode, Watcher}; @@ -338,24 +350,29 @@ impl DataUpdateService { Some(p) => p.to_path_buf(), None => return, }; + let file_name = path.file_name().map(ToOwned::to_owned); - let reg = Arc::clone(registration); - // The service is held alive by the caller for as long as updates are - // wanted, so the closure captures only the registration (which owns the - // engine and data file) plus a clone of the HTTP client is not needed - // here, the watcher only refreshes from the file on disk. - let client = self.client.clone(); - let service = ServiceHandle { - client, - inner: Arc::clone(&self.inner), - }; + // The file as it is now is the one the engine loaded, so only a later + // change refreshes it. + if registration.data_file.last_applied_modified().is_none() { + if let Some(modified) = file_modified_time(&path) { + registration.data_file.set_last_applied_modified(modified); + } + } + let reg = Arc::downgrade(registration); let watcher = notify::recommended_watcher(move |res: notify::Result| { - if let Ok(event) = res { - if event.kind.is_modify() || event.kind.is_create() { - // Debounce duplicate events by comparing the file's - // last-modified time with the last we applied. - service.on_file_changed(®); + let Ok(event) = res else { + return; + }; + // The watch is on the folder, so skip events for other files in it. + let names_the_file = event + .paths + .iter() + .any(|changed| changed.file_name() == file_name.as_deref()); + if names_the_file && (event.kind.is_modify() || event.kind.is_create()) { + if let Some(reg) = reg.upgrade() { + ServiceHandle::on_file_changed(®); } } }); @@ -535,7 +552,7 @@ impl ServiceHandle { /// Handle a file-system-watcher event by refreshing the engine from disk if /// the file is genuinely newer than the one already applied. - fn on_file_changed(&self, registration: &Arc) { + fn on_file_changed(registration: &Registration) { let path = match registration.data_file.data_file_path() { Some(p) => p.to_path_buf(), None => return, From 509640ee08b7228c59048dd7d8cd503dc2a2d68f Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 27 Oct 2026 00:03:52 +0000 Subject: [PATCH 3/3] Say in the update example that the watcher needs automatic updates, and format --- .../src/bin/dd-onprem-update-data-file.rs | 22 +++++++++++-------- pipeline-engines/tests/file_watcher.rs | 5 +---- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/examples/device-detection-examples/src/bin/dd-onprem-update-data-file.rs b/examples/device-detection-examples/src/bin/dd-onprem-update-data-file.rs index 21f7dc8..3e716aa 100644 --- a/examples/device-detection-examples/src/bin/dd-onprem-update-data-file.rs +++ b/examples/device-detection-examples/src/bin/dd-onprem-update-data-file.rs @@ -76,9 +76,11 @@ pub fn run(options: ExampleOptions) -> Result<()> { let config = DataFileConfiguration::builder(&options.data_file) // Off so `build` does not block; see the startup section in the docs. .update_on_startup(false) - // Watch the file on disk and reload when it is replaced. + // Watch the file on disk and reload when it is replaced, while + // automatic updates are on. .file_system_watcher_enabled(true) - // Poll the distributor for a newer file on an interval. + // Poll the distributor for a newer file on an interval. Automatic + // updates are the master switch for polling and the watcher alike. .automatic_updates_enabled(options.license_key.is_some()) // A short interval purely so the example does not appear idle; a real // deployment uses around 30 minutes. @@ -164,9 +166,11 @@ fn describe_update_mechanisms(license_key: Option<&str>) { an Enterprise license key and a writable file location." ); println!( - "2. File-system watcher. Set file_system_watcher_enabled(true). The \ - service watches the data file on disk and reloads the engine when the \ - file is replaced, which is how a manually-downloaded file is picked up." + "2. File-system watcher. Set file_system_watcher_enabled(true), with \ + automatic_updates_enabled(true), the master switch for automatic \ + updates. The service watches the data file on disk and reloads the \ + engine when the file is replaced, which is how a manually-downloaded \ + file is picked up." ); println!( "3. Remote-update polling. Set automatic_updates_enabled(true), a \ @@ -266,10 +270,10 @@ mod tests { * configuration, building the engine blocks while the service polls the * 51Degrees distributor once, so the engine starts with the most recent data. * This needs an Enterprise license key and a writable file location. - * - File-system watcher. With `file_system_watcher_enabled(true)`, the service - * watches the data file on disk and reloads the engine whenever the file is - * replaced. This is how a file you download and drop into place yourself is - * picked up automatically. + * - File-system watcher. With `file_system_watcher_enabled(true)` and + * automatic updates on, the service watches the data file on disk and + * reloads the engine whenever the file is replaced. This is how a file you + * download and drop into place yourself is picked up automatically. * - Remote-update polling. With `automatic_updates_enabled(true)`, a polling * interval (around 30 minutes in production) and a randomisation window (around * 10 minutes, to stagger many instances), the service periodically polls the diff --git a/pipeline-engines/tests/file_watcher.rs b/pipeline-engines/tests/file_watcher.rs index 32a5369..17d9d9a 100644 --- a/pipeline-engines/tests/file_watcher.rs +++ b/pipeline-engines/tests/file_watcher.rs @@ -90,10 +90,7 @@ impl OnPremiseAspectEngine for CountingEngine { /// A new, empty folder for one test, holding a data file. fn folder(test: &str) -> (PathBuf, PathBuf) { - let dir = std::env::temp_dir().join(format!( - "fiftyone-watcher-{test}-{}", - std::process::id() - )); + let dir = std::env::temp_dir().join(format!("fiftyone-watcher-{test}-{}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).unwrap(); let path = dir.join("device.dat");