ci: build macOS and Windows on a weekly schedule

Keep the 01:13 UTC cron on Linux and add a Sunday 16:00 UTC cron for macOS and Windows. Select that subset from the firing cron so a delayed runner cannot guess the weekday. Co-authored-by: hector <majinghe@users.noreply.github.com>
2026-11-11 12:43:54 +00:00 · 2026-11-11 01:46:03 +00:00
28 changed files with 413 additions and 1330 deletions
@@ -360,14 +360,15 @@
      "cadence": "nightly",
      "timezone": "UTC",
      "schedule": [
-        "13 1 * * *"
+        "13 1 * * *",
+        "0 16 * * 0"
      ],
      "owner": "@overtrue",
      "policy": "active",
      "monitor": "workflow",
      "max_age_hours": 36,
      "result_owner": "build.yml",
-      "disposition": "Builds artifacts only; public release writes belong to release-publish.yml.",
+      "disposition": "Daily cron builds Linux; Sunday 16:00 UTC builds macOS and Windows. Public release writes belong to release-publish.yml.",
      "failure_owner": "workflow",
      "origin": "baseline",
      "alert_contract": "scheduled-validation"
@@ -80,6 +80,6 @@ Rows using `macos-latest-xlarge` or a label starting with `sm-`, `dind-`, `ubicl

 ## Release stages

-The [candidate release pipeline](../../docs/operations/release-pipeline.md) builds Linux first and defers paid desktop capacity until acceptance. Daily development builds also select Linux; use manual platform selection for desktop development artifacts. All six platforms remain required before publication. The pipeline records and promotes the exact tested bytes. Builder workflows have no public-release credentials. Keep periodic development builds, PR verification and field-test chains separate when their contracts differ; do not re-enable disabled workflows as part of a source-file migration.
+The [candidate release pipeline](../../docs/operations/release-pipeline.md) builds Linux first and defers paid desktop capacity until acceptance. The daily `build.yml` schedule selects Linux; the Sunday 16:00 UTC schedule selects macOS and Windows. Manual dispatch can still select any platform list. All six platforms remain required before publication. The pipeline records and promotes the exact tested bytes. Builder workflows have no public-release credentials. Keep periodic development builds, PR verification and field-test chains separate when their contracts differ; do not re-enable disabled workflows as part of a source-file migration.

 Larger runners remain billed for public repositories; see [GitHub runner billing](https://docs.github.com/en/actions/concepts/runners/larger-runners#billing). Actual spending comes from organization billing, not runner labels or an estimated build duration.
@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-# Nightly development artifacts and reusable immutable-candidate builds.
+# Scheduled development artifacts and reusable immutable-candidate builds.
+# The daily cron builds Linux. Sunday 16:00 UTC builds macOS and Windows.
 # Public release writes live exclusively in release-publish.yml. Candidate callers
 # select Linux or desktop and pin release inputs before any matrix starts.

@@ -30,7 +31,8 @@ on:
        required: true
        type: string
  schedule:
-    - cron: "13 1 * * *" # Nightly 01:13 UTC
+    - cron: "13 1 * * *" # Daily Linux, 01:13 UTC
+    - cron: "0 16 * * 0" # Weekly macOS and Windows, Sunday 16:00 UTC (Monday 00:00 Asia/Shanghai)
  workflow_dispatch:
    inputs:
      publish:
@@ -219,16 +221,33 @@ jobs:
        env:
          # via env, not interpolation: a dispatch input is free-form text and
          # would otherwise be pasted into the script for bash to evaluate.
-          # Daily development builds cover Linux. Desktop remains available on
-          # manual dispatch and is required after candidate field acceptance.
+          # A direct schedule has no platforms input. Choose its subset from
+          # github.event.schedule, which is fixed when that cron fires, so a
+          # queued job cannot switch subsets because the runner clock moved.
+          # Callers that pass platforms keep that list, including a scheduled
+          # caller: inputs.platforms is set even though event_name stays schedule.
          RAW_PLATFORMS: ${{ inputs.platforms || 'linux' }}
+          HAS_PLATFORM_INPUT: ${{ inputs.platforms && 'true' || 'false' }}
+          EVENT_NAME: ${{ github.event_name }}
+          EVENT_SCHEDULE: ${{ github.event.schedule }}
        run: |
          set -euo pipefail

-          selected="$RAW_PLATFORMS"
-          selected="$(echo "${selected}" | tr -d '[:space:]')"
-          if [[ -z "${selected}" ]]; then
-            selected="all"
+          if [[ "${EVENT_NAME}" == "schedule" && "${HAS_PLATFORM_INPUT}" != "true" ]]; then
+            if [[ "${EVENT_SCHEDULE}" == "13 1 * * *" ]]; then
+              selected="linux"
+            elif [[ "${EVENT_SCHEDULE}" == "0 16 * * 0" ]]; then
+              selected="desktop"
+            else
+              echo "Unrecognized schedule cron: ${EVENT_SCHEDULE}" >&2
+              exit 1
+            fi
+          else
+            selected="${RAW_PLATFORMS}"
+            selected="$(echo "${selected}" | tr -d '[:space:]')"
+            if [[ -z "${selected}" ]]; then
+              selected="all"
+            fi
          fi

          # linux-aarch64-* need a real aarch64 host (placement rule: hardware)
@@ -322,9 +322,9 @@ dependencies = [

 [[package]]
 name = "archive-trait"
-version = "0.0.16"
+version = "0.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b5040ca935c09888e7dfb04383b4d62fa3571e25bc66b6f345b591e97958524"
+checksum = "999bf88cc483664cab77700ca2fdfb0415a73412abe284ef5bc2319d00938e00"
 dependencies = [
 "cap-std",
 "thiserror 2.0.21",
@@ -811,9 +811,9 @@ dependencies = [

 [[package]]
 name = "async-nats"
-version = "0.52.0"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "015f0a7a3b0b29701b78c76a4d4a122624c97c71fa30363381664eb729d619c7"
+checksum = "a4f92759f418774b78c36ef07885849150ce950c67d9bcdeff8a9c253f57d9a1"
 dependencies = [
 "aws-lc-rs",
 "base64 0.22.1",
@@ -3906,9 +3906,9 @@ dependencies = [

 [[package]]
 name = "dial9"
-version = "0.5.8"
+version = "0.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c83757d61dc75b3e9d3579a3004c7fe0c42aff2f779ca9eeec7f0c8e1ae3fc10"
+checksum = "2f3298a54fd82af243c9f122595762f5b0e48bd3d1ea8fe3f3eb5e95c130e881"
 dependencies = [
 "bon",
 "dial9-core",
@@ -3990,9 +3990,9 @@ dependencies = [

 [[package]]
 name = "dial9-tokio-telemetry"
-version = "0.5.8"
+version = "0.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f0eb05627c12c57e40510e63f9a3d5f628ab4b32fec9ccdb0ae98544c13f15e"
+checksum = "0428cba60fbcf37ab82666f3a7ef3b6d0671e90dfb6a352376813b471319f061"
 dependencies = [
 "bon",
 "bytes",
@@ -4525,9 +4525,9 @@ dependencies = [

 [[package]]
 name = "flate2"
-version = "1.1.11"
+version = "1.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "626d4928326b646e70f65ed36fc7cf9f206bdb68a0019ee7c6b4709da4e084c4"
+checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb"
 dependencies = [
 "crc32fast",
 "miniz_oxide",
@@ -4594,7 +4594,7 @@ checksum = "94e7099f6313ecacbe1256e8ff9d617b75d1bcb16a6fddef94866d225a01a14a"
 dependencies = [
 "io-lifetimes 2.0.4",
 "rustix",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -5385,9 +5385,9 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.31.1"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4039758c8bca06a34c475d17aaafc17ac1073ce2cecd8dba8c82e69edb300bbe"
+checksum = "d9632c1fd213e63c1b88d8f3e59543da4999b17f859cc1e1d67cf8f2207e8935"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -5420,15 +5420,15 @@ dependencies = [

 [[package]]
 name = "hotpath-drain"
-version = "0.31.1"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bbba0a3badfb9ebf95f22dc6cdb4afcd0ff07b91a85fa3196fd1060576ea3065"
+checksum = "738324f75aec847dea984f6806402f4f1bdfab437b4700fc8c3ebcad2dff6424"

 [[package]]
 name = "hotpath-macros"
-version = "0.31.1"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d980014875338771b00ac44f1c9e2ea1a10650aa68ce54226233bc767714361"
+checksum = "5d92fdfec803932e5764ea93c201cf7c6c2d8cff1871da44f64f5094c9c3ef8a"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -6412,9 +6412,9 @@ dependencies = [

 [[package]]
 name = "lzma-rust2"
-version = "0.25.0"
+version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef1d73c59a56771343712b22ea5615092c612c16ba291c0979e3105dfeec4890"
+checksum = "490750560fa7851fbbf5d3ab1a8750f09350ed361ed8aed85b1e1e40587812ee"
 dependencies = [
 "sha2 0.11.1",
 ]
@@ -10336,18 +10336,18 @@ dependencies = [

 [[package]]
 name = "rustfs-mimalloc"
-version = "0.7.4"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc7914232a854336c62058c24a8362c553ac8e7adba2e2395ce9097db24d1510"
+checksum = "216ecfc9341b5d6e2506d47ce57c9232ae1b8020bb10264ea05286fb234286e7"
 dependencies = [
 "rustfs-mimalloc-sys",
 ]

 [[package]]
 name = "rustfs-mimalloc-sys"
-version = "0.7.4"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8073624242781448512682f1931624ee67108b6b7873b22fb29c1a1df689ac0c"
+checksum = "fb16ad68f96883407a75cdb88252b43dc2ea592428f0fc54ff7885756de4b814"
 dependencies = [
 "cc",
 ]
@@ -11705,9 +11705,9 @@ dependencies = [

 [[package]]
 name = "serde_spanned"
-version = "1.1.3"
+version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c187e0a369471c550af028cba56d4f4fa2a489ad28151ae95a4d340a2a8b782"
+checksum = "4e7523beb55eece201a2356bee0bbca0d1ab466c14c07703b2e0ee6d42cb0c2c"
 dependencies = [
 "serde_core",
 ]
@@ -12037,9 +12037,9 @@ dependencies = [

 [[package]]
 name = "smallvec"
-version = "1.16.5"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "149309498a5f7334b12ddae0cad8aba81ce785fe91755068867e017cf98fee46"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
 "serde",
 ]
@@ -12234,9 +12234,9 @@ dependencies = [

 [[package]]
 name = "starshard"
-version = "2.4.1"
+version = "2.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b0883700b50fd0b62120ebc4c07230e02f0cf1ead474c3b3b44eaa13e0a999ea"
+checksum = "fc45de5060fab0de00d7a42230877543ae00c37caf6f49975e69851089b023ac"
 dependencies = [
 "async-trait",
 "hashbrown 0.17.1",
@@ -12485,9 +12485,9 @@ dependencies = [

 [[package]]
 name = "tar-codec"
-version = "0.0.16"
+version = "0.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6875ba507bbec6022500f46477159a03f07ebdffdb4ea043254974531df1dac2"
+checksum = "b9dad615e50d181afea4bafea23dacb8e33ed52dde06503813570634b932d467"
 dependencies = [
 "archive-trait",
 "tar-framing",
@@ -12497,9 +12497,9 @@ dependencies = [

 [[package]]
 name = "tar-framing"
-version = "0.0.16"
+version = "0.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "49727b1a300514960d425d0e741a7da3e693170f3db39de3180fdbbc07e542c5"
+checksum = "563e8181b47c9cdcce569eae3ad431eec71543d6269dd60c7bd2350f70bcd233"
 dependencies = [
 "thiserror 2.0.21",
 "tokio",
@@ -12873,9 +12873,9 @@ dependencies = [

 [[package]]
 name = "toml"
-version = "1.2.1+spec-1.1.0"
+version = "1.2.0+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d5755c51863cdac711f1956baf8eb9588e060755f7d3c4c979e70ea1e1f32cd7"
+checksum = "c5ffcdf8f50436dbbfd675c1fe1b8e0854f42b7defea8584bdebe1aee900e561"
 dependencies = [
 "indexmap 2.14.2",
 "serde_core",
@@ -12888,9 +12888,9 @@ dependencies = [

 [[package]]
 name = "toml_datetime"
-version = "1.1.3+spec-1.1.0"
+version = "1.1.2+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "173f25c883cd2176a2d8212155638c0a60d484a9d84546d270866f40f56b6e49"
+checksum = "2b86d767906c6c42421dcba507eb9d203e779497710a47782a224bb871653053"
 dependencies = [
 "serde_core",
 ]
@@ -13783,6 +13783,15 @@ dependencies = [
 "windows-targets 0.52.6",
 ]

+[[package]]
+name = "windows-sys"
+version = "0.59.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b"
+dependencies = [
+ "windows-targets 0.52.6",
+]
+
 [[package]]
 name = "windows-sys"
 version = "0.60.2"
@@ -13955,7 +13964,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f3fd376f71958b862e7afb20cfe5a22830e1963462f3a17f49d82a6c1d1f42d"
 dependencies = [
 "bitflags 2.13.2",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -14205,9 +14214,9 @@ dependencies = [

 [[package]]
 name = "zip"
-version = "9.5.0"
+version = "9.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0700e6c427aff64acdbf02e85fbfd946c2ee12319d1af013092f3a2178914ff8"
+checksum = "eaeb7f16d2605b49aef586996ded29e65642fab7259a3077e537351df6d80426"
 dependencies = [
 "aes 0.9.4",
 "bzip2",
@@ -149,7 +149,7 @@ async_zip = { default-features = false, version = "0.0.19" }
 mysql_async = { default-features = false, version = "0.37.1" }
 async-compression = { version = "0.4.52" }
 async-trait = "0.1.92"
-async-nats = { version = "0.52.0", default-features = false }
+async-nats = { version = "0.51.0", default-features = false }
 axum = "0.8.9"
 futures = "0.3.34"
 futures-core = "0.3.34"
@@ -249,8 +249,8 @@ anyhow = "1.0.104"
 arc-swap = "1.9.3"
 rustfs-tokio-tar = { version = "0.7.1" }
 # Candidate Snowball parser versions exercised by rustfs-zip compatibility fixtures.
-tar-codec = "0.0.16"
-tar-framing = "0.0.16"
+tar-codec = "0.0.15"
+tar-framing = "0.0.15"
 atoi = "3.1.1"
 atomic_enum = "0.3.0"
 aws-config = { version = "1.13.0" }
@@ -389,7 +389,7 @@ dav-server = "0.11.0"
 # Performance Analysis and Memory Profiling
 rustfs-mimalloc = { version = "0.7.3" }
 # Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
-hotpath = { version = "0.31.1", default-features = false }
+hotpath = { version = "0.30.3", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.49" }

@@ -91,12 +91,7 @@ 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 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_MAX_SINGLE_FILE_SIZE_BYTES: u64 = 0; // No single file limit
 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";
@@ -4320,28 +4320,20 @@ fn take_presynced_tmp_file(file: &Path) -> bool {
    PRESYNCED_TMP_FILES.remove(file).is_some()
 }

-fn open_lazy_local_file_std(
-    file_path: &Path,
-    volume_dir: &Path,
-    _reclaim_on_shutdown: bool,
-    parent_exists: bool,
-) -> std::io::Result<std::fs::File> {
-    let mut open_opts = std::fs::OpenOptions::new();
-    open_opts.write(true).create(true).truncate(true);
-    let file = match parent_exists.then(|| open_opts.open(file_path)) {
-        Some(Ok(file)) => file,
-        Some(Err(err)) if err.kind() != std::io::ErrorKind::NotFound => return Err(to_file_error(err)),
-        _ => {
-            if let Some(parent) = file_path.parent()
-                && (volume_dir.as_os_str().is_empty() || !volume_dir.starts_with(parent))
-                && let Err(err) = super::fs::create_dir_all_std(parent)
-                && err.kind() != std::io::ErrorKind::AlreadyExists
-            {
-                return Err(to_file_error(err));
-            }
-            open_opts.open(file_path).map_err(to_file_error)?
-        }
-    };
+fn open_lazy_local_file_std(file_path: &Path, volume_dir: &Path, _reclaim_on_shutdown: bool) -> std::io::Result<std::fs::File> {
+    if let Some(parent) = file_path.parent()
+        && (volume_dir.as_os_str().is_empty() || !volume_dir.starts_with(parent))
+        && let Err(err) = super::fs::create_dir_all_std(parent)
+        && err.kind() != std::io::ErrorKind::AlreadyExists
+    {
+        return Err(to_file_error(err));
+    }
+    let file = std::fs::OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(true)
+        .open(file_path)
+        .map_err(to_file_error)?;
    #[cfg(target_os = "macos")]
    if _reclaim_on_shutdown {
        let _ = set_std_fd_nocache(&file);
@@ -4356,7 +4348,6 @@ struct LazyLocalFileWriter {
    reclaim_on_shutdown: bool,
    sync_on_close: bool,
    mark_dir_presynced: bool,
-    parent_exists: bool,
    bytes_written: usize,
    state: LazyLocalFileWriterState,
 }
@@ -4377,7 +4368,6 @@ impl LazyLocalFileWriter {
        reclaim_on_shutdown: bool,
        sync_on_close: bool,
        mark_dir_presynced: bool,
-        parent_exists: bool,
    ) -> Self {
        Self {
            file_path,
@@ -4386,7 +4376,6 @@ impl LazyLocalFileWriter {
            reclaim_on_shutdown,
            sync_on_close,
            mark_dir_presynced,
-            parent_exists,
            bytes_written: 0,
            state: LazyLocalFileWriterState::Unopened,
        }
@@ -4399,13 +4388,12 @@ impl LazyLocalFileWriter {
        let reclaim_on_shutdown = self.reclaim_on_shutdown;
        let sync_on_close = self.sync_on_close;
        let mark_dir_presynced = self.mark_dir_presynced;
-        let parent_exists = self.parent_exists;
        match std::mem::replace(&mut self.state, LazyLocalFileWriterState::Closed) {
            LazyLocalFileWriterState::Unopened => {
                let file_path = self.file_path.clone();
                let volume_dir = self.volume_dir.clone();
                self.state = LazyLocalFileWriterState::Writing(tokio::task::spawn_blocking(move || {
-                    let mut file = open_lazy_local_file_std(&file_path, &volume_dir, reclaim_on_shutdown, parent_exists)?;
+                    let mut file = open_lazy_local_file_std(&file_path, &volume_dir, reclaim_on_shutdown)?;
                    std::io::Write::write_all(&mut file, &payload).map_err(to_file_error)?;
                    if close_after {
                        if sync_on_close {
@@ -4547,9 +4535,8 @@ impl AsyncWrite for LazyLocalFileWriter {
                    let file_path = self.file_path.clone();
                    let volume_dir = self.volume_dir.clone();
                    let reclaim_on_shutdown = self.reclaim_on_shutdown;
-                    let parent_exists = self.parent_exists;
                    self.state = LazyLocalFileWriterState::Closing(tokio::task::spawn_blocking(move || {
-                        open_lazy_local_file_std(&file_path, &volume_dir, reclaim_on_shutdown, parent_exists).map(drop)
+                        open_lazy_local_file_std(&file_path, &volume_dir, reclaim_on_shutdown).map(drop)
                    }));
                }
                LazyLocalFileWriterState::Open(file) => {
@@ -5598,14 +5585,11 @@ impl LocalIoBackend for StdBackend {
                let file_path = local_disk_object_path(self.io_root(), volume, path)?;
                check_path_length(file_path.to_string_lossy().as_ref())?;
                let mut mark_dir_presynced = false;
-                let mut parent_exists = false;
                if let Some(parent) = file_path.parent() {
                    check_path_length(parent.to_string_lossy().as_ref())?;
-                    parent_exists = parent == volume_dir;
                    if volume == RUSTFS_META_TMP_BUCKET {
                        unmark_presynced_tmp_file(&file_path);
                        let parent_was_presynced = unmark_presynced_tmp_data_dir(parent);
-                        parent_exists |= parent_was_presynced;
                        let file_name = file_path.file_name().and_then(|n| n.to_str()).unwrap_or_default();
                        mark_dir_presynced = sync_on_close
                            && (file_name == "part.1"
@@ -5656,7 +5640,6 @@ impl LocalIoBackend for StdBackend {
                    reclaim_on_shutdown,
                    sync_on_close,
                    mark_dir_presynced,
-                    parent_exists,
                )))
            }
            WriteMode::Append => {
@@ -286,15 +286,12 @@ pub(super) fn fsync_inline_commit_std(

    if destination_was_absent {
        let mut ancestor = dst.parent().and_then(Path::parent);
-        let mut ancestor_step = 0usize;
        while let Some(ancestor_dir) = ancestor {
            if !ancestor_dir.starts_with(dst_volume_dir) {
                break;
            }
            let fsync_started = rustfs_io_metrics::put_stage_timer();
-            if let Err(err) =
-                os::fsync_commit_ancestor_directory_limited_std(ancestor_dir, dst_volume_dir, ancestor_step, commit_guard)
-            {
+            if let Err(err) = os::fsync_dir_limited_std(ancestor_dir) {
                rustfs_io_metrics::record_put_object_stage_duration_from(
                    rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC,
                    fsync_started,
@@ -313,7 +310,6 @@ pub(super) fn fsync_inline_commit_std(
                break;
            }
            ancestor = ancestor_dir.parent();
-            ancestor_step += 1;
        }
    }
    Ok(())
@@ -41,53 +41,19 @@ struct DiscoverySession {
    touched: Instant,
 }

-/// 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.
+/// 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.
 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, DISCOVERY_STEP_ENTRIES, &|| false)
+        self.advance_bounded(limit, super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES)
            .map(|(transactions, next, _)| (transactions, next))
    }

-    /// 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)> {
+    fn advance_bounded(mut self, limit: u32, max_entries: u32) -> 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)),
@@ -100,11 +66,10 @@ impl DiscoverySession {
        let mut transactions = BTreeSet::new();
        let mut seen_candidates = HashSet::new();
        let mut visited = 0u32;
-        let mut skipped_by_name = 0u32;
+        let mut plain_children = 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
@@ -125,22 +90,13 @@ impl DiscoverySession {
                return Err(std::io::Error::from(ErrorKind::InvalidData));
            }
            // An object-scoped scan only ever descends into transaction (UUID)
-            // 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) {
+            // 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) {
                    visited += 1;
                }
                continue;
@@ -1172,8 +1128,6 @@ struct PreparedDeleteReceiptEntry {
    journal_raw: Vec<u8>,
    journal: DeleteRecoveryJournal,
    sole_transaction: bool,
-    had_old_data_dir: bool,
-    rollback_dir_fd: Option<Arc<std::fs::File>>,
 }

 #[derive(Clone)]
@@ -1183,8 +1137,6 @@ struct PublishedDeleteReceiptEntry {
    file_path: PathBuf,
    staging_root: PathBuf,
    journal_raw: Vec<u8>,
-    had_old_data_dir: bool,
-    rollback_dir_fd: Option<Arc<std::fs::File>>,
 }

 const PREPARED_DELETE_RECEIPTS_PER_SHARD: usize = 64;
@@ -1435,16 +1387,8 @@ 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.
-        // 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)?;
+        os::run_blocking_namespace_operation(lease, move || {
+            let (transactions, next, visited) = session.advance_bounded(256, max_entries)?;
            Ok(super::super::BoundedDeleteRecoveryDiscovery {
                transactions: if next.is_none() { transactions } else { Vec::new() },
                visited,
@@ -2551,9 +2495,8 @@ impl LocalDisk {
                    Err(err) if err.kind() == ErrorKind::NotFound => return Ok(Ok(None)),
                    Err(err) => return Err(err),
                };
-                type PreparedDeleteTuple =
-                    (Option<Uuid>, Option<Vec<u8>>, Vec<u8>, DeleteRecoveryJournal, bool, bool, Option<bool>);
-                let (old_dir, target, journal_raw, journal, sole_transaction, destination_missing, old_data_dir_state) =
+                type PreparedDeleteTuple = (Option<Uuid>, Option<Vec<u8>>, Vec<u8>, DeleteRecoveryJournal, bool, bool, bool);
+                let (old_dir, target, journal_raw, journal, sole_transaction, destination_missing, old_data_dir_present) =
                    match (|| -> Result<Option<PreparedDeleteTuple>> {
                        let mut meta = FileMeta::load(&buf)?;
                        let mut deleted_headers: Vec<_> = meta
@@ -2570,12 +2513,13 @@ impl LocalDisk {
                        if old_dir.is_some_and(|dir| dir == transaction || dir.is_nil()) {
                            return Err(DiskError::FileCorrupt);
                        }
-                        let mut old_path_to_verify = None;
                        if let Some(uuid) = old_dir {
                            meta.data.remove(vec![fi_clone.version_id.unwrap_or_default(), uuid])?;
                            let old_path = path_join(&[file_path.as_path(), Path::new(uuid.to_string().as_str())]);
                            check_path_length(old_path.to_string_lossy().as_ref())?;
-                            old_path_to_verify = Some(old_path);
+                            if old_path.join(STORAGE_FORMAT_FILE).exists() {
+                                return Ok(None);
+                            }
                        }
                        let surviving: HashSet<_> = meta
                            .versions
@@ -2657,12 +2601,6 @@ impl LocalDisk {
                                }
                            }
                        }
-                        if let Some(old_path) = old_path_to_verify
-                            && (!scanned_file_path || old_data_dir_exists)
-                            && old_path.join(STORAGE_FORMAT_FILE).exists()
-                        {
-                            return Ok(None);
-                        }
                        sole_transaction &= predecessors.is_empty();
                        let mut sparse = fi_clone;
                        sparse.data = None;
@@ -2688,21 +2626,15 @@ impl LocalDisk {
                            journal,
                            sole_transaction,
                            scanned_file_path && !destination_exists,
-                            scanned_file_path.then_some(old_data_dir_exists),
+                            scanned_file_path && old_data_dir_exists,
                        )))
                    })() {
                        Ok(Some(tuple)) => tuple,
                        Ok(None) => return Ok(Ok(None)),
                        Err(err) => return Ok(Err(err)),
                    };
-                let staged = match tempfile::Builder::new().prefix("delete-prep-").tempdir_in(&staging_root) {
-                    Ok(dir) => dir,
-                    Err(err) if err.kind() == ErrorKind::NotFound => {
-                        std::fs::create_dir_all(&staging_root)?;
-                        tempfile::Builder::new().prefix("delete-prep-").tempdir_in(&staging_root)?
-                    }
-                    Err(err) => return Err(err),
-                };
+                std::fs::create_dir_all(&staging_root)?;
+                let staged = tempfile::Builder::new().prefix("delete-prep-").tempdir_in(&staging_root)?;
                let staged_dir = staged.path().to_path_buf();
                let backup_path = staged_dir.join(STORAGE_FORMAT_FILE_BACKUP);
                if std::fs::hard_link(&xl_path, &backup_path).is_err() {
@@ -2726,12 +2658,8 @@ impl LocalDisk {
                for file in written_files {
                    file.sync_data()?;
                }
-                #[cfg(unix)]
-                let rollback_dir_fd = sync.then(|| std::fs::File::open(&staged_dir).ok().map(Arc::new)).flatten();
-                #[cfg(not(unix))]
-                let rollback_dir_fd: Option<Arc<std::fs::File>> = None;
                if sync {
-                    os::fsync_directory_handle_std(&staged_dir, rollback_dir_fd.as_deref())?;
+                    os::fsync_dir_std(&staged_dir)?;
                }
                if !destination_missing {
                    match std::fs::symlink_metadata(&destination) {
@@ -2753,12 +2681,8 @@ impl LocalDisk {
                }
                let old_data_path = old_dir
                    .map(|uuid| file_path.join(uuid.to_string()))
-                    .filter(|p| old_data_dir_state.unwrap_or_else(|| p.is_dir()));
-                #[cfg(unix)]
-                let file_path_dir = sync.then(|| std::fs::File::open(&file_path).ok()).flatten();
-                #[cfg(not(unix))]
-                let file_path_dir: Option<std::fs::File> = None;
-                if sync && let Err(err) = os::fsync_directory_handle_std(&file_path, file_path_dir.as_ref()) {
+                    .filter(|p| old_data_dir_present || p.is_dir());
+                if sync && let Err(err) = os::fsync_dir_std(&file_path) {
                    let _ = std::fs::remove_dir_all(&destination);
                    return Err(err);
                }
@@ -2767,15 +2691,11 @@ impl LocalDisk {
                let mut reserved_data_path = None;
                if let Some(old_path) = old_data_path {
                    let reserved_marker = old_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
-                    #[cfg(unix)]
-                    let old_dir_file = sync.then(|| std::fs::File::open(&old_path).ok()).flatten();
-                    #[cfg(not(unix))]
-                    let old_dir_file: Option<std::fs::File> = None;
                    let reserve_res = (|| -> std::io::Result<()> {
                        let marker = std::fs::File::create(&reserved_marker)?;
                        if sync {
                            marker.sync_all()?;
-                            os::fsync_directory_handle_std(&old_path, old_dir_file.as_ref())?;
+                            os::fsync_dir_std(&old_path)?;
                        }
                        Ok(())
                    })();
@@ -2784,7 +2704,7 @@ impl LocalDisk {
                        let _ = std::fs::remove_dir_all(&destination);
                        return Err(err);
                    }
-                    reserved_data_path = Some((old_path, reserved_marker, old_dir_file));
+                    reserved_data_path = Some((old_path, reserved_marker));
                }
                if let Some(ref target_bytes) = target {
                    // A provisional head must stay outside the published recovery transaction.
@@ -2815,7 +2735,7 @@ impl LocalDisk {
                        })()
                    };
                    if let Err(err) = write_res {
-                        if let Some((_, reserved_marker, _)) = &reserved_data_path {
+                        if let Some((_, reserved_marker)) = &reserved_data_path {
                            let _ = std::fs::remove_file(reserved_marker);
                        }
                        let _ = std::fs::remove_dir_all(&destination);
@@ -2824,20 +2744,19 @@ impl LocalDisk {
                } else if let Err(err) = std::fs::remove_file(&xl_path)
                    && err.kind() != ErrorKind::NotFound
                {
-                    if let Some((_, reserved_marker, _)) = &reserved_data_path {
+                    if let Some((_, reserved_marker)) = &reserved_data_path {
                        let _ = std::fs::remove_file(reserved_marker);
                    }
                    let _ = std::fs::remove_dir_all(&destination);
                    return Err(err);
                }
                if sync {
-                    os::fsync_directory_handle_std(&file_path, file_path_dir.as_ref())?;
+                    os::fsync_dir_std(&file_path)?;
                }
-                let had_old_data_dir = reserved_data_path.is_some();
-                if let Some((old_path, reserved_marker, old_dir_file)) = reserved_data_path {
+                if let Some((old_path, reserved_marker)) = reserved_data_path {
                    let recovery_marker = old_path.join(format!("{RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
                    match std::fs::rename(&reserved_marker, &recovery_marker) {
-                        Ok(()) if sync => os::fsync_directory_handle_std(&old_path, old_dir_file.as_ref())?,
+                        Ok(()) if sync => os::fsync_dir_std(&old_path)?,
                        Ok(()) => {}
                        Err(err) if err.kind() == ErrorKind::NotFound => {}
                        Err(err) => return Err(err),
@@ -2853,11 +2772,9 @@ impl LocalDisk {
                        journal_raw,
                        journal,
                        sole_transaction,
-                        had_old_data_dir,
-                        rollback_dir_fd,
                    },
                );
-                Ok(Ok(Some(old_dir.filter(|_| had_old_data_dir))))
+                Ok(Ok(Some(old_dir)))
            },
        )
        .await
@@ -2910,13 +2827,7 @@ impl LocalDisk {
                Some(entry.bucket_path),
                entry.file_path,
                entry.staging_root,
-                Some((
-                    entry.journal_raw,
-                    entry.journal,
-                    entry.sole_transaction,
-                    entry.had_old_data_dir,
-                    entry.rollback_dir_fd,
-                )),
+                Some((entry.journal_raw, entry.journal, entry.sole_transaction)),
            ),
            None => {
                self.reject_multipart_delete_recovery_mutation(volume, object)?;
@@ -2947,7 +2858,7 @@ impl LocalDisk {
            lease,
            sync,
            move || -> std::io::Result<Result<Option<(DeleteRecoveryJournal, bool, bool)>>> {
-                if let Some((journal_raw, journal, sole_transaction, had_old_data_dir, rollback_dir_fd)) = cached_receipt
+                if let Some((journal_raw, journal, sole_transaction)) = cached_receipt
                    && rollback.join(PREPARED).is_file()
                {
                    let tmp_receipt = staging_root.join(format!("receipt-{transaction}.tmp"));
@@ -2960,7 +2871,7 @@ impl LocalDisk {
                        }
                        std::fs::rename(&tmp_receipt, &receipt_path)?;
                        if sync {
-                            os::fsync_directory_handle_std(&rollback, rollback_dir_fd.as_deref())?;
+                            os::fsync_dir_std(&rollback)?;
                        }
                        Ok(())
                    })();
@@ -2977,8 +2888,6 @@ impl LocalDisk {
                                file_path: root,
                                staging_root,
                                journal_raw,
-                                had_old_data_dir,
-                                rollback_dir_fd,
                            },
                        );
                    }
@@ -3114,34 +3023,29 @@ impl LocalDisk {
            None
        };
        let cached_published = rollback_open_path.as_deref().and_then(take_published_delete_receipt);
-        let (object_path, bucket_path, staging_root, root, cached_journal_raw, cached_had_old_data_dir, mut rollback_dir_fd) =
-            match cached_published {
-                Some(entry) => (
-                    entry.object_path,
-                    entry.bucket_path,
-                    entry.staging_root,
-                    entry.file_path,
-                    Some(entry.journal_raw),
-                    Some(entry.had_old_data_dir),
-                    entry.rollback_dir_fd,
-                ),
-                None => {
-                    self.reject_multipart_delete_recovery_mutation(volume, object)?;
-                    if !safe_path(object, false) {
-                        return Err(DiskError::InvalidPath);
-                    }
-                    let obj_path = self.get_object_path(volume, object)?;
-                    (
-                        obj_path,
-                        self.io_get_bucket_path(volume)?,
-                        self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
-                        self.io_get_object_path(volume, object)?,
-                        None,
-                        None,
-                        None,
-                    )
+        let (object_path, bucket_path, staging_root, root, cached_journal_raw) = match cached_published {
+            Some(entry) => (
+                entry.object_path,
+                entry.bucket_path,
+                entry.staging_root,
+                entry.file_path,
+                Some(entry.journal_raw),
+            ),
+            None => {
+                self.reject_multipart_delete_recovery_mutation(volume, object)?;
+                if !safe_path(object, false) {
+                    return Err(DiskError::InvalidPath);
                }
-            };
+                let obj_path = self.get_object_path(volume, object)?;
+                (
+                    obj_path,
+                    self.io_get_bucket_path(volume)?,
+                    self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
+                    self.io_get_object_path(volume, object)?,
+                    None,
+                )
+            }
+        };
        let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await;
        let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
        let owner: Option<Arc<dyn Send + Sync>> = Some(Arc::new((metadata_owner, volume_guard)));
@@ -3214,61 +3118,56 @@ impl LocalDisk {
            let recovery_committed = format!("{RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
            let reserved = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
            let mut had_physical_data_dirs = false;
-            let fast_cleanup = cached_journal_raw.is_some();
-            if cached_had_old_data_dir != Some(false) {
-                let bases: &[&Path] = if fast_cleanup { &[&root] } else { &[&root, &rollback] };
-                for data_dir in &data_dirs {
-                    for base in bases {
-                        let dir_path = base.join(data_dir.to_string());
-                        match std::fs::read_dir(&dir_path) {
-                            Ok(entries) => {
-                                had_physical_data_dirs = true;
-                                let mut removable = Vec::new();
-                                let mut preserve_dir = false;
-                                for entry in recovery_entries(entries, work_budget.as_deref()) {
-                                    let entry = entry?;
-                                    let name = entry.file_name();
-                                    let Some(name) = name.to_str() else {
-                                        preserve_dir = true;
-                                        continue;
-                                    };
-                                    if entry.file_type()?.is_file()
-                                        && (delete_recovery_part_file(name)
-                                            || name == committed
-                                            || name == recovery_committed
-                                            || name == reserved
-                                            || is_delete_transaction_marker(name, RESERVED_DELETE_DATA_DIR_MARKER_PREFIX))
-                                    {
-                                        removable.push(entry.path());
-                                    } else {
-                                        preserve_dir = true;
-                                    }
-                                }
-                                for path in removable {
-                                    ignore_not_found(std::fs::remove_file(path))?;
-                                }
-                                let mut dir_removed = false;
-                                if !preserve_dir {
-                                    match std::fs::remove_dir(&dir_path) {
-                                        Ok(()) => dir_removed = true,
-                                        Err(err) if err.kind() == ErrorKind::NotFound => dir_removed = true,
-                                        Err(err) if is_dir_not_empty_error(&err) => {}
-                                        Err(err) => return Err(err),
-                                    }
-                                }
-                                if !dir_removed && sync {
-                                    os::fsync_dir_std(&dir_path)?;
+            for data_dir in &data_dirs {
+                for base in [&root, &rollback] {
+                    let dir_path = base.join(data_dir.to_string());
+                    match std::fs::read_dir(&dir_path) {
+                        Ok(entries) => {
+                            had_physical_data_dirs = true;
+                            let mut removable = Vec::new();
+                            let mut preserve_dir = false;
+                            for entry in recovery_entries(entries, work_budget.as_deref()) {
+                                let entry = entry?;
+                                let name = entry.file_name();
+                                let Some(name) = name.to_str() else {
+                                    preserve_dir = true;
+                                    continue;
+                                };
+                                if entry.file_type()?.is_file()
+                                    && (delete_recovery_part_file(name)
+                                        || name == committed
+                                        || name == recovery_committed
+                                        || name == reserved
+                                        || is_delete_transaction_marker(name, RESERVED_DELETE_DATA_DIR_MARKER_PREFIX))
+                                {
+                                    removable.push(entry.path());
+                                } else {
+                                    preserve_dir = true;
+                                }
+                            }
+                            for path in removable {
+                                ignore_not_found(std::fs::remove_file(path))?;
+                            }
+                            if sync {
+                                os::fsync_dir_std(&dir_path)?;
+                            }
+                            if !preserve_dir {
+                                match std::fs::remove_dir(&dir_path) {
+                                    Ok(()) => {}
+                                    Err(err) if err.kind() == ErrorKind::NotFound || is_dir_not_empty_error(&err) => {}
+                                    Err(err) => return Err(err),
                                }
                            }
-                            Err(err) if err.kind() == ErrorKind::NotFound => {}
-                            Err(err) => return Err(err),
                        }
+                        Err(err) if err.kind() == ErrorKind::NotFound => {}
+                        Err(err) => return Err(err),
                    }
                }
            }
            if had_physical_data_dirs && sync && root.is_dir() {
                os::fsync_dir_std(&root)?;
            }
+            let fast_cleanup = cached_journal_raw.is_some();
            if fast_cleanup {
                let backup_or_marker = if has_original {
                    Some(STORAGE_FORMAT_FILE_BACKUP)
@@ -3317,15 +3216,12 @@ impl LocalDisk {
                    }
                }
            }
-            let rollback_sync = rollback_dir_fd.take();
-            if sync && let Err(err) = os::fsync_directory_handle_std(&rollback, rollback_sync.as_deref()) {
-                drop(rollback_sync);
+            if sync && let Err(err) = os::fsync_dir_std(&rollback) {
                if has_target || had_physical_data_dirs {
                    let _ = persist_receipt(true);
                }
                return Err(err);
            }
-            drop(rollback_sync);
            #[cfg(all(any(test, feature = "test-util"), not(windows)))]
            os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &receipt_path);
            if let Err(err) = ignore_not_found(std::fs::remove_file(&receipt_path)) {
@@ -6608,93 +6504,6 @@ 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;
@@ -6797,11 +6606,8 @@ 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");
-        // 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())))
+        for index in 0..5000 {
+            fs::write(filler.join(format!("part-{index:04}")), [])
                .await
                .expect("non-evidence entry");
        }
@@ -2570,13 +2570,7 @@ pub struct BoundedDeleteRecoveryDiscovery {
    pub complete: bool,
 }

-/// 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_DISCOVERY_ENTRIES: u32 = 4096;
 pub const DELETE_RECOVERY_MAX_INSPECTION_ENTRIES: u32 = 65536;

 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -611,6 +611,10 @@ pub(crate) mod prepare_rename_test_hooks {
        }
    }

+    pub(crate) fn has_after_component_opened() -> bool {
+        !AFTER_COMPONENT_OPENED.lock().is_empty()
+    }
+
    /// Hooks keyed by a directory component name, run before a Windows
    /// directory pin retries a transient sharing conflict on it.
    #[cfg(windows)]
@@ -1280,28 +1284,6 @@ pub(crate) fn fsync_dir_limited_std(dir: &Path) -> io::Result<()> {
    fsync_dir_std(dir)
 }

-pub(crate) fn fsync_commit_ancestor_directory_limited_std(
-    dir: &Path,
-    base_dir: &Path,
-    step_from_destination: usize,
-    guard: &RenameCommitGuard,
-) -> io::Result<()> {
-    #[cfg(target_os = "linux")]
-    if base_dir == guard.directories.destination_base
-        && guard.directories.destination_path.ancestors().nth(1 + step_from_destination) == Some(dir)
-        && let Some(ancestors) = guard.directories.destination_ancestors.get()
-        && let Some(idx) = ancestors.len().checked_sub(1 + step_from_destination)
-        && let Some(file) = ancestors.get(idx)
-    {
-        #[cfg(test)]
-        fsync_dir_recorder::record_limited(dir);
-        return fsync_directory_handle_std(dir, Some(file));
-    }
-    #[cfg(not(target_os = "linux"))]
-    let _ = (base_dir, step_from_destination, guard);
-    fsync_dir_limited_std(dir)
-}
-
 /// Reuse the published directory without bypassing group commit or sync admission.
 pub(crate) async fn fsync_commit_directory(
    dir: &Path,
@@ -5766,7 +5748,6 @@ struct LinuxCommitDirectories {
    destination_path: PathBuf,
    source: LinuxCommitDirectory,
    destination: std::sync::OnceLock<LinuxCommitDirectory>,
-    destination_ancestors: std::sync::OnceLock<Vec<std::fs::File>>,
    destination_base: PathBuf,
 }

@@ -5783,8 +5764,6 @@ impl LinuxCommitDirectories {
                .ok_or_else(|| io::Error::other("missing destination directory guard"))?,
        );
        let directory = LinuxCommitDirectory::new(directory)?;
-        let ancestors = parents.into_iter().map(std::fs::File::from).collect();
-        let _ = self.destination_ancestors.set(ancestors);
        Ok(self.destination.get_or_init(|| directory))
    }

@@ -5794,7 +5773,7 @@ impl LinuxCommitDirectories {
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "file parent does not match its commit guard"))
    }

-    fn validate(&self, check_destination: bool) -> io::Result<()> {
+    fn validate(&self) -> io::Result<()> {
        use std::os::unix::fs::MetadataExt;
        // A pinned directory may have been unlinked or replaced. Do not report a
        // successful publication into a detached object tree.
@@ -5803,9 +5782,8 @@ impl LinuxCommitDirectories {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "commit directory identity changed"));
        }
        let destination = self.prepare_destination()?;
-        if check_destination
-            && ((self.source.device, self.source.inode) != (destination.device, destination.inode)
-                || self.source_path.as_os_str() != self.destination_path.as_os_str())
+        if (self.source.device, self.source.inode) != (destination.device, destination.inode)
+            || self.source_path.as_os_str() != self.destination_path.as_os_str()
        {
            let current = self.root.open_directory(&self.destination_path)?.metadata()?;
            if (current.dev(), current.ino()) != (destination.device, destination.inode) {
@@ -5818,9 +5796,8 @@ impl LinuxCommitDirectories {
    fn rename(&self, source: &Path, destination: &Path, prepared: Option<&PreparedRenameSource>) -> io::Result<()> {
        let source_name = self.file_name(source, true)?;
        let destination_name = self.file_name(destination, false)?;
-        let destination_was_prepared = self.destination.get().is_some();
        let destination_parent = self.prepare_destination()?;
-        self.validate(destination_was_prepared)?;
+        self.validate()?;
        let mut attempt = 0;
        loop {
            if let Some(prepared) = prepared {
@@ -6042,7 +6019,7 @@ pub(crate) fn multipart_payload_commit_guard_std(
    source_parent: &Path,
 ) -> io::Result<RenameCommitGuard> {
    let original = &metadata_guard.directories;
-    original.validate(true)?;
+    original.validate()?;
    let source = LinuxCommitDirectory::new(original.root.open_directory(source_parent)?)?;
    let destination = original.prepare_destination()?;
    // Clone the pinned file identity, never recapture the target pathname
@@ -6059,7 +6036,6 @@ pub(crate) fn multipart_payload_commit_guard_std(
            destination_path: original.destination_path.clone(),
            source,
            destination: std::sync::OnceLock::from(destination),
-            destination_ancestors: std::sync::OnceLock::new(),
            destination_base: original.destination_base.clone(),
        }),
    })
@@ -6141,7 +6117,6 @@ pub(crate) fn prepare_rename_commit_guard(
                destination_path: destination_parent.to_path_buf(),
                source,
                destination,
-                destination_ancestors: std::sync::OnceLock::new(),
                destination_base: destination_base.to_path_buf(),
            }),
        })
@@ -7251,7 +7226,6 @@ fn same_parent_commit_guard(
                    device: dir.device,
                    inode: dir.inode,
                }),
-                destination_ancestors: std::sync::OnceLock::new(),
                destination_base: base.to_path_buf(),
            }),
        });
@@ -8290,7 +8264,37 @@ pub(crate) fn mkdir_all_below_existing_base_std(
        let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC;
        let mode = Mode::RWXU | Mode::RWXG | Mode::RWXO;
        #[cfg(target_os = "linux")]
-        let base = publication_root.open_directory(base_dir)?.into();
+        let base = {
+            use rustix::fs::{ResolveFlags, openat2};
+            use std::os::fd::AsFd;
+
+            let owned_base = (base_dir != publication_root.io_path.as_path() && base_dir != publication_root.path.as_path())
+                .then(|| publication_root.open_directory(base_dir).map(std::os::fd::OwnedFd::from))
+                .transpose()?;
+            #[cfg(test)]
+            let has_hook = prepare_rename_test_hooks::has_after_component_opened();
+            #[cfg(not(test))]
+            let has_hook = false;
+            if !has_hook {
+                let base_fd = owned_base
+                    .as_ref()
+                    .map_or_else(|| publication_root.directory.as_fd(), AsFd::as_fd);
+                let target = if relative.as_os_str().is_empty() {
+                    Path::new(".")
+                } else {
+                    relative
+                };
+                match openat2(base_fd, target, flags, Mode::empty(), ResolveFlags::BENEATH | ResolveFlags::NO_SYMLINKS) {
+                    Ok(leaf) => return Ok(vec![leaf]),
+                    Err(Errno::NOENT | Errno::NOSYS) => {}
+                    Err(err) => return Err(err.into()),
+                }
+            }
+            match owned_base {
+                Some(base) => base,
+                None => publication_root.open_directory(base_dir)?.into(),
+            }
+        };
        #[cfg(not(target_os = "linux"))]
        let base = {
            // The configured endpoint is trusted; every component beneath it
@@ -975,11 +975,9 @@ pub struct DeleteRecoveryWorkBudget {
    remaining: std::sync::atomic::AtomicU32,
 }

-/// 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.
+/// 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.
 const DELETE_RECOVERY_RESERVATION_PER_DISK: u32 = 4096 + 65_536;

 impl Default for DeleteRecoveryWorkBudget {
@@ -11040,12 +11040,7 @@ mod tests {
        let live = Uuid::new_v4();
        let orphan = Uuid::new_v4();
        let object_dir = dir.path().join("bucket/obj");
-        // 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;
+        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).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");
@@ -11054,14 +11049,10 @@ 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");
@@ -11071,10 +11062,7 @@ mod tests {
            .await
            .expect("read first discovery page");
        assert!(first.transactions.is_empty());
-        assert!(
-            first.next.is_some(),
-            "the data directories of many versions must exhaust the first scan step"
-        );
+        assert!(first.next.is_some(), "ordinary multipart entries must exhaust the first scan budget");
        let last = disk
            .discover_delete_transactions("bucket", Some("obj"), first.next, 1)
            .await
@@ -1709,37 +1709,6 @@ 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;
@@ -145,7 +145,7 @@ With OTLP → Collector → Prometheus, the time from a failure to an alert also

 | Variable                        | Default      | Description                                                                                                                                                                                                                                                                                                                                                           |
 |---------------------------------|--------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `RUSTFS_OBS_LOGGER_LEVEL`       | `error`      | Log level; `RUST_LOG` syntax supported                                                                                                                                                                                                                                                                                                                                |
+| `RUSTFS_OBS_LOGGER_LEVEL`       | `info`       | 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,31 +157,23 @@ With OTLP → Collector → Prometheus, the time from a failure to an alert also

 | Variable                                        | Default      | Description                                                 |
 |-------------------------------------------------|--------------|-------------------------------------------------------------|
-| `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_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_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) – `22` (best ratio); `0` = codec default |
+| `RUSTFS_OBS_LOG_ZSTD_COMPRESSION_LEVEL`         | `8`          | Zstd level `1` (fastest) – `21` (best ratio)               |
 | `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; 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_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_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
@@ -149,7 +149,6 @@ impl LogCleaner {

        let LogScanResult {
            mut logs,
-            recent_logs,
            mut compressed_archives,
        } = scan_log_directory(
            &self.log_dir,
@@ -162,9 +161,8 @@ 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!(
@@ -180,35 +178,16 @@ impl LogCleaner {

            // Select the oldest files first, then additionally trim any files
            // that still violate configured size constraints.
-            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;
+            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;
+            }
        }

        if !compressed_archives.is_empty() {
@@ -238,44 +217,6 @@ 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
@@ -444,171 +444,6 @@ 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()?;
@@ -48,10 +48,6 @@ 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>,
 }
@@ -87,13 +83,11 @@ 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();

@@ -105,7 +99,6 @@ 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,
            });
        }
@@ -237,7 +230,10 @@ 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.
-        let recent = !is_compressed && !is_old_enough_for_cleanup(age, min_file_age_seconds);
+        if !is_compressed && !is_old_enough_for_cleanup(age, min_file_age_seconds) {
+            // Too young to be touched.
+            continue;
+        }

        let info = FileInfo {
            path,
@@ -248,9 +244,6 @@ 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);
        }
@@ -258,7 +251,6 @@ pub(super) fn scan_log_directory(

    Ok(LogScanResult {
        logs,
-        recent_logs,
        compressed_archives,
    })
 }
@@ -319,42 +311,6 @@ 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()?;
@@ -26,21 +26,22 @@ 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_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_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,
 };
 use rustfs_config::{
    APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOG_FILENAME,
@@ -188,13 +189,9 @@ pub struct OtelConfig {
    pub log_keep_files: Option<usize>,

    // ── Log cleanup ───────────────────────────────────────────────────────────
-    /// Byte budget for rolled log files, and separately for compressed
-    /// archives (default: 2 GiB); `0` means no budget. The active file is not
-    /// counted.
+    /// Hard ceiling on the total size (bytes) of all log files (default: 2 GiB).
    pub log_max_total_size_bytes: Option<u64>,
-    /// 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.
+    /// Per-file size ceiling (bytes); `0` means unlimited (default: `0`).
    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>,
@@ -347,8 +344,10 @@ 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)),
-            // 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_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,
+            )),
            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,
@@ -481,30 +480,6 @@ 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(|| {
@@ -40,13 +40,12 @@ 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_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,
+    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;
@@ -160,58 +159,6 @@ 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) => {
@@ -504,7 +451,12 @@ fn init_file_logging_internal(
        _ => Rotation::Daily,
    };

-    let (file_appender, roll_signal) = open_rolling_appender(config, log_directory, log_filename, rotation, match_mode)?;
+    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)?;
    validate_stdout_sink(&file_appender)?;

    let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
@@ -545,7 +497,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, roll_signal);
+    let cleanup_handle = spawn_cleanup_task(config, log_directory, log_filename, keep_files);

    info!(
        event = EVENT_LOCAL_LOGGING_STATE,
@@ -654,10 +606,6 @@ 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
@@ -675,9 +623,6 @@ const ROLL_TRIGGERED_CLEANUP_SPACING: Duration = Duration::from_secs(1);
 /// * `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.
@@ -686,7 +631,6 @@ 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`
@@ -701,7 +645,9 @@ 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 = resolve_cleanup_single_file_limit(config);
+    let max_single_file_size = config
+        .log_max_single_file_size_bytes
+        .unwrap_or(DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES);
    let compress = config.log_compress_old_files.unwrap_or(DEFAULT_OBS_LOG_COMPRESS_OLD_FILES);
    let gzip_level = config
        .log_gzip_compression_level
@@ -794,16 +740,9 @@ 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, 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;
-                }
-            }
+            // Wait for the next scheduled tick before dispatching another pass.
+            // The blocking filesystem work runs on a dedicated blocking thread.
+            interval.tick().await;
            let cleaner_clone = cleaner.clone();
            let result = tokio::task::spawn_blocking(move || cleaner_clone.cleanup()).await;

@@ -1314,153 +1253,6 @@ 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 {
@@ -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, open_rolling_appender, spawn_cleanup_task};
+use crate::telemetry::local::{build_json_log_layer, spawn_cleanup_task};
 use crate::telemetry::recorder::{Recorder, install_process_global_recorder};
 use crate::telemetry::resource::build_resource;
-use crate::telemetry::rolling::Rotation;
+use crate::telemetry::rolling::{RollingAppender, 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;
+use rustfs_config::observability::{DEFAULT_OBS_LOG_MATCH_MODE, DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES};
 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,12 +244,17 @@ pub(super) fn init_observability_http(
                "daily" => Rotation::Daily,
                _ => Rotation::Daily,
            };
-            let (file_appender, roll_signal) = open_rolling_appender(config, log_directory, log_filename, rotation, match_mode)?;
+            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)?;
            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, roll_signal);
+            let cleanup_handle = spawn_cleanup_task(config, log_directory, log_filename, keep_files);
            Ok((file_layer, guard, cleanup_handle, rotation_str))
        })();

@@ -29,7 +29,6 @@ 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};
@@ -83,9 +82,6 @@ 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 {
@@ -133,7 +129,6 @@ 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.
@@ -141,12 +136,6 @@ 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)
    }
@@ -300,9 +289,6 @@ 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) => {
@@ -466,80 +452,6 @@ 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();
@@ -425,10 +425,7 @@ enum FolderScanSource {
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum FolderScanMode {
    Namespace,
-    ErasureCandidate {
-        reuse_namespace: bool,
-        parent_has_metadata: bool,
-    },
+    ErasureCandidate { reuse_namespace: bool },
 }

 #[derive(Debug, PartialEq, Eq)]
@@ -2068,9 +2065,7 @@ 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()
                        {
@@ -2084,23 +2079,10 @@ impl FolderScanner {
                            return Ok(FolderScanOutcome::Namespace);
                        }
                        candidate_unclassified = false;
-                        // 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?;
+                        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?;
                    }
                    let Some(entry) = deferred_metadata.take() else { break };
                    (entry, true)
@@ -2570,6 +2552,20 @@ 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
@@ -2658,7 +2654,6 @@ 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
@@ -2726,10 +2721,7 @@ impl FolderScanner {
                                }
                                // Sampling may skip namespace work, never the
                                // classification that protects its parent object.
-                                child_mode = FolderScanMode::ErasureCandidate {
-                                    reuse_namespace: true,
-                                    parent_has_metadata: found_object_metadata,
-                                };
+                                child_mode = FolderScanMode::ErasureCandidate { reuse_namespace: true };
                            } else {
                                folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                            }
@@ -2824,11 +2816,6 @@ 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;
@@ -183,117 +183,6 @@ 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() {
@@ -315,10 +204,7 @@ async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
                ctx,
                folder,
                &mut DataUsageEntry::default(),
-                FolderScanMode::ErasureCandidate {
-                    reuse_namespace: false,
-                    parent_has_metadata: false,
-                },
+                FolderScanMode::ErasureCandidate { reuse_namespace: false },
                false,
            )
            .await
@@ -385,10 +271,7 @@ async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
            ctx.clone(),
            folder,
            &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate {
-                reuse_namespace: false,
-                parent_has_metadata: false,
-            },
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
            false,
        )
        .await;
@@ -428,10 +311,7 @@ async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
            ctx.clone(),
            folder,
            &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate {
-                reuse_namespace: false,
-                parent_has_metadata: false,
-            },
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
            false,
        )
        .await;
@@ -502,16 +382,7 @@ 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,
-                parent_has_metadata: false,
-            },
-            false,
-        )
+        .scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false }, false)
        .await
        .expect("classify candidate");
    assert_eq!(outcome, FolderScanOutcome::ErasureData);
@@ -723,10 +594,7 @@ async fn erasure_candidate_part_symlink_is_not_data_evidence() {
            CancellationToken::new(),
            folder,
            &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate {
-                reuse_namespace: false,
-                parent_has_metadata: false,
-            },
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
            false,
        )
        .await
@@ -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 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. |
+| `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). |
 | `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. |

@@ -10,7 +10,7 @@ RustFS builds the final product version before publication. A candidate is ident
 | `release-complete.yml` | Successful candidate run, accepted Linux manifest SHA-256, backlog acceptance report | Desktop archives, DEB/RPM, OCI images, Helm chart, full certification and sealed manifest | None |
 | `release-publish.yml` | Successful completion run and approved final manifest SHA-256 | Annotated tag, GitHub Release, R2 files, registry images, stable pointers and Helm index | Explicit publication |

-The builder workflows (`build.yml`, `package.yml`, `docker.yml`, `helm-package.yml`) are reusable stages. Only `release-publish.yml` has release write credentials. Development builds retain their artifact names and download channel. Daily development builds now select Linux only; macOS/Windows development artifacts are refreshed by an explicit manual build, and both remain mandatory for a completed release candidate. Tag pushes no longer start a second compilation.
+The builder workflows (`build.yml`, `package.yml`, `docker.yml`, `helm-package.yml`) are reusable stages. Only `release-publish.yml` has release write credentials. Development builds retain their artifact names and download channel. The daily schedule selects Linux only; the Sunday 16:00 UTC schedule selects macOS and Windows. Manual dispatch can still select any platform list, and both desktop targets remain mandatory for a completed release candidate. Tag pushes no longer start a second compilation.

 ## Freeze and build

@@ -87,7 +87,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 |---|---|---|---|---|
 | `ci.yml` (nightly) | full matrix, including the schedule/dispatch-only rio-v2 jobs `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` | strict aggregate; the full E2E lane runs on dispatch, merge groups, and main pushes | yes | dispatch `ci.yml` |
 | `cache-warm.yml` (every six hours, plus main pushes that change `Cargo.lock`, `rust-toolchain.toml` or the setup action) | `warm-ci-dev`, `warm-ci-feat-rio`, `warm-ci-feat-proto`, `warm-ci-uring` | no verdict; sole writer of the `rust-cache` keys that `ci.yml` restores | yes | dispatch `cache-warm.yml` |
-| `build.yml` (nightly) | `build-rustfs` over the six-target platform matrix in `prepare-platform-matrix` (four Linux, macOS aarch64, Windows x86_64) | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
+| `build.yml` (nightly Linux, weekly desktop) | `build-rustfs` over the `prepare-platform-matrix` subset for the firing cron: four Linux targets at 01:13 UTC, and macOS aarch64 plus Windows x86_64 at Sunday 16:00 UTC | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
 | `e2e-replication-nightly.yml` (nightly) | `repl-nightly`, `cluster-nightly`, `protocols-nightly` | three independent gates; JUnit, membership listing, server logs | yes | `python3 scripts/e2e_binary.py build --bins`, then `python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--features e2e-test-hooks` on both steps with `--profile e2e-nightly`; `--features ftps,webdav,sftp` on both steps with `-j 1 --profile e2e-protocols` |
 | `e2e-distributed.yml` (nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, run `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks`, then `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test` |
 | `e2e-s3tests.yml` (weekly) | `s3tests` (single and distributed, four shards each), `upstream-head-canary` | compatibility gate; report, JUnit, node IDs, server logs | yes | `scripts/s3-tests/run.sh` against an existing single or distributed target |
@@ -122,7 +122,7 @@ The versioned plan is `.config/release-validation-plan.json` (`schema_version=1`

 | Trigger | Workflow / job | Result |
 |---|---|---|
-| Nightly schedule, dispatch | `build.yml` `build-rustfs` (development build over the full platform matrix) | build artifacts and the uncertified dev download channel; no release publication |
+| Nightly Linux schedule, weekly desktop schedule, or dispatch | `build.yml` `build-rustfs` (development build of that schedule's platform subset, or the dispatch platform list) | build artifacts and the uncertified dev download channel; no release publication |
 | Valid release or preview tag | `build.yml` `build-rustfs`, `release-certification`, `create-release`, `upload-release-assets`, `publish-release`, `promote-release-artifacts` | staged binaries, draft release, checksummed assets, then certified publish and promotion of the same bytes |
 | Successful non-preview release-tag build (`workflow_run`) | `docker.yml` `release-certification`, `build-docker`, `scan-docker-image` | multi-architecture images staged, verified against the certification and promoted, plus vulnerability report |
 | Successful release-tag build (`workflow_run`) | `package.yml` `release-certification`, `package`, `promote-packages` | DEB/RPM packages with wrapped-binary verification, staged then promoted with checksums |
@@ -343,11 +343,69 @@ class PromotionTests(unittest.TestCase):


 class WorkflowContractTests(unittest.TestCase):
-    def test_nightly_avoids_desktop_and_manual_builds_keep_platform_selection(self):
+    def test_schedule_cron_selects_linux_or_desktop_without_dropping_manual_platforms(self):
        value = workflow("build.yml")
-        selection = value["jobs"]["prepare-platform-matrix"]["steps"][-1]
-        self.assertEqual(selection["env"]["RAW_PLATFORMS"], "${{ inputs.platforms || 'linux' }}")
+        self.assertEqual([item["cron"] for item in value["on"]["schedule"]], ["13 1 * * *", "0 16 * * 0"])
+        self.assertNotIn("push", value["on"])
+        self.assertNotIn("pull_request", value["on"])
        self.assertEqual(value["on"]["workflow_dispatch"]["inputs"]["platforms"]["default"], "all")
+        selection = value["jobs"]["prepare-platform-matrix"]["steps"][-1]
+        env = selection["env"]
+        self.assertEqual(env["RAW_PLATFORMS"], "${{ inputs.platforms || 'linux' }}")
+        self.assertEqual(env["HAS_PLATFORM_INPUT"], "${{ inputs.platforms && 'true' || 'false' }}")
+        self.assertEqual(env["EVENT_NAME"], "${{ github.event_name }}")
+        self.assertEqual(env["EVENT_SCHEDULE"], "${{ github.event.schedule }}")
+        linux = [
+            ("linux-x86_64-musl", "sm-standard-4"),
+            ("linux-aarch64-musl", "ubicloud-standard-16-arm"),
+            ("linux-x86_64-gnu", "sm-standard-4"),
+            ("linux-aarch64-gnu", "ubicloud-standard-16-arm"),
+        ]
+        desktop = [("macos-aarch64", "macos-latest-xlarge"), ("windows-x86_64", "windows-latest")]
+
+        def select(event, raw, schedule="", has_input="false"):
+            with tempfile.TemporaryDirectory() as temp:
+                output = Path(temp) / "github-output"
+                environment = {**os.environ, "EVENT_NAME": event, "EVENT_SCHEDULE": schedule,
+                               "RAW_PLATFORMS": raw, "HAS_PLATFORM_INPUT": has_input, "GITHUB_OUTPUT": str(output)}
+                result = subprocess.run(["bash", "-c", selection["run"]], env=environment, capture_output=True, text=True)
+                parsed = {}
+                if output.exists():
+                    for line in output.read_text().splitlines():
+                        key, item = line.split("=", 1)
+                        parsed[key] = item
+                return result, parsed
+
+        def rows(parsed):
+            return [(item["target_id"], item["os"]) for item in json.loads(parsed["matrix"])["include"]]
+
+        daily, daily_out = select("schedule", "all", "13 1 * * *")
+        self.assertEqual(daily.returncode, 0, daily.stderr)
+        self.assertEqual(rows(daily_out), linux)
+        self.assertEqual(daily_out["selected"], ",".join(item[0] for item in linux))
+        weekly, weekly_out = select("schedule", "linux", "0 16 * * 0")
+        self.assertEqual(weekly.returncode, 0, weekly.stderr)
+        self.assertEqual(rows(weekly_out), desktop)
+        self.assertEqual(weekly_out["selected"], ",".join(item[0] for item in desktop))
+        for schedule in ("", "13 1 * * 1", "0 16 * * 7", "13 1 extra stuff"):
+            rejected, _ = select("schedule", "all", schedule)
+            self.assertNotEqual(rejected.returncode, 0, schedule)
+            self.assertIn("Unrecognized schedule cron", rejected.stderr)
+        caller, caller_out = select("schedule", "linux", "0 16 * * 0", has_input="true")
+        self.assertEqual(caller.returncode, 0, caller.stderr)
+        self.assertEqual(rows(caller_out), linux)
+        dispatched, dispatched_out = select("workflow_dispatch", "all", has_input="true")
+        self.assertEqual(dispatched.returncode, 0, dispatched.stderr)
+        self.assertEqual(dispatched_out["selected"], "all")
+        self.assertEqual(rows(dispatched_out), linux + desktop)
+        custom, custom_out = select("workflow_dispatch", "linux-x86_64-musl,macos-aarch64", has_input="true")
+        self.assertEqual(custom.returncode, 0, custom.stderr)
+        self.assertEqual(custom_out["selected"], "linux-x86_64-musl,macos-aarch64")
+        self.assertEqual(rows(custom_out), [("linux-x86_64-musl", "sm-standard-4"), ("macos-aarch64", "macos-latest-xlarge")])
+        for event, raw, expected in (("workflow_call", "linux", linux), ("workflow_call", "desktop", desktop)):
+            called, called_out = select(event, raw, has_input="true")
+            self.assertEqual(called.returncode, 0, called.stderr)
+            self.assertEqual(rows(called_out), expected)

    def test_main_push_cannot_cancel_the_scheduled_oidc_gate(self):
        value = workflow("oidc-keycloak.yml")