diff --git a/Cargo.lock b/Cargo.lock index 5dd5a14..2f2fb75 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -783,6 +783,7 @@ dependencies = [ "fiftyone-native", "fiftyone-pipeline-core", "fiftyone-pipeline-engines", + "flate2", ] [[package]] diff --git a/device-detection-onpremise/src/builder.rs b/device-detection-onpremise/src/builder.rs index 721957f..0e7829f 100644 --- a/device-detection-onpremise/src/builder.rs +++ b/device-detection-onpremise/src/builder.rs @@ -38,6 +38,7 @@ use fiftyone_pipeline_engines::{ }; use crate::engine::{open_manager, DeviceDetectionOnPremiseEngine}; +use crate::operational_copy::OperationalCopy; /// The download type the Distributor serves Hash data files as, sent in the /// update URL's `Type` parameter. @@ -72,6 +73,7 @@ pub struct DeviceDetectionOnPremiseEngineBuilder { data_source_tier: Option, auto_update: Option, concurrency: Option, + temp_dir: Option, } /// The optional automatic-update settings captured by the builder. @@ -95,6 +97,7 @@ impl DeviceDetectionOnPremiseEngineBuilder { data_source_tier: None, auto_update: None, concurrency: None, + temp_dir: None, } } @@ -119,6 +122,15 @@ impl DeviceDetectionOnPremiseEngineBuilder { self } + /// Set the folder for the engine's temporary files. With automatic updates + /// on, every profile but [`PerformanceProfile::InMemory`] reads a copy of + /// the data file made there, so an update can replace the data file itself. + /// Defaults to the system's temporary folder, [`std::env::temp_dir`]. + pub fn temp_dir_path(mut self, path: impl Into) -> Self { + self.temp_dir = Some(path.into()); + self + } + /// Restrict the engine to a single named property, adding to any already /// requested. With no restriction the engine loads every property the data /// file supports. @@ -219,22 +231,35 @@ impl DeviceDetectionOnPremiseEngineBuilder { }; data_file_config.check_license_keys()?; - // Open the data set with the requested property restriction. - let manager = open_manager( + // Open the data set with the requested property restriction, from a + // copy of the data file when updates can replace the file while it is + // open. + let copy = OperationalCopy::when_needed( &self.data_file_path, + self.temp_dir.as_deref(), + self.profile, + &data_file_config, + ); + let manager = open_manager( + copy.as_ref() + .map_or(self.data_file_path.as_path(), OperationalCopy::path), self.profile, &self.requested_properties, self.concurrency, )?; - let engine = Arc::new(DeviceDetectionOnPremiseEngine::from_manager( - manager, - self.profile, - self.requested_properties, - data_file_config, - self.data_source_tier, - self.concurrency, - )); + let engine = Arc::new( + DeviceDetectionOnPremiseEngine::from_manager( + manager, + copy, + self.profile, + self.requested_properties, + data_file_config, + self.data_source_tier, + self.concurrency, + ) + .with_temp_dir(self.temp_dir), + ); // Register the data file with the update service if one was supplied. if let Some(service) = service { diff --git a/device-detection-onpremise/src/engine.rs b/device-detection-onpremise/src/engine.rs index 56329da..c0cea15 100644 --- a/device-detection-onpremise/src/engine.rs +++ b/device-detection-onpremise/src/engine.rs @@ -30,7 +30,7 @@ use std::collections::HashMap; use std::ffi::CStr; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, PoisonError}; use arc_swap::ArcSwap; @@ -51,6 +51,9 @@ use fiftyone_pipeline_engines::{ EngineDeployment, OnPremiseAspectEngine, }; +use crate::operational_copy::OperationalCopy; +use crate::retained::Retained; + /// The separator used to join multiple values of a list-typed property into a /// single string when reading from the native results. The other ports use a /// vertical bar between values, which the shared device-data list accessor then @@ -143,10 +146,11 @@ const METRIC_PROPERTIES: &[(&str, PropertyValueType, &str)] = &[ /// never caches results. Only the cloud path, whose data is plain owned values, /// may use a results cache. pub struct DeviceDetectionOnPremiseEngine { - /// The loaded Hash data set. Swapped atomically on refresh. In-flight + /// The loaded Hash data set and the properties a detection reads from it, + /// swapped together atomically on refresh. In-flight /// [`fiftyone_native::dd::Results`] hold their own [`Arc`] to the manager /// they were created from, so a swap does not invalidate them. - manager: ArcSwap, + loaded: ArcSwap, /// The performance profile used to (re)load the data file, retained so a /// refresh reloads with the same profile. @@ -156,55 +160,101 @@ pub struct DeviceDetectionOnPremiseEngine { /// properties. Used when reloading the data set on refresh. requested_properties: Vec, - /// Core property metadata for [`FlowElement::properties`]. - properties: Vec, + /// The property metadata and tier of the loaded data set, for + /// [`FlowElement::properties`], [`AspectEngine::aspect_properties`] and + /// [`AspectEngine::data_source_tier`]. + metadata: Retained, - /// Aspect property metadata for [`AspectEngine::aspect_properties`]. - aspect_properties: Vec, + /// The tier the builder set, which every data set loaded reports, or + /// `None` to report the tier each data file names. + tier_override: Option, - /// The data file properties a detection reads on request, shared by every - /// detection's reader. - readable: Arc, + /// Held while a refresh loads a data set, so refreshes run one at a time. + refresh_lock: Mutex<()>, /// The evidence keys this engine reads, replaced on refresh. evidence_key_filter: HashEvidenceKeyFilter, - /// The data-source tier, for example `Lite`. - data_source_tier: String, - /// The expected concurrency the data set's handle pool was sized for, or /// `None` for the profile default. Carried so a reload re-opens the data set /// with the same pool size. concurrency: Option, + /// The folder a refresh copies the data file into, or `None` for the + /// system's temporary folder. + temp_dir: Option, + /// The single data file's run-time state, shared with the update service. data_files: Vec>, } +/// A loaded Hash data set, the properties a detection reads from it and the +/// copy of the data file it was read from, if any. +struct LoadedDataSet { + manager: Arc, + readable: Arc, + /// Held only to be dropped with the data set, and declared after the + /// manager, so it is removed after the data set that has it open is freed. + _copy: Option, +} + +/// The property metadata and tier of a loaded data set. +#[derive(PartialEq)] +struct EngineMetadata { + properties: Vec, + aspect_properties: Vec, + tier: String, +} + +/// The data set `manager` holds, read from `copy` when there is one, and its +/// metadata, reporting `tier_override` when it is set, otherwise the tier the +/// data file names. +fn describe( + manager: Arc, + copy: Option, + tier_override: &Option, +) -> (LoadedDataSet, EngineMetadata) { + // An explicit builder override wins, then the tier read from the native + // data file header (Lite, Enterprise, TAC and so on), then the safe Lite + // default when the native name is absent. + let tier = tier_override + .clone() + .or_else(|| manager.data_set_name()) + .unwrap_or_else(|| DEFAULT_DATA_SOURCE_TIER.to_owned()); + let (properties, aspect_properties) = build_property_metadata(&manager, &Some(tier.clone())); + let readable = Arc::new(ReadableProperties::new(&aspect_properties)); + ( + LoadedDataSet { + manager, + readable, + _copy: copy, + }, + EngineMetadata { + properties, + aspect_properties, + tier, + }, + ) +} + impl DeviceDetectionOnPremiseEngine { /// Build an engine from an already-loaded manager and its configuration. /// - /// Called by the builder once it has opened the data file. Derives the - /// property and evidence-key metadata from the data set, records the - /// data-file publish time from the file on disk, and assembles the engine. + /// Called by the builder once it has opened the data file, or `copy` of + /// it. Derives the property and evidence-key metadata from the data set, + /// records the data-file publish time from the file on disk, and assembles + /// the engine. pub(crate) fn from_manager( manager: Arc, + copy: Option, profile: PerformanceProfile, requested_properties: Vec, data_file_config: DataFileConfiguration, data_source_tier: Option, concurrency: Option, ) -> Self { - // Resolve the data-source tier: an explicit builder override wins, then - // the tier read from the native data file header (Lite, Enterprise, TAC - // and so on), then the safe Lite default when the native name is absent. - let data_source_tier = data_source_tier - .or_else(|| manager.data_set_name()) - .unwrap_or_else(|| DEFAULT_DATA_SOURCE_TIER.to_owned()); - let (properties, aspect_properties) = - build_property_metadata(&manager, &Some(data_source_tier.clone())); - let readable = Arc::new(ReadableProperties::new(&aspect_properties)); let evidence_key_filter = HashEvidenceKeyFilter::new(named_evidence_keys(&manager)); + let (loaded, metadata) = describe(manager, copy, &data_source_tier); let data_file = Arc::new(AspectEngineDataFile::new(data_file_config)); // Record the publish time from the data file on disk, the best signal @@ -215,19 +265,26 @@ impl DeviceDetectionOnPremiseEngine { } DeviceDetectionOnPremiseEngine { - manager: ArcSwap::from(manager), + loaded: ArcSwap::from_pointee(loaded), profile, requested_properties, - properties, - aspect_properties, - readable, + metadata: Retained::new(metadata), + tier_override: data_source_tier, + refresh_lock: Mutex::new(()), evidence_key_filter, - data_source_tier, concurrency, + temp_dir: None, data_files: vec![data_file], } } + /// Copy the data file into `temp_dir` on refresh, the folder the builder + /// copied it into, or the system's temporary folder when `None`. + pub(crate) fn with_temp_dir(mut self, temp_dir: Option) -> Self { + self.temp_dir = temp_dir; + self + } + /// The performance profile the engine loads its data file with. pub fn performance_profile(&self) -> PerformanceProfile { self.profile @@ -238,13 +295,15 @@ impl DeviceDetectionOnPremiseEngine { /// the high-entropy values keys. The engine also accepts any `query.51d_` /// or `cookie.51d_` value, which the client-side JavaScript sends back. pub fn evidence_keys(&self) -> Vec { - named_evidence_keys(&self.manager.load()) + named_evidence_keys(&self.loaded.load().manager) } /// The names of the available (required) properties in the loaded data set, /// including the match-metric pseudo-properties. pub fn available_properties(&self) -> Vec { - self.aspect_properties + self.metadata + .current() + .aspect_properties .iter() .map(|p| p.name().to_owned()) .collect() @@ -260,8 +319,8 @@ impl DeviceDetectionOnPremiseEngine { // Snapshot the current data set. The snapshot is an `Arc`, so an // in-flight detection keeps the data set alive even if `refresh` swaps a // new one in mid-flight. - let manager = self.manager.load_full(); - let mut results = manager.create_results()?; + let loaded = self.loaded.load_full(); + let mut results = loaded.manager.create_results()?; // Marshal the evidence the engine understands, preferring decoded UACH // values, into one merged evidence set, then process it. @@ -308,9 +367,8 @@ impl DeviceDetectionOnPremiseEngine { ); Ok(device.with_reader(DetectionReader { - manager, results: Mutex::new(results), - readable: Arc::clone(&self.readable), + loaded, })) } @@ -356,18 +414,35 @@ impl DeviceDetectionOnPremiseEngine { /// Reload the data set from the data file on disk and atomically swap it in. /// /// Shared by [`OnPremiseAspectEngine::refresh`] and the builder's reload - /// path. A successful reload replaces the live manager; in-flight detections - /// keep using the manager they snapshotted. + /// path. A successful reload replaces the live data set, the properties a + /// detection reads, the evidence keys and the metadata, because a new data + /// file can carry different properties or a different tier. In-flight + /// detections keep using the data set they snapshotted. A failed reload + /// changes nothing. fn reload_from_path(&self, path: &Path) -> Result<()> { + let _refreshing = self + .refresh_lock + .lock() + .unwrap_or_else(PoisonError::into_inner); + let copy = self.data_files.first().and_then(|file| { + OperationalCopy::when_needed( + path, + self.temp_dir.as_deref(), + self.profile, + file.configuration(), + ) + }); let manager = open_manager( - path, + copy.as_ref().map_or(path, OperationalCopy::path), self.profile, &self.requested_properties, self.concurrency, )?; self.evidence_key_filter .replace(named_evidence_keys(&manager)); - self.manager.store(manager); + let (loaded, metadata) = describe(manager, copy, &self.tier_override); + self.loaded.store(Arc::new(loaded)); + self.metadata.replace(metadata); // Refresh the recorded publish time from the new file. if let Some(file) = self.data_files.first() { @@ -398,13 +473,13 @@ impl FlowElement for DeviceDetectionOnPremiseEngine { } fn properties(&self) -> &[PropertyMetaData] { - &self.properties + &self.metadata.current().properties } } impl AspectEngine for DeviceDetectionOnPremiseEngine { fn data_source_tier(&self) -> &str { - &self.data_source_tier + &self.metadata.current().tier } fn deployment(&self) -> EngineDeployment { @@ -412,7 +487,7 @@ impl AspectEngine for DeviceDetectionOnPremiseEngine { } fn aspect_properties(&self) -> &[AspectPropertyMetaData] { - &self.aspect_properties + &self.metadata.current().aspect_properties } } @@ -513,27 +588,27 @@ impl ReadableProperties { /// Reads the properties of one detection from its native results when they are /// requested. struct DetectionReader { - /// The data set snapshot the results came from, which the property indexes - /// belong to. - manager: Arc, /// The native results. Reading a value writes into them, so each read - /// holds the lock. + /// holds the lock. Declared first, so they are freed before the data set + /// they came from. results: Mutex, - readable: Arc, + /// The data set snapshot the results came from, which the property indexes + /// belong to. + loaded: Arc, } impl DevicePropertyReader for DetectionReader { fn property_names(&self) -> &[String] { - &self.readable.lowercase_names + &self.loaded.readable.lowercase_names } fn position(&self, name: &str) -> Option { - self.readable.by_name.get(name).copied() + self.loaded.readable.by_name.get(name).copied() } fn read_property(&self, position: usize) -> std::result::Result, String> { - let (name, value_type) = &self.readable.properties[position]; - let Some(index) = self.manager.required_property_index(name) else { + let (name, value_type) = &self.loaded.readable.properties[position]; + let Some(index) = self.loaded.manager.required_property_index(name) else { return Ok(None); }; let results = self.results.lock().unwrap_or_else(PoisonError::into_inner); diff --git a/device-detection-onpremise/src/lib.rs b/device-detection-onpremise/src/lib.rs index 6af7e57..2a8dc9d 100644 --- a/device-detection-onpremise/src/lib.rs +++ b/device-detection-onpremise/src/lib.rs @@ -105,6 +105,8 @@ mod builder; mod engine; +mod operational_copy; +mod retained; pub use builder::{DeviceDetectionOnPremiseEngineBuilder, DATA_DOWNLOAD_TYPE}; pub use engine::DeviceDetectionOnPremiseEngine; diff --git a/device-detection-onpremise/src/operational_copy.rs b/device-detection-onpremise/src/operational_copy.rs new file mode 100644 index 0000000..fe3b7cd --- /dev/null +++ b/device-detection-onpremise/src/operational_copy.rs @@ -0,0 +1,96 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! The copy of a data file an engine reads while updates can replace the file. + +use std::ffi::OsString; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use fiftyone_native::PerformanceProfile; +use fiftyone_pipeline_engines::DataFileConfiguration; + +/// A copy of a data file that a data set is read from, removed when dropped. +/// +/// Every profile but `InMemory` keeps the data file open while the data set +/// lives, and Windows refuses to replace a file that is open. With automatic +/// updates on, the engine therefore reads a copy in its temporary folder, +/// which the specification calls the operational data copy, and an update can +/// replace the data file itself. Drop the copy after the data set read from +/// it, which closes it. +pub(crate) struct OperationalCopy { + path: PathBuf, +} + +impl OperationalCopy { + /// A copy of `data_file` in `temp_dir`, or in the system's temporary + /// folder when that is `None`, when `profile` keeps the data file open and + /// automatic updates are on, otherwise `None`. + /// + /// When the copy cannot be made the engine reads the data file itself, so + /// a full temporary folder does not stop it from building, and on Windows + /// an update then cannot replace the file while it is open. + pub(crate) fn when_needed( + data_file: &Path, + temp_dir: Option<&Path>, + profile: PerformanceProfile, + config: &DataFileConfiguration, + ) -> Option { + if profile == PerformanceProfile::InMemory || !config.automatic_updates_enabled { + return None; + } + let temp_dir = temp_dir.map_or_else(std::env::temp_dir, Path::to_path_buf); + OperationalCopy::of(data_file, &temp_dir).ok() + } + + /// Copy `data_file` into `temp_dir`, under a name no other copy in this + /// process has. + fn of(data_file: &Path, temp_dir: &Path) -> std::io::Result { + static NEXT: AtomicU64 = AtomicU64::new(0); + let mut name = data_file + .file_name() + .map(OsString::from) + .unwrap_or_default(); + name.push(format!( + ".{}.{}.operational", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + // Made before the copy, so a copy that fails part way is removed. + let copy = OperationalCopy { + path: temp_dir.join(name), + }; + std::fs::copy(data_file, ©.path)?; + Ok(copy) + } + + /// Where the copy is. + pub(crate) fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for OperationalCopy { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} diff --git a/device-detection-onpremise/src/retained.rs b/device-detection-onpremise/src/retained.rs new file mode 100644 index 0000000..17e5064 --- /dev/null +++ b/device-detection-onpremise/src/retained.rs @@ -0,0 +1,87 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! Values a refresh replaces that the engine lends by reference. + +use std::sync::OnceLock; + +/// Every distinct value kept so far, oldest first, of which the newest is the +/// current one. +/// +/// The pipeline traits lend an engine's property lists and tier by reference +/// for as long as the engine lives, so a value a refresh replaces cannot be +/// freed while the engine lives. Each is kept instead. A refresh adds a value +/// only when it differs from the newest, which happens when a data file with +/// a different property set or tier is loaded. +pub(crate) struct Retained { + value: T, + next: OnceLock>>, +} + +impl Retained { + pub(crate) fn new(value: T) -> Self { + Retained { + value, + next: OnceLock::new(), + } + } + + fn newest_link(&self) -> &Retained { + let mut link = self; + while let Some(next) = link.next.get() { + link = next; + } + link + } + + /// The newest value. + pub(crate) fn current(&self) -> &T { + &self.newest_link().value + } + + /// Make `value` the current one, unless it equals the current one. The + /// caller serialises calls, so two never add at once. + pub(crate) fn replace(&self, value: T) { + let newest = self.newest_link(); + if newest.value != value { + let _ = newest.next.set(Box::new(Retained::new(value))); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_newest_distinct_value_is_current_and_older_ones_stay_lent() { + let retained = Retained::new(String::from("Lite")); + let lent = retained.current(); + retained.replace(String::from("Lite")); + assert!(retained.next.get().is_none(), "an equal value is not kept"); + retained.replace(String::from("Enterprise")); + assert_eq!(retained.current(), "Enterprise"); + assert_eq!(lent, "Lite"); + retained.replace(String::from("Lite")); + assert_eq!(retained.current(), "Lite"); + } +} diff --git a/device-detection-onpremise/tests/operational_copy.rs b/device-detection-onpremise/tests/operational_copy.rs new file mode 100644 index 0000000..e5ff92c --- /dev/null +++ b/device-detection-onpremise/tests/operational_copy.rs @@ -0,0 +1,254 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! A data update replaces the data file of a running engine under every +//! profile, and the copy the engine reads leaves with it. The tests need the +//! Lite file `51DEGREES_DD_PATH` names, and skip with a note without it. + +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use fiftyone_device_detection_onpremise::{ + DeviceDetectionOnPremiseEngine, DeviceDetectionOnPremiseEngineBuilder, +}; +use fiftyone_device_detection_shared::{DeviceData, DEVICE_DATA_KEY}; +use fiftyone_native::PerformanceProfile; +use fiftyone_pipeline_core::{Evidence, FlowData, FlowElement, Pipeline}; +use fiftyone_pipeline_engines::{ + AutoUpdateStatus, DataFileConfiguration, DataUpdateService, OnPremiseAspectEngine, +}; + +/// A mobile Safari on iPhone User-Agent. +const MOBILE_USER_AGENT: &str = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) \ + AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"; + +fn lite_data_file() -> Option { + std::env::var("51DEGREES_DD_PATH") + .ok() + .map(PathBuf::from) + .filter(|path| path.is_file()) +} + +/// A new, empty folder holding a copy of `lite` named `name`. +fn folder_with(lite: &Path, test: &str, name: &str) -> (PathBuf, PathBuf) { + let dir = std::env::temp_dir().join(format!("fiftyone-dd-copy-{test}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join(name); + std::fs::copy(lite, &path).unwrap(); + (dir, path) +} + +fn files_in(dir: &Path) -> Vec { + let mut names: Vec = std::fs::read_dir(dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + names +} + +/// Serve `body` to one request, and return the URL. +fn serve_once(body: Vec) -> (String, std::thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}/datafile", listener.local_addr().unwrap()); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0u8; 4096]; + let _ = stream.read(&mut request); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(&body); + }); + (url, server) +} + +/// The data file configuration of automatic updates from `url`, which the +/// test polls itself. +fn updating_config(path: &Path, url: String) -> DataFileConfiguration { + DataFileConfiguration::builder(path) + .data_update_url(url) + .automatic_updates_enabled(true) + .polling_interval_seconds(24 * 60 * 60) + .max_randomisation_seconds(0) + .file_system_watcher_enabled(false) + .decompress_content(false) + .verify_md5(false) + .license_key_required_for_updates(false) + .build() +} + +/// An engine with automatic updates on from `url`, which the test polls +/// itself. +fn updating_engine( + path: &Path, + profile: PerformanceProfile, + url: String, + service: &Arc, +) -> Arc { + DeviceDetectionOnPremiseEngineBuilder::new(path) + .performance_profile(profile) + .auto_update(updating_config(path, url), Some(Arc::clone(service))) + .build() + .expect("the data file loads") +} + +fn detect(engine: &Arc) -> FlowData { + let element: Arc = engine.clone(); + let pipeline = Pipeline::builder().add_element(element).build().unwrap(); + let mut data = pipeline.create_flow_data_with( + Evidence::builder() + .add("header.user-agent", MOBILE_USER_AGENT) + .build(), + ); + data.process().unwrap(); + data +} + +fn is_mobile(data: &FlowData) -> Option { + data.get(DEVICE_DATA_KEY) + .unwrap() + .is_mobile() + .value() + .ok() + .copied() +} + +/// With automatic updates on, an update replaces the data file of a running +/// engine under every profile, including the ones that read the data file +/// while the data set lives, and the engine moves to the new file. +#[test] +fn an_update_replaces_the_data_file_under_every_profile() { + let Some(lite) = lite_data_file() else { + eprintln!("skipping operational copies: set 51DEGREES_DD_PATH to the Lite file"); + return; + }; + for profile in [ + PerformanceProfile::Default, + PerformanceProfile::Balanced, + PerformanceProfile::LowMemory, + PerformanceProfile::HighPerformance, + PerformanceProfile::InMemory, + ] { + let (dir, path) = folder_with(&lite, profile.as_str(), "51Degrees.hash"); + let (url, server) = serve_once(std::fs::read(&lite).unwrap()); + let service = Arc::new(DataUpdateService::new()); + let engine = updating_engine(&path, profile, url, &service); + let as_engine: Arc = engine.clone(); + + let status = service.check_for_update(&as_engine, None); + server.join().unwrap(); + + assert_eq!( + status.map_err(|error| error.to_string()), + Ok(AutoUpdateStatus::Success), + "{profile:?}" + ); + assert_eq!(is_mobile(&detect(&engine)), Some(true), "{profile:?}"); + drop(as_engine); + drop(engine); + drop(service); + assert_eq!(files_in(&dir), ["51Degrees.hash"], "{profile:?}"); + let _ = std::fs::remove_dir_all(&dir); + } +} + +/// The engine reads one copy at a time, and a copy goes once nothing reads +/// the data set it holds, even when a detection outlives a refresh. None is +/// left after the engine is dropped. +#[test] +fn each_copy_goes_with_the_data_set_read_from_it() { + let Some(lite) = lite_data_file() else { + eprintln!("skipping operational copies: set 51DEGREES_DD_PATH to the Lite file"); + return; + }; + let name = format!("copied-{}.hash", std::process::id()); + let (dir, path) = folder_with(&lite, "copies", &name); + let copies = || { + files_in(&std::env::temp_dir()) + .into_iter() + .filter(|file| file.starts_with(&name) && file.ends_with(".operational")) + .count() + }; + let service = Arc::new(DataUpdateService::new()); + let engine = updating_engine( + &path, + PerformanceProfile::Default, + String::from("http://127.0.0.1:9/unused"), + &service, + ); + assert_eq!(copies(), 1, "one copy for the data set loaded"); + + let early = detect(&engine); + for _ in 0..3 { + engine.refresh(None).expect("the data file reloads"); + } + assert_eq!( + copies(), + 2, + "the copy the early detection reads, and the newest" + ); + assert_eq!(is_mobile(&early), Some(true)); + drop(early); + assert_eq!(copies(), 1, "the early detection's copy goes with it"); + + drop(engine); + drop(service); + assert_eq!(copies(), 0, "no copy outlives the engine"); + let _ = std::fs::remove_dir_all(&dir); +} + +/// The copy is made in the folder the builder names, on build and on each +/// refresh, and none is left there once the engine is dropped. +#[test] +fn the_copy_goes_in_the_folder_the_builder_names() { + let Some(lite) = lite_data_file() else { + eprintln!("skipping operational copies: set 51DEGREES_DD_PATH to the Lite file"); + return; + }; + let (dir, path) = folder_with(&lite, "named", "51Degrees.hash"); + let temp = dir.join("temp"); + std::fs::create_dir_all(&temp).unwrap(); + let copies = || files_in(&temp).len(); + + let engine = DeviceDetectionOnPremiseEngineBuilder::new(&path) + .temp_dir_path(&temp) + .auto_update( + updating_config(&path, String::from("http://127.0.0.1:9/unused")), + None, + ) + .build() + .expect("the data file loads"); + assert_eq!(copies(), 1, "the copy is in the named folder"); + + engine.refresh(None).expect("the data file reloads"); + assert_eq!(copies(), 1, "the refresh copies into the named folder too"); + + drop(engine); + assert_eq!(copies(), 0, "no copy outlives the engine"); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/device-detection-onpremise/tests/refresh.rs b/device-detection-onpremise/tests/refresh.rs new file mode 100644 index 0000000..c7fea76 --- /dev/null +++ b/device-detection-onpremise/tests/refresh.rs @@ -0,0 +1,127 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! A refresh to a data file with other properties or another tier leaves the +//! engine as an engine built on that file would be. +//! +//! The test needs the Lite file `51DEGREES_DD_PATH` names and a data file +//! with more properties beside it, `Enterprise-HashV41.hash` or +//! `TAC-HashV41.hash`, and skips with a note when either is missing. + +use std::path::PathBuf; +use std::sync::Arc; + +use fiftyone_device_detection_onpremise::{ + DeviceDetectionOnPremiseEngine, DeviceDetectionOnPremiseEngineBuilder, +}; +use fiftyone_device_detection_shared::DEVICE_DATA_KEY; +use fiftyone_native::PerformanceProfile; +use fiftyone_pipeline_core::{ElementData, Evidence, FlowElement, Pipeline}; +use fiftyone_pipeline_engines::{AspectEngine, OnPremiseAspectEngine}; + +/// A mobile Safari on iPhone User-Agent. +const MOBILE_USER_AGENT: &str = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) \ + AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"; + +/// A property the Lite file does not carry and the richer files do. +const RICHER_PROPERTY: &str = "HardwareModel"; + +/// The Lite file and a file with more properties beside it. +fn data_files() -> Option<(PathBuf, PathBuf)> { + let lite = PathBuf::from(std::env::var("51DEGREES_DD_PATH").ok()?); + let folder = lite.parent()?; + let richer = ["Enterprise-HashV41.hash", "TAC-HashV41.hash"] + .into_iter() + .map(|name| folder.join(name)) + .find(|path| path.is_file())?; + lite.is_file().then_some((lite, richer)) +} + +/// An engine that holds the whole data set in memory, so its data file can +/// be replaced on every platform. +fn engine(path: &PathBuf) -> Arc { + DeviceDetectionOnPremiseEngineBuilder::new(path) + .performance_profile(PerformanceProfile::InMemory) + .build() + .expect("the data file loads") +} + +/// The value of [`RICHER_PROPERTY`] for a mobile User-Agent, or the reason +/// there is none. +fn richer_value(engine: &Arc) -> Result { + let element: Arc = engine.clone(); + let pipeline = Pipeline::builder().add_element(element).build().unwrap(); + let mut data = pipeline.create_flow_data_with( + Evidence::builder() + .add("header.user-agent", MOBILE_USER_AGENT) + .build(), + ); + data.process().unwrap(); + let device = data.get(DEVICE_DATA_KEY).unwrap(); + device + .get(RICHER_PROPERTY) + .map(|value| format!("{value:?}")) + .map_err(|no_value| no_value.message) +} + +#[test] +fn a_refresh_reads_and_reports_the_properties_of_the_new_data_file() { + let Some((lite, richer)) = data_files() else { + eprintln!( + "skipping the refresh test: set 51DEGREES_DD_PATH to the Lite file, with \ + Enterprise-HashV41.hash or TAC-HashV41.hash beside it" + ); + return; + }; + let dir = std::env::temp_dir().join(format!("fiftyone-dd-refresh-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("51Degrees.hash"); + std::fs::copy(&lite, &path).unwrap(); + + let refreshed = engine(&path); + assert!( + richer_value(&refreshed).is_err(), + "the Lite file carries {RICHER_PROPERTY}" + ); + std::fs::copy(&richer, &path).unwrap(); + refreshed.refresh(None).expect("the richer file loads"); + let built = engine(&path); + + assert_eq!(refreshed.data_source_tier(), built.data_source_tier()); + assert_eq!( + refreshed.available_properties(), + built.available_properties() + ); + assert_eq!(refreshed.properties(), built.properties()); + assert_eq!(refreshed.aspect_properties(), built.aspect_properties()); + let value = richer_value(&refreshed); + assert!( + value.is_ok(), + "{RICHER_PROPERTY} after the refresh: {value:?}" + ); + assert_eq!(value, richer_value(&built)); + + drop(refreshed); + drop(built); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/ip-intelligence-onpremise/Cargo.toml b/ip-intelligence-onpremise/Cargo.toml index 6413b20..df7d952 100644 --- a/ip-intelligence-onpremise/Cargo.toml +++ b/ip-intelligence-onpremise/Cargo.toml @@ -21,6 +21,11 @@ fiftyone-native = { version = "4.5.2", path = "../native", default-features = fa arc-swap.workspace = true chrono.workspace = true +[dev-dependencies] +# Compresses a data file the way the Distributor serves it, for the update +# tests. +flate2.workspace = true + [package.metadata.docs.rs] # Build the documentation on docs.rs with every feature enabled, so the # cloud, on-premise and optional paths are all documented. diff --git a/ip-intelligence-onpremise/src/builder.rs b/ip-intelligence-onpremise/src/builder.rs index aa8512e..4e9e0ba 100644 --- a/ip-intelligence-onpremise/src/builder.rs +++ b/ip-intelligence-onpremise/src/builder.rs @@ -30,7 +30,6 @@ use std::path::PathBuf; use std::sync::Arc; -use chrono::{TimeZone, Utc}; use fiftyone_native::ipi::Manager; use fiftyone_native::PerformanceProfile; use fiftyone_pipeline_core::{Error, Result}; @@ -39,7 +38,8 @@ use fiftyone_pipeline_engines::{ OnPremiseAspectEngine, DISTRIBUTOR_URL, }; -use crate::engine::{IpIntelligenceOnPremiseEngine, DEFAULT_DATA_SOURCE_TIER}; +use crate::engine::{record_published, IpIntelligenceOnPremiseEngine}; +use crate::operational_copy::OperationalCopy; /// The download type the Distributor serves IP Intelligence data files as, /// sent in the update URL's `Type` parameter. @@ -85,6 +85,7 @@ pub struct IpIntelligenceOnPremiseEngineBuilder { polling_interval_seconds: Option, max_randomisation_seconds: Option, data_update_service: Option>, + temp_dir: Option, } impl IpIntelligenceOnPremiseEngineBuilder { @@ -111,6 +112,7 @@ impl IpIntelligenceOnPremiseEngineBuilder { polling_interval_seconds: None, max_randomisation_seconds: None, data_update_service: None, + temp_dir: None, } } @@ -120,6 +122,15 @@ impl IpIntelligenceOnPremiseEngineBuilder { self } + /// Set the folder for the engine's temporary files. With automatic updates + /// on, every profile but [`PerformanceProfile::InMemory`] reads a copy of + /// the data file made there, so an update can replace the data file itself. + /// Defaults to the system's temporary folder, [`std::env::temp_dir`]. + pub fn temp_dir_path(mut self, path: impl Into) -> Self { + self.temp_dir = Some(path.into()); + self + } + /// Restrict the engine to a single named property, adding to any already /// requested. With no properties requested the engine populates every /// property the shared model surfaces. @@ -144,7 +155,7 @@ impl IpIntelligenceOnPremiseEngineBuilder { /// /// By default the engine reads the tier from the data file header (for /// example `Lite`, `Enterprise` or `TAC`), falling back to - /// [`DEFAULT_DATA_SOURCE_TIER`] only when the header cannot be read. This + /// [`crate::DEFAULT_DATA_SOURCE_TIER`] only when the header cannot be read. This /// override forces a specific tier, which an application can use when it /// knows better than the file metadata. pub fn data_source_tier(mut self, tier: impl Into) -> Self { @@ -277,38 +288,33 @@ impl IpIntelligenceOnPremiseEngineBuilder { ) }; - let manager = Manager::open_with_properties( + // Open from a copy of the data file when updates can replace the file + // while it is open. + let copy = OperationalCopy::when_needed( &self.data_file_path, + self.temp_dir.as_deref(), + self.profile, + &configuration, + ); + let manager = Manager::open_with_properties( + copy.as_ref() + .map_or(self.data_file_path.as_path(), OperationalCopy::path), self.profile, property_refs.as_deref(), )?; - // Report the tier from an explicit override first, then the data file - // header, then the fixed default when the header carries no name. - let data_source_tier = self - .data_source_tier - .clone() - .or_else(|| manager.data_set_name()) - .unwrap_or_else(|| DEFAULT_DATA_SOURCE_TIER.to_owned()); - let data_file = Arc::new(AspectEngineDataFile::new(configuration)); - - // Record the data file's embedded publish date, read from its header, so - // the metadata reports it and the update service can use it for the - // `If-Modified-Since` decision. - if let Some((year, month, day)) = manager.data_set_published() { - if let Some(published) = Utc.with_ymd_and_hms(year, month, day, 0, 0, 0).single() { - data_file.set_data_published(published); - } - } + record_published(&manager, &data_file); Ok(IpIntelligenceOnPremiseEngine::from_parts( manager, + copy, self.profile, self.requested_properties, - data_source_tier, + self.data_source_tier, data_file, - )) + ) + .with_temp_dir(self.temp_dir)) } /// Build the data-file configuration that captures the update options. The diff --git a/ip-intelligence-onpremise/src/engine.rs b/ip-intelligence-onpremise/src/engine.rs index db39f5a..e412984 100644 --- a/ip-intelligence-onpremise/src/engine.rs +++ b/ip-intelligence-onpremise/src/engine.rs @@ -30,9 +30,11 @@ use std::collections::HashMap; use std::os::raw::c_int; +use std::path::PathBuf; use std::sync::{Arc, Mutex, PoisonError}; use arc_swap::ArcSwap; +use chrono::{TimeZone, Utc}; use fiftyone_ip_intelligence_shared::{ declared_property_value_type, generated_property_names, IpIntelligenceDataBase, @@ -50,6 +52,8 @@ use fiftyone_pipeline_engines::{ }; use crate::builder::IpIntelligenceOnPremiseEngineBuilder; +use crate::operational_copy::OperationalCopy; +use crate::retained::Retained; /// The default data-source tier reported by the engine. /// @@ -104,15 +108,57 @@ pub struct IpIntelligenceOnPremiseEngine { evidence_key_filter: DataFileEvidenceKeys, /// Core property metadata for the populated properties. properties: Vec, - /// Aspect property metadata for the populated properties. - aspect_properties: Vec, - /// The data-source tier reported by the engine. - data_source_tier: String, + /// The tier of the loaded data set and the aspect property metadata, which + /// names that tier, for [`AspectEngine::data_source_tier`] and + /// [`AspectEngine::aspect_properties`]. + metadata: Retained, + /// The tier the builder set, which every data set loaded reports, or + /// `None` to report the tier each data file names. + tier_override: Option, + /// Held while a refresh loads a data set, so refreshes run one at a time. + refresh_lock: Mutex<()>, + /// The folder a refresh copies the data file into, or `None` for the + /// system's temporary folder. + temp_dir: Option, /// The single data file this engine reads, as shared run-time state for the /// data update service. data_files: Vec>, } +/// The tier of a loaded data set and the aspect property metadata that names +/// it. +#[derive(PartialEq)] +struct TierMetadata { + tier: String, + aspect_properties: Vec, +} + +impl TierMetadata { + /// The metadata of the data set `manager` holds, reporting `tier_override` + /// when it is set, otherwise the tier the data file names, then + /// [`DEFAULT_DATA_SOURCE_TIER`] when it names none. + fn new(manager: &Manager, tier_override: &Option, populated: &[String]) -> Self { + let tier = tier_override + .clone() + .or_else(|| manager.data_set_name()) + .unwrap_or_else(|| DEFAULT_DATA_SOURCE_TIER.to_owned()); + TierMetadata { + aspect_properties: build_aspect_metadata(populated, &tier), + tier, + } + } +} + +/// Record the publish date in `manager`'s data file header on `data_file`, +/// for the metadata and the update service's `If-Modified-Since` header. +pub(crate) fn record_published(manager: &Manager, data_file: &AspectEngineDataFile) { + if let Some((year, month, day)) = manager.data_set_published() { + if let Some(published) = Utc.with_ymd_and_hms(year, month, day, 0, 0, 0).single() { + data_file.set_data_published(published); + } + } +} + impl IpIntelligenceOnPremiseEngine { /// Start building an engine for the `.ipi` data file at `data_file_path`. /// @@ -127,29 +173,43 @@ impl IpIntelligenceOnPremiseEngine { /// it has opened the data file and derived the metadata. pub(crate) fn from_parts( manager: Arc, + copy: Option, profile: PerformanceProfile, requested_properties: Vec, - data_source_tier: String, + tier_override: Option, data_file: Arc, ) -> Self { let populated_property_names = populated_property_names(&requested_properties); let properties = build_core_metadata(&populated_property_names); - let aspect_properties = build_aspect_metadata(&populated_property_names, &data_source_tier); + let metadata = TierMetadata::new(&manager, &tier_override, &populated_property_names); IpIntelligenceOnPremiseEngine { base: AspectEngineBase::new(), evidence_key_filter: DataFileEvidenceKeys::new(&manager), - loaded: ArcSwap::from_pointee(LoadedDataSet::new(manager, &populated_property_names)), + loaded: ArcSwap::from_pointee(LoadedDataSet::new( + manager, + copy, + &populated_property_names, + )), profile, requested_properties, populated_property_names, properties, - aspect_properties, - data_source_tier, + metadata: Retained::new(metadata), + tier_override, + refresh_lock: Mutex::new(()), + temp_dir: None, data_files: vec![data_file], } } + /// Copy the data file into `temp_dir` on refresh, the folder the builder + /// copied it into, or the system's temporary folder when `None`. + pub(crate) fn with_temp_dir(mut self, temp_dir: Option) -> Self { + self.temp_dir = temp_dir; + self + } + /// The evidence keys the loaded data file defines, in the order the native /// lookup tries them, lowercased. /// @@ -206,7 +266,7 @@ impl IpIntelligenceOnPremiseEngine { Ok( IpIntelligenceDataBase::new(IP_DATA_KEY_NAME).with_reader(LookupReader { results: Mutex::new(results), - readable: Arc::clone(&loaded.readable), + loaded, }), ) } @@ -214,8 +274,14 @@ impl IpIntelligenceOnPremiseEngine { /// Reopen the data file from disk and atomically swap it in. /// /// Shared by [`OnPremiseAspectEngine::refresh`]. A concurrent `process` call - /// keeps using the old data set until this swap completes. + /// keeps using the old data set until this swap completes. The evidence + /// keys, the tier and the publish date follow the new data file, which can + /// differ from the old one. A failed reload changes nothing. fn reload_from_disk(&self) -> Result<()> { + let _refreshing = self + .refresh_lock + .lock() + .unwrap_or_else(PoisonError::into_inner); let path = self .data_files .first() @@ -236,13 +302,35 @@ impl IpIntelligenceOnPremiseEngine { .collect(), ) }; - let manager = Manager::open_with_properties(&path, self.profile, property_refs.as_deref())?; + let copy = self.data_files.first().and_then(|file| { + OperationalCopy::when_needed( + &path, + self.temp_dir.as_deref(), + self.profile, + file.configuration(), + ) + }); + let manager = Manager::open_with_properties( + copy.as_ref().map_or(path.as_path(), OperationalCopy::path), + self.profile, + property_refs.as_deref(), + )?; let evidence_keys = DataFileEvidenceKeys::whitelist(&manager); + let metadata = TierMetadata::new( + &manager, + &self.tier_override, + &self.populated_property_names, + ); + if let Some(data_file) = self.data_files.first() { + record_published(&manager, data_file); + } self.loaded.store(Arc::new(LoadedDataSet::new( manager, + copy, &self.populated_property_names, ))); self.evidence_key_filter.keys.store(Arc::new(evidence_keys)); + self.metadata.replace(metadata); Ok(()) } } @@ -352,16 +440,24 @@ fn build_aspect_metadata(names: &[String], tier: &str) -> Vec, readable: Arc, + /// Held only to be dropped with the data set, and declared after the + /// manager, so it is removed after the data set that has it open is freed. + _copy: Option, } impl LoadedDataSet { - fn new(manager: Arc, populated: &[String]) -> Self { + fn new(manager: Arc, copy: Option, populated: &[String]) -> Self { let readable = Arc::new(ReadableNames::new(populated, &manager)); - LoadedDataSet { manager, readable } + LoadedDataSet { + manager, + readable, + _copy: copy, + } } } @@ -407,19 +503,20 @@ impl ReadableNames { /// requested. struct LookupReader { /// The native results. Reading a value writes into them, so each read - /// holds the lock. + /// holds the lock. Declared first, so they are freed before the data set + /// they came from. results: Mutex, - /// The properties of the data set the results came from. - readable: Arc, + /// The data set the results came from, with its properties. + loaded: Arc, } impl IpIntelligencePropertyReader for LookupReader { fn property_names(&self) -> &[String] { - &self.readable.lowercase_names + &self.loaded.readable.lowercase_names } fn position(&self, name: &str) -> Option { - self.readable.by_name.get(name).copied() + self.loaded.readable.by_name.get(name).copied() } fn read_property( @@ -427,8 +524,8 @@ impl IpIntelligencePropertyReader for LookupReader { position: usize, data: &mut IpIntelligenceDataBase, ) -> std::result::Result<(), String> { - let name = &self.readable.names[position]; - let index = self.readable.indexes[position]; + let name = &self.loaded.readable.names[position]; + let index = self.loaded.readable.indexes[position]; let results = self.results.lock().unwrap_or_else(PoisonError::into_inner); populate_property(data, &results, name, index).map_err(|error| error.to_string()) } @@ -552,7 +649,7 @@ impl FlowElement for IpIntelligenceOnPremiseEngine { impl AspectEngine for IpIntelligenceOnPremiseEngine { fn data_source_tier(&self) -> &str { - &self.data_source_tier + &self.metadata.current().tier } fn deployment(&self) -> EngineDeployment { @@ -560,7 +657,7 @@ impl AspectEngine for IpIntelligenceOnPremiseEngine { } fn aspect_properties(&self) -> &[AspectPropertyMetaData] { - &self.aspect_properties + &self.metadata.current().aspect_properties } } diff --git a/ip-intelligence-onpremise/src/lib.rs b/ip-intelligence-onpremise/src/lib.rs index 1151005..821fc3f 100644 --- a/ip-intelligence-onpremise/src/lib.rs +++ b/ip-intelligence-onpremise/src/lib.rs @@ -105,6 +105,8 @@ mod builder; mod engine; +mod operational_copy; +mod retained; pub use builder::{IpIntelligenceOnPremiseEngineBuilder, DATA_DOWNLOAD_TYPE}; #[allow(deprecated)] diff --git a/ip-intelligence-onpremise/src/operational_copy.rs b/ip-intelligence-onpremise/src/operational_copy.rs new file mode 100644 index 0000000..fe3b7cd --- /dev/null +++ b/ip-intelligence-onpremise/src/operational_copy.rs @@ -0,0 +1,96 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! The copy of a data file an engine reads while updates can replace the file. + +use std::ffi::OsString; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use fiftyone_native::PerformanceProfile; +use fiftyone_pipeline_engines::DataFileConfiguration; + +/// A copy of a data file that a data set is read from, removed when dropped. +/// +/// Every profile but `InMemory` keeps the data file open while the data set +/// lives, and Windows refuses to replace a file that is open. With automatic +/// updates on, the engine therefore reads a copy in its temporary folder, +/// which the specification calls the operational data copy, and an update can +/// replace the data file itself. Drop the copy after the data set read from +/// it, which closes it. +pub(crate) struct OperationalCopy { + path: PathBuf, +} + +impl OperationalCopy { + /// A copy of `data_file` in `temp_dir`, or in the system's temporary + /// folder when that is `None`, when `profile` keeps the data file open and + /// automatic updates are on, otherwise `None`. + /// + /// When the copy cannot be made the engine reads the data file itself, so + /// a full temporary folder does not stop it from building, and on Windows + /// an update then cannot replace the file while it is open. + pub(crate) fn when_needed( + data_file: &Path, + temp_dir: Option<&Path>, + profile: PerformanceProfile, + config: &DataFileConfiguration, + ) -> Option { + if profile == PerformanceProfile::InMemory || !config.automatic_updates_enabled { + return None; + } + let temp_dir = temp_dir.map_or_else(std::env::temp_dir, Path::to_path_buf); + OperationalCopy::of(data_file, &temp_dir).ok() + } + + /// Copy `data_file` into `temp_dir`, under a name no other copy in this + /// process has. + fn of(data_file: &Path, temp_dir: &Path) -> std::io::Result { + static NEXT: AtomicU64 = AtomicU64::new(0); + let mut name = data_file + .file_name() + .map(OsString::from) + .unwrap_or_default(); + name.push(format!( + ".{}.{}.operational", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + // Made before the copy, so a copy that fails part way is removed. + let copy = OperationalCopy { + path: temp_dir.join(name), + }; + std::fs::copy(data_file, ©.path)?; + Ok(copy) + } + + /// Where the copy is. + pub(crate) fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for OperationalCopy { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} diff --git a/ip-intelligence-onpremise/src/retained.rs b/ip-intelligence-onpremise/src/retained.rs new file mode 100644 index 0000000..17e5064 --- /dev/null +++ b/ip-intelligence-onpremise/src/retained.rs @@ -0,0 +1,87 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! Values a refresh replaces that the engine lends by reference. + +use std::sync::OnceLock; + +/// Every distinct value kept so far, oldest first, of which the newest is the +/// current one. +/// +/// The pipeline traits lend an engine's property lists and tier by reference +/// for as long as the engine lives, so a value a refresh replaces cannot be +/// freed while the engine lives. Each is kept instead. A refresh adds a value +/// only when it differs from the newest, which happens when a data file with +/// a different property set or tier is loaded. +pub(crate) struct Retained { + value: T, + next: OnceLock>>, +} + +impl Retained { + pub(crate) fn new(value: T) -> Self { + Retained { + value, + next: OnceLock::new(), + } + } + + fn newest_link(&self) -> &Retained { + let mut link = self; + while let Some(next) = link.next.get() { + link = next; + } + link + } + + /// The newest value. + pub(crate) fn current(&self) -> &T { + &self.newest_link().value + } + + /// Make `value` the current one, unless it equals the current one. The + /// caller serialises calls, so two never add at once. + pub(crate) fn replace(&self, value: T) { + let newest = self.newest_link(); + if newest.value != value { + let _ = newest.next.set(Box::new(Retained::new(value))); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_newest_distinct_value_is_current_and_older_ones_stay_lent() { + let retained = Retained::new(String::from("Lite")); + let lent = retained.current(); + retained.replace(String::from("Lite")); + assert!(retained.next.get().is_none(), "an equal value is not kept"); + retained.replace(String::from("Enterprise")); + assert_eq!(retained.current(), "Enterprise"); + assert_eq!(lent, "Lite"); + retained.replace(String::from("Lite")); + assert_eq!(retained.current(), "Lite"); + } +} diff --git a/ip-intelligence-onpremise/tests/operational_copy.rs b/ip-intelligence-onpremise/tests/operational_copy.rs new file mode 100644 index 0000000..7e5766a --- /dev/null +++ b/ip-intelligence-onpremise/tests/operational_copy.rs @@ -0,0 +1,185 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! A data update replaces the data file of a running engine under every +//! profile, and the copy the engine reads leaves with it. The test needs the +//! ASN file `51DEGREES_IPI_PATH` names, and skips with a note without it. + +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use fiftyone_ip_intelligence_onpremise::{IpIntelligenceOnPremiseEngine, IP_DATA_KEY}; +use fiftyone_native::PerformanceProfile; +use fiftyone_pipeline_core::{Evidence, FlowElement, Pipeline}; +use fiftyone_pipeline_engines::{AutoUpdateStatus, DataUpdateService, OnPremiseAspectEngine}; + +fn files_in(dir: &Path) -> Vec { + let mut names: Vec = std::fs::read_dir(dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + names +} + +/// Serve `body` to one request, and return the URL. +fn serve_once(body: Vec) -> (String, std::thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}/datafile", listener.local_addr().unwrap()); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0u8; 4096]; + let _ = stream.read(&mut request); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(&body); + }); + (url, server) +} + +/// With automatic updates on, an update replaces the data file of a running +/// engine under every profile, and the engine moves to the new file. No copy +/// is left once the engine is dropped. +#[test] +fn an_update_replaces_the_data_file_under_every_profile() { + let Some(asn) = std::env::var("51DEGREES_IPI_PATH") + .ok() + .map(PathBuf::from) + .filter(|path| path.is_file()) + else { + eprintln!("skipping operational copies: set 51DEGREES_IPI_PATH to the ASN file"); + return; + }; + for profile in [ + PerformanceProfile::Default, + PerformanceProfile::Balanced, + PerformanceProfile::LowMemory, + PerformanceProfile::HighPerformance, + PerformanceProfile::InMemory, + ] { + let dir = std::env::temp_dir().join(format!( + "fiftyone-ipi-copy-{}-{}", + profile.as_str(), + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let name = format!("copied-{}-{}.ipi", profile.as_str(), std::process::id()); + let path = dir.join(&name); + std::fs::copy(&asn, &path).unwrap(); + // Compressed, as the Distributor serves a data file. + let compressed = { + let mut encoder = + flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast()); + encoder.write_all(&std::fs::read(&asn).unwrap()).unwrap(); + encoder.finish().unwrap() + }; + let (url, server) = serve_once(compressed); + let service = Arc::new(DataUpdateService::new()); + let engine = IpIntelligenceOnPremiseEngine::builder(&path) + .performance_profile(profile) + .properties(["Asn"]) + .auto_update(true) + .file_system_watcher(false) + .polling_interval_seconds(24 * 60 * 60) + .max_randomisation_seconds(0) + .data_update_url(url) + .data_update_license_keys(["KEY"]) + .data_update_service(Arc::clone(&service)) + .build_shared() + .expect("the data file loads"); + let as_engine: Arc = engine.clone(); + + let status = service.check_for_update(&as_engine, None); + server.join().unwrap(); + + assert_eq!( + status.map_err(|error| error.to_string()), + Ok(AutoUpdateStatus::Success), + "{profile:?}" + ); + let element: Arc = engine.clone(); + let pipeline = Pipeline::builder().add_element(element).build().unwrap(); + let mut data = pipeline.create_flow_data_with( + Evidence::builder() + .add("server.client-ip", "1.1.1.1") + .build(), + ); + data.process().unwrap(); + assert!(data.get(IP_DATA_KEY).unwrap().string("Asn").value().is_ok()); + + drop(data); + drop(pipeline); + drop(as_engine); + drop(engine); + drop(service); + let left = files_in(&std::env::temp_dir()) + .into_iter() + .filter(|file| file.starts_with(&name)) + .count(); + assert_eq!(left, 0, "{profile:?}: copies left in the temporary folder"); + let _ = std::fs::remove_dir_all(&dir); + } +} + +/// The copy is made in the folder the builder names, on build and on each +/// refresh, and none is left there once the engine is dropped. +#[test] +fn the_copy_goes_in_the_folder_the_builder_names() { + let Some(asn) = std::env::var("51DEGREES_IPI_PATH") + .ok() + .map(PathBuf::from) + .filter(|path| path.is_file()) + else { + eprintln!("skipping operational copies: set 51DEGREES_IPI_PATH to the ASN file"); + return; + }; + let dir = std::env::temp_dir().join(format!("fiftyone-ipi-named-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let temp = dir.join("temp"); + std::fs::create_dir_all(&temp).unwrap(); + let path = dir.join("51Degrees.ipi"); + std::fs::copy(&asn, &path).unwrap(); + let copies = || files_in(&temp).len(); + + let engine = IpIntelligenceOnPremiseEngine::builder(&path) + .properties(["Asn"]) + .temp_dir_path(&temp) + .auto_update(true) + .file_system_watcher(false) + .data_update_license_keys(["KEY"]) + .build() + .expect("the data file loads"); + assert_eq!(copies(), 1, "the copy is in the named folder"); + + engine.refresh(None).expect("the data file reloads"); + assert_eq!(copies(), 1, "the refresh copies into the named folder too"); + + drop(engine); + assert_eq!(copies(), 0, "no copy outlives the engine"); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/ip-intelligence-onpremise/tests/refresh.rs b/ip-intelligence-onpremise/tests/refresh.rs new file mode 100644 index 0000000..342c910 --- /dev/null +++ b/ip-intelligence-onpremise/tests/refresh.rs @@ -0,0 +1,82 @@ +/* ********************************************************************* + * 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. + * ********************************************************************* */ + +//! A refresh to a data file of another tier leaves the engine reporting what +//! an engine built on that file reports. +//! +//! The test needs the ASN file `51DEGREES_IPI_PATH` names and the Lite file +//! `51Degrees-LiteV41.ipi` beside it, and skips with a note when either is +//! missing. + +use std::path::PathBuf; + +use fiftyone_ip_intelligence_onpremise::IpIntelligenceOnPremiseEngine; +use fiftyone_native::PerformanceProfile; +use fiftyone_pipeline_engines::{AspectEngine, OnPremiseAspectEngine}; + +/// The ASN file and the Lite file beside it. +fn data_files() -> Option<(PathBuf, PathBuf)> { + let asn = PathBuf::from(std::env::var("51DEGREES_IPI_PATH").ok()?); + let lite = asn.parent()?.join("51Degrees-LiteV41.ipi"); + (asn.is_file() && lite.is_file()).then_some((asn, lite)) +} + +/// An engine that holds the whole data set in memory, so its data file can +/// be replaced on every platform. +fn engine(path: &PathBuf) -> IpIntelligenceOnPremiseEngine { + IpIntelligenceOnPremiseEngine::builder(path) + .performance_profile(PerformanceProfile::InMemory) + .auto_update(false) + .build() + .expect("the data file loads") +} + +#[test] +fn a_refresh_reports_the_tier_and_publish_date_of_the_new_data_file() { + let Some((asn, lite)) = data_files() else { + eprintln!( + "skipping the refresh test: set 51DEGREES_IPI_PATH to the ASN file, with \ + 51Degrees-LiteV41.ipi beside it" + ); + return; + }; + let dir = std::env::temp_dir().join(format!("fiftyone-ipi-refresh-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("51Degrees.ipi"); + std::fs::copy(&asn, &path).unwrap(); + + let refreshed = engine(&path); + let asn_tier = refreshed.data_source_tier().to_owned(); + std::fs::copy(&lite, &path).unwrap(); + refreshed.refresh(None).expect("the Lite file loads"); + let built = engine(&path); + + assert_ne!(built.data_source_tier(), asn_tier, "the files share a tier"); + assert_eq!(refreshed.data_source_tier(), built.data_source_tier()); + assert_eq!(refreshed.aspect_properties(), built.aspect_properties()); + assert_eq!(refreshed.data_file_published(), built.data_file_published()); + + drop(refreshed); + drop(built); + let _ = std::fs::remove_dir_all(&dir); +}