diff --git a/seaweed-volume/src/remote_storage/mod.rs b/seaweed-volume/src/remote_storage/mod.rs index 58411bf83..b22393e50 100644 --- a/seaweed-volume/src/remote_storage/mod.rs +++ b/seaweed-volume/src/remote_storage/mod.rs @@ -7,9 +7,7 @@ pub mod endpoint_guard; pub mod s3; pub mod s3_tier; -pub use endpoint_guard::{ - guarded_tcp_connect, validate_remote_endpoint, validate_replica_target, -}; +pub use endpoint_guard::{guarded_tcp_connect, validate_remote_endpoint, validate_replica_target}; use crate::pb::remote_pb::{RemoteConf, RemoteStorageLocation}; diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 1b298e6fb..07f918336 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -906,8 +906,11 @@ impl VolumeServer for VolumeGrpcService { store.has_ec_volume(file_id.volume_id) }; - // Cookie validation (unless skip_cookie_check) - if !req.skip_cookie_check { + // Cookie validation (unless skip_cookie_check). EC volumes always + // take this branch: the distributed read is the only source of the + // on-disk cookie and size, and Go's DeleteEcShardNeedle compares + // the fid cookie against it even when the caller asked to skip. + if !req.skip_cookie_check || is_ec_volume { let original_cookie = n.cookie; if !is_ec_volume { let store = self.state.store.read().unwrap(); @@ -925,48 +928,43 @@ impl VolumeServer for VolumeGrpcService { } } } else { - // For EC volumes, verify needle exists in ecx index - let store = self.state.store.read().unwrap(); - if let Some(ec_vol) = store.find_ec_volume(file_id.volume_id) { - match ec_vol.find_needle_from_ecx(n.id) { - Ok(Some((_, size))) if !size.is_deleted() => { - // Needle exists and is not deleted — cookie check not possible - // for EC volumes without distributed read, so we accept it - n.data_size = size.0 as u32; - } - Ok(_) => { - results.push(volume_server_pb::DeleteResult { - file_id: fid_str.clone(), - status: 404, - error: format!("ec needle {} not found", fid_str), - size: 0, - version: 0, - }); - continue; - } - Err(e) => { - results.push(volume_server_pb::DeleteResult { - file_id: fid_str.clone(), - status: 404, - error: e.to_string(), - size: 0, - version: 0, - }); - continue; - } + // Go's ReadEcShardNeedle fills the needle — the local .ecx + // alone can't supply the cookie or the manifest flag. + match crate::server::store_ec::read_ec_shard_needle_distributed( + &self.state, + file_id.volume_id, + n.id, + ) + .await + { + Ok(Some(ec_needle)) => n = ec_needle, + Ok(None) => { + results.push(volume_server_pb::DeleteResult { + file_id: fid_str.clone(), + status: 404, + error: format!("ec needle {} not found", fid_str), + size: 0, + version: 0, + }); + continue; + } + Err(e) => { + results.push(volume_server_pb::DeleteResult { + file_id: fid_str.clone(), + status: 404, + error: e.to_string(), + size: 0, + version: 0, + }); + continue; } - } else { - results.push(volume_server_pb::DeleteResult { - file_id: fid_str.clone(), - status: 404, - error: format!("ec volume {} not found", file_id.volume_id), - size: 0, - version: 0, - }); - continue; } } - if n.cookie != original_cookie { + // Go's inner check is `cookie != 0 && cookie != n.Cookie`: a + // zero fid cookie skips validation, which can only happen + // here when skip_cookie_check was already requested. + if (!req.skip_cookie_check || original_cookie.0 != 0) && n.cookie != original_cookie + { results.push(volume_server_pb::DeleteResult { file_id: fid_str.clone(), status: 400, @@ -1068,42 +1066,50 @@ impl VolumeServer for VolumeGrpcService { } } } else { - // EC volume deletion: journal the delete locally (with cookie validation, matching Go) - let mut store = self.state.store.write().unwrap(); - if let Some(ec_vol) = store.find_ec_volume_mut(file_id.volume_id) { - let cookie = if req.skip_cookie_check { - crate::storage::types::Cookie(0) - } else { - n.cookie - }; - match ec_vol.journal_delete_with_cookie(n.id, cookie) { - Ok(()) => { - results.push(volume_server_pb::DeleteResult { - file_id: fid_str.clone(), - status: 202, - error: String::new(), - size: n.data_size, - version: 0, - }); - } - Err(e) => { - results.push(volume_server_pb::DeleteResult { - file_id: fid_str.clone(), - status: 500, - error: e.to_string(), - size: 0, - version: 0, - }); - } + // EC volume deletion: forward the tombstone to a holder of the + // needle's primary shard (Go's DeleteEcShardNeedle → + // VolumeEcBlobDelete). The cookie was already validated + // against the distributed read above. + match crate::server::store_ec::delete_ec_shard_needle_distributed( + &self.state, + file_id.volume_id, + n.id, + ) + .await + { + Ok(()) => { + results.push(volume_server_pb::DeleteResult { + file_id: fid_str.clone(), + status: 202, + error: String::new(), + size: n.data_size, + version: 0, + }); + } + Err(e) => { + // Needle vanished while the volume stays mounted: + // Go's ErrorDeleted → NotModified. The volume itself + // unmounting means no journal anywhere, so 500 (Go's + // generic error path) lets the caller retry. + let already_gone = e.kind() == std::io::ErrorKind::NotFound + && self + .state + .store + .read() + .unwrap() + .has_ec_volume(file_id.volume_id); + results.push(volume_server_pb::DeleteResult { + file_id: fid_str.clone(), + status: if already_gone { 304 } else { 500 }, + error: if already_gone { + String::new() + } else { + e.to_string() + }, + size: 0, + version: 0, + }); } - } else { - results.push(volume_server_pb::DeleteResult { - file_id: fid_str.clone(), - status: 404, - error: format!("ec volume {} not found", file_id.volume_id), - size: 0, - version: 0, - }); } } } @@ -3941,18 +3947,24 @@ impl VolumeServer for VolumeGrpcService { let vid = VolumeId(req.volume_id); let needle_id = NeedleId(req.file_key); - // Go checks if needle is already deleted (via ecx) before journaling. - // Search all locations for the EC volume. + // Go's handler locates the needle first: absent fails the RPC so the + // caller moves to the next holder; an existing tombstone is a no-op. let mut store = self.state.store.write().unwrap(); if let Some(ec_vol) = store.find_ec_volume_mut(vid) { - // Check if already deleted via ecx index - if let Ok(Some((_offset, size))) = ec_vol.find_needle_from_ecx(needle_id) - && size.is_deleted() - { - // Already deleted, no-op - return Ok(Response::new( - volume_server_pb::VolumeEcBlobDeleteResponse {}, - )); + match ec_vol.find_needle_from_ecx(needle_id) { + Ok(Some((_, size))) if size.is_deleted() => { + return Ok(Response::new( + volume_server_pb::VolumeEcBlobDeleteResponse {}, + )); + } + Ok(None) => { + return Err(Status::not_found(format!( + "needle {} not in ec volume {}", + needle_id, req.volume_id + ))); + } + Ok(Some(_)) => {} + Err(e) => return Err(Status::internal(e.to_string())), } ec_vol .journal_delete(needle_id) diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 929248719..5b8703add 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -2787,15 +2787,23 @@ pub async fn delete_handler( { let has_ec = state.store.read().unwrap().has_ec_volume(vid); if has_ec { - // Step 1: Read the EC needle to get its size and validate cookie - let ec_read_result = { - let store = state.store.read().unwrap(); - store - .find_ec_volume(vid) - .map(|ecv| ecv.read_ec_shard_needle(needle_id)) - }; + // Step 1: Read the EC needle to get its size and validate cookie. + // + // This must go through the *distributed* reader, as the GET path + // does. The local-only `EcVolume::read_ec_shard_needle` errors + // "ec shard N not available locally" for any interval that lives on + // a peer, so on a standard 10+4 spread every HTTP DELETE of an EC + // needle failed 500 and never appended to `.ecj`. The distributed + // reader does a local-first pass in its snapshot phase, so the + // all-shards-local case costs the same as before. + // + // No store guard is held across this call: the reader takes and + // releases its own, and `RwLockReadGuard` is `!Send`. + let ec_read_result = + crate::server::store_ec::read_ec_shard_needle_distributed(&state, vid, needle_id) + .await; match ec_read_result { - Some(Ok(Some(ec_needle))) => { + Ok(Some(ec_needle)) => { // Step 2: Validate cookie (Go: cookie != 0 && cookie != n.Cookie) if cookie.0 != 0 && ec_needle.cookie != cookie { return json_error_with_query( @@ -2805,26 +2813,50 @@ pub async fn delete_handler( ); } let count = ec_needle.data_size as i64; - // Step 3: Journal the delete - let mut store = state.store.write().unwrap(); - if let Some(ecv) = store.find_ec_volume_mut(vid) - && let Err(e) = ecv.journal_delete(needle_id) + // Step 3: Journal the delete on a holder of the needle's + // primary data shard — Go's DeleteEcShardNeedle forwards a + // VolumeEcBlobDelete there rather than journaling locally, + // so exactly one node carries the tombstone and + // delete_count stays consistent across replicas. + match crate::server::store_ec::delete_ec_shard_needle_distributed( + &state, vid, needle_id, + ) + .await { - return json_error_with_query( - StatusCode::INTERNAL_SERVER_ERROR, - format!("Deletion Failed: {}", e), - Some(&del_query), - ); + Ok(()) => { + let result = DeleteResult { size: count }; + return json_response_with_params( + StatusCode::ACCEPTED, + &result, + Some(&del_params), + ); + } + // Unmounted between the read and the append, or the + // needle went away in the same window: nothing was + // journalled, so answering 202 would lose the delete + // while reporting success. + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + let result = DeleteResult { size: 0 }; + return json_response_with_params( + StatusCode::NOT_FOUND, + &result, + Some(&del_params), + ); + } + Err(e) => { + return json_error_with_query( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Deletion Failed: {}", e), + Some(&del_query), + ); + } } - let result = DeleteResult { size: count }; - return json_response_with_params( - StatusCode::ACCEPTED, - &result, - Some(&del_params), - ); } - Some(Ok(None)) => { - // Needle not found in EC volume + Ok(None) => { + // Needle not in the EC index, or the volume disappeared + // between the `has_ec` check and the snapshot — the + // distributed reader reports both as `Ok(None)`, and both + // mean the same thing to a deleter. let result = DeleteResult { size: 0 }; return json_response_with_params( StatusCode::NOT_FOUND, @@ -2832,22 +2864,25 @@ pub async fn delete_handler( Some(&del_params), ); } - Some(Err(e)) => { + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + // A shard or interval the read needed is gone rather than + // merely remote. The GET path answers 404 here; answering + // 500 (as this handler did for every error) told callers to + // retry a delete that can never succeed. + let result = DeleteResult { size: 0 }; + return json_response_with_params( + StatusCode::NOT_FOUND, + &result, + Some(&del_params), + ); + } + Err(e) => { return json_error_with_query( StatusCode::INTERNAL_SERVER_ERROR, format!("Deletion Failed: {}", e), Some(&del_query), ); } - None => { - // EC volume disappeared between has_ec check and find - let result = DeleteResult { size: 0 }; - return json_response_with_params( - StatusCode::NOT_FOUND, - &result, - Some(&del_params), - ); - } } } } diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 0c60d01b8..a90effccc 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -26,7 +26,7 @@ //! cache write-back briefly reacquires the EcVolume's internal //! `RwLock` so we do not contend with the Store-level lock at all. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::fs; use std::io; use std::sync::Arc; @@ -39,7 +39,9 @@ use tokio::sync::Semaphore; use tonic::Request; use crate::pb::master_pb::{self, LookupEcVolumeRequest}; -use crate::pb::volume_server_pb::{CopyFileRequest, VolumeEcShardReadRequest}; +use crate::pb::volume_server_pb::{ + CopyFileRequest, VolumeEcBlobDeleteRequest, VolumeEcShardReadRequest, +}; use crate::server::grpc_client::{ GrpcDialOptions, connect_channel, master_client, parse_grpc_address, volume_server_client, }; @@ -253,6 +255,237 @@ pub async fn read_ec_shard_needle_distributed( Ok(Some(n)) } +/// What one EC delete RPC carries — `VolumeEcBlobDeleteRequest` minus tonic. +struct EcDeleteTarget<'a> { + vid: VolumeId, + collection: &'a str, + version: Version, + needle_id: NeedleId, +} + +/// `Store.doDeleteNeedleFromAtLeastOneRemoteEcShards` in Go: journal the +/// tombstone on one holder of the needle's primary data shard, falling back +/// to any other shard holder when the primary has none. Exactly one node +/// journals — replicas of a shard hold identical .ecx copies, so journaling +/// on more than one would double the reported delete count. +/// +/// `NotFound` means the volume or needle is gone; other errors mean every +/// reachable holder failed or no shard has a holder at all. +pub async fn delete_ec_shard_needle_distributed( + state: &Arc, + vid: VolumeId, + needle_id: NeedleId, +) -> io::Result<()> { + let ( + primary_shard_id, + collection, + version, + total_shards, + local_shards, + data_shards, + encode_ts_ns, + refreshed_at, + cached_locations, + ) = { + let store = state.store.read().unwrap(); + let ecv = store.find_ec_volume(vid).ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + format!("ec volume {} not mounted", vid.0), + ) + })?; + let (_, _, intervals) = ecv.locate_needle(needle_id)?.ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + format!("needle {} not in ec volume {}", needle_id, vid.0), + ) + })?; + let (shard_id, _) = intervals + .first() + .map(|i| ecv.interval_to_shard_id_and_offset(i)) + .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "no intervals for needle"))?; + let (cached_locations, refreshed_at) = ecv.shard_locations_snapshot(); + ( + shard_id, + ecv.collection.clone(), + ecv.version, + ecv.data_shards + ecv.parity_shards, + local_shard_ids(ecv), + ecv.data_shards as usize, + ecv.encode_ts_ns, + refreshed_at, + cached_locations, + ) + }; + let target = EcDeleteTarget { + vid, + collection: &collection, + version, + needle_id, + }; + + // Holder addresses come from the same staleness-gated cache as the read + // path: a master LookupEcVolume only when due, merged on a complete reply. + let mut locations = cached_locations; + if claim_shard_locations_refresh( + state, + vid, + &locations, + refreshed_at, + data_shards, + total_shards as usize, + ) { + match cached_lookup_ec_shard_locations(state, vid).await { + Ok(fresh) => { + match write_back_shard_locations(state, vid, fresh, data_shards, encode_ts_ns) { + Some(merged) => locations = merged, + None => mark_shard_locations_stale(state, vid), + } + } + Err(_) => mark_shard_locations_stale(state, vid), + } + } + + match delete_on_ec_shard_holders(state, &locations, &local_shards, primary_shard_id, &target) + .await + { + Ok(true) => return Ok(()), + Err(e) => return Err(e), + Ok(false) => {} + } + + for shard_id in 0..total_shards { + let Ok(shard_id) = shard_id_try_from(shard_id) else { + continue; + }; + if shard_id == primary_shard_id { + continue; + } + if let Ok(true) = + delete_on_ec_shard_holders(state, &locations, &local_shards, shard_id, &target).await + { + return Ok(()); + } + } + + Err(io::Error::other(format!( + "ec volume {}: no shard holder could journal the delete", + vid.0 + ))) +} + +/// `doDeleteNeedleFromRemoteEcShardServers` in Go. `Ok(false)` is the +/// shard-missing signal — no live holder anywhere — that triggers the +/// caller's fallback walk over the remaining shards. +async fn delete_on_ec_shard_holders( + state: &Arc, + locations: &HashMap>, + local_shards: &HashSet, + shard_id: ShardId, + target: &EcDeleteTarget<'_>, +) -> io::Result { + let addrs = locations.get(&shard_id); + if !local_shards.contains(&shard_id) && addrs.is_none_or(|a| a.is_empty()) { + return Ok(false); + } + + let mut last_err = None; + if local_shards.contains(&shard_id) { + match journal_delete_local(state, target.vid, target.needle_id) { + Ok(()) => return Ok(true), + // Nothing was committed — the volume unmounted or remounted + // without the needle — so it is safe to fall back to other + // shard holders, unlike an RPC failure which may have landed. + Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(false), + Err(e) => last_err = Some(e), + } + } + if let Some(addrs) = addrs { + let self_http = to_http_address(&state.self_url); + for addr in addrs { + // A stale self entry: the loopback RPC would journal on this same + // volume, which the local attempt above already covered. + if to_http_address(addr).as_ref() == self_http.as_ref() { + continue; + } + match delete_on_remote_ec_shard(state, addr, target).await { + Ok(()) => return Ok(true), + Err(e) => last_err = Some(e), + } + } + } + match last_err { + Some(e) => Err(e), + None => Ok(false), + } +} + +/// `doDeleteNeedleFromRemoteEcShard` in Go — one `VolumeEcBlobDelete` RPC. +async fn delete_on_remote_ec_shard( + state: &Arc, + addr: &str, + target: &EcDeleteTarget<'_>, +) -> io::Result<()> { + let grpc_addr = + parse_grpc_address(addr).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + let channel = connect_channel( + &grpc_addr, + state.outgoing_grpc_tls.as_ref(), + GrpcDialOptions::unary(), + ) + .await + .map_err(|e| io::Error::other(format!("connect to {}: {}", addr, e)))?; + let mut client = volume_server_client(channel); + client + .volume_ec_blob_delete(Request::new(VolumeEcBlobDeleteRequest { + volume_id: target.vid.0, + collection: target.collection.to_string(), + file_key: target.needle_id.0, + version: target.version.0 as u32, + })) + .await + .map_err(|e| io::Error::other(format!("volume_ec_blob_delete on {}: {}", addr, e)))?; + Ok(()) +} + +/// Journals on the local volume — what the `VolumeEcBlobDelete` handler runs +/// when this server is the shard holder. An absent needle is an error, not a +/// no-op: `journal_delete` would accept it silently, but here it means the +/// volume remounted as a different generation mid-delete and the tombstone +/// should go to a replica that still has the needle. +fn journal_delete_local( + state: &Arc, + vid: VolumeId, + needle_id: NeedleId, +) -> io::Result<()> { + let mut store = state.store.write().unwrap(); + let ecv = store.find_ec_volume_mut(vid).ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + format!("ec volume {} unmounted", vid.0), + ) + })?; + match ecv.find_needle_from_ecx(needle_id)? { + None => { + return Err(io::Error::new( + io::ErrorKind::NotFound, + format!("needle {} not in local ecx", needle_id), + )); + } + Some((_, size)) if size.is_deleted() => return Ok(()), + Some(_) => {} + } + ecv.journal_delete(needle_id) +} + +fn local_shard_ids(ecv: &crate::storage::erasure_coding::EcVolume) -> HashSet { + ecv.shards + .iter() + .enumerate() + .filter_map(|(i, s)| s.as_ref().map(|_| i as ShardId)) + .collect() +} + /// FULL EC scrub: verify every needle's bytes across local AND remote shards, /// without decoding (so genuine shard faults are reported rather than healed). /// Mirrors Go's `Store.ScrubEcVolume`. Returns (rows walked, broken shards, diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs index 53f358b8a..825f52955 100644 --- a/seaweed-volume/tests/http_integration.rs +++ b/seaweed-volume/tests/http_integration.rs @@ -1040,3 +1040,105 @@ async fn chunk_manifest_expands_chunk_stored_on_ec_volume() { assert_eq!(response.status(), StatusCode::OK); assert_eq!(body_bytes(response).await, chunk_data); } + +// ============================================================================ +// HTTP DELETE on an EC volume whose shards are not all mounted locally +// +// The delete handler used to validate the cookie with the local-only +// `EcVolume::read_ec_shard_needle`, which errors "ec shard N not available +// locally" for any interval held by a peer. Every such error was mapped to 500 +// and no `.ecj` tombstone was written, so on a standard 10+4 spread across 14 +// servers no HTTP delete of an EC needle could ever succeed. +// +// A single node can exercise that path without peers: mount 13 of the 14 +// shards, leaving out the one holding the needle's interval. The distributed +// reader seeds its Reed-Solomon buffers from locally mounted siblings (Phase 0 +// in `store_ec.rs`), so with >= 10 survivors it reconstructs without any peer +// fan-out — while the local-only read still fails outright. +// ============================================================================ + +#[tokio::test] +async fn delete_on_ec_volume_succeeds_when_the_needles_shard_is_not_mounted() { + use seaweed_volume::storage::erasure_coding::ec_encoder::write_ec_files; + use seaweed_volume::storage::erasure_coding::ec_shard::ShardId; + use seaweed_volume::storage::needle::needle::{FileId, Needle}; + use seaweed_volume::storage::types::{Cookie, NeedleId}; + use seaweed_volume::storage::volume::{Volume, VolumeSpec}; + + let (state, tmp) = test_state(); + let dir = tmp.path().to_str().unwrap(); + + let data: Vec = (0..4096u32).map(|i| (i % 251) as u8).collect(); + let nid = NeedleId(0x5c); + let cookie = Cookie(0x0badc0de); + + // Build regular volume 4, then EC-encode it. The volume is standalone and + // never registered in the store, so afterwards it exists only as shards. + { + let mut v = Volume::new( + dir, + dir, + VolumeId(4), + NeedleMapKind::InMemory, + &VolumeSpec::default(), + ) + .unwrap(); + let mut n = Needle { + id: nid, + cookie, + data: data.clone(), + data_size: data.len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + v.sync_to_disk().unwrap(); + v.close(); + } + write_ec_files(dir, dir, "", VolumeId(4), 10, 4).unwrap(); + + // Mount shards 1..=13 only. A needle at .dat offset 0 lives in shard 0's + // first small block, so the interval this delete needs is deliberately the + // one shard that is absent. + { + let mut store = state.store.write().unwrap(); + let shard_ids: Vec = (1..14).collect(); + store.mount_ec_shards(VolumeId(4), "", &shard_ids).unwrap(); + } + + let fid = FileId::new(VolumeId(4), nid, cookie).to_string(); + let app = build_admin_router(state.clone()); + let response = app + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/{}", fid)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!( + response.status(), + StatusCode::ACCEPTED, + "DELETE of an EC needle must reconstruct through the distributed \ + reader instead of failing 500 on a non-local shard" + ); + + // The tombstone must actually have landed: a later GET is a 404. + let app = build_admin_router(state.clone()); + let response = app + .oneshot( + Request::builder() + .uri(format!("/{}", fid)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::NOT_FOUND, + "the delete must have been journalled, not just answered 202" + ); +}