From ffccd8a4ed854b26080c66f06eef77d27974a553 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 1 Nov 2026 13:43:19 -0800 Subject: [PATCH] volume(rust): refuse a batch delete while a replica is offline Mirrors the Go volume server: batch_delete checks that the master lists every replica of a replicated volume before deleting, and answers 503 with the replica count error otherwise. Co-Authored-By: Claude Opus 5.5 --- seaweed-volume/src/server/grpc_server.rs | 112 +++++++++++++++++++++++ seaweed-volume/src/server/handlers.rs | 38 ++++++++ 2 files changed, 150 insertions(+) diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index b76b6ed7a3..42569223f5 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1093,6 +1093,8 @@ impl VolumeServer for VolumeGrpcService { self.state.check_maintenance()?; let req = request.into_inner(); let mut results = Vec::new(); + let mut replica_errs: std::collections::HashMap> = + std::collections::HashMap::new(); let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -1114,6 +1116,23 @@ impl VolumeServer for VolumeGrpcService { } }; + let vid = file_id.volume_id.0; + if !replica_errs.contains_key(&vid) { + let checked = + crate::server::handlers::check_delete_replicas(&self.state, vid).await; + replica_errs.insert(vid, checked); + } + if let Some(Err(e)) = replica_errs.get(&vid) { + results.push(volume_server_pb::DeleteResult { + file_id: fid_str.clone(), + status: 503, + error: e.clone(), + size: 0, + version: 0, + }); + continue; + } + let mut n = Needle { id: file_id.key, cookie: file_id.cookie, @@ -12965,6 +12984,99 @@ mod tests { ); } + /// Go's TestBatchDelete_WaitsForEveryReplica: a replica the master no + /// longer lists would miss the delete, so the delete waits for it. + #[tokio::test] + async fn test_batch_delete_waits_for_every_replica() { + use axum::routing::get; + + let (service, _tmp) = make_local_service_with_volume("", None); + { + let mut store = service.state.store.write().unwrap(); + store + .add_volume( + VolumeId(2), + DiskType::HardDrive, + &VolumeSpec { + replica_placement: Some( + crate::storage::super_block::ReplicaPlacement::from_string("001") + .unwrap(), + ), + ..Default::default() + }, + ) + .unwrap(); + let mut n = Needle { + id: NeedleId(1), + cookie: Cookie(0x3344), + data: b"hello".to_vec(), + data_size: 5, + ..Needle::default() + }; + store + .write_volume_needle(VolumeId(2), &mut n, false) + .unwrap(); + } + + let urls = Arc::new(std::sync::Mutex::new(vec!["127.0.0.1:1"])); + let app = axum::Router::new().route( + "/dir/lookup", + get({ + let urls = urls.clone(); + move || { + let locations: Vec<_> = urls + .lock() + .unwrap() + .iter() + .map(|url| serde_json::json!({ "url": url })) + .collect(); + async move { + axum::Json(serde_json::json!({ + "volumeOrFileId": "2", + "locations": locations + })) + } + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + *service.state.current_master_url.write().await = format!("127.0.0.1:{port}"); + + let fid = needle::FileId::new(VolumeId(2), NeedleId(1), Cookie(0x3344)).to_string(); + let delete = || async { + let resp = service + .batch_delete(Request::new(volume_server_pb::BatchDeleteRequest { + file_ids: vec![fid.clone()], + skip_cookie_check: true, + })) + .await + .unwrap() + .into_inner(); + assert_eq!(resp.results.len(), 1, "{:?}", resp.results); + (resp.results[0].status, resp.results[0].error.clone()) + }; + + let (status, error) = delete().await; + assert_eq!(status, 503, "{error}"); + assert!(error.contains("replication copy count"), "{error}"); + { + let store = service.state.store.read().unwrap(); + let (_, v) = store.find_volume(VolumeId(2)).unwrap(); + let mut n = Needle { + id: NeedleId(1), + ..Needle::default() + }; + v.read_needle(&mut n) + .expect("needle deleted while a replica was offline"); + } + + urls.lock().unwrap().push("127.0.0.2:8080"); + let (status, error) = delete().await; + assert_eq!(status, 202, "{error}"); + } + const GO_EC_NEEDLE_NOT_FOUND: &str = "locate in local ec volume: FindNeedleFromEcx: needle not found"; diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index d2d02d35b0..a1ceefd012 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -695,6 +695,44 @@ async fn batch_delete_file_ids( Ok(()) } +/// Mirrors Go's checkDeleteReplicas: refuses a delete the master cannot route +/// to every replica, as the HTTP delete path does, so an offline replica does +/// not keep the needle once it returns. +pub(crate) async fn check_delete_replicas( + state: &VolumeServerState, + vid: u32, +) -> Result<(), String> { + let copy_count = { + let store = state.store.read().unwrap(); + match store.find_volume(VolumeId(vid)) { + Some((_, v)) => v.super_block().replica_placement.get_copy_count(), + None => return Ok(()), + } + }; + if copy_count <= 1 { + return Ok(()); + } + let locations = lookup_volume( + &state.http_client, + &state.outgoing_http_scheme, + &state.current_master().await, + vid, + ) + .await + .map_err(|e| format!("replicating lookup failed for {}: {}", vid, e))?; + let deletable = locations + .iter() + .filter(|loc| !loc.read_only || loc.read_only_can_delete) + .count(); + if deletable < copy_count as usize { + return Err(format!( + "replication locations [{}] is less than volume {} replication copy count [{}]", + deletable, vid, copy_count + )); + } + Ok(()) +} + /// Helper to synchronously replicate a request to peer volume servers. async fn do_replicated_request( state: &VolumeServerState,