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/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/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, diff --git a/pipeline-engines/tests/file_watcher.rs b/pipeline-engines/tests/file_watcher.rs new file mode 100644 index 0000000..17d9d9a --- /dev/null +++ b/pipeline-engines/tests/file_watcher.rs @@ -0,0 +1,198 @@ +/* ********************************************************************* + * 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); +}