Making caching support optional in Wasmtime (#2119)

This commit moves all of the caching support that currently lives in
`wasmtime-environ` into a `wasmtime-cache` crate and makes it optional. The
goal here is to slim down the `wasmtime-environ` crate and clearly separate
boundaries where caching is a standalone and optional feature, not intertwined
with other crates.

crates/cache/src/config.rs

@@ -0,0 +1,584 @@
+//! Module for configuring the cache system.
+
+use super::Worker;
+use anyhow::{anyhow, bail, Context, Result};
+use directories::ProjectDirs;
+use log::{trace, warn};
+use serde::{
+    de::{self, Deserializer},
+    Deserialize,
+};
+use std::fmt::Debug;
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
+use std::sync::Arc;
+use std::time::Duration;
+
+// wrapped, so we have named section in config,
+// also, for possible future compatibility
+#[derive(Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
+struct Config {
+    cache: CacheConfig,
+}
+
+/// Global configuration for how the cache is managed
+#[derive(Deserialize, Debug, Clone)]
+#[serde(deny_unknown_fields)]
+pub struct CacheConfig {
+    enabled: bool,
+    directory: Option<PathBuf>,
+    #[serde(
+        default,
+        rename = "worker-event-queue-size",
+        deserialize_with = "deserialize_si_prefix"
+    )]
+    worker_event_queue_size: Option<u64>,
+    #[serde(rename = "baseline-compression-level")]
+    baseline_compression_level: Option<i32>,
+    #[serde(rename = "optimized-compression-level")]
+    optimized_compression_level: Option<i32>,
+    #[serde(
+        default,
+        rename = "optimized-compression-usage-counter-threshold",
+        deserialize_with = "deserialize_si_prefix"
+    )]
+    optimized_compression_usage_counter_threshold: Option<u64>,
+    #[serde(
+        default,
+        rename = "cleanup-interval",
+        deserialize_with = "deserialize_duration"
+    )]
+    cleanup_interval: Option<Duration>,
+    #[serde(
+        default,
+        rename = "optimizing-compression-task-timeout",
+        deserialize_with = "deserialize_duration"
+    )]
+    optimizing_compression_task_timeout: Option<Duration>,
+    #[serde(
+        default,
+        rename = "allowed-clock-drift-for-files-from-future",
+        deserialize_with = "deserialize_duration"
+    )]
+    allowed_clock_drift_for_files_from_future: Option<Duration>,
+    #[serde(
+        default,
+        rename = "file-count-soft-limit",
+        deserialize_with = "deserialize_si_prefix"
+    )]
+    file_count_soft_limit: Option<u64>,
+    #[serde(
+        default,
+        rename = "files-total-size-soft-limit",
+        deserialize_with = "deserialize_disk_space"
+    )]
+    files_total_size_soft_limit: Option<u64>,
+    #[serde(
+        default,
+        rename = "file-count-limit-percent-if-deleting",
+        deserialize_with = "deserialize_percent"
+    )]
+    file_count_limit_percent_if_deleting: Option<u8>,
+    #[serde(
+        default,
+        rename = "files-total-size-limit-percent-if-deleting",
+        deserialize_with = "deserialize_percent"
+    )]
+    files_total_size_limit_percent_if_deleting: Option<u8>,
+
+    #[serde(skip)]
+    worker: Option<Worker>,
+    #[serde(skip)]
+    state: Arc<CacheState>,
+}
+
+#[derive(Default, Debug)]
+struct CacheState {
+    hits: AtomicUsize,
+    misses: AtomicUsize,
+}
+
+/// Creates a new configuration file at specified path, or default path if None is passed.
+/// Fails if file already exists.
+pub fn create_new_config<P: AsRef<Path> + Debug>(config_file: Option<P>) -> Result<PathBuf> {
+    trace!("Creating new config file, path: {:?}", config_file);
+
+    let config_file = match config_file {
+        Some(path) => path.as_ref().to_path_buf(),
+        None => default_config_path()?,
+    };
+
+    if config_file.exists() {
+        bail!(
+            "Configuration file '{}' already exists.",
+            config_file.display()
+        );
+    }
+
+    let parent_dir = config_file
+        .parent()
+        .ok_or_else(|| anyhow!("Invalid cache config path: {}", config_file.display()))?;
+
+    fs::create_dir_all(parent_dir).with_context(|| {
+        format!(
+            "Failed to create config directory, config path: {}",
+            config_file.display(),
+        )
+    })?;
+
+    let content = "\
+# Comment out certain settings to use default values.
+# For more settings, please refer to the documentation:
+# https://bytecodealliance.github.io/wasmtime/cli-cache.html
+
+[cache]
+enabled = true
+";
+
+    fs::write(&config_file, &content).with_context(|| {
+        format!(
+            "Failed to flush config to the disk, path: {}",
+            config_file.display(),
+        )
+    })?;
+
+    Ok(config_file.to_path_buf())
+}
+
+// permitted levels from: https://docs.rs/zstd/0.4.28+zstd.1.4.3/zstd/stream/write/struct.Encoder.html
+const ZSTD_COMPRESSION_LEVELS: std::ops::RangeInclusive<i32> = 0..=21;
+
+// Default settings, you're welcome to tune them!
+// TODO: what do we want to warn users about?
+
+// At the moment of writing, the modules couldn't depend on anothers,
+// so we have at most one module per wasmtime instance
+// if changed, update cli-cache.md
+const DEFAULT_WORKER_EVENT_QUEUE_SIZE: u64 = 0x10;
+const WORKER_EVENT_QUEUE_SIZE_WARNING_TRESHOLD: u64 = 3;
+// should be quick and provide good enough compression
+// if changed, update cli-cache.md
+const DEFAULT_BASELINE_COMPRESSION_LEVEL: i32 = zstd::DEFAULT_COMPRESSION_LEVEL;
+// should provide significantly better compression than baseline
+// if changed, update cli-cache.md
+const DEFAULT_OPTIMIZED_COMPRESSION_LEVEL: i32 = 20;
+// shouldn't be to low to avoid recompressing too many files
+// if changed, update cli-cache.md
+const DEFAULT_OPTIMIZED_COMPRESSION_USAGE_COUNTER_THRESHOLD: u64 = 0x100;
+// if changed, update cli-cache.md
+const DEFAULT_CLEANUP_INTERVAL: Duration = Duration::from_secs(60 * 60);
+// if changed, update cli-cache.md
+const DEFAULT_OPTIMIZING_COMPRESSION_TASK_TIMEOUT: Duration = Duration::from_secs(30 * 60);
+// the default assumes problems with timezone configuration on network share + some clock drift
+// please notice 24 timezones = max 23h difference between some of them
+// if changed, update cli-cache.md
+const DEFAULT_ALLOWED_CLOCK_DRIFT_FOR_FILES_FROM_FUTURE: Duration =
+    Duration::from_secs(60 * 60 * 24);
+// if changed, update cli-cache.md
+const DEFAULT_FILE_COUNT_SOFT_LIMIT: u64 = 0x10_000;
+// if changed, update cli-cache.md
+const DEFAULT_FILES_TOTAL_SIZE_SOFT_LIMIT: u64 = 1024 * 1024 * 512;
+// if changed, update cli-cache.md
+const DEFAULT_FILE_COUNT_LIMIT_PERCENT_IF_DELETING: u8 = 70;
+// if changed, update cli-cache.md
+const DEFAULT_FILES_TOTAL_SIZE_LIMIT_PERCENT_IF_DELETING: u8 = 70;
+
+fn project_dirs() -> Option<ProjectDirs> {
+    ProjectDirs::from("", "BytecodeAlliance", "wasmtime")
+}
+
+fn default_config_path() -> Result<PathBuf> {
+    match project_dirs() {
+        Some(dirs) => Ok(dirs.config_dir().join("config.toml")),
+        None => bail!("config file not specified and failed to get the default"),
+    }
+}
+
+// Deserializers of our custom formats
+// can be replaced with const generics later
+macro_rules! generate_deserializer {
+    ($name:ident($numname:ident: $numty:ty, $unitname:ident: &str) -> $retty:ty {$body:expr}) => {
+        fn $name<'de, D>(deserializer: D) -> Result<$retty, D::Error>
+        where
+            D: Deserializer<'de>,
+        {
+            let text = Option::<String>::deserialize(deserializer)?;
+            let text = match text {
+                None => return Ok(None),
+                Some(text) => text,
+            };
+            let text = text.trim();
+            let split_point = text.find(|c: char| !c.is_numeric());
+            let (num, unit) = split_point.map_or_else(|| (text, ""), |p| text.split_at(p));
+            let deserialized = (|| {
+                let $numname = num.parse::<$numty>().ok()?;
+                let $unitname = unit.trim();
+                $body
+            })();
+            if deserialized.is_some() {
+                Ok(deserialized)
+            } else {
+                Err(de::Error::custom(
+                    "Invalid value, please refer to the documentation",
+                ))
+            }
+        }
+    };
+}
+
+generate_deserializer!(deserialize_duration(num: u64, unit: &str) -> Option<Duration> {
+    match unit {
+        "s" => Some(Duration::from_secs(num)),
+        "m" => Some(Duration::from_secs(num * 60)),
+        "h" => Some(Duration::from_secs(num * 60 * 60)),
+        "d" => Some(Duration::from_secs(num * 60 * 60 * 24)),
+        _ => None,
+    }
+});
+
+generate_deserializer!(deserialize_si_prefix(num: u64, unit: &str) -> Option<u64> {
+    match unit {
+        "" => Some(num),
+        "K" => num.checked_mul(1_000),
+        "M" => num.checked_mul(1_000_000),
+        "G" => num.checked_mul(1_000_000_000),
+        "T" => num.checked_mul(1_000_000_000_000),
+        "P" => num.checked_mul(1_000_000_000_000_000),
+        _ => None,
+    }
+});
+
+generate_deserializer!(deserialize_disk_space(num: u64, unit: &str) -> Option<u64> {
+    match unit {
+        "" => Some(num),
+        "K" => num.checked_mul(1_000),
+        "Ki" => num.checked_mul(1u64 << 10),
+        "M" => num.checked_mul(1_000_000),
+        "Mi" => num.checked_mul(1u64 << 20),
+        "G" => num.checked_mul(1_000_000_000),
+        "Gi" => num.checked_mul(1u64 << 30),
+        "T" => num.checked_mul(1_000_000_000_000),
+        "Ti" => num.checked_mul(1u64 << 40),
+        "P" => num.checked_mul(1_000_000_000_000_000),
+        "Pi" => num.checked_mul(1u64 << 50),
+        _ => None,
+    }
+});
+
+generate_deserializer!(deserialize_percent(num: u8, unit: &str) -> Option<u8> {
+    match unit {
+        "%" => Some(num),
+        _ => None,
+    }
+});
+
+static CACHE_IMPROPER_CONFIG_ERROR_MSG: &str =
+    "Cache system should be enabled and all settings must be validated or defaulted";
+
+macro_rules! generate_setting_getter {
+    ($setting:ident: $setting_type:ty) => {
+        /// Returns `$setting`.
+        ///
+        /// Panics if the cache is disabled.
+        pub fn $setting(&self) -> $setting_type {
+            self.$setting.expect(CACHE_IMPROPER_CONFIG_ERROR_MSG)
+        }
+    };
+}
+
+impl CacheConfig {
+    generate_setting_getter!(worker_event_queue_size: u64);
+    generate_setting_getter!(baseline_compression_level: i32);
+    generate_setting_getter!(optimized_compression_level: i32);
+    generate_setting_getter!(optimized_compression_usage_counter_threshold: u64);
+    generate_setting_getter!(cleanup_interval: Duration);
+    generate_setting_getter!(optimizing_compression_task_timeout: Duration);
+    generate_setting_getter!(allowed_clock_drift_for_files_from_future: Duration);
+    generate_setting_getter!(file_count_soft_limit: u64);
+    generate_setting_getter!(files_total_size_soft_limit: u64);
+    generate_setting_getter!(file_count_limit_percent_if_deleting: u8);
+    generate_setting_getter!(files_total_size_limit_percent_if_deleting: u8);
+
+    /// Returns true if and only if the cache is enabled.
+    pub fn enabled(&self) -> bool {
+        self.enabled
+    }
+
+    /// Returns path to the cache directory.
+    ///
+    /// Panics if the cache is disabled.
+    pub fn directory(&self) -> &PathBuf {
+        self.directory
+            .as_ref()
+            .expect(CACHE_IMPROPER_CONFIG_ERROR_MSG)
+    }
+
+    /// Creates a new set of configuration which represents a disabled cache
+    pub fn new_cache_disabled() -> Self {
+        Self {
+            enabled: false,
+            directory: None,
+            worker_event_queue_size: None,
+            baseline_compression_level: None,
+            optimized_compression_level: None,
+            optimized_compression_usage_counter_threshold: None,
+            cleanup_interval: None,
+            optimizing_compression_task_timeout: None,
+            allowed_clock_drift_for_files_from_future: None,
+            file_count_soft_limit: None,
+            files_total_size_soft_limit: None,
+            file_count_limit_percent_if_deleting: None,
+            files_total_size_limit_percent_if_deleting: None,
+            worker: None,
+            state: Arc::new(CacheState::default()),
+        }
+    }
+
+    fn new_cache_enabled_template() -> Self {
+        let mut conf = Self::new_cache_disabled();
+        conf.enabled = true;
+        conf
+    }
+
+    /// Parses cache configuration from the file specified
+    pub fn from_file(config_file: Option<&Path>) -> Result<Self> {
+        let mut config = Self::load_and_parse_file(config_file)?;
+
+        // validate values and fill in defaults
+        config.validate_directory_or_default()?;
+        config.validate_worker_event_queue_size_or_default();
+        config.validate_baseline_compression_level_or_default()?;
+        config.validate_optimized_compression_level_or_default()?;
+        config.validate_optimized_compression_usage_counter_threshold_or_default();
+        config.validate_cleanup_interval_or_default();
+        config.validate_optimizing_compression_task_timeout_or_default();
+        config.validate_allowed_clock_drift_for_files_from_future_or_default();
+        config.validate_file_count_soft_limit_or_default();
+        config.validate_files_total_size_soft_limit_or_default();
+        config.validate_file_count_limit_percent_if_deleting_or_default()?;
+        config.validate_files_total_size_limit_percent_if_deleting_or_default()?;
+        config.spawn_worker();
+
+        Ok(config)
+    }
+
+    fn spawn_worker(&mut self) {
+        if self.enabled {
+            self.worker = Some(Worker::start_new(self, None));
+        }
+    }
+
+    pub(super) fn worker(&self) -> &Worker {
+        assert!(self.enabled);
+        self.worker.as_ref().unwrap()
+    }
+
+    /// Returns the number of cache hits seen so far
+    pub fn cache_hits(&self) -> usize {
+        self.state.hits.load(SeqCst)
+    }
+
+    /// Returns the number of cache misses seen so far
+    pub fn cache_misses(&self) -> usize {
+        self.state.misses.load(SeqCst)
+    }
+
+    pub(crate) fn on_cache_get_async(&self, path: impl AsRef<Path>) {
+        self.state.hits.fetch_add(1, SeqCst);
+        self.worker().on_cache_get_async(path)
+    }
+
+    pub(crate) fn on_cache_update_async(&self, path: impl AsRef<Path>) {
+        self.state.misses.fetch_add(1, SeqCst);
+        self.worker().on_cache_update_async(path)
+    }
+
+    fn load_and_parse_file(config_file: Option<&Path>) -> Result<Self> {
+        // get config file path
+        let (config_file, user_custom_file) = match config_file {
+            Some(path) => (path.to_path_buf(), true),
+            None => (default_config_path()?, false),
+        };
+
+        // read config, or use default one
+        let entity_exists = config_file.exists();
+        match (entity_exists, user_custom_file) {
+            (false, false) => Ok(Self::new_cache_enabled_template()),
+            _ => {
+                let bytes = fs::read(&config_file).context(format!(
+                    "failed to read config file: {}",
+                    config_file.display()
+                ))?;
+                let config = toml::from_slice::<Config>(&bytes[..]).context(format!(
+                    "failed to parse config file: {}",
+                    config_file.display()
+                ))?;
+                Ok(config.cache)
+            }
+        }
+    }
+
+    fn validate_directory_or_default(&mut self) -> Result<()> {
+        if self.directory.is_none() {
+            match project_dirs() {
+                Some(proj_dirs) => self.directory = Some(proj_dirs.cache_dir().to_path_buf()),
+                None => {
+                    bail!("Cache directory not specified and failed to get the default");
+                }
+            }
+        }
+
+        // On Windows, if we want long paths, we need '\\?\' prefix, but it doesn't work
+        // with relative paths. One way to get absolute path (the only one?) is to use
+        // fs::canonicalize, but it requires that given path exists. The extra advantage
+        // of this method is fact that the method prepends '\\?\' on Windows.
+        let cache_dir = self.directory.as_ref().unwrap();
+
+        if !cache_dir.is_absolute() {
+            bail!(
+                "Cache directory path has to be absolute, path: {}",
+                cache_dir.display(),
+            );
+        }
+
+        fs::create_dir_all(cache_dir).context(format!(
+            "failed to create cache directory: {}",
+            cache_dir.display()
+        ))?;
+        let canonical = fs::canonicalize(cache_dir).context(format!(
+            "failed to canonicalize cache directory: {}",
+            cache_dir.display()
+        ))?;
+        self.directory = Some(canonical);
+        Ok(())
+    }
+
+    fn validate_worker_event_queue_size_or_default(&mut self) {
+        if self.worker_event_queue_size.is_none() {
+            self.worker_event_queue_size = Some(DEFAULT_WORKER_EVENT_QUEUE_SIZE);
+        }
+
+        if self.worker_event_queue_size.unwrap() < WORKER_EVENT_QUEUE_SIZE_WARNING_TRESHOLD {
+            warn!("Detected small worker event queue size. Some messages might be lost.");
+        }
+    }
+
+    fn validate_baseline_compression_level_or_default(&mut self) -> Result<()> {
+        if self.baseline_compression_level.is_none() {
+            self.baseline_compression_level = Some(DEFAULT_BASELINE_COMPRESSION_LEVEL);
+        }
+
+        if !ZSTD_COMPRESSION_LEVELS.contains(&self.baseline_compression_level.unwrap()) {
+            bail!(
+                "Invalid baseline compression level: {} not in {:#?}",
+                self.baseline_compression_level.unwrap(),
+                ZSTD_COMPRESSION_LEVELS
+            );
+        }
+        Ok(())
+    }
+
+    // assumption: baseline compression level has been verified
+    fn validate_optimized_compression_level_or_default(&mut self) -> Result<()> {
+        if self.optimized_compression_level.is_none() {
+            self.optimized_compression_level = Some(DEFAULT_OPTIMIZED_COMPRESSION_LEVEL);
+        }
+
+        let opt_lvl = self.optimized_compression_level.unwrap();
+        let base_lvl = self.baseline_compression_level.unwrap();
+
+        if !ZSTD_COMPRESSION_LEVELS.contains(&opt_lvl) {
+            bail!(
+                "Invalid optimized compression level: {} not in {:#?}",
+                opt_lvl,
+                ZSTD_COMPRESSION_LEVELS
+            );
+        }
+
+        if opt_lvl < base_lvl {
+            bail!(
+                "Invalid optimized compression level is lower than baseline: {} < {}",
+                opt_lvl,
+                base_lvl
+            );
+        }
+        Ok(())
+    }
+
+    fn validate_optimized_compression_usage_counter_threshold_or_default(&mut self) {
+        if self.optimized_compression_usage_counter_threshold.is_none() {
+            self.optimized_compression_usage_counter_threshold =
+                Some(DEFAULT_OPTIMIZED_COMPRESSION_USAGE_COUNTER_THRESHOLD);
+        }
+    }
+
+    fn validate_cleanup_interval_or_default(&mut self) {
+        if self.cleanup_interval.is_none() {
+            self.cleanup_interval = Some(DEFAULT_CLEANUP_INTERVAL);
+        }
+    }
+
+    fn validate_optimizing_compression_task_timeout_or_default(&mut self) {
+        if self.optimizing_compression_task_timeout.is_none() {
+            self.optimizing_compression_task_timeout =
+                Some(DEFAULT_OPTIMIZING_COMPRESSION_TASK_TIMEOUT);
+        }
+    }
+
+    fn validate_allowed_clock_drift_for_files_from_future_or_default(&mut self) {
+        if self.allowed_clock_drift_for_files_from_future.is_none() {
+            self.allowed_clock_drift_for_files_from_future =
+                Some(DEFAULT_ALLOWED_CLOCK_DRIFT_FOR_FILES_FROM_FUTURE);
+        }
+    }
+
+    fn validate_file_count_soft_limit_or_default(&mut self) {
+        if self.file_count_soft_limit.is_none() {
+            self.file_count_soft_limit = Some(DEFAULT_FILE_COUNT_SOFT_LIMIT);
+        }
+    }
+
+    fn validate_files_total_size_soft_limit_or_default(&mut self) {
+        if self.files_total_size_soft_limit.is_none() {
+            self.files_total_size_soft_limit = Some(DEFAULT_FILES_TOTAL_SIZE_SOFT_LIMIT);
+        }
+    }
+
+    fn validate_file_count_limit_percent_if_deleting_or_default(&mut self) -> Result<()> {
+        if self.file_count_limit_percent_if_deleting.is_none() {
+            self.file_count_limit_percent_if_deleting =
+                Some(DEFAULT_FILE_COUNT_LIMIT_PERCENT_IF_DELETING);
+        }
+
+        let percent = self.file_count_limit_percent_if_deleting.unwrap();
+        if percent > 100 {
+            bail!(
+                "Invalid files count limit percent if deleting: {} not in range 0-100%",
+                percent
+            );
+        }
+        Ok(())
+    }
+
+    fn validate_files_total_size_limit_percent_if_deleting_or_default(&mut self) -> Result<()> {
+        if self.files_total_size_limit_percent_if_deleting.is_none() {
+            self.files_total_size_limit_percent_if_deleting =
+                Some(DEFAULT_FILES_TOTAL_SIZE_LIMIT_PERCENT_IF_DELETING);
+        }
+
+        let percent = self.files_total_size_limit_percent_if_deleting.unwrap();
+        if percent > 100 {
+            bail!(
+                "Invalid files total size limit percent if deleting: {} not in range 0-100%",
+                percent
+            );
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+#[macro_use]
+pub mod tests;

crates/cache/src/config/tests.rs

@@ -0,0 +1,525 @@
+use super::CacheConfig;
+use std::fs;
+use std::path::PathBuf;
+use std::time::Duration;
+use tempfile::{self, TempDir};
+
+// note: config loading during validation creates cache directory to canonicalize its path,
+//       that's why these function and macro always use custom cache directory
+// note: tempdir removes directory when being dropped, so we need to return it to the caller,
+//       so the paths are valid
+pub fn test_prolog() -> (TempDir, PathBuf, PathBuf) {
+    let _ = pretty_env_logger::try_init();
+    let temp_dir = tempfile::tempdir().expect("Can't create temporary directory");
+    let cache_dir = temp_dir.path().join("cache-dir");
+    let config_path = temp_dir.path().join("cache-config.toml");
+    (temp_dir, cache_dir, config_path)
+}
+
+macro_rules! load_config {
+    ($config_path:ident, $content_fmt:expr, $cache_dir:ident) => {{
+        let config_path = &$config_path;
+        let content = format!(
+            $content_fmt,
+            cache_dir = toml::to_string_pretty(&format!("{}", $cache_dir.display())).unwrap()
+        );
+        fs::write(config_path, content).expect("Failed to write test config file");
+        CacheConfig::from_file(Some(config_path)).unwrap()
+    }};
+}
+
+macro_rules! bad_config {
+    ($config_path:ident, $content_fmt:expr, $cache_dir:ident) => {{
+        let config_path = &$config_path;
+        let content = format!(
+            $content_fmt,
+            cache_dir = toml::to_string_pretty(&format!("{}", $cache_dir.display())).unwrap()
+        );
+        fs::write(config_path, content).expect("Failed to write test config file");
+        assert!(CacheConfig::from_file(Some(config_path)).is_err());
+    }};
+}
+
+// test without macros to test being disabled
+#[test]
+fn test_disabled() {
+    let dir = tempfile::tempdir().expect("Can't create temporary directory");
+    let config_path = dir.path().join("cache-config.toml");
+    let config_content = "[cache]\n\
+                          enabled = false\n";
+    fs::write(&config_path, config_content).expect("Failed to write test config file");
+    let conf = CacheConfig::from_file(Some(&config_path)).unwrap();
+    assert!(!conf.enabled());
+}
+
+#[test]
+fn test_unrecognized_settings() {
+    let (_td, cd, cp) = test_prolog();
+    bad_config!(
+        cp,
+        "unrecognized-setting = 42\n\
+         [cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         unrecognized-setting = 42",
+        cd
+    );
+}
+
+#[test]
+fn test_all_settings() {
+    let (_td, cd, cp) = test_prolog();
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         baseline-compression-level = 3\n\
+         optimized-compression-level = 20\n\
+         optimized-compression-usage-counter-threshold = '256'\n\
+         cleanup-interval = '1h'\n\
+         optimizing-compression-task-timeout = '30m'\n\
+         allowed-clock-drift-for-files-from-future = '1d'\n\
+         file-count-soft-limit = '65536'\n\
+         files-total-size-soft-limit = '512Mi'\n\
+         file-count-limit-percent-if-deleting = '70%'\n\
+         files-total-size-limit-percent-if-deleting = '70%'",
+        cd
+    );
+    check_conf(&conf, &cd);
+
+    let conf = load_config!(
+        cp,
+        // added some white spaces
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size  = ' 16\t'\n\
+         baseline-compression-level = 3\n\
+         optimized-compression-level =\t 20\n\
+         optimized-compression-usage-counter-threshold = '256'\n\
+         cleanup-interval = ' 1h'\n\
+         optimizing-compression-task-timeout = '30  m'\n\
+         allowed-clock-drift-for-files-from-future = '1\td'\n\
+         file-count-soft-limit = '\t \t65536\t'\n\
+         files-total-size-soft-limit = '512\t\t Mi '\n\
+         file-count-limit-percent-if-deleting = '70\t%'\n\
+         files-total-size-limit-percent-if-deleting = ' 70 %'",
+        cd
+    );
+    check_conf(&conf, &cd);
+
+    fn check_conf(conf: &CacheConfig, cd: &PathBuf) {
+        assert!(conf.enabled());
+        assert_eq!(
+            conf.directory(),
+            &fs::canonicalize(cd).expect("canonicalize failed")
+        );
+        assert_eq!(conf.worker_event_queue_size(), 0x10);
+        assert_eq!(conf.baseline_compression_level(), 3);
+        assert_eq!(conf.optimized_compression_level(), 20);
+        assert_eq!(conf.optimized_compression_usage_counter_threshold(), 0x100);
+        assert_eq!(conf.cleanup_interval(), Duration::from_secs(60 * 60));
+        assert_eq!(
+            conf.optimizing_compression_task_timeout(),
+            Duration::from_secs(30 * 60)
+        );
+        assert_eq!(
+            conf.allowed_clock_drift_for_files_from_future(),
+            Duration::from_secs(60 * 60 * 24)
+        );
+        assert_eq!(conf.file_count_soft_limit(), 0x10_000);
+        assert_eq!(conf.files_total_size_soft_limit(), 512 * (1u64 << 20));
+        assert_eq!(conf.file_count_limit_percent_if_deleting(), 70);
+        assert_eq!(conf.files_total_size_limit_percent_if_deleting(), 70);
+    }
+}
+
+#[test]
+fn test_compression_level_settings() {
+    let (_td, cd, cp) = test_prolog();
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         baseline-compression-level = 1\n\
+         optimized-compression-level = 21",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.baseline_compression_level(), 1);
+    assert_eq!(conf.optimized_compression_level(), 21);
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         baseline-compression-level = -1\n\
+         optimized-compression-level = 21",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         baseline-compression-level = 15\n\
+         optimized-compression-level = 10",
+        cd
+    );
+}
+
+#[test]
+fn test_si_prefix_settings() {
+    let (_td, cd, cp) = test_prolog();
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '42'\n\
+         optimized-compression-usage-counter-threshold = '4K'\n\
+         file-count-soft-limit = '3M'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.worker_event_queue_size(), 42);
+    assert_eq!(conf.optimized_compression_usage_counter_threshold(), 4_000);
+    assert_eq!(conf.file_count_soft_limit(), 3_000_000);
+
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '2K'\n\
+         optimized-compression-usage-counter-threshold = '4444T'\n\
+         file-count-soft-limit = '1P'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.worker_event_queue_size(), 2_000);
+    assert_eq!(
+        conf.optimized_compression_usage_counter_threshold(),
+        4_444_000_000_000_000
+    );
+    assert_eq!(conf.file_count_soft_limit(), 1_000_000_000_000_000);
+
+    // different errors
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '2g'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         file-count-soft-limit = 1",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         file-count-soft-limit = '-31337'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         file-count-soft-limit = '3.14M'",
+        cd
+    );
+}
+
+#[test]
+fn test_disk_space_settings() {
+    let (_td, cd, cp) = test_prolog();
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '76'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.files_total_size_soft_limit(), 76);
+
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '42 Mi'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.files_total_size_soft_limit(), 42 * (1u64 << 20));
+
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '2 Gi'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.files_total_size_soft_limit(), 2 * (1u64 << 30));
+
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '31337 Ti'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.files_total_size_soft_limit(), 31337 * (1u64 << 40));
+
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '7 Pi'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.files_total_size_soft_limit(), 7 * (1u64 << 50));
+
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '7M'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.files_total_size_soft_limit(), 7_000_000);
+
+    // different errors
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '7 mi'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = 1",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '-31337'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-soft-limit = '3.14Ki'",
+        cd
+    );
+}
+
+#[test]
+fn test_duration_settings() {
+    let (_td, cd, cp) = test_prolog();
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         cleanup-interval = '100s'\n\
+         optimizing-compression-task-timeout = '3m'\n\
+         allowed-clock-drift-for-files-from-future = '4h'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.cleanup_interval(), Duration::from_secs(100));
+    assert_eq!(
+        conf.optimizing_compression_task_timeout(),
+        Duration::from_secs(3 * 60)
+    );
+    assert_eq!(
+        conf.allowed_clock_drift_for_files_from_future(),
+        Duration::from_secs(4 * 60 * 60)
+    );
+
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         cleanup-interval = '2d'\n\
+         optimizing-compression-task-timeout = '333 m'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(
+        conf.cleanup_interval(),
+        Duration::from_secs(2 * 24 * 60 * 60)
+    );
+    assert_eq!(
+        conf.optimizing_compression_task_timeout(),
+        Duration::from_secs(333 * 60)
+    );
+
+    // different errors
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         optimizing-compression-task-timeout = '333'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         optimizing-compression-task-timeout = 333",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         optimizing-compression-task-timeout = '10 M'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         optimizing-compression-task-timeout = '10 min'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         optimizing-compression-task-timeout = '-10s'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         optimizing-compression-task-timeout = '1.5m'",
+        cd
+    );
+}
+
+#[test]
+fn test_percent_settings() {
+    let (_td, cd, cp) = test_prolog();
+    let conf = load_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         file-count-limit-percent-if-deleting = '62%'\n\
+         files-total-size-limit-percent-if-deleting = '23 %'",
+        cd
+    );
+    assert!(conf.enabled());
+    assert_eq!(conf.file_count_limit_percent_if_deleting(), 62);
+    assert_eq!(conf.files_total_size_limit_percent_if_deleting(), 23);
+
+    // different errors
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-limit-percent-if-deleting = '23'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-limit-percent-if-deleting = '22.5%'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-limit-percent-if-deleting = '0.5'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-limit-percent-if-deleting = '-1%'",
+        cd
+    );
+
+    bad_config!(
+        cp,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         files-total-size-limit-percent-if-deleting = '101%'",
+        cd
+    );
+}

crates/cache/src/lib.rs

@@ -0,0 +1,212 @@
+use log::{debug, trace, warn};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::fs;
+use std::hash::Hash;
+use std::hash::Hasher;
+use std::io::Write;
+use std::path::{Path, PathBuf};
+
+#[macro_use] // for tests
+mod config;
+mod worker;
+
+pub use config::{create_new_config, CacheConfig};
+use worker::Worker;
+
+/// Module level cache entry.
+pub struct ModuleCacheEntry<'config>(Option<ModuleCacheEntryInner<'config>>);
+
+struct ModuleCacheEntryInner<'config> {
+    root_path: PathBuf,
+    cache_config: &'config CacheConfig,
+}
+
+struct Sha256Hasher(Sha256);
+
+impl<'config> ModuleCacheEntry<'config> {
+    /// Create the cache entry.
+    pub fn new<'data>(compiler_name: &str, cache_config: &'config CacheConfig) -> Self {
+        if cache_config.enabled() {
+            Self(Some(ModuleCacheEntryInner::new(
+                compiler_name,
+                cache_config,
+            )))
+        } else {
+            Self(None)
+        }
+    }
+
+    #[cfg(test)]
+    fn from_inner(inner: ModuleCacheEntryInner<'config>) -> Self {
+        Self(Some(inner))
+    }
+
+    /// Gets cached data if state matches, otherwise calls the `compute`.
+    pub fn get_data<T, U, E>(&self, state: T, compute: fn(T) -> Result<U, E>) -> Result<U, E>
+    where
+        T: Hash,
+        U: Serialize + for<'a> Deserialize<'a>,
+    {
+        let mut hasher = Sha256Hasher(Sha256::new());
+        state.hash(&mut hasher);
+        let hash: [u8; 32] = hasher.0.result().into();
+        // standard encoding uses '/' which can't be used for filename
+        let hash = base64::encode_config(&hash, base64::URL_SAFE_NO_PAD);
+
+        let inner = match &self.0 {
+            Some(inner) => inner,
+            None => return compute(state),
+        };
+
+        if let Some(cached_val) = inner.get_data(&hash) {
+            let mod_cache_path = inner.root_path.join(&hash);
+            inner.cache_config.on_cache_get_async(&mod_cache_path); // call on success
+            return Ok(cached_val);
+        }
+        let val_to_cache = compute(state)?;
+        if inner.update_data(&hash, &val_to_cache).is_some() {
+            let mod_cache_path = inner.root_path.join(&hash);
+            inner.cache_config.on_cache_update_async(&mod_cache_path); // call on success
+        }
+        Ok(val_to_cache)
+    }
+}
+
+impl<'config> ModuleCacheEntryInner<'config> {
+    fn new<'data>(compiler_name: &str, cache_config: &'config CacheConfig) -> Self {
+        // If debug assertions are enabled then assume that we're some sort of
+        // local build. We don't want local builds to stomp over caches between
+        // builds, so just use a separate cache directory based on the mtime of
+        // our executable, which should roughly correlate with "you changed the
+        // source code so you get a different directory".
+        //
+        // Otherwise if this is a release build we use the `GIT_REV` env var
+        // which is either the git rev if installed from git or the crate
+        // version if installed from crates.io.
+        let compiler_dir = if cfg!(debug_assertions) {
+            fn self_mtime() -> Option<String> {
+                let path = std::env::current_exe().ok()?;
+                let metadata = path.metadata().ok()?;
+                let mtime = metadata.modified().ok()?;
+                Some(match mtime.duration_since(std::time::UNIX_EPOCH) {
+                    Ok(dur) => format!("{}", dur.as_millis()),
+                    Err(err) => format!("m{}", err.duration().as_millis()),
+                })
+            }
+            let self_mtime = self_mtime().unwrap_or("no-mtime".to_string());
+            format!(
+                "{comp_name}-{comp_ver}-{comp_mtime}",
+                comp_name = compiler_name,
+                comp_ver = env!("GIT_REV"),
+                comp_mtime = self_mtime,
+            )
+        } else {
+            format!(
+                "{comp_name}-{comp_ver}",
+                comp_name = compiler_name,
+                comp_ver = env!("GIT_REV"),
+            )
+        };
+        let root_path = cache_config.directory().join("modules").join(compiler_dir);
+
+        Self {
+            root_path,
+            cache_config,
+        }
+    }
+
+    fn get_data<T>(&self, hash: &str) -> Option<T>
+    where
+        T: for<'a> Deserialize<'a>,
+    {
+        let mod_cache_path = self.root_path.join(hash);
+        trace!("get_data() for path: {}", mod_cache_path.display());
+        let compressed_cache_bytes = fs::read(&mod_cache_path).ok()?;
+        let cache_bytes = zstd::decode_all(&compressed_cache_bytes[..])
+            .map_err(|err| warn!("Failed to decompress cached code: {}", err))
+            .ok()?;
+        bincode::deserialize(&cache_bytes[..])
+            .map_err(|err| warn!("Failed to deserialize cached code: {}", err))
+            .ok()
+    }
+
+    fn update_data<T: Serialize>(&self, hash: &str, data: &T) -> Option<()> {
+        let mod_cache_path = self.root_path.join(hash);
+        trace!("update_data() for path: {}", mod_cache_path.display());
+        let serialized_data = bincode::serialize(&data)
+            .map_err(|err| warn!("Failed to serialize cached code: {}", err))
+            .ok()?;
+        let compressed_data = zstd::encode_all(
+            &serialized_data[..],
+            self.cache_config.baseline_compression_level(),
+        )
+        .map_err(|err| warn!("Failed to compress cached code: {}", err))
+        .ok()?;
+
+        // Optimize syscalls: first, try writing to disk. It should succeed in most cases.
+        // Otherwise, try creating the cache directory and retry writing to the file.
+        if fs_write_atomic(&mod_cache_path, "mod", &compressed_data) {
+            return Some(());
+        }
+
+        debug!(
+            "Attempting to create the cache directory, because \
+             failed to write cached code to disk, path: {}",
+            mod_cache_path.display(),
+        );
+
+        let cache_dir = mod_cache_path.parent().unwrap();
+        fs::create_dir_all(cache_dir)
+            .map_err(|err| {
+                warn!(
+                    "Failed to create cache directory, path: {}, message: {}",
+                    cache_dir.display(),
+                    err
+                )
+            })
+            .ok()?;
+
+        if fs_write_atomic(&mod_cache_path, "mod", &compressed_data) {
+            Some(())
+        } else {
+            None
+        }
+    }
+}
+
+impl Hasher for Sha256Hasher {
+    fn finish(&self) -> u64 {
+        panic!("Sha256Hasher doesn't support finish!");
+    }
+
+    fn write(&mut self, bytes: &[u8]) {
+        self.0.input(bytes);
+    }
+}
+
+// Assumption: path inside cache directory.
+// Then, we don't have to use sound OS-specific exclusive file access.
+// Note: there's no need to remove temporary file here - cleanup task will do it later.
+fn fs_write_atomic(path: &Path, reason: &str, contents: &[u8]) -> bool {
+    let lock_path = path.with_extension(format!("wip-atomic-write-{}", reason));
+    fs::OpenOptions::new()
+        .create_new(true) // atomic file creation (assumption: no one will open it without this flag)
+        .write(true)
+        .open(&lock_path)
+        .and_then(|mut file| file.write_all(contents))
+        // file should go out of scope and be closed at this point
+        .and_then(|()| fs::rename(&lock_path, &path)) // atomic file rename
+        .map_err(|err| {
+            warn!(
+                "Failed to write file with rename, lock path: {}, target path: {}, err: {}",
+                lock_path.display(),
+                path.display(),
+                err
+            )
+        })
+        .is_ok()
+}
+
+#[cfg(test)]
+mod tests;

crates/cache/src/tests.rs

@@ -0,0 +1,92 @@
+use super::config::tests::test_prolog;
+use super::*;
+use std::fs;
+
+// Since cache system is a global thing, each test needs to be run in seperate process.
+// So, init() tests are run as integration tests.
+// However, caching is a private thing, an implementation detail, and needs to be tested
+// from the inside of the module.
+// We test init() in exactly one test, rest of the tests doesn't rely on it.
+
+#[test]
+fn test_cache_init() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let baseline_compression_level = 4;
+    let config_content = format!(
+        "[cache]\n\
+         enabled = true\n\
+         directory = {}\n\
+         baseline-compression-level = {}\n",
+        toml::to_string_pretty(&format!("{}", cache_dir.display())).unwrap(),
+        baseline_compression_level,
+    );
+    fs::write(&config_path, config_content).expect("Failed to write test config file");
+
+    let cache_config = CacheConfig::from_file(Some(&config_path)).unwrap();
+
+    // test if we can use config
+    assert!(cache_config.enabled());
+    // assumption: config init creates cache directory and returns canonicalized path
+    assert_eq!(
+        *cache_config.directory(),
+        fs::canonicalize(cache_dir).unwrap()
+    );
+    assert_eq!(
+        cache_config.baseline_compression_level(),
+        baseline_compression_level
+    );
+
+    // test if we can use worker
+    cache_config.worker().on_cache_update_async(config_path);
+}
+
+#[test]
+fn test_write_read_cache() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         baseline-compression-level = 3\n",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+
+    // assumption: config load creates cache directory and returns canonicalized path
+    assert_eq!(
+        *cache_config.directory(),
+        fs::canonicalize(cache_dir).unwrap()
+    );
+
+    let compiler1 = "test-1";
+    let compiler2 = "test-2";
+
+    let entry1 = ModuleCacheEntry::from_inner(ModuleCacheEntryInner::new(compiler1, &cache_config));
+    let entry2 = ModuleCacheEntry::from_inner(ModuleCacheEntryInner::new(compiler2, &cache_config));
+
+    entry1.get_data::<_, i32, i32>(1, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+
+    entry1.get_data::<_, i32, i32>(2, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
+
+    entry1.get_data::<_, i32, i32>(3, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(3, |_| panic!()).unwrap();
+
+    entry1.get_data::<_, i32, i32>(4, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(3, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(4, |_| panic!()).unwrap();
+
+    entry2.get_data::<_, i32, i32>(1, |_| Ok(100)).unwrap();
+    entry1.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(2, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(3, |_| panic!()).unwrap();
+    entry1.get_data::<_, i32, i32>(4, |_| panic!()).unwrap();
+    entry2.get_data::<_, i32, i32>(1, |_| panic!()).unwrap();
+}

crates/cache/src/worker.rs

@@ -0,0 +1,909 @@
+//! Background worker that watches over the cache.
+//!
+//! It cleans up old cache, updates statistics and optimizes the cache.
+//! We allow losing some messages (it doesn't hurt) and some races,
+//! but we guarantee eventual consistency and fault tolerancy.
+//! Background tasks can be CPU intensive, but the worker thread has low priority.
+
+use super::{fs_write_atomic, CacheConfig};
+use log::{debug, info, trace, warn};
+use serde::{Deserialize, Serialize};
+use std::cmp;
+use std::collections::HashMap;
+use std::ffi::OsStr;
+use std::fmt;
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
+#[cfg(test)]
+use std::sync::{Arc, Condvar, Mutex};
+use std::thread;
+use std::time::Duration;
+#[cfg(not(test))]
+use std::time::SystemTime;
+#[cfg(test)]
+use tests::system_time_stub::SystemTimeStub as SystemTime;
+
+#[derive(Clone)]
+pub(super) struct Worker {
+    sender: SyncSender<CacheEvent>,
+    #[cfg(test)]
+    stats: Arc<(Mutex<WorkerStats>, Condvar)>,
+}
+
+struct WorkerThread {
+    receiver: Receiver<CacheEvent>,
+    cache_config: CacheConfig,
+    #[cfg(test)]
+    stats: Arc<(Mutex<WorkerStats>, Condvar)>,
+}
+
+#[cfg(test)]
+#[derive(Default)]
+struct WorkerStats {
+    dropped: u32,
+    sent: u32,
+    handled: u32,
+}
+
+#[derive(Debug, Clone)]
+enum CacheEvent {
+    OnCacheGet(PathBuf),
+    OnCacheUpdate(PathBuf),
+}
+
+impl Worker {
+    pub(super) fn start_new(
+        cache_config: &CacheConfig,
+        init_file_per_thread_logger: Option<&'static str>,
+    ) -> Self {
+        let queue_size = match cache_config.worker_event_queue_size() {
+            num if num <= usize::max_value() as u64 => num as usize,
+            _ => usize::max_value(),
+        };
+        let (tx, rx) = sync_channel(queue_size);
+
+        #[cfg(test)]
+        let stats = Arc::new((Mutex::new(WorkerStats::default()), Condvar::new()));
+
+        let worker_thread = WorkerThread {
+            receiver: rx,
+            cache_config: cache_config.clone(),
+            #[cfg(test)]
+            stats: stats.clone(),
+        };
+
+        // when self is dropped, sender will be dropped, what will cause the channel
+        // to hang, and the worker thread to exit -- it happens in the tests
+        // non-tests binary has only a static worker, so Rust doesn't drop it
+        thread::spawn(move || worker_thread.run(init_file_per_thread_logger));
+
+        Self {
+            sender: tx,
+            #[cfg(test)]
+            stats,
+        }
+    }
+
+    pub(super) fn on_cache_get_async(&self, path: impl AsRef<Path>) {
+        let event = CacheEvent::OnCacheGet(path.as_ref().to_path_buf());
+        self.send_cache_event(event);
+    }
+
+    pub(super) fn on_cache_update_async(&self, path: impl AsRef<Path>) {
+        let event = CacheEvent::OnCacheUpdate(path.as_ref().to_path_buf());
+        self.send_cache_event(event);
+    }
+
+    #[inline]
+    fn send_cache_event(&self, event: CacheEvent) {
+        let sent_event = self.sender.try_send(event.clone());
+
+        if let Err(ref err) = sent_event {
+            info!(
+                "Failed to send asynchronously message to worker thread, \
+                 event: {:?}, error: {}",
+                event, err
+            );
+        }
+
+        #[cfg(test)]
+        {
+            let mut stats = self
+                .stats
+                .0
+                .lock()
+                .expect("Failed to acquire worker stats lock");
+
+            if sent_event.is_ok() {
+                stats.sent += 1;
+            } else {
+                stats.dropped += 1;
+            }
+        }
+    }
+
+    #[cfg(test)]
+    pub(super) fn events_dropped(&self) -> u32 {
+        let stats = self
+            .stats
+            .0
+            .lock()
+            .expect("Failed to acquire worker stats lock");
+        stats.dropped
+    }
+
+    #[cfg(test)]
+    pub(super) fn wait_for_all_events_handled(&self) {
+        let (stats, condvar) = &*self.stats;
+        let mut stats = stats.lock().expect("Failed to acquire worker stats lock");
+        while stats.handled != stats.sent {
+            stats = condvar
+                .wait(stats)
+                .expect("Failed to reacquire worker stats lock");
+        }
+    }
+}
+
+impl fmt::Debug for Worker {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Worker").finish()
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+struct ModuleCacheStatistics {
+    pub usages: u64,
+    #[serde(rename = "optimized-compression")]
+    pub compression_level: i32,
+}
+
+impl ModuleCacheStatistics {
+    fn default(cache_config: &CacheConfig) -> Self {
+        Self {
+            usages: 0,
+            compression_level: cache_config.baseline_compression_level(),
+        }
+    }
+}
+
+enum CacheEntry {
+    Recognized {
+        path: PathBuf,
+        mtime: SystemTime,
+        size: u64,
+    },
+    Unrecognized {
+        path: PathBuf,
+        is_dir: bool,
+    },
+}
+
+macro_rules! unwrap_or_warn {
+    ($result:expr, $cont:stmt, $err_msg:expr, $path:expr) => {
+        match $result {
+            Ok(val) => val,
+            Err(err) => {
+                warn!("{}, path: {}, msg: {}", $err_msg, $path.display(), err);
+                $cont
+            }
+        }
+    };
+}
+
+impl WorkerThread {
+    fn run(self, init_file_per_thread_logger: Option<&'static str>) {
+        if let Some(prefix) = init_file_per_thread_logger {
+            file_per_thread_logger::initialize(prefix);
+        }
+
+        debug!("Cache worker thread started.");
+
+        Self::lower_thread_priority();
+
+        #[cfg(test)]
+        let (stats, condvar) = &*self.stats;
+
+        for event in self.receiver.iter() {
+            match event {
+                CacheEvent::OnCacheGet(path) => self.handle_on_cache_get(path),
+                CacheEvent::OnCacheUpdate(path) => self.handle_on_cache_update(path),
+            }
+
+            #[cfg(test)]
+            {
+                let mut stats = stats.lock().expect("Failed to acquire worker stats lock");
+                stats.handled += 1;
+                condvar.notify_all();
+            }
+        }
+    }
+
+    #[cfg(target_os = "fuchsia")]
+    fn lower_thread_priority() {
+        // TODO This needs to use Fuchsia thread profiles
+        // https://fuchsia.dev/fuchsia-src/reference/kernel_objects/profile
+        warn!(
+            "Lowering thread priority on Fuchsia is currently a noop. It might affect application performance."
+        );
+    }
+
+    #[cfg(target_os = "windows")]
+    fn lower_thread_priority() {
+        use std::convert::TryInto;
+        use winapi::um::processthreadsapi::{GetCurrentThread, SetThreadPriority};
+        use winapi::um::winbase::THREAD_MODE_BACKGROUND_BEGIN;
+
+        // https://docs.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-setthreadpriority
+        // https://docs.microsoft.com/en-us/windows/win32/procthread/scheduling-priorities
+
+        if unsafe {
+            SetThreadPriority(
+                GetCurrentThread(),
+                THREAD_MODE_BACKGROUND_BEGIN.try_into().unwrap(),
+            )
+        } == 0
+        {
+            warn!(
+                "Failed to lower worker thread priority. It might affect application performance."
+            );
+        }
+    }
+
+    #[cfg(not(any(target_os = "windows", target_os = "fuchsia")))]
+    fn lower_thread_priority() {
+        // http://man7.org/linux/man-pages/man7/sched.7.html
+
+        const NICE_DELTA_FOR_BACKGROUND_TASKS: i32 = 3;
+
+        errno::set_errno(errno::Errno(0));
+        let current_nice = unsafe { libc::nice(NICE_DELTA_FOR_BACKGROUND_TASKS) };
+        let errno_val = errno::errno().0;
+
+        if errno_val != 0 {
+            warn!(
+                "Failed to lower worker thread priority. It might affect application performance. \
+                 errno: {}",
+                errno_val
+            );
+        } else {
+            debug!("New nice value of worker thread: {}", current_nice);
+        }
+    }
+
+    /// Increases the usage counter and recompresses the file
+    /// if the usage counter reached configurable treshold.
+    fn handle_on_cache_get(&self, path: PathBuf) {
+        trace!("handle_on_cache_get() for path: {}", path.display());
+
+        // construct .stats file path
+        let filename = path.file_name().unwrap().to_str().unwrap();
+        let stats_path = path.with_file_name(format!("{}.stats", filename));
+
+        // load .stats file (default if none or error)
+        let mut stats = read_stats_file(stats_path.as_ref())
+            .unwrap_or_else(|| ModuleCacheStatistics::default(&self.cache_config));
+
+        // step 1: update the usage counter & write to the disk
+        //         it's racy, but it's fine (the counter will be just smaller,
+        //         sometimes will retrigger recompression)
+        stats.usages += 1;
+        if !write_stats_file(stats_path.as_ref(), &stats) {
+            return;
+        }
+
+        // step 2: recompress if there's a need
+        let opt_compr_lvl = self.cache_config.optimized_compression_level();
+        if stats.compression_level >= opt_compr_lvl
+            || stats.usages
+                < self
+                    .cache_config
+                    .optimized_compression_usage_counter_threshold()
+        {
+            return;
+        }
+
+        let lock_path = if let Some(p) = acquire_task_fs_lock(
+            path.as_ref(),
+            self.cache_config.optimizing_compression_task_timeout(),
+            self.cache_config
+                .allowed_clock_drift_for_files_from_future(),
+        ) {
+            p
+        } else {
+            return;
+        };
+
+        trace!("Trying to recompress file: {}", path.display());
+
+        // recompress, write to other file, rename (it's atomic file content exchange)
+        // and update the stats file
+        let compressed_cache_bytes = unwrap_or_warn!(
+            fs::read(&path),
+            return,
+            "Failed to read old cache file",
+            path
+        );
+
+        let cache_bytes = unwrap_or_warn!(
+            zstd::decode_all(&compressed_cache_bytes[..]),
+            return,
+            "Failed to decompress cached code",
+            path
+        );
+
+        let recompressed_cache_bytes = unwrap_or_warn!(
+            zstd::encode_all(&cache_bytes[..], opt_compr_lvl),
+            return,
+            "Failed to compress cached code",
+            path
+        );
+
+        unwrap_or_warn!(
+            fs::write(&lock_path, &recompressed_cache_bytes),
+            return,
+            "Failed to write recompressed cache",
+            lock_path
+        );
+
+        unwrap_or_warn!(
+            fs::rename(&lock_path, &path),
+            {
+                if let Err(error) = fs::remove_file(&lock_path) {
+                    warn!(
+                        "Failed to clean up (remove) recompressed cache, path {}, err: {}",
+                        lock_path.display(),
+                        error
+                    );
+                }
+
+                return;
+            },
+            "Failed to rename recompressed cache",
+            lock_path
+        );
+
+        // update stats file (reload it! recompression can take some time)
+        if let Some(mut new_stats) = read_stats_file(stats_path.as_ref()) {
+            if new_stats.compression_level >= opt_compr_lvl {
+                // Rare race:
+                //    two instances with different opt_compr_lvl: we don't know in which order they updated
+                //    the cache file and the stats file (they are not updated together atomically)
+                // Possible solution is to use directories per cache entry, but it complicates the system
+                // and is not worth it.
+                debug!(
+                    "DETECTED task did more than once (or race with new file): \
+                     recompression of {}. Note: if optimized compression level setting \
+                     has changed in the meantine, the stats file might contain \
+                     inconsistent compression level due to race.",
+                    path.display()
+                );
+            } else {
+                new_stats.compression_level = opt_compr_lvl;
+                let _ = write_stats_file(stats_path.as_ref(), &new_stats);
+            }
+
+            if new_stats.usages < stats.usages {
+                debug!(
+                    "DETECTED lower usage count (new file or race with counter \
+                     increasing): file {}",
+                    path.display()
+                );
+            }
+        } else {
+            debug!(
+                "Can't read stats file again to update compression level (it might got \
+                 cleaned up): file {}",
+                stats_path.display()
+            );
+        }
+
+        trace!("Task finished: recompress file: {}", path.display());
+    }
+
+    fn handle_on_cache_update(&self, path: PathBuf) {
+        trace!("handle_on_cache_update() for path: {}", path.display());
+
+        // ---------------------- step 1: create .stats file
+
+        // construct .stats file path
+        let filename = path
+            .file_name()
+            .expect("Expected valid cache file name")
+            .to_str()
+            .expect("Expected valid cache file name");
+        let stats_path = path.with_file_name(format!("{}.stats", filename));
+
+        // create and write stats file
+        let mut stats = ModuleCacheStatistics::default(&self.cache_config);
+        stats.usages += 1;
+        write_stats_file(&stats_path, &stats);
+
+        // ---------------------- step 2: perform cleanup task if needed
+
+        // acquire lock for cleanup task
+        // Lock is a proof of recent cleanup task, so we don't want to delete them.
+        // Expired locks will be deleted by the cleanup task.
+        let cleanup_file = self.cache_config.directory().join(".cleanup"); // some non existing marker file
+        if acquire_task_fs_lock(
+            &cleanup_file,
+            self.cache_config.cleanup_interval(),
+            self.cache_config
+                .allowed_clock_drift_for_files_from_future(),
+        )
+        .is_none()
+        {
+            return;
+        }
+
+        trace!("Trying to clean up cache");
+
+        let mut cache_index = self.list_cache_contents();
+        let future_tolerance = SystemTime::now()
+            .checked_add(
+                self.cache_config
+                    .allowed_clock_drift_for_files_from_future(),
+            )
+            .expect("Brace your cache, the next Big Bang is coming (time overflow)");
+        cache_index.sort_unstable_by(|lhs, rhs| {
+            // sort by age
+            use CacheEntry::*;
+            match (lhs, rhs) {
+                (Recognized { mtime: lhs_mt, .. }, Recognized { mtime: rhs_mt, .. }) => {
+                    match (*lhs_mt > future_tolerance, *rhs_mt > future_tolerance) {
+                        // later == younger
+                        (false, false) => rhs_mt.cmp(lhs_mt),
+                        // files from far future are treated as oldest recognized files
+                        // we want to delete them, so the cache keeps track of recent files
+                        // however, we don't delete them uncodintionally,
+                        // because .stats file can be overwritten with a meaningful mtime
+                        (true, false) => cmp::Ordering::Greater,
+                        (false, true) => cmp::Ordering::Less,
+                        (true, true) => cmp::Ordering::Equal,
+                    }
+                }
+                // unrecognized is kind of infinity
+                (Recognized { .. }, Unrecognized { .. }) => cmp::Ordering::Less,
+                (Unrecognized { .. }, Recognized { .. }) => cmp::Ordering::Greater,
+                (Unrecognized { .. }, Unrecognized { .. }) => cmp::Ordering::Equal,
+            }
+        });
+
+        // find "cut" boundary:
+        // - remove unrecognized files anyway,
+        // - remove some cache files if some quota has been exceeded
+        let mut total_size = 0u64;
+        let mut start_delete_idx = None;
+        let mut start_delete_idx_if_deleting_recognized_items: Option<usize> = None;
+
+        let total_size_limit = self.cache_config.files_total_size_soft_limit();
+        let file_count_limit = self.cache_config.file_count_soft_limit();
+        let tsl_if_deleting = total_size_limit
+            .checked_mul(
+                self.cache_config
+                    .files_total_size_limit_percent_if_deleting() as u64,
+            )
+            .unwrap()
+            / 100;
+        let fcl_if_deleting = file_count_limit
+            .checked_mul(self.cache_config.file_count_limit_percent_if_deleting() as u64)
+            .unwrap()
+            / 100;
+
+        for (idx, item) in cache_index.iter().enumerate() {
+            let size = if let CacheEntry::Recognized { size, .. } = item {
+                size
+            } else {
+                start_delete_idx = Some(idx);
+                break;
+            };
+
+            total_size += size;
+            if start_delete_idx_if_deleting_recognized_items.is_none()
+                && (total_size > tsl_if_deleting || (idx + 1) as u64 > fcl_if_deleting)
+            {
+                start_delete_idx_if_deleting_recognized_items = Some(idx);
+            }
+
+            if total_size > total_size_limit || (idx + 1) as u64 > file_count_limit {
+                start_delete_idx = start_delete_idx_if_deleting_recognized_items;
+                break;
+            }
+        }
+
+        if let Some(idx) = start_delete_idx {
+            for item in &cache_index[idx..] {
+                let (result, path, entity) = match item {
+                    CacheEntry::Recognized { path, .. }
+                    | CacheEntry::Unrecognized {
+                        path,
+                        is_dir: false,
+                    } => (fs::remove_file(path), path, "file"),
+                    CacheEntry::Unrecognized { path, is_dir: true } => {
+                        (fs::remove_dir_all(path), path, "directory")
+                    }
+                };
+                if let Err(err) = result {
+                    warn!(
+                        "Failed to remove {} during cleanup, path: {}, err: {}",
+                        entity,
+                        path.display(),
+                        err
+                    );
+                }
+            }
+        }
+
+        trace!("Task finished: clean up cache");
+    }
+
+    // Be fault tolerant: list as much as you can, and ignore the rest
+    fn list_cache_contents(&self) -> Vec<CacheEntry> {
+        fn enter_dir(
+            vec: &mut Vec<CacheEntry>,
+            dir_path: &Path,
+            level: u8,
+            cache_config: &CacheConfig,
+        ) {
+            macro_rules! add_unrecognized {
+                (file: $path:expr) => {
+                    add_unrecognized!(false, $path)
+                };
+                (dir: $path:expr) => {
+                    add_unrecognized!(true, $path)
+                };
+                ($is_dir:expr, $path:expr) => {
+                    vec.push(CacheEntry::Unrecognized {
+                        path: $path.to_path_buf(),
+                        is_dir: $is_dir,
+                    });
+                };
+            }
+            macro_rules! add_unrecognized_and {
+                ([ $( $ty:ident: $path:expr ),* ], $cont:stmt) => {{
+                    $( add_unrecognized!($ty: $path); )*
+                        $cont
+                }};
+            }
+
+            macro_rules! unwrap_or {
+                ($result:expr, $cont:stmt, $err_msg:expr) => {
+                    unwrap_or!($result, $cont, $err_msg, dir_path)
+                };
+                ($result:expr, $cont:stmt, $err_msg:expr, $path:expr) => {
+                    unwrap_or_warn!(
+                        $result,
+                        $cont,
+                        format!("{}, level: {}", $err_msg, level),
+                        $path
+                    )
+                };
+            }
+
+            // If we fail to list a directory, something bad is happening anyway
+            // (something touches our cache or we have disk failure)
+            // Try to delete it, so we can stay within soft limits of the cache size.
+            // This comment applies later in this function, too.
+            let it = unwrap_or!(
+                fs::read_dir(dir_path),
+                add_unrecognized_and!([dir: dir_path], return),
+                "Failed to list cache directory, deleting it"
+            );
+
+            let mut cache_files = HashMap::new();
+            for entry in it {
+                // read_dir() returns an iterator over results - in case some of them are errors
+                // we don't know their names, so we can't delete them. We don't want to delete
+                // the whole directory with good entries too, so we just ignore the erroneous entries.
+                let entry = unwrap_or!(
+                    entry,
+                    continue,
+                    "Failed to read a cache dir entry (NOT deleting it, it still occupies space)"
+                );
+                let path = entry.path();
+                match (level, path.is_dir()) {
+                    (0..=1, true) => enter_dir(vec, &path, level + 1, cache_config),
+                    (0..=1, false) => {
+                        if level == 0
+                            && path.file_stem() == Some(OsStr::new(".cleanup"))
+                                && path.extension().is_some()
+                                // assume it's cleanup lock
+                                && !is_fs_lock_expired(
+                                    Some(&entry),
+                                    &path,
+                                    cache_config.cleanup_interval(),
+                                    cache_config.allowed_clock_drift_for_files_from_future(),
+                                )
+                        {
+                            continue; // skip active lock
+                        }
+                        add_unrecognized!(file: path);
+                    }
+                    (2, false) => {
+                        match path.extension().and_then(OsStr::to_str) {
+                            // mod or stats file
+                            None | Some("stats") => {
+                                cache_files.insert(path, entry);
+                            }
+
+                            Some(ext) => {
+                                // check if valid lock
+                                let recognized = ext.starts_with("wip-")
+                                    && !is_fs_lock_expired(
+                                        Some(&entry),
+                                        &path,
+                                        cache_config.optimizing_compression_task_timeout(),
+                                        cache_config.allowed_clock_drift_for_files_from_future(),
+                                    );
+
+                                if !recognized {
+                                    add_unrecognized!(file: path);
+                                }
+                            }
+                        }
+                    }
+                    (_, is_dir) => add_unrecognized!(is_dir, path),
+                }
+            }
+
+            // associate module with its stats & handle them
+            // assumption: just mods and stats
+            for (path, entry) in cache_files.iter() {
+                let path_buf: PathBuf;
+                let (mod_, stats_, is_mod) = match path.extension() {
+                    Some(_) => {
+                        path_buf = path.with_extension("");
+                        (
+                            cache_files.get(&path_buf).map(|v| (&path_buf, v)),
+                            Some((path, entry)),
+                            false,
+                        )
+                    }
+                    None => {
+                        path_buf = path.with_extension("stats");
+                        (
+                            Some((path, entry)),
+                            cache_files.get(&path_buf).map(|v| (&path_buf, v)),
+                            true,
+                        )
+                    }
+                };
+
+                // construct a cache entry
+                match (mod_, stats_, is_mod) {
+                    (Some((mod_path, mod_entry)), Some((stats_path, stats_entry)), true) => {
+                        let mod_metadata = unwrap_or!(
+                            mod_entry.metadata(),
+                            add_unrecognized_and!([file: stats_path, file: mod_path], continue),
+                            "Failed to get metadata, deleting BOTH module cache and stats files",
+                            mod_path
+                        );
+                        let stats_mtime = unwrap_or!(
+                            stats_entry.metadata().and_then(|m| m.modified()),
+                            add_unrecognized_and!(
+                                [file: stats_path],
+                                unwrap_or!(
+                                    mod_metadata.modified(),
+                                    add_unrecognized_and!(
+                                        [file: stats_path, file: mod_path],
+                                        continue
+                                    ),
+                                    "Failed to get mtime, deleting BOTH module cache and stats \
+                                     files",
+                                    mod_path
+                                )
+                            ),
+                            "Failed to get metadata/mtime, deleting the file",
+                            stats_path
+                        );
+                        // .into() called for the SystemTimeStub if cfg(test)
+                        #[allow(clippy::identity_conversion)]
+                        vec.push(CacheEntry::Recognized {
+                            path: mod_path.to_path_buf(),
+                            mtime: stats_mtime.into(),
+                            size: mod_metadata.len(),
+                        })
+                    }
+                    (Some(_), Some(_), false) => (), // was or will be handled by previous branch
+                    (Some((mod_path, mod_entry)), None, _) => {
+                        let (mod_metadata, mod_mtime) = unwrap_or!(
+                            mod_entry
+                                .metadata()
+                                .and_then(|md| md.modified().map(|mt| (md, mt))),
+                            add_unrecognized_and!([file: mod_path], continue),
+                            "Failed to get metadata/mtime, deleting the file",
+                            mod_path
+                        );
+                        // .into() called for the SystemTimeStub if cfg(test)
+                        #[allow(clippy::identity_conversion)]
+                        vec.push(CacheEntry::Recognized {
+                            path: mod_path.to_path_buf(),
+                            mtime: mod_mtime.into(),
+                            size: mod_metadata.len(),
+                        })
+                    }
+                    (None, Some((stats_path, _stats_entry)), _) => {
+                        debug!("Found orphaned stats file: {}", stats_path.display());
+                        add_unrecognized!(file: stats_path);
+                    }
+                    _ => unreachable!(),
+                }
+            }
+        }
+
+        let mut vec = Vec::new();
+        enter_dir(
+            &mut vec,
+            self.cache_config.directory(),
+            0,
+            &self.cache_config,
+        );
+        vec
+    }
+}
+
+fn read_stats_file(path: &Path) -> Option<ModuleCacheStatistics> {
+    fs::read(path)
+        .map_err(|err| {
+            trace!(
+                "Failed to read stats file, path: {}, err: {}",
+                path.display(),
+                err
+            )
+        })
+        .and_then(|bytes| {
+            toml::from_slice::<ModuleCacheStatistics>(&bytes[..]).map_err(|err| {
+                trace!(
+                    "Failed to parse stats file, path: {}, err: {}",
+                    path.display(),
+                    err,
+                )
+            })
+        })
+        .ok()
+}
+
+fn write_stats_file(path: &Path, stats: &ModuleCacheStatistics) -> bool {
+    toml::to_string_pretty(&stats)
+        .map_err(|err| {
+            warn!(
+                "Failed to serialize stats file, path: {}, err: {}",
+                path.display(),
+                err
+            )
+        })
+        .and_then(|serialized| {
+            if fs_write_atomic(path, "stats", serialized.as_bytes()) {
+                Ok(())
+            } else {
+                Err(())
+            }
+        })
+        .is_ok()
+}
+
+/// Tries to acquire a lock for specific task.
+///
+/// Returns Some(path) to the lock if succeeds. The task path must not
+/// contain any extension and have file stem.
+///
+/// To release a lock you need either manually rename or remove it,
+/// or wait until it expires and cleanup task removes it.
+///
+/// Note: this function is racy. Main idea is: be fault tolerant and
+///       never block some task. The price is that we rarely do some task
+///       more than once.
+fn acquire_task_fs_lock(
+    task_path: &Path,
+    timeout: Duration,
+    allowed_future_drift: Duration,
+) -> Option<PathBuf> {
+    assert!(task_path.extension().is_none());
+    assert!(task_path.file_stem().is_some());
+
+    // list directory
+    let dir_path = task_path.parent()?;
+    let it = fs::read_dir(dir_path)
+        .map_err(|err| {
+            warn!(
+                "Failed to list cache directory, path: {}, err: {}",
+                dir_path.display(),
+                err
+            )
+        })
+        .ok()?;
+
+    // look for existing locks
+    for entry in it {
+        let entry = entry
+            .map_err(|err| {
+                warn!(
+                    "Failed to list cache directory, path: {}, err: {}",
+                    dir_path.display(),
+                    err
+                )
+            })
+            .ok()?;
+
+        let path = entry.path();
+        if path.is_dir() || path.file_stem() != task_path.file_stem() {
+            continue;
+        }
+
+        // check extension and mtime
+        match path.extension() {
+            None => continue,
+            Some(ext) => {
+                if let Some(ext_str) = ext.to_str() {
+                    // if it's None, i.e. not valid UTF-8 string, then that's not our lock for sure
+                    if ext_str.starts_with("wip-")
+                        && !is_fs_lock_expired(Some(&entry), &path, timeout, allowed_future_drift)
+                    {
+                        return None;
+                    }
+                }
+            }
+        }
+    }
+
+    // create the lock
+    let lock_path = task_path.with_extension(format!("wip-{}", std::process::id()));
+    let _file = fs::OpenOptions::new()
+        .create_new(true)
+        .write(true)
+        .open(&lock_path)
+        .map_err(|err| {
+            warn!(
+                "Failed to create lock file (note: it shouldn't exists): path: {}, err: {}",
+                lock_path.display(),
+                err
+            )
+        })
+        .ok()?;
+
+    Some(lock_path)
+}
+
+// we have either both, or just path; dir entry is desirable since on some platforms we can get
+// metadata without extra syscalls
+// futhermore: it's better to get a path if we have it instead of allocating a new one from the dir entry
+fn is_fs_lock_expired(
+    entry: Option<&fs::DirEntry>,
+    path: &PathBuf,
+    threshold: Duration,
+    allowed_future_drift: Duration,
+) -> bool {
+    let mtime = match entry
+        .map_or_else(|| path.metadata(), |e| e.metadata())
+        .and_then(|metadata| metadata.modified())
+    {
+        Ok(mt) => mt,
+        Err(err) => {
+            warn!(
+                "Failed to get metadata/mtime, treating as an expired lock, path: {}, err: {}",
+                path.display(),
+                err
+            );
+            return true; // can't read mtime, treat as expired, so this task will not be starved
+        }
+    };
+
+    // DON'T use: mtime.elapsed() -- we must call SystemTime directly for the tests to be deterministic
+    match SystemTime::now().duration_since(mtime) {
+        Ok(elapsed) => elapsed >= threshold,
+        Err(err) => {
+            trace!(
+                "Found mtime in the future, treating as a not expired lock, path: {}, err: {}",
+                path.display(),
+                err
+            );
+            // the lock is expired if the time is too far in the future
+            // it is fine to have network share and not synchronized clocks,
+            // but it's not good when user changes time in their system clock
+            err.duration() > allowed_future_drift
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/cache/src/worker/tests.rs

@@ -0,0 +1,768 @@
+use super::*;
+use crate::config::tests::test_prolog;
+use more_asserts::{assert_ge, assert_gt, assert_lt};
+use std::iter::repeat;
+use std::process;
+// load_config! comes from crate::cache(::config::tests);
+
+// when doing anything with the tests, make sure they are DETERMINISTIC
+// -- the result shouldn't rely on system time!
+pub mod system_time_stub;
+
+#[test]
+fn test_on_get_create_stats_file() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+
+    let mod_file = cache_dir.join("some-mod");
+    worker.on_cache_get_async(mod_file);
+    worker.wait_for_all_events_handled();
+    assert_eq!(worker.events_dropped(), 0);
+
+    let stats_file = cache_dir.join("some-mod.stats");
+    let stats = read_stats_file(&stats_file).expect("Failed to read stats file");
+    assert_eq!(stats.usages, 1);
+    assert_eq!(
+        stats.compression_level,
+        cache_config.baseline_compression_level()
+    );
+}
+
+#[test]
+fn test_on_get_update_usage_counter() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+
+    let mod_file = cache_dir.join("some-mod");
+    let stats_file = cache_dir.join("some-mod.stats");
+    let default_stats = ModuleCacheStatistics::default(&cache_config);
+    assert!(write_stats_file(&stats_file, &default_stats));
+
+    let mut usages = 0;
+    for times_used in &[4, 7, 2] {
+        for _ in 0..*times_used {
+            worker.on_cache_get_async(mod_file.clone());
+            usages += 1;
+        }
+
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        let stats = read_stats_file(&stats_file).expect("Failed to read stats file");
+        assert_eq!(stats.usages, usages);
+    }
+}
+
+#[test]
+fn test_on_get_recompress_no_mod_file() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         baseline-compression-level = 3\n\
+         optimized-compression-level = 7\n\
+         optimized-compression-usage-counter-threshold = '256'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+
+    let mod_file = cache_dir.join("some-mod");
+    let stats_file = cache_dir.join("some-mod.stats");
+    let mut start_stats = ModuleCacheStatistics::default(&cache_config);
+    start_stats.usages = 250;
+    assert!(write_stats_file(&stats_file, &start_stats));
+
+    let mut usages = start_stats.usages;
+    for times_used in &[4, 7, 2] {
+        for _ in 0..*times_used {
+            worker.on_cache_get_async(mod_file.clone());
+            usages += 1;
+        }
+
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        let stats = read_stats_file(&stats_file).expect("Failed to read stats file");
+        assert_eq!(stats.usages, usages);
+        assert_eq!(
+            stats.compression_level,
+            cache_config.baseline_compression_level()
+        );
+    }
+}
+
+#[test]
+fn test_on_get_recompress_with_mod_file() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         baseline-compression-level = 3\n\
+         optimized-compression-level = 7\n\
+         optimized-compression-usage-counter-threshold = '256'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+
+    let mod_file = cache_dir.join("some-mod");
+    let mod_data = "some test data to be compressed";
+    let data = zstd::encode_all(
+        mod_data.as_bytes(),
+        cache_config.baseline_compression_level(),
+    )
+    .expect("Failed to compress sample mod file");
+    fs::write(&mod_file, &data).expect("Failed to write sample mod file");
+
+    let stats_file = cache_dir.join("some-mod.stats");
+    let mut start_stats = ModuleCacheStatistics::default(&cache_config);
+    start_stats.usages = 250;
+    assert!(write_stats_file(&stats_file, &start_stats));
+
+    // scenarios:
+    // 1. Shouldn't be recompressed
+    // 2. Should be recompressed
+    // 3. After lowering compression level, should be recompressed
+    let scenarios = [(4, false), (7, true), (2, false)];
+
+    let mut usages = start_stats.usages;
+    assert_lt!(
+        usages,
+        cache_config.optimized_compression_usage_counter_threshold()
+    );
+    let mut tested_higher_opt_compr_lvl = false;
+    for (times_used, lower_compr_lvl) in &scenarios {
+        for _ in 0..*times_used {
+            worker.on_cache_get_async(mod_file.clone());
+            usages += 1;
+        }
+
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        let mut stats = read_stats_file(&stats_file).expect("Failed to read stats file");
+        assert_eq!(stats.usages, usages);
+        assert_eq!(
+            stats.compression_level,
+            if usages < cache_config.optimized_compression_usage_counter_threshold() {
+                cache_config.baseline_compression_level()
+            } else {
+                cache_config.optimized_compression_level()
+            }
+        );
+        let compressed_data = fs::read(&mod_file).expect("Failed to read mod file");
+        let decoded_data =
+            zstd::decode_all(&compressed_data[..]).expect("Failed to decompress mod file");
+        assert_eq!(decoded_data, mod_data.as_bytes());
+
+        if *lower_compr_lvl {
+            assert_ge!(
+                usages,
+                cache_config.optimized_compression_usage_counter_threshold()
+            );
+            tested_higher_opt_compr_lvl = true;
+            stats.compression_level -= 1;
+            assert!(write_stats_file(&stats_file, &stats));
+        }
+    }
+    assert_ge!(
+        usages,
+        cache_config.optimized_compression_usage_counter_threshold()
+    );
+    assert!(tested_higher_opt_compr_lvl);
+}
+
+#[test]
+fn test_on_get_recompress_lock() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         baseline-compression-level = 3\n\
+         optimized-compression-level = 7\n\
+         optimized-compression-usage-counter-threshold = '256'\n\
+         optimizing-compression-task-timeout = '30m'\n\
+         allowed-clock-drift-for-files-from-future = '1d'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+
+    let mod_file = cache_dir.join("some-mod");
+    let mod_data = "some test data to be compressed";
+    let data = zstd::encode_all(
+        mod_data.as_bytes(),
+        cache_config.baseline_compression_level(),
+    )
+    .expect("Failed to compress sample mod file");
+    fs::write(&mod_file, &data).expect("Failed to write sample mod file");
+
+    let stats_file = cache_dir.join("some-mod.stats");
+    let mut start_stats = ModuleCacheStatistics::default(&cache_config);
+    start_stats.usages = 255;
+
+    let lock_file = cache_dir.join("some-mod.wip-lock");
+
+    let scenarios = [
+        // valid lock
+        (true, "past", Duration::from_secs(30 * 60 - 1)),
+        // valid future lock
+        (true, "future", Duration::from_secs(24 * 60 * 60)),
+        // expired lock
+        (false, "past", Duration::from_secs(30 * 60)),
+        // expired future lock
+        (false, "future", Duration::from_secs(24 * 60 * 60 + 1)),
+    ];
+
+    for (lock_valid, duration_sign, duration) in &scenarios {
+        assert!(write_stats_file(&stats_file, &start_stats)); // restore usage & compression level
+        create_file_with_mtime(&lock_file, "", duration_sign, &duration);
+
+        worker.on_cache_get_async(mod_file.clone());
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        let stats = read_stats_file(&stats_file).expect("Failed to read stats file");
+        assert_eq!(stats.usages, start_stats.usages + 1);
+        assert_eq!(
+            stats.compression_level,
+            if *lock_valid {
+                cache_config.baseline_compression_level()
+            } else {
+                cache_config.optimized_compression_level()
+            }
+        );
+        let compressed_data = fs::read(&mod_file).expect("Failed to read mod file");
+        let decoded_data =
+            zstd::decode_all(&compressed_data[..]).expect("Failed to decompress mod file");
+        assert_eq!(decoded_data, mod_data.as_bytes());
+    }
+}
+
+#[test]
+fn test_on_update_fresh_stats_file() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         baseline-compression-level = 3\n\
+         optimized-compression-level = 7\n\
+         cleanup-interval = '1h'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+
+    let mod_file = cache_dir.join("some-mod");
+    let stats_file = cache_dir.join("some-mod.stats");
+    let cleanup_certificate = cache_dir.join(".cleanup.wip-done");
+    create_file_with_mtime(&cleanup_certificate, "", "future", &Duration::from_secs(0));
+    // the below created by the worker if it cleans up
+    let worker_lock_file = cache_dir.join(format!(".cleanup.wip-{}", process::id()));
+
+    // scenarios:
+    // 1. Create new stats file
+    // 2. Overwrite existing file
+    for update_file in &[true, false] {
+        worker.on_cache_update_async(mod_file.clone());
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        let mut stats = read_stats_file(&stats_file).expect("Failed to read stats file");
+        assert_eq!(stats.usages, 1);
+        assert_eq!(
+            stats.compression_level,
+            cache_config.baseline_compression_level()
+        );
+
+        if *update_file {
+            stats.usages += 42;
+            stats.compression_level += 1;
+            assert!(write_stats_file(&stats_file, &stats));
+        }
+
+        assert!(!worker_lock_file.exists());
+    }
+}
+
+#[test]
+fn test_on_update_cleanup_limits_trash_locks() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         cleanup-interval = '30m'\n\
+         optimizing-compression-task-timeout = '30m'\n\
+         allowed-clock-drift-for-files-from-future = '1d'\n\
+         file-count-soft-limit = '5'\n\
+         files-total-size-soft-limit = '30K'\n\
+         file-count-limit-percent-if-deleting = '70%'\n\
+         files-total-size-limit-percent-if-deleting = '70%'
+         ",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+    let content_1k = "a".repeat(1_000);
+    let content_10k = "a".repeat(10_000);
+
+    let mods_files_dir = cache_dir.join("target-triple").join("compiler-version");
+    let mod_with_stats = mods_files_dir.join("mod-with-stats");
+    let trash_dirs = [
+        mods_files_dir.join("trash"),
+        mods_files_dir.join("trash").join("trash"),
+    ];
+    let trash_files = [
+        cache_dir.join("trash-file"),
+        cache_dir.join("trash-file.wip-lock"),
+        cache_dir.join("target-triple").join("trash.txt"),
+        cache_dir.join("target-triple").join("trash.txt.wip-lock"),
+        mods_files_dir.join("trash.ogg"),
+        mods_files_dir.join("trash").join("trash.doc"),
+        mods_files_dir.join("trash").join("trash.doc.wip-lock"),
+        mods_files_dir.join("trash").join("trash").join("trash.xls"),
+        mods_files_dir
+            .join("trash")
+            .join("trash")
+            .join("trash.xls.wip-lock"),
+    ];
+    let mod_locks = [
+        // valid lock
+        (
+            mods_files_dir.join("mod0.wip-lock"),
+            true,
+            "past",
+            Duration::from_secs(30 * 60 - 1),
+        ),
+        // valid future lock
+        (
+            mods_files_dir.join("mod1.wip-lock"),
+            true,
+            "future",
+            Duration::from_secs(24 * 60 * 60),
+        ),
+        // expired lock
+        (
+            mods_files_dir.join("mod2.wip-lock"),
+            false,
+            "past",
+            Duration::from_secs(30 * 60),
+        ),
+        // expired future lock
+        (
+            mods_files_dir.join("mod3.wip-lock"),
+            false,
+            "future",
+            Duration::from_secs(24 * 60 * 60 + 1),
+        ),
+    ];
+    // the below created by the worker if it cleans up
+    let worker_lock_file = cache_dir.join(format!(".cleanup.wip-{}", process::id()));
+
+    let scenarios = [
+        // Close to limits, but not reached, only trash deleted
+        (2, 2, 4),
+        // File count limit exceeded
+        (1, 10, 3),
+        // Total size limit exceeded
+        (4, 0, 2),
+        // Both limits exceeded
+        (3, 5, 3),
+    ];
+
+    for (files_10k, files_1k, remaining_files) in &scenarios {
+        let mut secs_ago = 100;
+
+        for d in &trash_dirs {
+            fs::create_dir_all(d).expect("Failed to create directories");
+        }
+        for f in &trash_files {
+            create_file_with_mtime(f, "", "past", &Duration::from_secs(0));
+        }
+        for (f, _, sign, duration) in &mod_locks {
+            create_file_with_mtime(f, "", sign, &duration);
+        }
+
+        let mut mods_paths = vec![];
+        for content in repeat(&content_10k)
+            .take(*files_10k)
+            .chain(repeat(&content_1k).take(*files_1k))
+        {
+            mods_paths.push(mods_files_dir.join(format!("test-mod-{}", mods_paths.len())));
+            create_file_with_mtime(
+                mods_paths.last().unwrap(),
+                content,
+                "past",
+                &Duration::from_secs(secs_ago),
+            );
+            assert_gt!(secs_ago, 0);
+            secs_ago -= 1;
+        }
+
+        // creating .stats file updates mtime what affects test results
+        // so we use a separate nonexistent module here (orphaned .stats will be removed anyway)
+        worker.on_cache_update_async(mod_with_stats.clone());
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        for ent in trash_dirs.iter().chain(trash_files.iter()) {
+            assert!(!ent.exists());
+        }
+        for (f, valid, ..) in &mod_locks {
+            assert_eq!(f.exists(), *valid);
+        }
+        for (idx, path) in mods_paths.iter().enumerate() {
+            let should_exist = idx >= mods_paths.len() - *remaining_files;
+            assert_eq!(path.exists(), should_exist);
+            if should_exist {
+                // cleanup before next iteration
+                fs::remove_file(path).expect("Failed to remove a file");
+            }
+        }
+        fs::remove_file(&worker_lock_file).expect("Failed to remove lock file");
+    }
+}
+
+#[test]
+fn test_on_update_cleanup_lru_policy() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         file-count-soft-limit = '5'\n\
+         files-total-size-soft-limit = '30K'\n\
+         file-count-limit-percent-if-deleting = '80%'\n\
+         files-total-size-limit-percent-if-deleting = '70%'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+    let content_1k = "a".repeat(1_000);
+    let content_5k = "a".repeat(5_000);
+    let content_10k = "a".repeat(10_000);
+
+    let mods_files_dir = cache_dir.join("target-triple").join("compiler-version");
+    fs::create_dir_all(&mods_files_dir).expect("Failed to create directories");
+    let nonexistent_mod_file = cache_dir.join("nonexistent-mod");
+    let orphaned_stats_file = cache_dir.join("orphaned-mod.stats");
+    let worker_lock_file = cache_dir.join(format!(".cleanup.wip-{}", process::id()));
+
+    // content, how long ago created, how long ago stats created (if created), should be alive
+    let scenarios = [
+        &[
+            (&content_10k, 29, None, false),
+            (&content_10k, 28, None, false),
+            (&content_10k, 27, None, false),
+            (&content_1k, 26, None, true),
+            (&content_10k, 25, None, true),
+            (&content_1k, 24, None, true),
+        ],
+        &[
+            (&content_10k, 29, None, false),
+            (&content_10k, 28, None, false),
+            (&content_10k, 27, None, true),
+            (&content_1k, 26, None, true),
+            (&content_5k, 25, None, true),
+            (&content_1k, 24, None, true),
+        ],
+        &[
+            (&content_10k, 29, Some(19), true),
+            (&content_10k, 28, None, false),
+            (&content_10k, 27, None, false),
+            (&content_1k, 26, Some(18), true),
+            (&content_5k, 25, None, true),
+            (&content_1k, 24, None, true),
+        ],
+        &[
+            (&content_10k, 29, Some(19), true),
+            (&content_10k, 28, Some(18), true),
+            (&content_10k, 27, None, false),
+            (&content_1k, 26, Some(17), true),
+            (&content_5k, 25, None, false),
+            (&content_1k, 24, None, false),
+        ],
+        &[
+            (&content_10k, 29, Some(19), true),
+            (&content_10k, 28, None, false),
+            (&content_1k, 27, None, false),
+            (&content_5k, 26, Some(18), true),
+            (&content_1k, 25, None, false),
+            (&content_10k, 24, None, false),
+        ],
+    ];
+
+    for mods in &scenarios {
+        let filenames = (0..mods.len())
+            .map(|i| {
+                (
+                    mods_files_dir.join(format!("mod-{}", i)),
+                    mods_files_dir.join(format!("mod-{}.stats", i)),
+                )
+            })
+            .collect::<Vec<_>>();
+
+        for ((content, mod_secs_ago, create_stats, _), (mod_filename, stats_filename)) in
+            mods.iter().zip(filenames.iter())
+        {
+            create_file_with_mtime(
+                mod_filename,
+                content,
+                "past",
+                &Duration::from_secs(*mod_secs_ago),
+            );
+            if let Some(stats_secs_ago) = create_stats {
+                create_file_with_mtime(
+                    stats_filename,
+                    "cleanup doesn't care",
+                    "past",
+                    &Duration::from_secs(*stats_secs_ago),
+                );
+            }
+        }
+        create_file_with_mtime(
+            &orphaned_stats_file,
+            "cleanup doesn't care",
+            "past",
+            &Duration::from_secs(0),
+        );
+
+        worker.on_cache_update_async(nonexistent_mod_file.clone());
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        assert!(!orphaned_stats_file.exists());
+        for ((_, _, create_stats, alive), (mod_filename, stats_filename)) in
+            mods.iter().zip(filenames.iter())
+        {
+            assert_eq!(mod_filename.exists(), *alive);
+            assert_eq!(stats_filename.exists(), *alive && create_stats.is_some());
+
+            // cleanup for next iteration
+            if *alive {
+                fs::remove_file(&mod_filename).expect("Failed to remove a file");
+                if create_stats.is_some() {
+                    fs::remove_file(&stats_filename).expect("Failed to remove a file");
+                }
+            }
+        }
+
+        fs::remove_file(&worker_lock_file).expect("Failed to remove lock file");
+    }
+}
+
+// clock drift should be applied to mod cache & stats, too
+// however, postpone deleting files to as late as possible
+#[test]
+fn test_on_update_cleanup_future_files() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         allowed-clock-drift-for-files-from-future = '1d'\n\
+         file-count-soft-limit = '3'\n\
+         files-total-size-soft-limit = '1M'\n\
+         file-count-limit-percent-if-deleting = '70%'\n\
+         files-total-size-limit-percent-if-deleting = '70%'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+    let content_1k = "a".repeat(1_000);
+
+    let mods_files_dir = cache_dir.join("target-triple").join("compiler-version");
+    fs::create_dir_all(&mods_files_dir).expect("Failed to create directories");
+    let nonexistent_mod_file = cache_dir.join("nonexistent-mod");
+    // the below created by the worker if it cleans up
+    let worker_lock_file = cache_dir.join(format!(".cleanup.wip-{}", process::id()));
+
+    let scenarios: [&[_]; 5] = [
+        // NOT cleaning up, everythings ok
+        &[
+            (Duration::from_secs(0), None, true),
+            (Duration::from_secs(24 * 60 * 60), None, true),
+        ],
+        // NOT cleaning up, everythings ok
+        &[
+            (Duration::from_secs(0), None, true),
+            (Duration::from_secs(24 * 60 * 60 + 1), None, true),
+        ],
+        // cleaning up, removing files from oldest
+        &[
+            (Duration::from_secs(0), None, false),
+            (Duration::from_secs(24 * 60 * 60), None, true),
+            (Duration::from_secs(1), None, false),
+            (Duration::from_secs(2), None, true),
+        ],
+        // cleaning up, removing files from oldest; deleting file from far future
+        &[
+            (Duration::from_secs(0), None, false),
+            (Duration::from_secs(1), None, true),
+            (Duration::from_secs(24 * 60 * 60 + 1), None, false),
+            (Duration::from_secs(2), None, true),
+        ],
+        // cleaning up, removing files from oldest; file from far future should have .stats from +-now => it's a legitimate file
+        &[
+            (Duration::from_secs(0), None, false),
+            (Duration::from_secs(1), None, false),
+            (
+                Duration::from_secs(24 * 60 * 60 + 1),
+                Some(Duration::from_secs(3)),
+                true,
+            ),
+            (Duration::from_secs(2), None, true),
+        ],
+    ];
+
+    for mods in &scenarios {
+        let filenames = (0..mods.len())
+            .map(|i| {
+                (
+                    mods_files_dir.join(format!("mod-{}", i)),
+                    mods_files_dir.join(format!("mod-{}.stats", i)),
+                )
+            })
+            .collect::<Vec<_>>();
+
+        for ((duration, opt_stats_duration, _), (mod_filename, stats_filename)) in
+            mods.iter().zip(filenames.iter())
+        {
+            create_file_with_mtime(mod_filename, &content_1k, "future", duration);
+            if let Some(stats_duration) = opt_stats_duration {
+                create_file_with_mtime(stats_filename, "", "future", stats_duration);
+            }
+        }
+
+        worker.on_cache_update_async(nonexistent_mod_file.clone());
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        for ((_, opt_stats_duration, alive), (mod_filename, stats_filename)) in
+            mods.iter().zip(filenames.iter())
+        {
+            assert_eq!(mod_filename.exists(), *alive);
+            assert_eq!(
+                stats_filename.exists(),
+                *alive && opt_stats_duration.is_some()
+            );
+            if *alive {
+                fs::remove_file(mod_filename).expect("Failed to remove a file");
+                if opt_stats_duration.is_some() {
+                    fs::remove_file(stats_filename).expect("Failed to remove a file");
+                }
+            }
+        }
+
+        fs::remove_file(&worker_lock_file).expect("Failed to remove lock file");
+    }
+}
+
+// this tests if worker triggered cleanup or not when some cleanup lock/certificate was out there
+#[test]
+fn test_on_update_cleanup_self_lock() {
+    let (_tempdir, cache_dir, config_path) = test_prolog();
+    let cache_config = load_config!(
+        config_path,
+        "[cache]\n\
+         enabled = true\n\
+         directory = {cache_dir}\n\
+         worker-event-queue-size = '16'\n\
+         cleanup-interval = '30m'\n\
+         allowed-clock-drift-for-files-from-future = '1d'",
+        cache_dir
+    );
+    assert!(cache_config.enabled());
+    let worker = Worker::start_new(&cache_config, None);
+
+    let mod_file = cache_dir.join("some-mod");
+    let trash_file = cache_dir.join("trash-file.txt");
+
+    let lock_file = cache_dir.join(".cleanup.wip-lock");
+    // the below created by the worker if it cleans up
+    let worker_lock_file = cache_dir.join(format!(".cleanup.wip-{}", process::id()));
+
+    let scenarios = [
+        // valid lock
+        (true, "past", Duration::from_secs(30 * 60 - 1)),
+        // valid future lock
+        (true, "future", Duration::from_secs(24 * 60 * 60)),
+        // expired lock
+        (false, "past", Duration::from_secs(30 * 60)),
+        // expired future lock
+        (false, "future", Duration::from_secs(24 * 60 * 60 + 1)),
+    ];
+
+    for (lock_valid, duration_sign, duration) in &scenarios {
+        create_file_with_mtime(
+            &trash_file,
+            "with trash content",
+            "future",
+            &Duration::from_secs(0),
+        );
+        create_file_with_mtime(&lock_file, "", duration_sign, &duration);
+
+        worker.on_cache_update_async(mod_file.clone());
+        worker.wait_for_all_events_handled();
+        assert_eq!(worker.events_dropped(), 0);
+
+        assert_eq!(trash_file.exists(), *lock_valid);
+        assert_eq!(lock_file.exists(), *lock_valid);
+        if *lock_valid {
+            assert!(!worker_lock_file.exists());
+        } else {
+            fs::remove_file(&worker_lock_file).expect("Failed to remove lock file");
+        }
+    }
+}
+
+fn create_file_with_mtime(filename: &Path, contents: &str, offset_sign: &str, offset: &Duration) {
+    fs::write(filename, contents).expect("Failed to create a file");
+    let mtime = match offset_sign {
+        "past" => system_time_stub::NOW
+            .checked_sub(*offset)
+            .expect("Failed to calculate new mtime"),
+        "future" => system_time_stub::NOW
+            .checked_add(*offset)
+            .expect("Failed to calculate new mtime"),
+        _ => unreachable!(),
+    };
+    filetime::set_file_mtime(filename, mtime.into()).expect("Failed to set mtime");
+}

crates/cache/src/worker/tests/system_time_stub.rs

@@ -0,0 +1,29 @@
+use lazy_static::lazy_static;
+use std::time::{Duration, SystemTime, SystemTimeError};
+
+lazy_static! {
+    pub static ref NOW: SystemTime = SystemTime::now(); // no need for RefCell and set_now() for now
+}
+
+#[derive(PartialOrd, PartialEq, Ord, Eq)]
+pub struct SystemTimeStub(SystemTime);
+
+impl SystemTimeStub {
+    pub fn now() -> Self {
+        Self(*NOW)
+    }
+
+    pub fn checked_add(&self, duration: Duration) -> Option<Self> {
+        self.0.checked_add(duration).map(|t| t.into())
+    }
+
+    pub fn duration_since(&self, earlier: SystemTime) -> Result<Duration, SystemTimeError> {
+        self.0.duration_since(earlier)
+    }
+}
+
+impl From<SystemTime> for SystemTimeStub {
+    fn from(time: SystemTime) -> Self {
+        Self(time)
+    }
+}