[DiskCache] No-growth reopen is no-op (#10049)

* add `ScheduleReopen::Unchanged` to avoid retrying to reopen blockingly

* wrap `ScheduledReopen` in `Option`

lib/common/common/src/universal_io/simple_disk_cache/file/mod.rs

@@ -73,7 +73,7 @@ pub(crate) enum State<R: UniversalRead + 'static> {
         ///
         /// [`reopen`]: UniversalRead::reopen
         /// [`schedule_reopen`]: UniversalRead::schedule_reopen
-        scheduled_reopen: ScheduledReopen<R>,
+        scheduled_reopen: Option<ScheduledReopen<R>>,
     },
     /// Eager open-time prefill: an in-flight whole-object read scheduled at open;
     /// init waits on it and writes the whole mirror. For `Populate::Blocking` /
@@ -95,10 +95,9 @@ pub(crate) enum State<R: UniversalRead + 'static> {
 /// the staged targets, hence the `target_len` on every variant.
 #[derive(Debug)]
 pub(crate) enum ScheduledReopen<R: UniversalRead + 'static> {
-    /// Nothing staged — the default at every [`State::Ready`] construction
-    /// site, and what a no-growth schedule leaves behind. `reopen` then
-    /// schedules and waits inline.
-    No,
+    /// Scheduling detected that the file size has not changed, so no need to
+    /// reopen.
+    Unchanged,
     /// Lazy populate (`No` / `Auto` / `Partial`): apply resizes the mirror to
     /// `target_len` and lets the new blocks fault in on demand.
     Resize { target_len: u64 },
@@ -117,7 +116,7 @@ impl<R: UniversalRead + 'static> ScheduledReopen<R> {
     /// staged.
     pub(super) fn target_len(&self) -> Option<u64> {
         match self {
-            ScheduledReopen::No => None,
+            ScheduledReopen::Unchanged => None,
             ScheduledReopen::Resize { target_len }
             | ScheduledReopen::Tail {
                 target_len,
@@ -134,7 +133,7 @@ impl<R: UniversalRead + 'static> State<R> {
         State::Ready {
             remote,
             local,
-            scheduled_reopen: ScheduledReopen::No,
+            scheduled_reopen: None,
         }
     }
 

lib/common/common/src/universal_io/simple_disk_cache/file/reopen.rs

@@ -45,16 +45,20 @@ where
             return Ok(false);
         };
 
-        match std::mem::replace(scheduled_reopen, ScheduledReopen::No) {
-            // Nothing staged
-            ScheduledReopen::No => Ok(false),
+        let Some(scheduled_reopen) = scheduled_reopen.take() else {
+            // There isn't anything scheduled.
+            return Ok(false);
+        };
+
+        // Handle the scheduled reopen.
+        match scheduled_reopen {
+            // It was staged without changes.
+            ScheduledReopen::Unchanged => {}
             ScheduledReopen::Resize { target_len } => {
                 // reopen remote, so we can read up to the new length.
                 remote.reopen()?;
 
                 local.resize(&self.local_path, target_len)?;
-
-                Ok(true)
             }
             ScheduledReopen::Tail {
                 mut pipeline,
@@ -78,10 +82,10 @@ where
 
                 // replace remote with the one from the owned pipeline
                 *remote = pipeline.into_inner();
-
-                Ok(true)
             }
         }
+
+        Ok(true)
     }
 
     pub(super) fn schedule_reopen_impl<F: FnOnce(&Path) -> Option<FileInfo>>(
@@ -142,34 +146,44 @@ where
             )));
         }
 
-        // Check if length has grown
-        if scheduled_reopen.target_len() == Some(remote_len) || remote_len == local_len {
+        // Check if staged length has grown
+        if scheduled_reopen
+            .as_ref()
+            .is_some_and(|r| r.target_len() == Some(remote_len))
+        {
             return Ok(());
         }
 
-        let new_scheduled_reopen = match self.open_options.populate {
-            Populate::Blocking | Populate::PreferBackground => {
-                // Schedule the read of the new tail blocks.
-                let (blocks_range, byte_range) =
-                    block_aligned_fetch(local_len..remote_len, remote_len)
-                        .expect("the byte range is non-empty");
+        let new_scheduled_reopen = if remote_len == local_len {
+            ScheduledReopen::Unchanged
+        } else {
+            match self.open_options.populate {
+                Populate::Blocking | Populate::PreferBackground => {
+                    // Schedule the read of the new tail blocks.
+                    let (blocks_range, byte_range) =
+                        block_aligned_fetch(local_len..remote_len, remote_len)
+                            .expect("the byte range is non-empty");
 
-                // Fresh handle: the staged fetch must not share a mapping with
-                // the held remote, which later reopens would remap.
-                let new_remote = self.open_remote()?;
-                let mut pipeline = OwnedPipeline::new(new_remote)?;
-                // FIXME: check can_schedule in a loop?
-                pipeline.schedule::<Sequential>(blocks_range, byte_range, REMOTE_READ_ALIGNMENT)?;
+                    // Fresh remote handle
+                    let new_remote = self.open_remote()?;
+                    let mut pipeline = OwnedPipeline::new(new_remote)?;
+                    // FIXME: check can_schedule in a loop?
+                    pipeline.schedule::<Sequential>(
+                        blocks_range,
+                        byte_range,
+                        REMOTE_READ_ALIGNMENT,
+                    )?;
 
-                ScheduledReopen::Tail {
-                    pipeline,
-                    target_len: remote_len,
+                    ScheduledReopen::Tail {
+                        pipeline,
+                        target_len: remote_len,
+                    }
                 }
+                // No prefetch for lazy population
+                Populate::Auto | Populate::No | Populate::Partial(_) => ScheduledReopen::Resize {
+                    target_len: remote_len,
+                },
             }
-            // No prefetch for lazy population
-            Populate::Auto | Populate::No | Populate::Partial(_) => ScheduledReopen::Resize {
-                target_len: remote_len,
-            },
         };
 
         // Re-borrow the state: `open_remote` above needs `&self`.
@@ -181,7 +195,7 @@ where
         else {
             unreachable!("state was Ready above");
         };
-        *scheduled_reopen = new_scheduled_reopen;
+        *scheduled_reopen = Some(new_scheduled_reopen);
 
         Ok(())
     }