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/ecstore/src/disk/local/delete_recovery.rs

@@ -41,19 +41,53 @@ struct DiscoverySession {
     touched: Instant,
 }
 
-/// Plain (non-transaction) children of an object-scoped discovery root cost one
-/// budget entry per this many names: enough to bound a hostile prefix without
-/// making a real folder prefix undeletable.
+/// Names an object-scoped scan skips without a stat (plain children of the
+/// key's directory, and entries of a UUID child that cannot be delete evidence)
+/// cost one budget entry per this many names: enough to bound a hostile prefix
+/// without making a real folder prefix undeletable.
 const PLAIN_CHILDREN_PER_BUDGET_ENTRY: u32 = 64;
 
+/// Entries one step of the resumable (background) discovery may visit. The
+/// foreground bounded protocol has its own, larger per-call allowance.
+const DISCOVERY_STEP_ENTRIES: u32 = 4096;
+
+/// Wall-clock allowance of one bounded discovery call: a third of the drive
+/// timeout that wraps the call (including its wait for a lane worker), capped at
+/// ten seconds, so an incomplete walk returns before the call can time out.
+fn bounded_discovery_time_limit() -> std::time::Duration {
+    (crate::disk::disk_store::get_max_timeout_duration() / 3).min(std::time::Duration::from_secs(10))
+}
+
+/// Whether an entry inside a UUID child of an object-scoped root can name one
+/// of the key's delete transactions: a journal file of a rollback directory or
+/// a data-directory marker. Every other name is inert to discovery.
+fn may_be_delete_evidence(name: &str) -> bool {
+    [PREPARED, STORAGE_FORMAT_FILE_BACKUP, DELETE_QUORUM_COMMITTED_FILE].contains(&name)
+        || [
+            DELETE_DATA_DIR_MARKER_PREFIX,
+            RESERVED_DELETE_DATA_DIR_MARKER_PREFIX,
+            RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX,
+        ]
+        .iter()
+        .any(|prefix| delete_transaction_marker_id(name, prefix).is_some())
+}
+
 impl DiscoverySession {
     /// Consume one bounded step, returning ownership only when traversal remains.
     fn advance(self, limit: u32) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>)> {
-        self.advance_bounded(limit, super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES)
+        self.advance_bounded(limit, DISCOVERY_STEP_ENTRIES, &|| false)
             .map(|(transactions, next, _)| (transactions, next))
     }
 
-    fn advance_bounded(mut self, limit: u32, max_entries: u32) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>, u32)> {
+    /// Walk until `limit` transactions, `max_entries` charged entries, the end
+    /// of the scope, or `should_stop` (checked before every entry); any early
+    /// stop returns the remaining traversal as continuation state.
+    fn advance_bounded(
+        mut self,
+        limit: u32,
+        max_entries: u32,
+        should_stop: &dyn Fn() -> bool,
+    ) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>, u32)> {
         if self.frames.is_empty() {
             match std::fs::read_dir(&self.root) {
                 Ok(entries) => self.frames.push((String::new(), entries)),
@@ -66,10 +100,11 @@ impl DiscoverySession {
         let mut transactions = BTreeSet::new();
         let mut seen_candidates = HashSet::new();
         let mut visited = 0u32;
-        let mut plain_children = 0u32;
+        let mut skipped_by_name = 0u32;
         while !self.frames.is_empty()
             && transactions.len() < usize::try_from(limit).map_err(|_| std::io::Error::from(ErrorKind::InvalidInput))?
             && visited < max_entries
+            && !should_stop()
         {
             let (parent, entries) = self
                 .frames
@@ -90,13 +125,22 @@ impl DiscoverySession {
                 return Err(std::io::Error::from(ErrorKind::InvalidData));
             }
             // An object-scoped scan only ever descends into transaction (UUID)
-            // directories at the root. Plain child keys of a prefix directory are
-            // skipped by name without a stat; they are charged at one entry per
-            // readdir batch so the work stays bounded while deleting a missing key
-            // that names a prefix with thousands of children can still complete.
-            if self.object.is_some() && parent.is_empty() && !canonical_uuid(&name).is_some_and(|id| !id.is_nil()) {
-                plain_children += 1;
-                if plain_children.is_multiple_of(PLAIN_CHILDREN_PER_BUDGET_ENTRY) {
+            // directories at the root, and inside them only journal files and
+            // data-directory markers can be evidence. Every other name (plain
+            // child keys of a prefix directory, and the `xl.meta`, data
+            // directories or children of a UUID-named child key) is skipped by
+            // name without a stat and charged at one entry per readdir batch, so
+            // the work stays bounded while deleting a missing key that names a
+            // folder prefix with many children, UUID-named or not, can complete.
+            if self.object.is_some()
+                && if parent.is_empty() {
+                    !canonical_uuid(&name).is_some_and(|id| !id.is_nil())
+                } else {
+                    !may_be_delete_evidence(&name)
+                }
+            {
+                skipped_by_name += 1;
+                if skipped_by_name.is_multiple_of(PLAIN_CHILDREN_PER_BUDGET_ENTRY) {
                     visited += 1;
                 }
                 continue;
@@ -1391,8 +1435,16 @@ impl LocalDisk {
         self.validate_delete_discovery_owner(&session)?;
         // The blocking owner drops every directory handle even if the RPC waiter disappears.
         // A partial inventory is never evidence of absence and never enters DISCOVERY_SESSIONS.
-        os::run_blocking_namespace_operation(lease, move || {
-            let (transactions, next, visited) = session.advance_bounded(256, max_entries)?;
+        // A large folder prefix may take a while to walk, so the probe rides the
+        // durable-mutation lane like bounded inspection, never the one-worker
+        // foreground lane that every other mutation on the drive queues behind.
+        // It also stops, incomplete, when its waiter goes away or well before the
+        // drive timeout, so a slow walk answers SlowDown instead of marking the
+        // drive failed or holding a lane worker after the request is gone.
+        let deadline = Instant::now() + bounded_discovery_time_limit();
+        os::run_cancellable_blocking_namespace_operation_with_sync(lease, true, move |is_cancelled| {
+            let should_stop = || is_cancelled() || Instant::now() >= deadline;
+            let (transactions, next, visited) = session.advance_bounded(256, max_entries, &should_stop)?;
             Ok(super::super::BoundedDeleteRecoveryDiscovery {
                 transactions: if next.is_none() { transactions } else { Vec::new() },
                 visited,
@@ -6556,6 +6608,93 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn delete_recovery_bounded_discovery_charges_uuid_named_children_once_and_keeps_evidence() {
+        let (_temp, disk, _, _) = fixture().await;
+        let path = disk.delete_discovery_path("recovery", Some("prefix")).expect("prefix path");
+        std::fs::create_dir_all(&path).expect("physical prefix");
+        // UUID-named child keys of the folder: each holds its own `xl.meta` and a
+        // data directory, neither of which can be evidence for the folder's key.
+        const CHILDREN: u32 = 2000;
+        for _ in 0..CHILDREN {
+            let child = path.join(Uuid::new_v4().to_string());
+            std::fs::create_dir_all(child.join(Uuid::new_v4().to_string())).expect("child key data dir");
+            std::fs::write(child.join(STORAGE_FORMAT_FILE), b"meta").expect("child key metadata");
+        }
+        // A UUID-named sub-prefix with children of its own.
+        const SUB_PREFIX_CHILDREN: u32 = 640;
+        let sub_prefix = path.join(Uuid::new_v4().to_string());
+        for index in 0..SUB_PREFIX_CHILDREN {
+            std::fs::create_dir_all(sub_prefix.join(format!("child-{index}"))).expect("sub-prefix child");
+        }
+        // Evidence of two interrupted transactions of the key itself: a journaled
+        // rollback directory, and a data-directory marker whose rollback is gone.
+        let journaled = Uuid::new_v4();
+        std::fs::create_dir_all(path.join(journaled.to_string())).expect("rollback directory");
+        std::fs::write(path.join(journaled.to_string()).join(PREPARED), b"{}").expect("journal");
+        let marker_only = Uuid::new_v4();
+        let data_dir = path.join(Uuid::new_v4().to_string());
+        std::fs::create_dir_all(&data_dir).expect("residual data directory");
+        std::fs::write(data_dir.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{marker_only}")), [])
+            .expect("reservation marker");
+
+        let page = disk
+            .discover_delete_transactions_bounded(
+                "recovery",
+                "prefix",
+                super::super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES,
+            )
+            .await
+            .expect("bounded probe");
+        assert!(page.complete, "a folder of UUID-named keys is a complete inventory");
+        // One entry per descent (children, sub-prefix, rollback and data
+        // directory) and per evidence file; inert names at 1/64.
+        let inert = 2 * CHILDREN + SUB_PREFIX_CHILDREN;
+        assert_eq!(page.visited, CHILDREN + 3 + 2 + inert / PLAIN_CHILDREN_PER_BUDGET_ENTRY);
+        let found = page.transactions.iter().map(|id| id.transaction).collect::<BTreeSet<_>>();
+        assert!(found.contains(&journaled), "a journaled rollback directory is evidence");
+        assert!(found.contains(&marker_only), "a marker-only residue is evidence");
+        assert_eq!(found.len(), 2, "no child key is mistaken for a transaction");
+    }
+
+    #[tokio::test]
+    async fn delete_recovery_bounded_discovery_stops_on_request_and_skips_only_inert_irregular_entries() {
+        let (_temp, disk, _, _) = fixture().await;
+        let path = disk.delete_discovery_path("recovery", Some("prefix")).expect("prefix path");
+        let child = path.join(Uuid::new_v4().to_string());
+        std::fs::create_dir_all(&child).expect("UUID-named child key");
+        // A stop request (cancellation or the wall-clock limit) ends the walk
+        // before any entry, leaving continuation state: never evidence of absence.
+        let session = disk
+            .new_delete_discovery_session("recovery", Some("prefix"), &path)
+            .expect("private session");
+        let (transactions, next, visited) = session
+            .advance_bounded(256, super::super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES, &|| true)
+            .expect("stopped walk");
+        assert!(transactions.is_empty());
+        assert_eq!(visited, 0);
+        assert!(next.is_some(), "a stopped walk is incomplete");
+        drop(next);
+        #[cfg(unix)]
+        {
+            // An irregular entry whose name cannot be evidence is skipped unread.
+            std::os::unix::fs::symlink(&path, child.join("inert-link")).expect("inert symlink");
+            let page = disk
+                .discover_delete_transactions_bounded("recovery", "prefix", 4096)
+                .await
+                .expect("inert irregular entries are not evidence");
+            assert!(page.complete && page.transactions.is_empty());
+            // One with an evidence name still fails closed.
+            std::os::unix::fs::symlink(&path, child.join(PREPARED)).expect("evidence-named symlink");
+            assert!(
+                disk.discover_delete_transactions_bounded("recovery", "prefix", 4096)
+                    .await
+                    .is_err(),
+                "an evidence-named symlink is corruption, not a skipped name"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn delete_recovery_bounded_discovery_refuses_partial_inventory_without_public_cursor() {
         let (_temp, disk, _, _) = fixture().await;
@@ -6658,8 +6797,11 @@ mod tests {
         let filler_id = Uuid::new_v4().to_string();
         let filler = path.join(&filler_id);
         fs::create_dir_all(&filler).await.expect("non-evidence directory");
-        for index in 0..5000 {
-            fs::write(filler.join(format!("part-{index:04}")), [])
+        // Inert names are skipped at 1/64, so fill the step with entries that
+        // pass the evidence-name filter but are directories: each costs a full
+        // budget entry and names no transaction.
+        for _ in 0..5000 {
+            fs::create_dir(filler.join(format!("{DELETE_DATA_DIR_MARKER_PREFIX}{}", Uuid::new_v4())))
                 .await
                 .expect("non-evidence entry");
         }

crates/ecstore/src/disk/mod.rs

@@ -2570,7 +2570,13 @@ pub struct BoundedDeleteRecoveryDiscovery {
     pub complete: bool,
 }
 
-pub const DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES: u32 = 4096;
+/// Entries one bounded discovery call may charge on one drive. Object-scoped
+/// discovery charges each transaction-shaped (UUID) child of the key's
+/// directory once and every name that cannot be delete evidence at
+/// 1/`PLAIN_CHILDREN_PER_BUDGET_ENTRY`, so this admits a missing-key delete of a
+/// folder prefix with this many UUID-named children per drive. The request
+/// budget (`DeleteRecoveryWorkBudget`) still bounds the total across keys.
+pub const DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES: u32 = 262_144;
 pub const DELETE_RECOVERY_MAX_INSPECTION_ENTRIES: u32 = 65536;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

crates/ecstore/src/object_api/types.rs

@@ -975,9 +975,11 @@ pub struct DeleteRecoveryWorkBudget {
     remaining: std::sync::atomic::AtomicU32,
 }
 
-/// Entries one disk may visit per recovery step (discovery plus inspection).
-/// Every disk of every set a request fans out to is admitted up to this much
-/// and charged on completion, so a budget must cover that many disks.
+/// Per-drive unit a request budget is sized by (a typical discovery plus one
+/// inspection). Admission charges what a call visited, not this unit, so it only
+/// scales the request total with the drives a request can reach; it is kept
+/// independent of the per-call caps, which may let one key's probe use more of
+/// the same total on a large folder.
 const DELETE_RECOVERY_RESERVATION_PER_DISK: u32 = 4096 + 65_536;
 
 impl Default for DeleteRecoveryWorkBudget {

crates/ecstore/src/set_disk/mod.rs

@@ -11040,7 +11040,12 @@ mod tests {
         let live = Uuid::new_v4();
         let orphan = Uuid::new_v4();
         let object_dir = dir.path().join("bucket/obj");
-        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await;
+        // Part files are inert to discovery and cost 1/64 of an entry, so the
+        // first step is exhausted by what still costs a full entry each: the
+        // data directories of many versions, all referenced by the metadata.
+        let versions = (0..4200).map(|_| Uuid::new_v4()).collect::<Vec<_>>();
+        let referenced = std::iter::once(live).chain(versions.iter().copied()).collect::<Vec<_>>();
+        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &referenced).await;
         let live_dir = object_dir.join(live.to_string());
         let orphan_dir = object_dir.join(orphan.to_string());
         fs::create_dir_all(&live_dir).await.expect("create live multipart directory");
@@ -11049,10 +11054,14 @@ mod tests {
             .await
             .expect("write orphan payload");
         let parts_dir = live_dir.clone();
+        let versions_root = object_dir.clone();
         tokio::task::spawn_blocking(move || {
             for part in 1..=4200 {
                 std::fs::write(parts_dir.join(format!("part.{part}")), b"live").expect("write live multipart entry");
             }
+            for version in versions {
+                std::fs::create_dir(versions_root.join(version.to_string())).expect("create referenced version directory");
+            }
         })
         .await
         .expect("create multipart fixture");
@@ -11062,7 +11071,10 @@ mod tests {
             .await
             .expect("read first discovery page");
         assert!(first.transactions.is_empty());
-        assert!(first.next.is_some(), "ordinary multipart entries must exhaust the first scan budget");
+        assert!(
+            first.next.is_some(),
+            "the data directories of many versions must exhaust the first scan step"
+        );
         let last = disk
             .discover_delete_transactions("bucket", Some("obj"), first.next, 1)
             .await

crates/ecstore/src/set_disk/ops/delete_recovery.rs

@@ -1709,6 +1709,37 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn delete_recovery_missing_key_naming_a_folder_of_uuid_named_keys_completes() {
+        let bucket = "recover-uuid-named-folder";
+        let (dirs, store, _) = fixture(bucket).await;
+        // More UUID-named child keys per drive than the former 4096-entry probe
+        // could cover once each child's `xl.meta` and data directory were charged
+        // (three entries per child).
+        const CHILDREN: usize = 1500;
+        for dir in &dirs {
+            let prefix = dir.path().join(bucket).join("uploads");
+            for _ in 0..CHILDREN {
+                let child = prefix.join(Uuid::new_v4().to_string());
+                std::fs::create_dir_all(child.join(Uuid::new_v4().to_string())).expect("child key data dir");
+                std::fs::write(child.join(crate::disk::STORAGE_FORMAT_FILE), b"meta").expect("child key metadata");
+            }
+        }
+        let error = store
+            .delete_object(bucket, "uploads", ObjectOptions::default())
+            .await
+            .expect_err("the folder's own key does not exist");
+        assert!(
+            crate::error::is_err_object_not_found(&error),
+            "a complete inventory keeps the missing-key no-op semantics: {error:?}"
+        );
+        assert!(dirs.iter().all(|dir| {
+            std::fs::read_dir(dir.path().join(bucket).join("uploads"))
+                .map(|entries| entries.count() == CHILDREN)
+                .unwrap_or(false)
+        }));
+    }
+
     #[tokio::test]
     async fn delete_recovery_batch_budget_survives_pool_and_set_fanout() {
         use crate::layout::endpoint::Endpoint;

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();

crates/scanner/src/scanner_folder.rs

@@ -425,7 +425,10 @@ enum FolderScanSource {
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum FolderScanMode {
     Namespace,
-    ErasureCandidate { reuse_namespace: bool },
+    ErasureCandidate {
+        reuse_namespace: bool,
+        parent_has_metadata: bool,
+    },
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -2065,7 +2068,9 @@ impl FolderScanner {
                     if candidate_unclassified {
                         // Apply namespace reuse only after the current listing
                         // proves there is no direct erasure part.
-                        if let FolderScanMode::ErasureCandidate { reuse_namespace: true } = mode
+                        if let FolderScanMode::ErasureCandidate {
+                            reuse_namespace: true, ..
+                        } = mode
                             && !into.compacted
                             && self.old_cache.find(&folder.name).is_some()
                         {
@@ -2079,10 +2084,23 @@ impl FolderScanner {
                             return Ok(FolderScanOutcome::Namespace);
                         }
                         candidate_unclassified = false;
-                        if !self.budget.try_start_directory() {
-                            return Err(ScannerError::Other("Operation cancelled".to_string()));
+                        // An empty UUID directory beside object metadata has
+                        // no namespace work. Its listing was already charged;
+                        // nonempty namespaces still need directory admission.
+                        let empty_object_child = matches!(
+                            mode,
+                            FolderScanMode::ErasureCandidate {
+                                parent_has_metadata: true,
+                                ..
+                            }
+                        ) && deferred_metadata.is_none()
+                            && queued_folders.is_empty();
+                        if !empty_object_child {
+                            if !self.budget.try_start_directory() {
+                                return Err(ScannerError::Other("Operation cancelled".to_string()));
+                            }
+                            wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
                         }
-                        wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
                     }
                     let Some(entry) = deferred_metadata.take() else { break };
                     (entry, true)
@@ -2552,20 +2570,6 @@ impl FolderScanner {
                 return Err(ScannerError::Other("Operation cancelled".to_string()));
             }
 
-            if found_object_metadata && self.is_erasure_mode {
-                // If we found an object in erasure mode, we skip subdirs (only datadirs)...
-                debug!(
-                    target: "rustfs::scanner::folder",
-                    event = EVENT_SCANNER_FOLDER_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_FOLDER,
-                    folder = %folder.name,
-                    state = "erasure_object_found",
-                    "Scanner folder descent stopped after erasure object"
-                );
-                break;
-            }
-
             // Checkpointed sweeps need child entries to validate their resume
             // frontier (#8292). Outside that, also compact a wide bucket root:
             // excluding it left #8961's ~170k top-level prefixes unbounded until
@@ -2654,6 +2658,7 @@ impl FolderScanner {
                 let mut child_mode = if self.is_erasure_mode && is_erasure_candidate(&folder_item.name) {
                     FolderScanMode::ErasureCandidate {
                         reuse_namespace: reuse_covered,
+                        parent_has_metadata: found_object_metadata,
                     }
                 } else {
                     FolderScanMode::Namespace
@@ -2721,7 +2726,10 @@ impl FolderScanner {
                                 }
                                 // Sampling may skip namespace work, never the
                                 // classification that protects its parent object.
-                                child_mode = FolderScanMode::ErasureCandidate { reuse_namespace: true };
+                                child_mode = FolderScanMode::ErasureCandidate {
+                                    reuse_namespace: true,
+                                    parent_has_metadata: found_object_metadata,
+                                };
                             } else {
                                 folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                             }
@@ -2816,6 +2824,11 @@ impl FolderScanner {
                         updates.cache.delete_recursive(&h);
                     }
                     into.children.remove(&clone_cache_value(&h.0));
+                    // Object keys can also be prefixes. A classified payload
+                    // directory must not hide their other namespace children.
+                    if found_object_metadata {
+                        continue;
+                    }
                     // Earlier namespace siblings may have completed, but none
                     // can certify coverage of this unresolved parent object.
                     self.coverage_gap |= forward_sweep;

crates/scanner/src/scanner_folder/tests/erasure_probe.rs

@@ -183,6 +183,117 @@ fn charged_ops(root: &std::path::Path) -> u64 {
     status.charged_ops
 }
 
+#[tokio::test]
+#[serial]
+async fn erasure_object_metadata_preserves_child_object_namespaces() {
+    for (corrupt_parent, cached_parent) in [(false, false), (true, false), (false, true)] {
+        let (mut scanner, root) = build_test_scanner().await;
+        let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
+        scanner.old_cache.info.name = "bucket".into();
+        scanner.new_cache.info.name = "bucket".into();
+        scanner.is_erasure_mode = true;
+        bind_budget(&mut scanner, &root, 1_000_000);
+
+        let namespace = format!("parent/{}/nested", Uuid::new_v4());
+        for object in ["parent", "parent/child", namespace.as_str()] {
+            write_test_object_metadata_bytes(&root, "bucket", object, &metadata_for_object_version("bucket", object, None)).await;
+        }
+        let payload = root.join("bucket/parent").join(Uuid::new_v4().to_string());
+        tokio::fs::create_dir_all(&payload)
+            .await
+            .expect("create parent payload directory");
+        tokio::fs::write(payload.join("part.1"), b"shard")
+            .await
+            .expect("create actual payload evidence");
+        if corrupt_parent {
+            tokio::fs::write(root.join("bucket/parent/xl.meta"), b"")
+                .await
+                .expect("corrupt only the parent metadata");
+        }
+        if cached_parent {
+            scanner.old_cache.replace("bucket", "", DataUsageEntry::default());
+            scanner.old_cache.replace(
+                "bucket/parent",
+                "bucket",
+                DataUsageEntry {
+                    objects: 1,
+                    size: 1,
+                    compacted: true,
+                    ..Default::default()
+                },
+            );
+            let interval = data_usage_update_dir_cycles();
+            scanner.old_cache.info.next_cycle = u64::from(
+                (0..interval)
+                    .find(|cycle| hash_path("bucket/parent").mod_(*cycle, interval))
+                    .expect("one normal sampling cycle revisits the old parent-only cache"),
+            );
+        }
+
+        let mut into = DataUsageEntry::default();
+        scanner
+            .scan_folder(
+                CancellationToken::new(),
+                CachedFolder {
+                    name: "bucket".into(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut into,
+            )
+            .await
+            .expect("scan object and child namespaces independently");
+        let total = scanner.new_cache.size_recursive("bucket").expect("retain bucket accounting");
+        assert_eq!(total.objects, if corrupt_parent { 2 } else { 3 });
+        assert_eq!(total.size, if corrupt_parent { 2 } else { 3 });
+        assert_eq!(total.failed_objects, usize::from(corrupt_parent));
+        assert_eq!(
+            scanner.new_cache.info.failed_objects.len(),
+            usize::from(corrupt_parent),
+            "payload evidence must not invent missing parent metadata"
+        );
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_object_prefix_children_keep_the_object_budget() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
+    scanner.old_cache.info.name = "bucket".into();
+    scanner.new_cache.info.name = "bucket".into();
+    scanner.is_erasure_mode = true;
+    bind_budget(&mut scanner, &root, 1_000_000);
+    for object in ["parent", "parent/child", "parent/other"] {
+        write_test_object_metadata(&root, "bucket", object).await;
+    }
+    let visits = metadata_visits::observe(&mut scanner);
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        crate::scanner_budget::ScannerCycleBudgetConfig {
+            max_objects: Some(2),
+            ..Default::default()
+        },
+    );
+    scanner.budget = budget.clone();
+    let result = scanner
+        .scan_folder(
+            budget.token(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        )
+        .await;
+    assert!(result.is_err(), "parent metadata cannot hide children or bypass their budget");
+    assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Objects));
+    assert_eq!(budget.progress().0, 2);
+    assert_eq!(visits.load(Ordering::Relaxed), 2);
+}
+
 #[tokio::test]
 #[serial]
 async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
@@ -204,7 +315,10 @@ async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
                 ctx,
                 folder,
                 &mut DataUsageEntry::default(),
-                FolderScanMode::ErasureCandidate { reuse_namespace: false },
+                FolderScanMode::ErasureCandidate {
+                    reuse_namespace: false,
+                    parent_has_metadata: false,
+                },
                 false,
             )
             .await
@@ -271,7 +385,10 @@ async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await;
@@ -311,7 +428,10 @@ async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await;
@@ -382,7 +502,16 @@ async fn erasure_candidate_defers_metadata_until_part_classification_finishes()
     scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
     let mut into = DataUsageEntry::default();
     let outcome = scanner
-        .scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false }, false)
+        .scan_folder_inner(
+            ctx,
+            folder,
+            &mut into,
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
+            false,
+        )
         .await
         .expect("classify candidate");
     assert_eq!(outcome, FolderScanOutcome::ErasureData);
@@ -594,7 +723,10 @@ async fn erasure_candidate_part_symlink_is_not_data_evidence() {
             CancellationToken::new(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await

docs/architecture/s3-compatibility-matrix.md

@@ -130,7 +130,7 @@ These behaviors are client-visible and differ from AWS S3 or MinIO. They bound h
 | Bucket tag key or value outside the AWS tag character class (`[\p{L}\p{N}\p{Z}+\-=._:/@]`) | `400 InvalidTag` on `PutBucketTagging` | AWS enforces the same class; MinIO is more permissive | AWS parity. Legacy stored tags still load. |
 | `CopyObject` with the default `x-amz-tagging-directive: COPY` from a source that carries tags | `403 AccessDenied` unless the caller also holds `s3:GetObjectTagging` (or `s3:GetObjectVersionTagging`) on the source; the destination's `s3:RequestObjectTag/*` conditions are evaluated against the copied tag set | AWS and MinIO copy the tags with `s3:GetObject` alone | A copy would otherwise disclose source tags to a principal that may not read them, and source tags could satisfy destination conditions the caller cannot assert (rustfs/backlog#3034). Use `REPLACE` with an explicit `x-amz-tagging` to copy without the tagging grant. |
 | `ListObjectVersions` / `ListObjectsV2` with the include-deleted extension | Requires `s3:ListBucketVersions` | Vendor extension | Delete markers and non-current versions are version-listing data. |
-| `DeleteObject` of a missing key that names a prefix | `204 NoContent` when the recovery probe of the prefix fits the request budget (`DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES`, 4096 entries per drive, one entry per 64 plain children); `503 SlowDown` beyond that, deterministically, and for the duration of a rolling upgrade while a drive in the set still runs a release without the bounded recovery RPC | `204 NoContent` | The probe looks for interrupted delete transactions under the prefix and is bounded so a hostile prefix cannot pin the request. Prefixes with more than 262,144 direct children per drive, or roughly 1,400 to 4,096 UUID-named children, exceed the bound (rustfs/backlog#3038 residual). |
+| `DeleteObject` of a missing key that names a prefix | `204 NoContent` when the recovery probe of the prefix fits its per-drive allowance (`DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES`, 262,144 entries: one per UUID-named direct child, one per 64 other names), its per-call time limit (a third of the drive timeout, at most 10 seconds) and the request budget (`DeleteRecoveryWorkBudget`, shared by every key and drive of the request); `503 SlowDown` beyond that, and for the duration of a rolling upgrade while a drive in the set still runs a release without the bounded recovery RPC | `204 NoContent` | The probe looks for interrupted delete transactions of the key and is bounded so a hostile prefix cannot pin the request. A folder completes with up to 262,144 UUID-named direct children or about 16.7 million other direct children per drive when the drive can walk them within the time limit (a warm cache walks roughly 25,000 UUID-named children per second; a cold HDD far fewer); inside a UUID-named child only journal files and data-directory markers are examined, so a child key's own `xl.meta`, data directories or children cost 1/64 each. |
 | `PutObject` / `UploadPart` body delivering less than 64 KiB per 60 seconds of server wait | `400 RequestTimeout` (the minimum-progress floor, logged as `upload_body_progress_stalled`); independent of `RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT`, including when that timeout is 0 | No minimum upload rate | A trickle upload otherwise pins upload admission and its buffers indefinitely; roughly 1.1 KiB/s is below any practical client. |
 | First write after `SetBucketQuota`, or after the quota changes, on a bucket without an authoritative usage ledger | `503 SlowDown` (retryable) until a background scan of the bucket publishes the exact ledger; a failed scan retries after 30 seconds | AWS has no bucket quota; MinIO admits with unknown usage | Quota admission uses only the authoritative ledger (scanner reporting is informational, see `reporting_retirement` tests), and the request must not hold its bucket metadata guard across a whole-bucket listing. The window is one bucket listing long. |