From d51454adf469ddb2ce7227549bbe80fa0088c878 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 16 May 2026 20:44:28 -0700 Subject: [PATCH] rust(seaweed-volume): distributed EC read across peer servers (#9516) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(seaweed-volume): distributed EC read across peer servers EcVolume::read_ec_shard_needle previously errored with NotFound when any interval's shard wasn't local. In an RS(10,4)-across-N deployment each server holds one shard, so every read needed >=9 peer fetches and post-EC GETs returned 404 on volumes whose shards lived on more than one server. Mirror of weed/storage/store_ec.go's readOneEcShardInterval -> readRemoteEcShardInterval -> recoverOneRemoteEcShardInterval chain: * server/store_ec.rs (new): entry point read_ec_shard_needle_distributed. Snapshots locate-needle + local reads under the Store sync lock, drops the lock, then async-fetches missing intervals via the peer's VolumeEcShardRead RPC. Falls back to Reed-Solomon reconstruction (read every other shard at the same (shard_offset, size) and run rs.reconstruct) when the direct peer read fails. Refreshes the per-EcVolume shard_locations cache from the master's LookupEcVolume RPC using Go's freshness thresholds (11s / 7min / 37min). * erasure_coding/ec_volume.rs: shard_locations now sits behind a std::sync::RwLock so the read path can refresh the map without holding the Store write lock. Adds shard_locations_refresh_time (Mutex>) for the staleness heuristic. Mirrors Go's ShardLocationsLock / ShardLocationsRefreshTime fields. set/get helpers updated for interior mutability. * server/handlers.rs: GET handler now tries the local-only fast path first, then falls through to the distributed path on NotFound. * review: address PR 9516 feedback on distributed EC read Five of the six PR-review comments addressed; the sixth (JWT on outgoing peer gRPC) is deferred with an explicit TODO because the crate-wide outgoing-JWT signing surface doesn't exist yet — adding it in this one call site would split the credential plumbing across peer paths that already lack it (copy_file_from_source, batch_delete, …). Revisit when an outgoing-JWT helper lands. Fixed in this commit: * Handlers: drop the two-tier (local-first, then distributed) read in handlers.rs. read_ec_shard_needle_distributed already does the local-first pass under the same store read lock; the redundant outer attempt re-read local intervals twice for any needle that spanned mixed-locality shards. * Scanner snapshot: replace inline locate-needle math with `ecv.locate_needle(needle_id)`. Same routine the local-only read path uses, so byte-identical on shard-size + interval boundaries. * EcVolume::set_shard_locations also advances shard_locations_refresh_time so the staleness check honors callers that populate the cache directly without going through the master LookupEcVolume RPC. * parse_grpc_address moved from grpc_server.rs into grpc_client.rs as `pub` and is reused by both grpc_server.rs and the new store_ec module. Single source of truth for the HTTP↔gRPC port-offset convention. * Reconstruction (recover_one_remote_ec_shard_interval) now seeds bufs from locally-mounted survivor shards BEFORE the remote fan-out. Previously the fan-out was remote-only, so when the shard_locations cache was cold or the master lookup failed, reconstruction errored even though enough siblings were on local disk to recover the missing interval. * review: tighten parse_grpc_address; atomic shard-locations cache swap Two follow-up findings from the PR 9516 review round 2: * `parse_grpc_address` now validates BOTH port components in the dotted form (`host:port.grpcPort`) — previously a non-numeric HTTP port like `host:abc.18080` slipped through and tripped a less-useful downstream URI parse error. The implicit form (`host:port` → port + 10000) also gains an overflow check so inputs like `host:60000` (which silently wrap past u16) are rejected here instead of producing an opaque connection failure later. Six unit tests cover each rejection path. * `EcVolume::set_shard_locations` no longer bumps the per-volume refresh timestamp. The previous fix introduced a freshness race: a multi-shard population that inserts shard-by-shard would flip `needs_refresh == false` on the first write, letting a concurrent reader observe a half-populated map already marked "fresh" and return NotFound for the not-yet-inserted shards. Added `EcVolume::replace_shard_locations(map)` for the atomic bulk swap; `write_back_shard_locations` in the distributed-read path uses it so the cache transitions old → fresh in a single observable step. --- seaweed-volume/src/server/grpc_client.rs | 97 +++ seaweed-volume/src/server/grpc_server.rs | 29 +- seaweed-volume/src/server/handlers.rs | 63 +- seaweed-volume/src/server/mod.rs | 1 + seaweed-volume/src/server/store_ec.rs | 678 ++++++++++++++++++ .../src/storage/erasure_coding/ec_volume.rs | 58 +- 6 files changed, 862 insertions(+), 64 deletions(-) create mode 100644 seaweed-volume/src/server/store_ec.rs diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index 2eee9d5dd..e222478fd 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/seaweed-volume/src/server/grpc_client.rs @@ -107,6 +107,57 @@ pub fn build_grpc_endpoint( Ok(endpoint) } +/// Parse a SeaweedFS server address (`"ip:port.grpcPort"` or +/// `"ip:port"`) into the `host:grpcPort` form `build_grpc_endpoint` +/// expects. With the trailing `.grpcPort` segment, that segment IS +/// the gRPC port; without it, the gRPC port is `port + 10000` +/// (SeaweedFS's HTTP↔gRPC port-offset convention). +/// +/// Shared between `grpc_server.rs` and the distributed-EC-read path +/// in `store_ec.rs` — keep this as the single source of truth so the +/// HTTP↔gRPC port translation can't drift between callers. +pub fn parse_grpc_address(source: &str) -> Result { + let colon_idx = source + .rfind(':') + .ok_or_else(|| format!("cannot parse address: {}", source))?; + let host = &source[..colon_idx]; + let port_part = &source[colon_idx + 1..]; + + if let Some(dot_idx) = port_part.rfind('.') { + // Format: "ip:port.grpcPort". Validate BOTH ports as u16 + // so a malformed HTTP port (e.g. `host:abc.18080`) is + // rejected here rather than tripping a downstream + // `build_grpc_endpoint` URI parse failure with a less + // useful error. + let http_port = &port_part[..dot_idx]; + let grpc_port = &port_part[dot_idx + 1..]; + http_port + .parse::() + .map_err(|e| format!("invalid http port {:?}: {}", http_port, e))?; + grpc_port + .parse::() + .map_err(|e| format!("invalid grpc port {:?}: {}", grpc_port, e))?; + return Ok(format!("{}:{}", host, grpc_port)); + } + + // Format: "ip:port" → grpc = port + 10000. Reject inputs whose + // implicit grpc port would overflow the TCP port range (e.g. + // `host:60000` produces 70000 — invalid). Without this check + // the cast silently wraps and the endpoint call later fails + // with an opaque connection error. + let port: u16 = port_part + .parse() + .map_err(|e| format!("invalid port {:?}: {}", port_part, e))?; + let grpc_port = port as u32 + 10000; + if grpc_port > u16::MAX as u32 { + return Err(format!( + "implicit grpc port out of range: {} + 10000 = {}", + port, grpc_port + )); + } + Ok(format!("{}:{}", host, grpc_port)) +} + #[cfg(test)] mod tests { use super::{build_grpc_endpoint, grpc_endpoint_uri, load_outgoing_grpc_tls}; @@ -203,4 +254,50 @@ mod tests { let endpoint = build_grpc_endpoint("127.0.0.1:19333", None).unwrap(); assert_eq!(endpoint.uri().scheme_str(), Some("http")); } + + #[test] + fn test_parse_grpc_address_dotted_form() { + use super::parse_grpc_address; + assert_eq!( + parse_grpc_address("127.0.0.1:8080.18080").unwrap(), + "127.0.0.1:18080" + ); + } + + #[test] + fn test_parse_grpc_address_implicit_form_adds_10000() { + use super::parse_grpc_address; + assert_eq!( + parse_grpc_address("127.0.0.1:8080").unwrap(), + "127.0.0.1:18080" + ); + } + + #[test] + fn test_parse_grpc_address_rejects_non_numeric_http_port_in_dotted_form() { + use super::parse_grpc_address; + let err = parse_grpc_address("host:abc.18080").unwrap_err(); + assert!(err.contains("invalid http port"), "{}", err); + } + + #[test] + fn test_parse_grpc_address_rejects_non_numeric_grpc_port_in_dotted_form() { + use super::parse_grpc_address; + let err = parse_grpc_address("host:8080.xyz").unwrap_err(); + assert!(err.contains("invalid grpc port"), "{}", err); + } + + #[test] + fn test_parse_grpc_address_rejects_implicit_port_that_overflows() { + use super::parse_grpc_address; + let err = parse_grpc_address("127.0.0.1:60000").unwrap_err(); + assert!(err.contains("out of range"), "{}", err); + } + + #[test] + fn test_parse_grpc_address_rejects_input_without_colon() { + use super::parse_grpc_address; + let err = parse_grpc_address("hostname").unwrap_err(); + assert!(err.contains("cannot parse"), "{}", err); + } } diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index ebb8c885a..10f84af13 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -4051,29 +4051,12 @@ async fn ping_filer_target( Ok(resp.into_inner().start_time_ns) } -/// Parse a SeaweedFS server address ("ip:port.grpcPort" or "ip:port") into a gRPC address. -fn parse_grpc_address(source: &str) -> Result { - if let Some(colon_idx) = source.rfind(':') { - let port_part = &source[colon_idx + 1..]; - if let Some(dot_idx) = port_part.rfind('.') { - // Format: "ip:port.grpcPort" - let host = &source[..colon_idx]; - let grpc_port = &port_part[dot_idx + 1..]; - grpc_port - .parse::() - .map_err(|e| format!("invalid grpc port: {}", e))?; - return Ok(format!("{}:{}", host, grpc_port)); - } - // Format: "ip:port" → grpc = port + 10000 - let port: u16 = port_part - .parse() - .map_err(|e| format!("invalid port: {}", e))?; - let grpc_port = port as u32 + 10000; - let host = &source[..colon_idx]; - return Ok(format!("{}:{}", host, grpc_port)); - } - Err(format!("cannot parse address: {}", source)) -} +// parse_grpc_address moved to super::grpc_client::parse_grpc_address +// for sharing with the distributed-EC-read path in server/store_ec.rs. +// In-file callers below still write `parse_grpc_address(...)`; this +// `use` makes them resolve to the new home without churning every +// call site. +use super::grpc_client::parse_grpc_address; /// Set the modification time of a file from nanoseconds since Unix epoch. fn set_file_mtime(path: &str, modified_ts_ns: i64) { diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index d324a587b..633df25bf 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -1019,43 +1019,44 @@ async fn get_or_head_handler_inner( if has_ec_volume && !has_volume { // ---- EC volume read path (always full read, no streaming) ---- - let store = state.store.read().unwrap(); - match store.find_ec_volume(vid) { - Some(ecv) => match ecv.read_ec_shard_needle(needle_id) { - Ok(Some(ec_needle)) => { - n = ec_needle; - } - Ok(None) => { - metrics::HANDLER_COUNTER - .with_label_values(&[metrics::ERROR_GET_NOT_FOUND]) - .inc(); - return StatusCode::NOT_FOUND.into_response(); - } - Err(e) => { - if e.kind() == std::io::ErrorKind::NotFound { - metrics::HANDLER_COUNTER - .with_label_values(&[metrics::ERROR_GET_NOT_FOUND]) - .inc(); - return StatusCode::NOT_FOUND.into_response(); - } - metrics::HANDLER_COUNTER - .with_label_values(&[metrics::ERROR_GET_INTERNAL]) - .inc(); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - format!("read ec error: {}", e), - ) - .into_response(); - } - }, - None => { + // + // The distributed read path already does a local-first pass + // in its Snapshot phase under the same store read lock the + // legacy code would have taken — so calling it directly + // serves both the "all shards local" fast case and the + // "some intervals need peer fetch + reconstruct" general + // case without paying for the local interval reads twice. + match crate::server::store_ec::read_ec_shard_needle_distributed( + &state, vid, needle_id, + ) + .await + { + Ok(Some(ec_needle)) => { + n = ec_needle; + } + Ok(None) => { metrics::HANDLER_COUNTER .with_label_values(&[metrics::ERROR_GET_NOT_FOUND]) .inc(); return StatusCode::NOT_FOUND.into_response(); } + Err(e) => { + let kind = if e.kind() == std::io::ErrorKind::NotFound { + metrics::ERROR_GET_NOT_FOUND + } else { + metrics::ERROR_GET_INTERNAL + }; + metrics::HANDLER_COUNTER.with_label_values(&[kind]).inc(); + if e.kind() == std::io::ErrorKind::NotFound { + return StatusCode::NOT_FOUND.into_response(); + } + return ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("ec read: {}", e), + ) + .into_response(); + } } - drop(store); // Validate cookie (matches Go behavior after ReadEcShardNeedle) if n.cookie != cookie { diff --git a/seaweed-volume/src/server/mod.rs b/seaweed-volume/src/server/mod.rs index 4056ccfce..092e43c35 100644 --- a/seaweed-volume/src/server/mod.rs +++ b/seaweed-volume/src/server/mod.rs @@ -9,6 +9,7 @@ pub mod memory_status; pub mod profiling; pub mod request_id; pub mod server_stats; +pub mod store_ec; pub mod ui; pub mod volume_server; pub mod write_queue; diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs new file mode 100644 index 000000000..37d957458 --- /dev/null +++ b/seaweed-volume/src/server/store_ec.rs @@ -0,0 +1,678 @@ +//! Distributed EC read path. Mirror of `weed/storage/store_ec.go`'s +//! `readEcShardIntervals` → `readOneEcShardInterval` → +//! `readRemoteEcShardInterval` → `recoverOneRemoteEcShardInterval` +//! chain. +//! +//! The existing `EcVolume::read_ec_shard_needle` reads only locally- +//! mounted shards and returns `NotFound` if any interval requires a +//! shard held on a peer server. In a standard RS(10,4)-across-14 +//! deployment each server holds one shard, so every read needs >=9 +//! peer fetches. This module fills the gap by: +//! +//! 1. Locating the needle in `.ecx` (under the Store read lock) and +//! computing the per-interval (shard_id, shard_offset, size). +//! 2. Reading the local-resident intervals while still holding the +//! lock — same path the local-only helper uses. +//! 3. Dropping the lock and, for any remaining intervals, fetching +//! from peer volume servers via `VolumeEcShardRead`. If the +//! direct peer read fails, fan-out reads to other shards at the +//! same (shard_offset, size) and rebuild the missing shard via +//! Reed-Solomon — exactly Go's flow. +//! 4. Refreshing the per-EcVolume `shard_locations` cache from the +//! master's `LookupEcVolume` RPC when the cached map is stale. +//! +//! All gRPC IO is async; the file IO portion runs under the sync +//! Store read lock, matching Go's `readLocalEcShardInterval`. The +//! 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::io; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use futures::future::join_all; +use reed_solomon_erasure::galois_8::ReedSolomon; +use tonic::Request; + +use crate::pb::master_pb::{self, seaweed_client::SeaweedClient, LookupEcVolumeRequest}; +use crate::pb::volume_server_pb::{ + volume_server_client::VolumeServerClient, VolumeEcShardReadRequest, +}; +use crate::server::grpc_client::{build_grpc_endpoint, parse_grpc_address, GRPC_MAX_MESSAGE_SIZE}; +use crate::server::request_id::outgoing_request_id_interceptor; +use crate::server::volume_server::VolumeServerState; +use crate::storage::erasure_coding::ec_shard::ShardId; +use crate::storage::needle::needle::{get_actual_size, Needle}; +use crate::storage::types::*; + +/// One interval's data after Phase A. +enum IntervalResult { + /// Already read from a locally-mounted shard. + Local(Vec), + /// Shard not local (or local read failed). Must be fetched. + NeedRemote { + shard_id: ShardId, + shard_offset: i64, + size: usize, + }, +} + +/// Snapshot extracted under the Store read lock so Phases B/C can run +/// without holding any sync lock across `.await`. +struct Snapshot { + data_shards: u32, + parity_shards: u32, + version: Version, + actual_size: usize, + offset: Offset, + size_for_parse: Size, + intervals: Vec, + cached_locations: HashMap>, + cache_refreshed_at: Option, +} + +/// Top-level entry point. Returns `Ok(None)` for "not found" (matches +/// Go's `ReadEcShardNeedle`); errors propagate as `io::Error`. +pub async fn read_ec_shard_needle_distributed( + state: &Arc, + vid: VolumeId, + needle_id: NeedleId, +) -> io::Result> { + // Phase A — under the Store read lock, locate the needle, compute + // intervals, and read any locally-mounted shard intervals. We must + // not `.await` while holding this guard (std::sync::RwLockReadGuard + // is !Send). + let snapshot = match snapshot_under_lock(state, vid, needle_id)? { + Some(s) => s, + None => return Ok(None), + }; + + // Phase B — refresh the shard_locations cache from the master if + // it is stale. Do this lazily: if every needed interval was read + // locally we can skip the master RPC entirely. + let any_remote = snapshot + .intervals + .iter() + .any(|r| matches!(r, IntervalResult::NeedRemote { .. })); + let total_shards = (snapshot.data_shards + snapshot.parity_shards) as usize; + + let mut shard_locations = snapshot.cached_locations.clone(); + if any_remote + && needs_refresh( + &shard_locations, + snapshot.cache_refreshed_at, + snapshot.data_shards as usize, + total_shards, + ) + { + match cached_lookup_ec_shard_locations(state, vid).await { + Ok(fresh) => { + shard_locations = fresh.clone(); + write_back_shard_locations(state, vid, fresh); + } + Err(e) => { + // Lookup failed — proceed with cached values. If cache + // is empty, the remote fetch below will fail and we + // surface a NotFound (matching Go's behavior when no + // locations are known). + tracing::warn!( + "ec lookup failed for volume {}: {} — using cached locations ({} entries)", + vid.0, + e, + shard_locations.len(), + ); + } + } + } + + // Phase C — fetch missing intervals, reconstructing when the + // direct peer read fails. + let mut assembled: Vec> = Vec::with_capacity(snapshot.intervals.len()); + for res in snapshot.intervals { + match res { + IntervalResult::Local(buf) => assembled.push(buf), + IntervalResult::NeedRemote { + shard_id, + shard_offset, + size, + } => { + let buf = fetch_one_interval( + state, + vid, + needle_id, + shard_id, + shard_offset, + size, + &shard_locations, + snapshot.data_shards as usize, + snapshot.parity_shards as usize, + ) + .await?; + assembled.push(buf); + } + } + } + + // Phase D — assemble and parse the Needle. Mirrors the tail of + // `EcVolume::read_ec_shard_needle`. + let mut bytes = Vec::with_capacity(snapshot.actual_size); + for chunk in assembled { + bytes.extend_from_slice(&chunk); + } + bytes.truncate(snapshot.actual_size); + if bytes.len() < snapshot.actual_size { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "read {} bytes but need {} for needle {}", + bytes.len(), + snapshot.actual_size, + needle_id + ), + )); + } + + let mut n = Needle::default(); + n.id = needle_id; + n.read_bytes( + &bytes, + snapshot.offset.to_actual_offset(), + snapshot.size_for_parse, + snapshot.version, + ) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?; + Ok(Some(n)) +} + +fn snapshot_under_lock( + state: &Arc, + vid: VolumeId, + needle_id: NeedleId, +) -> io::Result> { + let store = state.store.read().unwrap(); + let ecv = match store.find_ec_volume(vid) { + Some(v) => v, + None => return Ok(None), + }; + + // Reuse EcVolume::locate_needle for offset/size resolution AND + // the per-needle shard-interval math — it's the same routine the + // local-only read path uses, so we stay byte-identical on the + // shard-size + interval boundaries. + let (offset, size, intervals) = match ecv.locate_needle(needle_id)? { + Some(v) => v, + None => return Ok(None), + }; + if intervals.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "no intervals for needle", + )); + } + let actual = get_actual_size(size, ecv.version); + + // Phase A.local: for each interval read the local shard if we + // hold it; otherwise fall through to remote. We accumulate the + // results in interval order so the assembly step is just a + // concat. + let mut interval_results = Vec::with_capacity(intervals.len()); + for interval in &intervals { + let (shard_id, shard_offset) = interval.to_shard_id_and_offset(ecv.data_shards); + let buf_size = interval.size as usize; + let local = ecv + .shards + .get(shard_id as usize) + .and_then(|s| s.as_ref()); + match local { + Some(shard) => { + let mut buf = vec![0u8; buf_size]; + match shard.read_at(&mut buf, shard_offset as u64) { + Ok(n) if n == buf_size => { + interval_results.push(IntervalResult::Local(buf)); + } + _ => interval_results.push(IntervalResult::NeedRemote { + shard_id, + shard_offset, + size: buf_size, + }), + } + } + None => interval_results.push(IntervalResult::NeedRemote { + shard_id, + shard_offset, + size: buf_size, + }), + } + } + + let cached_locations = ecv.shard_locations.read().unwrap().clone(); + let cache_refreshed_at = *ecv.shard_locations_refresh_time.lock().unwrap(); + + Ok(Some(Snapshot { + data_shards: ecv.data_shards, + parity_shards: ecv.parity_shards, + version: ecv.version, + actual_size: actual as usize, + offset, + size_for_parse: size, + intervals: interval_results, + cached_locations, + cache_refreshed_at, + })) +} + +/// Master `LookupEcVolume` freshness rules — match Go's +/// `cachedLookupEcShardLocations` thresholds in store_ec.go. +fn needs_refresh( + locations: &HashMap>, + refreshed_at: Option, + data_shards: usize, + total_shards: usize, +) -> bool { + let now = Instant::now(); + let age = match refreshed_at { + Some(t) => now.saturating_duration_since(t), + None => return true, + }; + let shard_count = locations.len(); + if shard_count < data_shards && age < Duration::from_secs(11) { + return false; + } + if shard_count == total_shards && age < Duration::from_secs(37 * 60) { + return false; + } + if shard_count >= data_shards && age < Duration::from_secs(7 * 60) { + return false; + } + true +} + +async fn cached_lookup_ec_shard_locations( + state: &Arc, + vid: VolumeId, +) -> io::Result>> { + let master = { + let live = state.current_master_url.read().await.clone(); + if !live.is_empty() { + live + } else { + state.master_url.clone() + } + }; + if master.is_empty() { + return Err(io::Error::new( + io::ErrorKind::Other, + "no master configured for ec shard lookup", + )); + } + + let grpc_addr = parse_grpc_address(&master) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref()) + .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?; + let channel = endpoint + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(10)) + .connect() + .await + .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("master connect: {}", e)))?; + + let mut client = SeaweedClient::with_interceptor(channel, outgoing_request_id_interceptor) + .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) + .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + + let resp = client + .lookup_ec_volume(Request::new(LookupEcVolumeRequest { volume_id: vid.0 })) + .await + .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("lookup_ec_volume: {}", e)))?; + let resp = resp.into_inner(); + + let mut out = HashMap::new(); + for entry in resp.shard_id_locations { + let addrs: Vec = entry + .locations + .iter() + .map(format_location_as_server_address) + .collect(); + out.insert(entry.shard_id as ShardId, addrs); + } + Ok(out) +} + +fn write_back_shard_locations( + state: &Arc, + vid: VolumeId, + locations: HashMap>, +) { + let store = state.store.read().unwrap(); + if let Some(ecv) = store.find_ec_volume(vid) { + // Atomic swap + freshness stamp so a concurrent reader sees + // either the prior cache or the fresh one — never an + // intermediate half-replaced map with the freshness flag + // already flipped. + ecv.replace_shard_locations(locations); + } +} + +/// Build a SeaweedFS-style `host:httpPort.grpcPort` address from a +/// master `Location` so the result is what `parse_grpc_address` (and +/// the heartbeat path) already understand. +fn format_location_as_server_address(loc: &master_pb::Location) -> String { + let raw = loc + .url + .trim_start_matches("http://") + .trim_start_matches("https://"); + if loc.grpc_port > 0 { + if let Some((host, http_port)) = raw.rsplit_once(':') { + return format!("{}:{}.{}", host, http_port, loc.grpc_port); + } + } + raw.to_string() +} + +/// Try direct peer read; on failure, reconstruct via Reed-Solomon +/// from the other shards. Mirrors `readOneEcShardInterval`'s tail. +async fn fetch_one_interval( + state: &Arc, + vid: VolumeId, + needle_id: NeedleId, + shard_id: ShardId, + shard_offset: i64, + size: usize, + shard_locations: &HashMap>, + data_shards: usize, + parity_shards: usize, +) -> io::Result> { + // Direct peer read against the cached locations for this shard. + if let Some(sources) = shard_locations.get(&shard_id) { + if !sources.is_empty() { + match read_remote_ec_shard_interval( + state, + sources, + vid, + needle_id, + shard_id, + shard_offset, + size, + ) + .await + { + Ok(buf) => return Ok(buf), + Err(e) => { + tracing::debug!( + "direct read ec shard {}.{} from {:?} failed: {} — will reconstruct", + vid.0, + shard_id, + sources, + e + ); + } + } + } + } + + // Reconstruct: fan-out reads to every other shard at the same + // (shard_offset, size). Mirrors `recoverOneRemoteEcShardInterval`. + recover_one_remote_ec_shard_interval( + state, + vid, + needle_id, + shard_id, + shard_offset, + size, + shard_locations, + data_shards, + parity_shards, + ) + .await +} + +async fn read_remote_ec_shard_interval( + state: &Arc, + sources: &[String], + vid: VolumeId, + needle_id: NeedleId, + shard_id: ShardId, + shard_offset: i64, + size: usize, +) -> io::Result> { + let mut last_err: Option = None; + for src in sources { + match do_read_remote_ec_shard_interval( + state, + src, + vid, + needle_id, + shard_id, + shard_offset, + size, + ) + .await + { + Ok(buf) => return Ok(buf), + Err(e) => last_err = Some(e), + } + } + Err(last_err.unwrap_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + format!("no source for ec shard {}.{}", vid.0, shard_id), + ) + })) +} + +async fn do_read_remote_ec_shard_interval( + state: &Arc, + source: &str, + vid: VolumeId, + needle_id: NeedleId, + shard_id: ShardId, + shard_offset: i64, + size: usize, +) -> io::Result> { + let grpc_addr = + parse_grpc_address(source).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref()) + .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?; + let channel = endpoint + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(30)) + .connect() + .await + .map_err(|e| { + io::Error::new( + io::ErrorKind::Other, + format!("connect to {}: {}", source, e), + ) + })?; + + // TODO(grpc-jwt): clusters with `jwt.signing.key` configured will + // reject peer-to-peer VolumeEcShardRead calls until the Rust + // crate grows an outgoing-JWT interceptor. The gap is shared + // with every other peer gRPC call from this binary + // (`copy_file_from_source`, `batch_delete`, …) — handling it + // here in isolation would split the credential plumbing across + // call sites. Re-visit when outgoing JWT signing lands as a + // server-wide helper. + let mut client = VolumeServerClient::with_interceptor(channel, outgoing_request_id_interceptor) + .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE) + .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE); + + let req = VolumeEcShardReadRequest { + volume_id: vid.0, + shard_id: shard_id as u32, + offset: shard_offset, + size: size as i64, + file_key: needle_id.0, + }; + let resp = client + .volume_ec_shard_read(Request::new(req)) + .await + .map_err(|e| { + io::Error::new( + io::ErrorKind::Other, + format!("volume_ec_shard_read {}.{} from {}: {}", vid.0, shard_id, source, e), + ) + })?; + let mut stream = resp.into_inner(); + + let mut out = Vec::with_capacity(size); + while let Some(msg) = stream + .message() + .await + .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("recv: {}", e)))? + { + if !msg.data.is_empty() { + out.extend_from_slice(&msg.data); + } + } + if out.len() < size { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "short read from {} for ec shard {}.{}: got {} want {}", + source, + vid.0, + shard_id, + out.len(), + size + ), + )); + } + out.truncate(size); + Ok(out) +} + +async fn recover_one_remote_ec_shard_interval( + state: &Arc, + vid: VolumeId, + needle_id: NeedleId, + shard_id_to_recover: ShardId, + shard_offset: i64, + size: usize, + shard_locations: &HashMap>, + data_shards: usize, + parity_shards: usize, +) -> io::Result> { + let total_shards = data_shards + parity_shards; + let rs = ReedSolomon::new(data_shards, parity_shards).map_err(|e| { + io::Error::new( + io::ErrorKind::Other, + format!("reed-solomon init: {:?}", e), + ) + })?; + + let mut bufs: Vec>> = vec![None; total_shards]; + + // Phase 0: seed bufs from LOCALLY mounted shards. If this node + // already holds enough sibling shards, reconstruction completes + // without any peer fan-out — and even with a cold/incomplete + // shard_locations cache or a failed master lookup, local + // survivors still contribute. Mirrors Go's + // recoverOneRemoteEcShardInterval behaviour, which is implicitly + // local-aware because the Store fan-out targets ALL known + // locations (including the caller's own server address); the + // Rust port had been remote-only, so reconstructing with a cold + // cache failed even when enough siblings were on disk. + { + let store = state.store.read().unwrap(); + if let Some(ecv) = store.find_ec_volume(vid) { + for sid in 0..total_shards { + if sid as ShardId == shard_id_to_recover { + continue; + } + if let Some(Some(shard)) = ecv.shards.get(sid) { + let mut buf = vec![0u8; size]; + if shard.read_at(&mut buf, shard_offset as u64).map(|n| n == size).unwrap_or(false) { + bufs[sid] = Some(buf); + } + } + } + } + } + + // Phase 1: remote fan-out — one task per known shard location + // we DON'T already have locally and DON'T need to recover. + let mut tasks = Vec::new(); + for (sid, locs) in shard_locations { + if *sid == shard_id_to_recover || locs.is_empty() { + continue; + } + if bufs[*sid as usize].is_some() { + continue; + } + let sid = *sid; + let locs = locs.clone(); + let state = state.clone(); + tasks.push(async move { + let res = read_remote_ec_shard_interval( + &state, + &locs, + vid, + needle_id, + sid, + shard_offset, + size, + ) + .await; + (sid, res) + }); + } + let results = join_all(tasks).await; + + for (sid, res) in results { + match res { + Ok(buf) => { + if (sid as usize) < total_shards { + bufs[sid as usize] = Some(buf); + } + } + Err(e) => { + tracing::debug!( + "recover: read {}.{} for needle {} failed: {}", + vid.0, + sid, + needle_id, + e + ); + } + } + } + + let available = bufs.iter().filter(|b| b.is_some()).count(); + if available < data_shards { + return Err(io::Error::new( + io::ErrorKind::Other, + format!( + "cannot recover ec shard {}.{}: only {} shards available, need at least {}", + vid.0, shard_id_to_recover, available, data_shards + ), + )); + } + + rs.reconstruct(&mut bufs).map_err(|e| { + io::Error::new( + io::ErrorKind::Other, + format!( + "reed-solomon reconstruct ec shard {}.{}: {:?}", + vid.0, shard_id_to_recover, e + ), + ) + })?; + + match bufs.into_iter().nth(shard_id_to_recover as usize).flatten() { + Some(buf) => Ok(buf), + None => Err(io::Error::new( + io::ErrorKind::Other, + format!( + "reconstructed buffer for shard {}.{} missing after RS reconstruct", + vid.0, shard_id_to_recover + ), + )), + } +} + +// parse_grpc_address lives in `grpc_client.rs` and is re-exported +// here via the use above so this module shares a single +// HTTP↔gRPC port-translation routine with grpc_server.rs. diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index e070fe9e9..5d137542a 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -44,8 +44,17 @@ pub struct EcVolume { /// Directory where .ecx/.ecj were actually found (may differ from dir_idx after fallback). ecx_actual_dir: String, /// Maps shard ID -> list of server addresses where that shard exists. - /// Used for distributed EC reads across the cluster. - pub shard_locations: HashMap>, + /// Used for distributed EC reads across the cluster. Wrapped in + /// `RwLock` so the read path can refresh the map (under master + /// lookup) without holding the Store write lock — mirrors Go's + /// `ShardLocationsLock sync.RWMutex` in `weed/storage/erasure_coding/ec_volume.go`. + pub shard_locations: std::sync::RwLock>>, + /// Wall-clock timestamp of the most recent successful + /// `LookupEcVolume` refresh of `shard_locations`. `None` until the + /// first refresh. Drives the staleness heuristic in + /// `cached_lookup_ec_shard_locations` (mirrors Go's + /// `ShardLocationsRefreshTime`). + pub shard_locations_refresh_time: std::sync::Mutex>, /// EC volume expiration time (unix epoch seconds), set during EC encode from TTL. pub expire_at_sec: u64, } @@ -165,7 +174,8 @@ impl EcVolume { deleted_needles: RwLock::new(HashSet::new()), disk_type: DiskType::default(), ecx_actual_dir: dir_idx.to_string(), - shard_locations: HashMap::new(), + shard_locations: std::sync::RwLock::new(HashMap::new()), + shard_locations_refresh_time: std::sync::Mutex::new(None), expire_at_sec, }; @@ -430,17 +440,45 @@ impl EcVolume { // ---- Shard locations (distributed tracking) ---- - /// Set the list of server addresses for a given shard ID. - pub fn set_shard_locations(&mut self, shard_id: ShardId, locations: Vec) { - self.shard_locations.insert(shard_id, locations); + /// Set the list of server addresses for a single shard ID. Does + /// NOT touch `shard_locations_refresh_time` — a per-shard write + /// from inside a multi-shard population (e.g. iterating the + /// `LookupEcVolume` response shard-by-shard) would otherwise + /// flip the staleness flag while the map is still incomplete, + /// letting a concurrent reader observe `needs_refresh == false` + /// against a half-populated cache and return NotFound for the + /// not-yet-inserted shards. + /// + /// Callers populating the whole cache atomically should use + /// [`Self::replace_shard_locations`] instead — it swaps the + /// entire map under the write lock and advances the refresh + /// timestamp in one step. + pub fn set_shard_locations(&self, shard_id: ShardId, locations: Vec) { + self.shard_locations + .write() + .unwrap() + .insert(shard_id, locations); } - /// Get the list of server addresses for a given shard ID. - pub fn get_shard_locations(&self, shard_id: ShardId) -> &[String] { + /// Atomically replace the entire shard-locations map and stamp + /// the refresh time. Used by the distributed-read path's + /// post-`LookupEcVolume` write-back so the cache transitions + /// from old → fresh in a single observable step — concurrent + /// readers either see the full prior map or the full new map, + /// never an intermediate state with the freshness flag flipped. + pub fn replace_shard_locations(&self, locations: HashMap>) { + *self.shard_locations.write().unwrap() = locations; + *self.shard_locations_refresh_time.lock().unwrap() = Some(std::time::Instant::now()); + } + + /// Get a cloned list of server addresses for a given shard ID. + pub fn get_shard_locations(&self, shard_id: ShardId) -> Vec { self.shard_locations + .read() + .unwrap() .get(&shard_id) - .map(|v| v.as_slice()) - .unwrap_or(&[]) + .cloned() + .unwrap_or_default() } // ---- Index operations ----