[BUG][wal3] Fix a double-load that leads to inconsistent views of the world. (#5781)

## Description of changes

This fixes a bug with the following behavior:
- Writer A creates empty log L_1
- Writer A calls copy(L_1) -> L_2
- Writer A reads the empty manifest in copy
- Concurrenty, writer B appends to L_1
- Writer A re-reads the log and uses B's current offsets to populate its
  next-to-fill-in slots.

The result is that when used in forking, the compaction cursor is now
behind the next-to-fill-in-slot.

Tests from Claude.

## Test plan

CI + cargo test -p wal3 locally

## Migration plan

N/A

## Observability plan

N/A

## Documentation Changes

N/A

Author: rescrv

rust/wal3/src/copy.rs

@@ -15,11 +15,18 @@ pub async fn copy(
     offset: LogPosition,
     target: String,
 ) -> Result<(), Error> {
-    let fragments = match reader.scan(offset, Limits::UNLIMITED).await {
-        Ok(fragments) => fragments,
-        Err(Error::UninitializedLog) => vec![],
-        Err(err) => return Err(err),
-    };
+    let reference = reader
+        .manifest()
+        .await?
+        .unwrap_or(Manifest::new_empty("zero-copy task"));
+    let mut short_read = false;
+    let fragments = reader
+        .scan_with_cache(&reference, offset, Limits::UNLIMITED, &mut short_read)
+        .await?;
+    if short_read {
+        tracing::error!("short_read in unlimited copy");
+        return Err(Error::Internal);
+    }
     if !fragments.is_empty() {
         let mut futures = vec![];
         for fragment in fragments.into_iter() {
@@ -63,10 +70,6 @@ pub async fn copy(
         };
         Manifest::initialize_from_manifest(options, storage, &target, manifest).await?;
     } else {
-        let reference = reader
-            .manifest()
-            .await?
-            .unwrap_or(Manifest::new_empty("zero-copy task"));
         let setsum = Setsum::default();
         let collected = Setsum::default();
         let acc_bytes = 0;

rust/wal3/src/reader.rs

@@ -132,13 +132,20 @@ impl LogReader {
             return Err(Error::UninitializedLog);
         };
         let mut short_read = false;
-        self.scan_with_cache(manifest, from, limits, &mut short_read)
+        self.scan_with_cache(&manifest, from, limits, &mut short_read)
             .await
     }
 
-    async fn scan_with_cache(
+    /// Scan up to:
+    /// 1. Up to, but not including, the offset of the log position.  This makes it a half-open
+    ///    interval.
+    /// 2. Up to, and including, the number of files to return.
+    ///
+    /// This differs from scan in that it takes a loaded manifest.
+    /// This differs from scan_from_manifest because it will load snapshots.
+    pub async fn scan_with_cache(
         &self,
-        manifest: Manifest,
+        manifest: &Manifest,
         from: LogPosition,
         limits: Limits,
         short_read: &mut bool,
@@ -344,7 +351,7 @@ impl LogReader {
         let from = manifest.oldest_timestamp();
         let mut short_read = false;
         let fragments = self
-            .scan_with_cache(manifest, from, limits, &mut short_read)
+            .scan_with_cache(&manifest, from, limits, &mut short_read)
             .await
             .map_err(|x| vec![x])?;
         let futures = fragments