fix(obs): trim only what the log byte budget needs

Lifting the minimum file age for an over-budget pass merged fresh files into the regular selection, so the keep_files ceiling and the per-file limit applied to them too: one byte over the budget could process every fresh file but keep_files of them. The override is now a byte trim on the files that would remain, oldest first, until the budget holds.

An unset single-file limit now rolls at a quarter of a small byte budget instead of 256 MiB, so a freshly rolled file is not larger than the whole budget. Both logging setups open the appender through one function that returns the roll signal the cleanup task listens on. The README states that a log directory must not be shared by processes whose file names match the same pattern.

crates/config/src/observability/mod.rs

@@ -91,7 +91,12 @@ pub const ENV_OBS_LOG_MATCH_MODE: &str = "RUSTFS_OBS_LOG_MATCH_MODE";
 
 /// Default values for log cleanup
 pub const DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES: u64 = 2 * 1024 * 1024 * 1024; // 2 GiB
-pub const DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES: u64 = 0; // No single file limit
+pub const DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES: u64 = 0; // No per-file trim by the cleaner
+/// Size at which the active log file is rolled when
+/// `RUSTFS_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES` is unset. Without a size roll
+/// the active file is bounded only by the rotation period, and the cleaner
+/// never touches the active file. An explicit `0` keeps size rolling off.
+pub const DEFAULT_OBS_LOG_ACTIVE_FILE_ROLL_SIZE_BYTES: u64 = 256 * 1024 * 1024; // 256 MiB
 pub const DEFAULT_OBS_LOG_COMPRESS_OLD_FILES: bool = true;
 pub const DEFAULT_OBS_LOG_GZIP_COMPRESSION_LEVEL: u32 = 6;
 pub const DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM_GZIP: &str = "gzip";

crates/obs/README.md

@@ -145,7 +145,7 @@ With OTLP → Collector → Prometheus, the time from a failure to an alert also
 
 | Variable                        | Default      | Description                                                                                                                                                                                                                                                                                                                                                           |
 |---------------------------------|--------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `RUSTFS_OBS_LOGGER_LEVEL`       | `info`       | Log level; `RUST_LOG` syntax supported                                                                                                                                                                                                                                                                                                                                |
+| `RUSTFS_OBS_LOGGER_LEVEL`       | `error`      | Log level; `RUST_LOG` syntax supported                                                                                                                                                                                                                                                                                                                                |
 | `RUSTFS_OBS_LOG_STDOUT_ENABLED` | `false`      | When file logging is active, also mirror to stdout                                                                                                                                                                                                                                                                                                                    |
 | `RUSTFS_OBS_LOG_DIRECTORY`      | _(empty)_    | **Directory for rolling log files. When empty, logs go to stdout only**                                                                                                                                                                                                                                                                                               |
 | `RUSTFS_OBS_LOG_FILENAME`       | `rustfs.log` | Base filename for rolling logs. Rotated archives include a high-precision timestamp and counter. With the default `RUSTFS_OBS_LOG_MATCH_MODE=suffix`, names look like `<timestamp>-<counter>.rustfs.log` (e.g., `20231027103001.123456-0.rustfs.log`); with `prefix`, they look like `rustfs.log.<timestamp>-<counter>` (e.g., `rustfs.log.20231027103001.123456-0`). |
@@ -157,23 +157,31 @@ With OTLP → Collector → Prometheus, the time from a failure to an alert also
 
 | Variable                                        | Default      | Description                                                 |
 |-------------------------------------------------|--------------|-------------------------------------------------------------|
-| `RUSTFS_OBS_LOG_MAX_TOTAL_SIZE_BYTES`           | `2147483648` | Hard cap on total log directory size (2 GiB)                |
-| `RUSTFS_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES`     | `0`          | Per-file size cap; `0` = unlimited                          |
+| `RUSTFS_OBS_LOG_MAX_TOTAL_SIZE_BYTES`           | `2147483648` | Byte budget (2 GiB), applied to rolled logs and to compressed archives separately; `0` = no budget. See [Disk usage bound](#disk-usage-bound) |
+| `RUSTFS_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES`     | unset        | Unset or unparsable: the active file rolls at 256 MiB, or at a quarter of the byte budget when that is smaller (at least 1 MiB). A value: the active file rolls at that size and the cleaner also processes rolled files above it. `0` = no size roll |
 | `RUSTFS_OBS_LOG_COMPRESS_OLD_FILES`             | `true`       | Compress files before deleting                              |
 | `RUSTFS_OBS_LOG_GZIP_COMPRESSION_LEVEL`         | `6`          | Gzip level `1` (fastest) – `9` (best)                       |
 | `RUSTFS_OBS_LOG_COMPRESSION_ALGORITHM`          | `zstd`       | Compression codec: `zstd` or `gzip`                         |
 | `RUSTFS_OBS_LOG_PARALLEL_COMPRESS`              | `true`       | Enable work-stealing parallel compression                   |
 | `RUSTFS_OBS_LOG_PARALLEL_WORKERS`               | `6`          | Number of cleaner worker threads                            |
-| `RUSTFS_OBS_LOG_ZSTD_COMPRESSION_LEVEL`         | `8`          | Zstd level `1` (fastest) – `21` (best ratio)               |
+| `RUSTFS_OBS_LOG_ZSTD_COMPRESSION_LEVEL`         | `8`          | Zstd level `1` (fastest) – `22` (best ratio); `0` = codec default |
 | `RUSTFS_OBS_LOG_ZSTD_FALLBACK_TO_GZIP`          | `true`       | Fallback to gzip when zstd compression fails                |
 | `RUSTFS_OBS_LOG_ZSTD_WORKERS`                   | `1`          | zstdmt worker threads per compression task                  |
 | `RUSTFS_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS` | `30`         | Delete `.gz` / `.zst` archives older than N days; `0` = keep forever |
 | `RUSTFS_OBS_LOG_EXCLUDE_PATTERNS`               | _(empty)_    | Comma-separated glob patterns to never clean up             |
 | `RUSTFS_OBS_LOG_DELETE_EMPTY_FILES`             | `true`       | Remove zero-byte files                                      |
-| `RUSTFS_OBS_LOG_MIN_FILE_AGE_SECONDS`           | `3600`       | Minimum file age (seconds) before cleanup                   |
-| `RUSTFS_OBS_LOG_CLEANUP_INTERVAL_SECONDS`       | `1800`       | How often the cleanup task runs (0.5 hours)                 |
+| `RUSTFS_OBS_LOG_MIN_FILE_AGE_SECONDS`           | `3600`       | Minimum file age (seconds) before cleanup; not applied while rolled logs exceed the byte budget |
+| `RUSTFS_OBS_LOG_CLEANUP_INTERVAL_SECONDS`       | `1800`       | How often the cleanup task runs (0.5 hours); a pass also runs after every roll |
 | `RUSTFS_OBS_LOG_DRY_RUN`                        | `false`      | Report deletions without actually removing files            |
 
+### Disk usage bound
+
+With the defaults the log directory holds at most the byte budget of rolled logs, the same budget of compressed archives, and one active file of up to 256 MiB. The active file is rolled by size as well as by time, and the cleaner runs a pass after every roll in addition to its interval, so the budget is enforced when a file is rolled. While the rolled logs exceed the budget the cleaner processes the oldest ones, as many as the budget requires, even if they are younger than `RUSTFS_OBS_LOG_MIN_FILE_AGE_SECONDS`; `RUSTFS_OBS_LOG_KEEP_FILES` still applies only to files past that age.
+
+The bound is not instantaneous: the directory can exceed it by what is written while one cleanup pass runs, archives created by a pass are trimmed by the next one, and a cleaner that compresses more slowly than logs are produced falls behind. Setting `RUSTFS_OBS_LOG_MAX_TOTAL_SIZE_BYTES=0` removes the budget and `RUSTFS_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES=0` turns size rolling off, which leaves the active file bounded only by the rotation period. Neither setting limits how fast logs are produced.
+
+The cleaner recognizes only its own process's active file. Do not let two processes write log files that match the same pattern in one directory: a file another process is still writing looks like a rolled log, and the minimum file age no longer shields it once the budget is exceeded.
+
 ---
 
 ## Cleaner & Rotation Metrics

crates/obs/src/cleaner/core.rs

@@ -149,6 +149,7 @@ impl LogCleaner {
 
         let LogScanResult {
             mut logs,
+            recent_logs,
             mut compressed_archives,
         } = scan_log_directory(
             &self.log_dir,
@@ -161,8 +162,9 @@ impl LogCleaner {
             self.dry_run,
         )?;
 
+        logs.sort_by_key(|f| f.modified);
+        let mut to_delete = Vec::new();
         if !logs.is_empty() {
-            logs.sort_by_key(|f| f.modified);
             let total_size: u64 = logs.iter().map(|f| f.size).sum();
 
             info!(
@@ -178,16 +180,35 @@ impl LogCleaner {
 
             // Select the oldest files first, then additionally trim any files
             // that still violate configured size constraints.
-            let to_delete = self.select_files_to_process(&logs, total_size);
-            if !to_delete.is_empty() {
-                let (deleted, freed) = if self.parallel_compress && self.compress_old_files {
-                    self.parallel_stealing_compress(&to_delete)?
-                } else {
-                    self.serial_compress_and_delete(&to_delete)?
-                };
-                total_deleted += deleted;
-                total_freed += freed;
-            }
+            to_delete = self.select_files_to_process(&logs, total_size);
+        }
+
+        // `min_file_age_seconds` keeps freshly rolled files readable for a
+        // while, but those files fill the disk like any other. Once the rolled
+        // files that would remain exceed the byte budget, the budget wins:
+        // the oldest of them go, fresh or not, and only as many as it takes.
+        let over_budget = self.select_files_over_budget(&logs, &to_delete, recent_logs);
+        if !over_budget.is_empty() {
+            warn!(
+                event = EVENT_LOG_CLEANER_STATE,
+                component = LOG_COMPONENT_OBS,
+                subsystem = LOG_SUBSYSTEM_LOG_CLEANER,
+                result = "byte_budget_overrides_min_file_age",
+                log_files = over_budget.len(),
+                max_total_size_bytes = self.max_total_size_bytes,
+                "log cleaner state changed"
+            );
+            to_delete.extend(over_budget);
+        }
+
+        if !to_delete.is_empty() {
+            let (deleted, freed) = if self.parallel_compress && self.compress_old_files {
+                self.parallel_stealing_compress(&to_delete)?
+            } else {
+                self.serial_compress_and_delete(&to_delete)?
+            };
+            total_deleted += deleted;
+            total_freed += freed;
         }
 
         if !compressed_archives.is_empty() {
@@ -217,6 +238,44 @@ impl LogCleaner {
         Ok((total_deleted, total_freed))
     }
 
+    /// Byte-budget trim that also counts the rolled files still inside
+    /// `min_file_age_seconds`.
+    ///
+    /// `aged` is sorted oldest first and `selected` is what the regular
+    /// selection already took from it. When the files that would remain,
+    /// aged and recent together, exceed `max_total_size_bytes`, the oldest of
+    /// them are returned until the rest fits. `keep_files` and the per-file
+    /// limit are not applied to recent files: only the bytes matter here.
+    pub(super) fn select_files_over_budget(
+        &self,
+        aged: &[FileInfo],
+        selected: &[FileInfo],
+        mut recent: Vec<FileInfo>,
+    ) -> Vec<FileInfo> {
+        if self.max_total_size_bytes == 0 || recent.is_empty() {
+            return Vec::new();
+        }
+        recent.sort_by_key(|file| file.modified);
+        let selected: std::collections::HashSet<&std::path::Path> = selected.iter().map(|file| file.path.as_path()).collect();
+        // Aged files are older than every recent one, so chaining keeps the
+        // oldest-first order.
+        let remaining: Vec<&FileInfo> = aged
+            .iter()
+            .filter(|file| !selected.contains(file.path.as_path()))
+            .chain(recent.iter())
+            .collect();
+        let mut occupied = remaining.iter().fold(0u64, |total, file| total.saturating_add(file.size));
+        let mut over_budget = Vec::new();
+        for file in remaining {
+            if occupied <= self.max_total_size_bytes {
+                break;
+            }
+            occupied = occupied.saturating_sub(file.size);
+            over_budget.push(file.clone());
+        }
+        over_budget
+    }
+
     /// Choose regular log files that should be compressed and/or deleted.
     ///
     /// The `files` slice must already be sorted from oldest to newest. The

crates/obs/src/cleaner/mod.rs

@@ -444,6 +444,171 @@ mod tests {
         Ok(())
     }
 
+    /// Rolled files inside `min_file_age_seconds` occupy the disk too: once
+    /// they push the directory over the byte budget, the oldest are processed
+    /// regardless of age until the budget holds again.
+    #[test]
+    fn byte_budget_overrides_min_age_for_fresh_logs() -> std::io::Result<()> {
+        let tmp = TempDir::new()?;
+        let dir = tmp.path().to_path_buf();
+        for i in 0..6u64 {
+            let name = format!("app.log.2024-01-0{i}");
+            create_log_file(&dir, &name, 1024)?;
+            // All well inside the age gate, oldest first by name.
+            set_mtime_ago(&dir.join(&name), Duration::from_secs(60 - i * 5));
+        }
+        create_log_file(&dir, "app.log", 1024)?;
+
+        let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string())
+            .match_mode(FileMatchMode::Prefix)
+            .keep_files(30)
+            .max_total_size_bytes(3 * 1024)
+            .min_file_age_seconds(3600)
+            .build();
+        let (deleted, freed) = cleaner.cleanup()?;
+
+        assert_eq!((deleted, freed), (3, 3 * 1024), "the three oldest fresh logs go; the budget then holds");
+        for i in 0..3 {
+            assert!(!dir.join(format!("app.log.2024-01-0{i}")).exists());
+        }
+        for i in 3..6 {
+            assert!(dir.join(format!("app.log.2024-01-0{i}")).exists(), "the newest fresh logs stay");
+        }
+        assert!(dir.join("app.log").exists(), "the active file is never touched");
+        Ok(())
+    }
+
+    /// Within the byte budget the age gate is unchanged: fresh logs are not
+    /// touched, even above `keep_files`.
+    #[test]
+    fn min_age_still_protects_fresh_logs_within_the_byte_budget() -> std::io::Result<()> {
+        let tmp = TempDir::new()?;
+        let dir = tmp.path().to_path_buf();
+        for i in 0..4 {
+            create_log_file(&dir, &format!("app.log.2024-01-0{i}"), 1024)?;
+        }
+
+        let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string())
+            .match_mode(FileMatchMode::Prefix)
+            .keep_files(1)
+            .max_total_size_bytes(4 * 1024)
+            .min_file_age_seconds(3600)
+            .build();
+        assert_eq!(cleaner.cleanup()?, (0, 0), "exactly at the budget nothing is over it");
+        for i in 0..4 {
+            assert!(dir.join(format!("app.log.2024-01-0{i}")).exists());
+        }
+
+        // No byte budget configured: fresh files are never counted.
+        let unbounded = LogCleaner::builder(dir, "app.log.".to_string(), "app.log".to_string())
+            .match_mode(FileMatchMode::Prefix)
+            .keep_files(1)
+            .max_total_size_bytes(0)
+            .min_file_age_seconds(3600)
+            .build();
+        assert_eq!(unbounded.cleanup()?, (0, 0));
+        Ok(())
+    }
+
+    /// The budget override is a byte trim only: `keep_files` is not applied
+    /// to fresh files, so one byte over the budget costs one file, not all
+    /// but `keep_files` of them.
+    #[test]
+    fn byte_budget_override_removes_only_what_the_budget_needs() -> std::io::Result<()> {
+        let tmp = TempDir::new()?;
+        let dir = tmp.path().to_path_buf();
+        for i in 0..5u64 {
+            let name = format!("app.log.2024-01-0{i}");
+            create_log_file(&dir, &name, 1024)?;
+            set_mtime_ago(&dir.join(&name), Duration::from_secs(50 - i * 10));
+        }
+
+        let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string())
+            .match_mode(FileMatchMode::Prefix)
+            .keep_files(1)
+            .max_single_file_size_bytes(512)
+            .max_total_size_bytes(4 * 1024)
+            .min_file_age_seconds(3600)
+            .build();
+        let (deleted, freed) = cleaner.cleanup()?;
+
+        assert_eq!(
+            (deleted, freed),
+            (1, 1024),
+            "neither keep_files nor the per-file limit applies to fresh files"
+        );
+        assert!(!dir.join("app.log.2024-01-00").exists(), "the oldest fresh file goes");
+        for i in 1..5 {
+            assert!(dir.join(format!("app.log.2024-01-0{i}")).exists());
+        }
+        Ok(())
+    }
+
+    /// Aged files keep their regular selection; the byte trim then counts the
+    /// fresh files and takes the oldest remaining files first.
+    #[test]
+    fn regular_selection_and_budget_override_compose() -> std::io::Result<()> {
+        let tmp = TempDir::new()?;
+        let dir = tmp.path().to_path_buf();
+        for i in 0..3u64 {
+            let name = format!("app.log.aged-{i}");
+            create_log_file(&dir, &name, 1024)?;
+            set_mtime_ago(&dir.join(&name), Duration::from_secs((5 - i) * 3600));
+        }
+        for i in 0..3u64 {
+            let name = format!("app.log.fresh-{i}");
+            create_log_file(&dir, &name, 1024)?;
+            set_mtime_ago(&dir.join(&name), Duration::from_secs(30 - i * 10));
+        }
+
+        // keep_files(2) takes aged-0. Five files remain against a budget of
+        // three, so the two oldest of them go as well: aged-1 and aged-2.
+        let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string())
+            .match_mode(FileMatchMode::Prefix)
+            .keep_files(2)
+            .max_total_size_bytes(3 * 1024)
+            .min_file_age_seconds(3600)
+            .build();
+        let (deleted, freed) = cleaner.cleanup()?;
+
+        assert_eq!((deleted, freed), (3, 3 * 1024));
+        for i in 0..3 {
+            assert!(!dir.join(format!("app.log.aged-{i}")).exists());
+            assert!(dir.join(format!("app.log.fresh-{i}")).exists());
+        }
+        Ok(())
+    }
+
+    /// Aged and fresh logs share one budget: aged files alone fit, the fresh
+    /// ones put the directory over it, and the oldest files overall go first.
+    #[test]
+    fn aged_and_fresh_logs_share_the_byte_budget() -> std::io::Result<()> {
+        let tmp = TempDir::new()?;
+        let dir = tmp.path().to_path_buf();
+        create_log_file(&dir, "app.log.aged-0", 1024)?;
+        create_log_file(&dir, "app.log.aged-1", 1024)?;
+        set_mtime_ago(&dir.join("app.log.aged-0"), Duration::from_secs(3 * 3600));
+        set_mtime_ago(&dir.join("app.log.aged-1"), Duration::from_secs(2 * 3600));
+        create_log_file(&dir, "app.log.fresh-0", 1024)?;
+        create_log_file(&dir, "app.log.fresh-1", 1024)?;
+        set_mtime_ago(&dir.join("app.log.fresh-0"), Duration::from_secs(30));
+
+        let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string())
+            .match_mode(FileMatchMode::Prefix)
+            .keep_files(30)
+            .max_total_size_bytes(2 * 1024)
+            .min_file_age_seconds(3600)
+            .build();
+        let (deleted, _) = cleaner.cleanup()?;
+
+        assert_eq!(deleted, 2);
+        assert!(!dir.join("app.log.aged-0").exists());
+        assert!(!dir.join("app.log.aged-1").exists());
+        assert!(dir.join("app.log.fresh-0").exists());
+        assert!(dir.join("app.log.fresh-1").exists());
+        Ok(())
+    }
+
     #[test]
     fn active_file_matching_pattern_is_protected() -> std::io::Result<()> {
         let tmp = TempDir::new()?;

crates/obs/src/cleaner/scanner.rs

@@ -48,6 +48,10 @@ const ORPHAN_TMP_GRACE_SECONDS: u64 = 60;
 pub(super) struct LogScanResult {
     /// Regular log files eligible for deletion/compression.
     pub logs: Vec<FileInfo>,
+    /// Regular log files younger than `min_file_age_seconds`. They occupy the
+    /// directory like any other rolled file, so the caller counts them against
+    /// the byte budget and may process them when that budget is exceeded.
+    pub recent_logs: Vec<FileInfo>,
     /// Already compressed files eligible for expiry deletion.
     pub compressed_archives: Vec<FileInfo>,
 }
@@ -83,11 +87,13 @@ pub(super) fn scan_log_directory(
     if file_pattern.is_empty() {
         return Ok(LogScanResult {
             logs: Vec::new(),
+            recent_logs: Vec::new(),
             compressed_archives: Vec::new(),
         });
     }
 
     let mut logs = Vec::new();
+    let mut recent_logs = Vec::new();
     let mut compressed_archives = Vec::new();
     let now = SystemTime::now();
 
@@ -99,6 +105,7 @@ pub(super) fn scan_log_directory(
             // as "no files found" instead of failing the whole cleanup pass.
             return Ok(LogScanResult {
                 logs,
+                recent_logs,
                 compressed_archives,
             });
         }
@@ -230,10 +237,7 @@ pub(super) fn scan_log_directory(
         // 6. Age gate regular logs only.
         // Compressed files deliberately bypass this check because archive
         // expiry is driven by a dedicated retention horizon in the caller.
-        if !is_compressed && !is_old_enough_for_cleanup(age, min_file_age_seconds) {
-            // Too young to be touched.
-            continue;
-        }
+        let recent = !is_compressed && !is_old_enough_for_cleanup(age, min_file_age_seconds);
 
         let info = FileInfo {
             path,
@@ -244,6 +248,9 @@ pub(super) fn scan_log_directory(
 
         if is_compressed {
             compressed_archives.push(info);
+        } else if recent {
+            // Too young to be touched while the directory is within budget.
+            recent_logs.push(info);
         } else {
             logs.push(info);
         }
@@ -251,6 +258,7 @@ pub(super) fn scan_log_directory(
 
     Ok(LogScanResult {
         logs,
+        recent_logs,
         compressed_archives,
     })
 }
@@ -311,6 +319,42 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn fresh_nonempty_log_is_reported_as_recent_not_as_a_candidate() -> std::io::Result<()> {
+        let tmp = TempDir::new()?;
+        let dir = tmp.path();
+        std::fs::write(dir.join("fresh.rustfs.log"), b"fresh")?;
+        std::fs::write(dir.join("aged.rustfs.log"), b"aged!!")?;
+        let old = SystemTime::now() - Duration::from_secs(7200);
+        File::options()
+            .write(true)
+            .open(dir.join("aged.rustfs.log"))?
+            .set_modified(old)?;
+        std::fs::write(dir.join("active.rustfs.log"), b"active")?;
+
+        let result = scan_log_directory(
+            dir,
+            ".rustfs.log",
+            Some("active.rustfs.log"),
+            FileMatchMode::Suffix,
+            &[],
+            3600,
+            true,
+            false,
+        )?;
+
+        let names = |files: &[FileInfo]| -> Vec<String> {
+            files
+                .iter()
+                .map(|file| file.path.file_name().expect("file name").to_string_lossy().into_owned())
+                .collect()
+        };
+        assert_eq!(names(&result.logs), vec!["aged.rustfs.log"]);
+        assert_eq!(names(&result.recent_logs), vec!["fresh.rustfs.log"], "the active file is in neither list");
+        assert_eq!(result.recent_logs[0].size, 5);
+        Ok(())
+    }
+
     #[test]
     fn orphan_tmp_archive_is_deleted() -> std::io::Result<()> {
         let tmp = TempDir::new()?;

crates/obs/src/config.rs

@@ -26,22 +26,21 @@ use rustfs_config::observability::{
     DEFAULT_OBS_ENVIRONMENT_PRODUCTION, DEFAULT_OBS_LOG_CLEANUP_INTERVAL_SECONDS, DEFAULT_OBS_LOG_COMPRESS_OLD_FILES,
     DEFAULT_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM, DEFAULT_OBS_LOG_DELETE_EMPTY_FILES,
     DEFAULT_OBS_LOG_DRY_RUN, DEFAULT_OBS_LOG_GZIP_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_MATCH_MODE,
-    DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MIN_FILE_AGE_SECONDS,
-    DEFAULT_OBS_LOG_PARALLEL_COMPRESS, DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL,
-    DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_ENDPOINT, ENV_OBS_ENDPOINT_HEADERS,
-    ENV_OBS_ENDPOINT_LOGS_HEADERS, ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_METRICS_HEADERS,
-    ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TRACES_HEADERS,
-    ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS, ENV_OBS_ENVIRONMENT, ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS,
-    ENV_OBS_LOG_COMPRESS_OLD_FILES, ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, ENV_OBS_LOG_COMPRESSION_ALGORITHM,
-    ENV_OBS_LOG_DELETE_EMPTY_FILES, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN, ENV_OBS_LOG_ENDPOINT,
-    ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL, ENV_OBS_LOG_KEEP_FILES,
-    ENV_OBS_LOG_MATCH_MODE, ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES,
-    ENV_OBS_LOG_MIN_FILE_AGE_SECONDS, ENV_OBS_LOG_PARALLEL_COMPRESS, ENV_OBS_LOG_PARALLEL_WORKERS, ENV_OBS_LOG_ROTATION_TIME,
-    ENV_OBS_LOG_STDOUT_ENABLED, ENV_OBS_LOG_ZSTD_COMPRESSION_LEVEL, ENV_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, ENV_OBS_LOG_ZSTD_WORKERS,
-    ENV_OBS_LOGGER_LEVEL, ENV_OBS_LOGS_EXPORT_ENABLED, ENV_OBS_METER_INTERVAL, ENV_OBS_METRIC_ENDPOINT,
-    ENV_OBS_METRICS_EXPORT_ENABLED, ENV_OBS_PROFILING_ENDPOINT, ENV_OBS_PROFILING_EXPORT_ENABLED, ENV_OBS_SAMPLE_RATIO,
-    ENV_OBS_SERVICE_NAME, ENV_OBS_SERVICE_VERSION, ENV_OBS_TLS_CA_FILE, ENV_OBS_TRACE_ENDPOINT, ENV_OBS_TRACES_EXPORT_ENABLED,
-    ENV_OBS_USE_STDOUT,
+    DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MIN_FILE_AGE_SECONDS, DEFAULT_OBS_LOG_PARALLEL_COMPRESS,
+    DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP,
+    DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_ENDPOINT, ENV_OBS_ENDPOINT_HEADERS, ENV_OBS_ENDPOINT_LOGS_HEADERS,
+    ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_METRICS_HEADERS, ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS,
+    ENV_OBS_ENDPOINT_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TRACES_HEADERS, ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS,
+    ENV_OBS_ENVIRONMENT, ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS, ENV_OBS_LOG_COMPRESS_OLD_FILES,
+    ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, ENV_OBS_LOG_COMPRESSION_ALGORITHM, ENV_OBS_LOG_DELETE_EMPTY_FILES,
+    ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN, ENV_OBS_LOG_ENDPOINT, ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME,
+    ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL, ENV_OBS_LOG_KEEP_FILES, ENV_OBS_LOG_MATCH_MODE, ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES,
+    ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES, ENV_OBS_LOG_MIN_FILE_AGE_SECONDS, ENV_OBS_LOG_PARALLEL_COMPRESS,
+    ENV_OBS_LOG_PARALLEL_WORKERS, ENV_OBS_LOG_ROTATION_TIME, ENV_OBS_LOG_STDOUT_ENABLED, ENV_OBS_LOG_ZSTD_COMPRESSION_LEVEL,
+    ENV_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, ENV_OBS_LOG_ZSTD_WORKERS, ENV_OBS_LOGGER_LEVEL, ENV_OBS_LOGS_EXPORT_ENABLED,
+    ENV_OBS_METER_INTERVAL, ENV_OBS_METRIC_ENDPOINT, ENV_OBS_METRICS_EXPORT_ENABLED, ENV_OBS_PROFILING_ENDPOINT,
+    ENV_OBS_PROFILING_EXPORT_ENABLED, ENV_OBS_SAMPLE_RATIO, ENV_OBS_SERVICE_NAME, ENV_OBS_SERVICE_VERSION, ENV_OBS_TLS_CA_FILE,
+    ENV_OBS_TRACE_ENDPOINT, ENV_OBS_TRACES_EXPORT_ENABLED, ENV_OBS_USE_STDOUT,
 };
 use rustfs_config::{
     APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOG_FILENAME,
@@ -189,9 +188,13 @@ pub struct OtelConfig {
     pub log_keep_files: Option<usize>,
 
     // ── Log cleanup ───────────────────────────────────────────────────────────
-    /// Hard ceiling on the total size (bytes) of all log files (default: 2 GiB).
+    /// Byte budget for rolled log files, and separately for compressed
+    /// archives (default: 2 GiB); `0` means no budget. The active file is not
+    /// counted.
     pub log_max_total_size_bytes: Option<u64>,
-    /// Per-file size ceiling (bytes); `0` means unlimited (default: `0`).
+    /// Per-file size ceiling (bytes). Unset or unparsable: the active file
+    /// rolls at 256 MiB (less under a small byte budget) and the cleaner
+    /// applies no per-file trim. `0`: no size roll and no trim.
     pub log_max_single_file_size_bytes: Option<u64>,
     /// Compress eligible files with gzip before deletion (default: `true`).
     pub log_compress_old_files: Option<bool>,
@@ -344,10 +347,8 @@ impl OtelConfig {
             log_keep_files,
             // Log cleanup
             log_max_total_size_bytes: Some(get_env_u64(ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES)),
-            log_max_single_file_size_bytes: Some(get_env_u64(
-                ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES,
-                DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES,
-            )),
+            // Unset stays `None`: the appender and the cleaner default differently.
+            log_max_single_file_size_bytes: get_env_opt_u64(ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES),
             log_compress_old_files: Some(get_env_bool(ENV_OBS_LOG_COMPRESS_OLD_FILES, DEFAULT_OBS_LOG_COMPRESS_OLD_FILES)),
             log_gzip_compression_level: Some(get_env_u64(
                 ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL,
@@ -480,6 +481,30 @@ mod tests {
         OtelConfig::extract_otel_config_from_env(None).profiling_export_enabled
     }
 
+    #[test]
+    fn single_file_size_limit_distinguishes_unset_from_an_explicit_zero() {
+        with_profiling_env_lock(|| {
+            temp_env::with_var_unset(ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, || {
+                let config = OtelConfig::extract_otel_config_from_env(None);
+                assert_eq!(config.log_max_single_file_size_bytes, None, "unset leaves the default to each consumer");
+            });
+            temp_env::with_var(ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, Some("0"), || {
+                let config = OtelConfig::extract_otel_config_from_env(None);
+                assert_eq!(config.log_max_single_file_size_bytes, Some(0), "an explicit 0 is kept as unlimited");
+            });
+            for unparsable in ["", "256MiB"] {
+                temp_env::with_var(ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, Some(unparsable), || {
+                    let config = OtelConfig::extract_otel_config_from_env(None);
+                    assert_eq!(config.log_max_single_file_size_bytes, None, "{unparsable:?} is treated as unset");
+                });
+            }
+            temp_env::with_var(ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, Some("1048576"), || {
+                let config = OtelConfig::extract_otel_config_from_env(None);
+                assert_eq!(config.log_max_single_file_size_bytes, Some(1_048_576));
+            });
+        });
+    }
+
     #[test]
     fn service_version_uses_application_version_when_environment_is_unset() {
         with_profiling_env_lock(|| {

crates/obs/src/telemetry/local.rs

@@ -40,12 +40,13 @@ use crate::telemetry::filter::{build_env_filter, effective_verbose_logging, pyro
 use crate::telemetry::rolling::{RollingAppender, Rotation};
 use metrics::counter;
 use rustfs_config::observability::{
-    DEFAULT_OBS_LOG_CLEANUP_INTERVAL_SECONDS, DEFAULT_OBS_LOG_COMPRESS_OLD_FILES, DEFAULT_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS,
-    DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM, DEFAULT_OBS_LOG_DELETE_EMPTY_FILES, DEFAULT_OBS_LOG_DRY_RUN,
-    DEFAULT_OBS_LOG_GZIP_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_MATCH_MODE, DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES,
-    DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MIN_FILE_AGE_SECONDS, DEFAULT_OBS_LOG_PARALLEL_COMPRESS,
-    DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_SPAN_EVENTS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL,
-    DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_LOG_SPAN_EVENTS,
+    DEFAULT_OBS_LOG_ACTIVE_FILE_ROLL_SIZE_BYTES, DEFAULT_OBS_LOG_CLEANUP_INTERVAL_SECONDS, DEFAULT_OBS_LOG_COMPRESS_OLD_FILES,
+    DEFAULT_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM, DEFAULT_OBS_LOG_DELETE_EMPTY_FILES,
+    DEFAULT_OBS_LOG_DRY_RUN, DEFAULT_OBS_LOG_GZIP_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_MATCH_MODE,
+    DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MIN_FILE_AGE_SECONDS,
+    DEFAULT_OBS_LOG_PARALLEL_COMPRESS, DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_SPAN_EVENTS,
+    DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS,
+    ENV_OBS_LOG_SPAN_EVENTS,
 };
 use rustfs_config::{APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME};
 use rustfs_utils::get_env_str;
@@ -159,6 +160,58 @@ fn validate_distinct_file_identities(
     Ok(())
 }
 
+/// Smallest roll size derived from a small byte budget.
+const MIN_DERIVED_ROLL_SIZE_BYTES: u64 = 1024 * 1024;
+
+/// Size at which the active log file is rolled. An explicit limit is used as
+/// is, and an explicit `0` keeps size rolling off. An unset limit rolls at the
+/// shipped default so the active file is bounded, lowered to a quarter of the
+/// byte budget when the budget is small: a rolled file larger than the whole
+/// budget would be removed by the pass its own roll starts.
+pub(crate) fn resolve_active_file_roll_size(config: &OtelConfig) -> u64 {
+    if let Some(configured) = config.log_max_single_file_size_bytes {
+        return configured;
+    }
+    match config
+        .log_max_total_size_bytes
+        .unwrap_or(DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES)
+    {
+        0 => DEFAULT_OBS_LOG_ACTIVE_FILE_ROLL_SIZE_BYTES,
+        budget => DEFAULT_OBS_LOG_ACTIVE_FILE_ROLL_SIZE_BYTES.min((budget / 4).max(MIN_DERIVED_ROLL_SIZE_BYTES)),
+    }
+}
+
+/// Per-file limit above which the cleaner processes a rolled file. Unset stays
+/// off: the appender's default roll size is not a reason to trim files that an
+/// earlier release rolled by time only.
+pub(crate) fn resolve_cleanup_single_file_limit(config: &OtelConfig) -> u64 {
+    config
+        .log_max_single_file_size_bytes
+        .unwrap_or(DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES)
+}
+
+/// Open the rolling appender for the configured log file together with the
+/// signal it raises after every roll. The caller hands that signal to
+/// [`spawn_cleanup_task`].
+pub(crate) fn open_rolling_appender(
+    config: &OtelConfig,
+    log_directory: &str,
+    log_filename: &str,
+    rotation: Rotation,
+    match_mode: FileMatchMode,
+) -> std::io::Result<(RollingAppender, Arc<tokio::sync::Notify>)> {
+    let roll_signal = Arc::new(tokio::sync::Notify::new());
+    let appender = RollingAppender::new(
+        log_directory,
+        log_filename.to_string(),
+        rotation,
+        resolve_active_file_roll_size(config),
+        match_mode,
+    )?
+    .with_roll_signal(roll_signal.clone());
+    Ok((appender, roll_signal))
+}
+
 fn resolve_log_cleanup_interval_seconds(config: &OtelConfig) -> u64 {
     match config.log_cleanup_interval_seconds {
         Some(0) => {
@@ -451,12 +504,7 @@ fn init_file_logging_internal(
         _ => Rotation::Daily,
     };
 
-    let max_single_file_size = config
-        .log_max_single_file_size_bytes
-        .unwrap_or(DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES);
-
-    let file_appender =
-        RollingAppender::new(log_directory, log_filename.to_string(), rotation, max_single_file_size, match_mode)?;
+    let (file_appender, roll_signal) = open_rolling_appender(config, log_directory, log_filename, rotation, match_mode)?;
     validate_stdout_sink(&file_appender)?;
 
     let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
@@ -497,7 +545,7 @@ fn init_file_logging_internal(
     set_observability_metric_enabled(false);
 
     // ── 5. Start background cleanup task ─────────────────────────────────────
-    let cleanup_handle = spawn_cleanup_task(config, log_directory, log_filename, keep_files);
+    let cleanup_handle = spawn_cleanup_task(config, log_directory, log_filename, keep_files, roll_signal);
 
     info!(
         event = EVENT_LOCAL_LOGGING_STATE,
@@ -606,6 +654,10 @@ pub(super) fn emit_file_logging_fallback_warning(log_directory: &str, error: &Te
 
 // ─── Cleanup task ─────────────────────────────────────────────────────────────
 
+/// Shortest pause between two cleanup passes started by rolls. A pass that the
+/// timer starts is never delayed by it.
+const ROLL_TRIGGERED_CLEANUP_SPACING: Duration = Duration::from_secs(1);
+
 /// Spawn a background task that periodically cleans up old log files.
 ///
 /// All cleanup parameters are derived from [`OtelConfig`] fields, with
@@ -623,6 +675,9 @@ pub(super) fn emit_file_logging_fallback_warning(log_directory: &str, error: &Te
 /// * `log_filename` - Base filename (used as the file prefix for matching).
 /// * `keep_files` - Legacy keep-files count; used as fallback when the new
 ///   `log_keep_count` field is absent.
+/// * `roll_signal` - Notified by the appender after every roll. The loop then
+///   runs a pass right away, so the byte budget is enforced when a file is
+///   rolled instead of up to one cleanup interval later.
 ///
 /// # Returns
 /// A [`tokio::task::JoinHandle`] for the spawned cleanup loop.
@@ -631,6 +686,7 @@ pub fn spawn_cleanup_task(
     log_directory: &str,
     log_filename: &str,
     keep_files: usize,
+    roll_signal: Arc<tokio::sync::Notify>,
 ) -> tokio::task::JoinHandle<()> {
     let log_dir = std::path::PathBuf::from(log_directory);
     // Use suffix matching for log files like `2026-03-01-06-21.rustfs.log`
@@ -645,9 +701,7 @@ pub fn spawn_cleanup_task(
     let max_total_size = config
         .log_max_total_size_bytes
         .unwrap_or(DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES);
-    let max_single_file_size = config
-        .log_max_single_file_size_bytes
-        .unwrap_or(DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES);
+    let max_single_file_size = resolve_cleanup_single_file_limit(config);
     let compress = config.log_compress_old_files.unwrap_or(DEFAULT_OBS_LOG_COMPRESS_OLD_FILES);
     let gzip_level = config
         .log_gzip_compression_level
@@ -740,9 +794,16 @@ pub fn spawn_cleanup_task(
     tokio::spawn(async move {
         let mut interval = tokio::time::interval(Duration::from_secs(cleanup_interval));
         loop {
-            // Wait for the next scheduled tick before dispatching another pass.
-            // The blocking filesystem work runs on a dedicated blocking thread.
-            interval.tick().await;
+            // Wait for the next scheduled tick, or for the appender to roll a
+            // file, before dispatching another pass. The blocking filesystem
+            // work runs on a dedicated blocking thread.
+            tokio::select! {
+                _ = interval.tick() => {}
+                _ = roll_signal.notified() => {
+                    // Rolls can come in bursts; one pass covers all of them.
+                    tokio::time::sleep(ROLL_TRIGGERED_CLEANUP_SPACING).await;
+                }
+            }
             let cleaner_clone = cleaner.clone();
             let result = tokio::task::spawn_blocking(move || cleaner_clone.cleanup()).await;
 
@@ -1253,6 +1314,153 @@ mod tests {
         assert_eq!(resolve_log_cleanup_interval_seconds(&config), DEFAULT_OBS_LOG_CLEANUP_INTERVAL_SECONDS);
     }
 
+    #[test]
+    fn active_file_rolls_by_size_unless_explicitly_disabled() {
+        let unset = OtelConfig {
+            log_max_single_file_size_bytes: None,
+            log_max_total_size_bytes: None,
+            ..OtelConfig::default()
+        };
+        assert_eq!(resolve_active_file_roll_size(&unset), DEFAULT_OBS_LOG_ACTIVE_FILE_ROLL_SIZE_BYTES);
+        assert_eq!(DEFAULT_OBS_LOG_ACTIVE_FILE_ROLL_SIZE_BYTES, 256 * 1024 * 1024);
+        assert_eq!(
+            resolve_cleanup_single_file_limit(&unset),
+            0,
+            "the appender's default roll size is not a per-file trim for the cleaner"
+        );
+
+        let disabled = OtelConfig {
+            log_max_single_file_size_bytes: Some(0),
+            ..OtelConfig::default()
+        };
+        assert_eq!(resolve_active_file_roll_size(&disabled), 0, "an explicit 0 keeps size rolling off");
+        assert_eq!(resolve_cleanup_single_file_limit(&disabled), 0);
+
+        let configured = OtelConfig {
+            log_max_single_file_size_bytes: Some(64 * 1024 * 1024),
+            log_max_total_size_bytes: Some(1024),
+            ..OtelConfig::default()
+        };
+        assert_eq!(
+            resolve_active_file_roll_size(&configured),
+            64 * 1024 * 1024,
+            "an explicit limit is never derived"
+        );
+        assert_eq!(resolve_cleanup_single_file_limit(&configured), 64 * 1024 * 1024);
+    }
+
+    #[test]
+    fn default_roll_size_follows_a_small_byte_budget() {
+        let with_budget = |budget: u64| OtelConfig {
+            log_max_single_file_size_bytes: None,
+            log_max_total_size_bytes: Some(budget),
+            ..OtelConfig::default()
+        };
+        const MIB: u64 = 1024 * 1024;
+        assert_eq!(
+            resolve_active_file_roll_size(&with_budget(2048 * MIB)),
+            256 * MIB,
+            "the default budget keeps 256 MiB"
+        );
+        assert_eq!(resolve_active_file_roll_size(&with_budget(1024 * MIB)), 256 * MIB);
+        assert_eq!(
+            resolve_active_file_roll_size(&with_budget(256 * MIB)),
+            64 * MIB,
+            "a quarter of a small budget"
+        );
+        assert_eq!(resolve_active_file_roll_size(&with_budget(MIB)), MIB, "never below 1 MiB");
+        assert_eq!(resolve_active_file_roll_size(&with_budget(0)), 256 * MIB, "no budget, no derivation");
+    }
+
+    /// Both logging setups open the appender through this function, so the
+    /// cleanup task they start is always the one the appender wakes.
+    #[test]
+    fn opened_appender_reports_its_rolls_on_the_returned_signal() {
+        use std::io::Write as _;
+
+        let tmp = tempfile::tempdir().expect("log directory");
+        let config = OtelConfig {
+            log_max_single_file_size_bytes: Some(8),
+            ..OtelConfig::default()
+        };
+        let (mut appender, roll_signal) = open_rolling_appender(
+            &config,
+            tmp.path().to_str().expect("UTF-8 path"),
+            "rustfs.log",
+            Rotation::Daily,
+            FileMatchMode::Suffix,
+        )
+        .expect("appender");
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_time()
+            .build()
+            .expect("runtime");
+        let reported = || runtime.block_on(async { tokio::time::timeout(Duration::ZERO, roll_signal.notified()).await.is_ok() });
+
+        appender.write_all(b"12345678").expect("first write");
+        assert!(!reported());
+        appender.write_all(b"9").expect("write that rolls at the configured size");
+        assert!(reported(), "the configured roll size is in effect and the roll reaches the signal");
+    }
+
+    /// The cleanup loop runs a pass when the appender reports a roll, long
+    /// before its interval elapses, and that pass enforces the byte budget on
+    /// files younger than the age gate.
+    #[tokio::test]
+    async fn cleanup_loop_enforces_the_byte_budget_when_a_roll_is_reported() {
+        let tmp = tempfile::tempdir().expect("log directory");
+        let dir = tmp.path();
+        let config = OtelConfig {
+            log_max_total_size_bytes: Some(2 * 1024),
+            log_max_single_file_size_bytes: Some(0),
+            log_compress_old_files: Some(false),
+            log_min_file_age_seconds: Some(3600),
+            log_cleanup_interval_seconds: Some(3600),
+            log_match_mode: Some("suffix".to_string()),
+            log_filename: Some("rustfs.log".to_string()),
+            ..OtelConfig::default()
+        };
+        let roll_signal = Arc::new(tokio::sync::Notify::new());
+        let rolled_files = || {
+            fs::read_dir(dir)
+                .expect("log directory")
+                .filter_map(Result::ok)
+                .filter(|entry| entry.file_name() != "rustfs.log")
+                .count()
+        };
+        let write_rolled = |range: std::ops::Range<u32>| {
+            for index in range {
+                fs::write(dir.join(format!("2026-01-01-00-{index:02}.rustfs.log")), vec![b'x'; 1024]).expect("rolled file");
+            }
+        };
+        let wait_for_two_rolled_files = |what: &'static str| async move {
+            tokio::time::timeout(Duration::from_secs(10), async {
+                while rolled_files() > 2 {
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                }
+            })
+            .await
+            .unwrap_or_else(|_| panic!("{what}"));
+        };
+
+        // The pass the interval starts at once is observed through its effect,
+        // so what follows cannot be that pass arriving late.
+        fs::write(dir.join("rustfs.log"), vec![b'a'; 1024]).expect("active file");
+        write_rolled(0..6);
+        let cleanup = spawn_cleanup_task(&config, dir.to_str().expect("UTF-8 path"), "rustfs.log", 30, roll_signal.clone());
+        wait_for_two_rolled_files("the startup pass enforces the budget").await;
+
+        write_rolled(10..14);
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        assert_eq!(rolled_files(), 6, "without a roll report nothing runs before the next interval");
+
+        roll_signal.notify_one();
+        wait_for_two_rolled_files("the reported roll starts a pass that brings the rolled files back under the budget").await;
+        assert_eq!(rolled_files(), 2);
+        assert!(dir.join("rustfs.log").exists(), "the active file is never removed");
+        cleanup.abort();
+    }
+
     #[test]
     fn empty_cleanup_pattern_falls_back_to_runtime_log_filename() {
         let config = OtelConfig {

crates/obs/src/telemetry/otel.rs

@@ -41,10 +41,10 @@ use crate::config::OtelConfig;
 use crate::global::set_observability_metric_enabled;
 use crate::telemetry::filter::{build_env_filter, pyroscope_log_filter, s3s_log_filter};
 use crate::telemetry::guard::{OtelGuard, ProfilingAgent};
-use crate::telemetry::local::{build_json_log_layer, spawn_cleanup_task};
+use crate::telemetry::local::{build_json_log_layer, open_rolling_appender, spawn_cleanup_task};
 use crate::telemetry::recorder::{Recorder, install_process_global_recorder};
 use crate::telemetry::resource::build_resource;
-use crate::telemetry::rolling::{RollingAppender, Rotation};
+use crate::telemetry::rolling::Rotation;
 // Import helper functions from local.rs (sibling module)
 use crate::TelemetryError;
 use metrics::counter;
@@ -58,7 +58,7 @@ use opentelemetry_sdk::{
     trace::{RandomIdGenerator, Sampler, SdkTracerProvider},
 };
 use percent_encoding::percent_decode_str;
-use rustfs_config::observability::{DEFAULT_OBS_LOG_MATCH_MODE, DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES};
+use rustfs_config::observability::DEFAULT_OBS_LOG_MATCH_MODE;
 use rustfs_config::{
     APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOGS_EXPORT_ENABLED,
     DEFAULT_OBS_METRICS_EXPORT_ENABLED, DEFAULT_OBS_TRACES_EXPORT_ENABLED, METER_INTERVAL, SAMPLE_RATIO,
@@ -244,17 +244,12 @@ pub(super) fn init_observability_http(
                 "daily" => Rotation::Daily,
                 _ => Rotation::Daily,
             };
-            let max_single_file_size = config
-                .log_max_single_file_size_bytes
-                .unwrap_or(DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES);
-
-            let file_appender =
-                RollingAppender::new(log_directory, log_filename.to_string(), rotation, max_single_file_size, match_mode)?;
+            let (file_appender, roll_signal) = open_rolling_appender(config, log_directory, log_filename, rotation, match_mode)?;
             crate::telemetry::local::validate_stdout_sink(&file_appender)?;
 
             let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
             let file_layer = build_json_log_layer(non_blocking, false, span_events.clone(), span_list);
-            let cleanup_handle = spawn_cleanup_task(config, log_directory, log_filename, keep_files);
+            let cleanup_handle = spawn_cleanup_task(config, log_directory, log_filename, keep_files, roll_signal);
             Ok((file_layer, guard, cleanup_handle, rotation_str))
         })();
 

crates/obs/src/telemetry/rolling.rs

@@ -29,6 +29,7 @@ use metrics::{counter, gauge, histogram};
 use std::fs::{self, File};
 use std::io::{self, Write};
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::thread;
 use std::time::{Duration, Instant};
@@ -82,6 +83,9 @@ pub struct RollingAppender {
     // Store as seconds since Unix epoch
     last_roll_ts: i64,
     rotation_retry_cooldown_until: Option<Instant>,
+    /// Woken after every successful roll so the cleaner can enforce the
+    /// directory budget on the file that was just rolled.
+    roll_signal: Option<Arc<tokio::sync::Notify>>,
 }
 
 impl RollingAppender {
@@ -129,6 +133,7 @@ impl RollingAppender {
             size: 0,
             last_roll_ts: Zoned::now().timestamp().as_second(),
             rotation_retry_cooldown_until: None,
+            roll_signal: None,
         };
         // Eagerly open the file to validate the path and capture accurate
         // initial size / last-roll timestamp.
@@ -136,6 +141,12 @@ impl RollingAppender {
         Ok(appender)
     }
 
+    /// Notify `signal` after every successful roll.
+    pub(crate) fn with_roll_signal(mut self, signal: Arc<tokio::sync::Notify>) -> Self {
+        self.roll_signal = Some(signal);
+        self
+    }
+
     pub(crate) fn active_file_path(&self) -> PathBuf {
         self.dir.join(&self.filename)
     }
@@ -289,6 +300,9 @@ impl RollingAppender {
                     counter!(METRIC_LOG_CLEANER_ROTATION_TOTAL).increment(1);
                     histogram!(METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS).record(rotate_started.elapsed().as_secs_f64());
                     gauge!(METRIC_LOG_CLEANER_ACTIVE_FILE_SIZE_BYTES).set(self.size as f64);
+                    if let Some(signal) = &self.roll_signal {
+                        signal.notify_one();
+                    }
                     return Ok(());
                 }
                 Err(e) => {
@@ -452,6 +466,80 @@ mod tests {
 
     // ── Size-based rotation ────────────────────────────────────────────────────
 
+    /// A storm with the shipped roll behavior: the appender rolls by size and
+    /// wakes the cleaner, and the directory stays within the byte budget plus
+    /// the active file although every rolled file is younger than the age gate.
+    #[test]
+    fn log_directory_stays_bounded_under_a_storm() -> io::Result<()> {
+        use crate::cleaner::LogCleaner;
+
+        const ROLL_SIZE: u64 = 4 * 1024;
+        const BUDGET: u64 = 16 * 1024;
+        let tmp = TempDir::new()?;
+        let dir = tmp.path().to_path_buf();
+        let signal = Arc::new(tokio::sync::Notify::new());
+        let mut appender =
+            RollingAppender::new(&dir, "rustfs.log".to_string(), Rotation::Never, ROLL_SIZE, FileMatchMode::Suffix)?
+                .with_roll_signal(signal.clone());
+        let cleaner = LogCleaner::builder(dir.clone(), "rustfs.log".to_string(), "rustfs.log".to_string())
+            .match_mode(FileMatchMode::Suffix)
+            .keep_files(30)
+            .max_total_size_bytes(BUDGET)
+            .compress_old_files(false)
+            .min_file_age_seconds(3600)
+            .build();
+        let directory_bytes = |dir: &Path| -> io::Result<u64> {
+            let mut total = 0;
+            for entry in fs::read_dir(dir)? {
+                total += entry?.metadata()?.len();
+            }
+            Ok(total)
+        };
+        let runtime = tokio::runtime::Builder::new_current_thread().enable_time().build()?;
+
+        let line = [b'x'; 512];
+        let mut passes = 0;
+        let mut peak = 0;
+        // 512 KiB of log lines, 32 times the budget.
+        for _ in 0..1024 {
+            appender.write_all(&line)?;
+            // The cleanup loop runs one pass per reported roll.
+            let rolled = runtime.block_on(async { tokio::time::timeout(Duration::ZERO, signal.notified()).await.is_ok() });
+            if rolled {
+                cleaner.cleanup()?;
+                passes += 1;
+            }
+            peak = peak.max(directory_bytes(&dir)?);
+        }
+        appender.flush()?;
+
+        assert!(passes > 100, "size rolls woke the cleaner {passes} times");
+        // One freshly rolled file may sit above the budget until its pass runs.
+        assert!(
+            peak <= BUDGET + 2 * ROLL_SIZE,
+            "peak directory size {peak} exceeded the budget {BUDGET} plus the active and one rolled file"
+        );
+        assert!(directory_bytes(&dir)? <= BUDGET + ROLL_SIZE);
+        Ok(())
+    }
+
+    #[test]
+    fn every_successful_roll_wakes_the_roll_signal() {
+        let tmp = TempDir::new().unwrap();
+        let signal = Arc::new(tokio::sync::Notify::new());
+        let mut appender = RollingAppender::new(tmp.path(), "app.log".to_string(), Rotation::Never, 5, FileMatchMode::Suffix)
+            .unwrap()
+            .with_roll_signal(signal.clone());
+        let runtime = tokio::runtime::Builder::new_current_thread().enable_time().build().unwrap();
+        let woken = || runtime.block_on(async { tokio::time::timeout(Duration::ZERO, signal.notified()).await.is_ok() });
+
+        appender.write_all(b"12345").expect("write should succeed");
+        assert!(!woken(), "no roll yet");
+        appender.write_all(b"abcde").expect("write after rotation should succeed");
+        assert!(woken(), "the size roll is reported");
+        assert!(!woken(), "one roll is one wake-up");
+    }
+
     #[test]
     fn test_size_rotation_creates_archive() {
         let tmp = TempDir::new().unwrap();