rust volume: keep a copied replica on deleted-count drift, as Go does (#12251)

* rust volume: keep a copied replica on deleted-count drift, as Go does

check_copy_counts rejected a VolumeCopy whenever the target's deleted count
differed from the source's. On a redb needle map that count is an estimate:
metrics_from_idx rebuilds it at load through a bloom filter (0.1% false
positives per entry, each counted as an overwrite), and the source holds the
estimate from its own last load plus exact runtime updates. A source written
since its load and a fresh load of the same files disagree by a few, so good
copies were refused with "target deleted count [N+1] is not same as origin
deleted count [N]". An index of unique keys 1..=128 already reports one
deletion.

Go fixed the same rejection in #11701 (issue #11678) by warning instead of
failing. Do the same for the deleted count. The file count stays strict: it
is exact on both sides (every non-tombstone .idx row is one file), and the
.dat/.idx size checks still gate integrity.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Et288CnR1mxDyQNdYNMBNq

* test(volume): expect 409 when a read-only upload is refused

x-seaweed-write-outcome: refused-before-append. The HTTP integration
test still expected 500, so Rust unit tests failed on Linux and Windows.
Deletes stay 500.

* rust volume: keep the strict count check when only deletes drift

Deletes append tombstone rows without moving the file count, so a copy whose source only deleted mid-flight still has an exactly comparable file count. Gate the strict check on file-count stability alone; the deleted-count drift is already a warning inside check_copy_counts.

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: user.email <chrislusf@users.noreply.github.com>

seaweed-volume/src/server/grpc_server.rs

@@ -6523,6 +6523,7 @@ impl VolumeCopyJob {
             if validate_counts
                 && let Some((_, v)) = store.find_volume(vid)
                 && let Err(e) = check_copy_counts(
+                    vid,
                     source_status_after,
                     v.file_count() as u64,
                     v.deleted_count() as u64,
@@ -6999,7 +7000,18 @@ where
 }
 
 /// Compare a copied replica's record counts with the source's.
+///
+/// The file count is exact on both sides (every non-tombstone .idx row is one
+/// file, at load and at runtime), so a mismatch rejects the copy. The deleted
+/// count is not: a redb needle map rebuilds it at load from the .idx through
+/// a bloom filter (`metrics_from_idx`, 0.1% false positives per entry, each
+/// one counted as an overwrite), and the source's figure is that estimate from
+/// its own last load plus exact runtime updates. A source written since its
+/// load and the target's fresh load of the same files therefore disagree by a
+/// few for identical data. Go warns instead of rejecting on this drift since
+/// #11701 (issue #11678); the .dat/.idx size checks already gate integrity.
 fn check_copy_counts(
+    vid: VolumeId,
     origin: &volume_server_pb::VolumeStatusResponse,
     target_file_count: u64,
     target_deleted_count: u64,
@@ -7011,25 +7023,26 @@ fn check_copy_counts(
         ));
     }
     if origin.file_deleted_count != target_deleted_count {
-        return Err(format!(
-            "target deleted count [{}] is not same as origin deleted count [{}]",
-            target_deleted_count, origin.file_deleted_count
-        ));
+        tracing::warn!(
+            "copied volume {} deleted count [{}] differs from origin deleted count [{}] (counter drift, copy kept)",
+            vid,
+            target_deleted_count,
+            origin.file_deleted_count
+        );
     }
     Ok(())
 }
 
-/// A writable source may take writes or deletes while its files are copied;
-/// then the before/after counts do not describe one snapshot and a strict
-/// count check would report a false mismatch.
+/// A writable source may take writes while its files are copied; then the
+/// before/after file counts do not describe one snapshot and a strict count
+/// check would report a false mismatch. Deletes do not move the file count —
+/// they append tombstone rows the deleted counter records — so drift there
+/// leaves the strict check exact, and check_copy_counts warns on it instead.
 fn copy_counts_stable(
     before: Option<&volume_server_pb::VolumeStatusResponse>,
     after: &volume_server_pb::VolumeStatusResponse,
 ) -> bool {
-    before.is_some_and(|before| {
-        before.file_count == after.file_count
-            && before.file_deleted_count == after.file_deleted_count
-    })
+    before.is_some_and(|before| before.file_count == after.file_count)
 }
 
 /// Verify that a copied file has the expected size.
@@ -9934,6 +9947,75 @@ mod tests {
         );
     }
 
+    // A deleted-count difference alone is counter drift, not a bad copy: the
+    // count is a bloom-filter estimate on redb needle maps, so a source written
+    // since its last load and a fresh load of the same files can disagree.
+    // The copy must be kept and mounted, as Go does since #11701.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn test_volume_copy_keeps_replica_on_deleted_count_drift() {
+        let (source_service, _source_tmp, _dat_bytes) = make_local_service_with_large_volume();
+        {
+            let store = source_service.state.store.read().unwrap();
+            let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
+            vol.add_deletion_count_for_test(1);
+        }
+        let (port, _shutdown) = serve_source(source_service).await;
+
+        let (dest_service, _dest_tmp) = make_local_service_with_volume("", None);
+        {
+            let mut store = dest_service.state.store.write().unwrap();
+            store
+                .delete_volume(VolumeId(1), false, false, false)
+                .unwrap();
+            for loc in &store.locations {
+                loc.check_disk_space();
+            }
+        }
+
+        run_volume_copy(&dest_service, port)
+            .await
+            .expect("a deleted-count difference alone must not fail the copy");
+
+        let store = dest_service.state.store.read().unwrap();
+        let (_, vol) = store
+            .find_volume(VolumeId(1))
+            .expect("the copied replica must stay mounted");
+        assert_eq!(
+            vol.file_count(),
+            2,
+            "the replica must hold the source's needles"
+        );
+        assert_eq!(
+            vol.deleted_count(),
+            0,
+            "the target's own count is kept; the drift is only logged"
+        );
+    }
+
+    // Writes move the file count; deletes move only the deleted count. The
+    // strict check still applies when only the latter drifts: the copied
+    // file count remains exact, and check_copy_counts warns on the drift.
+    #[test]
+    fn test_copy_counts_stable_gates_on_file_count() {
+        let before = volume_server_pb::VolumeStatusResponse {
+            file_count: 10,
+            file_deleted_count: 3,
+            ..Default::default()
+        };
+        let mut after = before.clone();
+        assert!(copy_counts_stable(Some(&before), &after));
+
+        after.file_deleted_count = 5;
+        assert!(
+            copy_counts_stable(Some(&before), &after),
+            "deleted-count drift alone must not disable the strict check"
+        );
+
+        after.file_count = 9;
+        assert!(!copy_counts_stable(Some(&before), &after));
+        assert!(!copy_counts_stable(None, &after));
+    }
+
     // A copy whose record counts disagree with a stable source must be
     // rejected, left unmounted, and cleaned off disk, even though the .dat and
     // .idx sizes match.

seaweed-volume/src/storage/needle_map.rs

@@ -1486,6 +1486,16 @@ impl NeedleMap {
         metric.file_count.fetch_add(delta, Ordering::Relaxed);
     }
 
+    #[cfg(test)]
+    pub(crate) fn add_deletion_count_for_test(&self, delta: i64) {
+        let metric = match self {
+            NeedleMap::InMemory(nm) => &nm.metric,
+            NeedleMap::Redb(nm) => &nm.metric,
+            NeedleMap::SortedFile(_) => panic!("sorted-file needle maps are read-only"),
+        };
+        metric.deletion_count.fetch_add(delta, Ordering::Relaxed);
+    }
+
     /// Largest (offset + actual size) seen during the load walk; 0 if the
     /// map is empty. Used at volume load to detect .idx entries that
     /// reference past the end of .dat (issue #8928) without a second scan.

seaweed-volume/src/storage/needle_map/idx_metric.rs

@@ -156,4 +156,35 @@ mod tests {
             .count();
         assert!(false_positives < 500, "false positives: {false_positives}");
     }
+
+    // The deleted count rebuilt from an .idx is an estimate: an index of
+    // unique keys, with no overwrite and no tombstone, still reports a
+    // deletion once the bloom filter yields a false positive. VolumeCopy
+    // therefore cannot demand that a fresh load's deleted count equal the
+    // source's (check_copy_counts in server/grpc_server.rs).
+    #[test]
+    fn test_unique_keys_can_still_report_a_deletion() {
+        let mut idx = Vec::new();
+        let mut first_drift = None;
+        for n in 1u64..=2000 {
+            let mut entry = [0u8; NEEDLE_MAP_ENTRY_SIZE];
+            idx_entry_to_bytes(
+                &mut entry,
+                NeedleId(n),
+                Offset::from_actual_offset(n as i64 * 128),
+                Size(100),
+            );
+            idx.extend_from_slice(&entry);
+            let metric = metrics_from_idx(&mut std::io::Cursor::new(&idx), Version(3)).unwrap();
+            assert_eq!(metric.file_count.load(Ordering::Relaxed), n as i64);
+            if metric.deletion_count.load(Ordering::Relaxed) > 0 {
+                first_drift = Some(n);
+                break;
+            }
+        }
+        assert!(
+            first_drift.is_some(),
+            "expected a bloom false positive to count a deletion among 2000 unique keys"
+        );
+    }
 }

seaweed-volume/src/storage/volume.rs

@@ -6068,6 +6068,13 @@ impl Volume {
             nm.add_file_count_for_test(delta);
         }
     }
+
+    #[cfg(test)]
+    pub(crate) fn add_deletion_count_for_test(&self, delta: i64) {
+        if let Some(nm) = self.io.read().unwrap().nm.as_ref() {
+            nm.add_deletion_count_for_test(delta);
+        }
+    }
 }
 
 // ============================================================================

seaweed-volume/tests/http_integration.rs

@@ -521,6 +521,13 @@ async fn write_and_delete_on_read_only_volume_say_is_read_only() {
         .await
         .unwrap();
     assert_eq!(response.status(), StatusCode::CONFLICT);
+    assert_eq!(
+        response
+            .headers()
+            .get("x-seaweed-write-outcome")
+            .and_then(|value| value.to_str().ok()),
+        Some("refused-before-append")
+    );
     let error = error_of(body_bytes(response).await);
     assert!(error.contains("volume 1 is read only"), "{error}");