diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index a0696853a..e0446d9d0 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -19,6 +19,7 @@ use crate::pb::volume_server_pb::volume_server_server::VolumeServer; use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT; use crate::storage::needle::needle::{self, Needle}; use crate::storage::types::*; +use crate::storage::volume::VolumeSpec; use super::grpc_client::{build_grpc_endpoint, GRPC_MAX_MESSAGE_SIZE}; use super::volume_server::VolumeServerState; @@ -1276,12 +1277,14 @@ impl VolumeServer for VolumeGrpcService { store .add_volume( vid, - &req.collection, - Some(rp), - ttl, - req.preallocate as u64, disk_type, - version, + &VolumeSpec { + collection: &req.collection, + replica_placement: Some(rp), + ttl, + preallocate: req.preallocate as u64, + version, + }, ) .map_err(|e| Status::internal(e.to_string()))?; self.state.volume_state_notify.notify_one(); @@ -1978,24 +1981,29 @@ impl VolumeServer for VolumeGrpcService { } // Copy .dat file + let mut progress = CopyProgress { + tx: &tx, + next_report_target: &mut next_report_target, + report_interval, + throttler: &mut throttler, + }; if !has_remote_dat { let dat_path = format!("{}.dat", data_base_name); let dat_modified_ts_ns = copy_file_from_source( &mut client, - false, - &req.collection, - req.volume_id, - vol_info.compaction_revision, - vol_info.dat_file_size, - &dat_path, - ".dat", - false, - true, - &tx, - true, - &mut next_report_target, - report_interval, - &mut throttler, + &CopyFileSpec { + is_ec_volume: false, + collection: &req.collection, + volume_id: req.volume_id, + compaction_revision: vol_info.compaction_revision, + stop_offset: vol_info.dat_file_size, + dest_path: &dat_path, + ext: ".dat", + is_append: false, + ignore_source_not_found: true, + report_progress: true, + }, + &mut progress, ) .await?; if dat_modified_ts_ns > 0 { @@ -2007,20 +2015,19 @@ impl VolumeServer for VolumeGrpcService { let idx_path = format!("{}.idx", idx_base_name); let idx_modified_ts_ns = copy_file_from_source( &mut client, - false, - &req.collection, - req.volume_id, - vol_info.compaction_revision, - vol_info.idx_file_size, - &idx_path, - ".idx", - false, - false, - &tx, - false, - &mut next_report_target, - report_interval, - &mut throttler, + &CopyFileSpec { + is_ec_volume: false, + collection: &req.collection, + volume_id: req.volume_id, + compaction_revision: vol_info.compaction_revision, + stop_offset: vol_info.idx_file_size, + dest_path: &idx_path, + ext: ".idx", + is_append: false, + ignore_source_not_found: false, + report_progress: false, + }, + &mut progress, ) .await?; if idx_modified_ts_ns > 0 { @@ -2031,20 +2038,19 @@ impl VolumeServer for VolumeGrpcService { let vif_path = format!("{}.vif", data_base_name); let vif_modified_ts_ns = copy_file_from_source( &mut client, - false, - &req.collection, - req.volume_id, - vol_info.compaction_revision, - 1024 * 1024, - &vif_path, - ".vif", - false, - true, - &tx, - false, - &mut next_report_target, - report_interval, - &mut throttler, + &CopyFileSpec { + is_ec_volume: false, + collection: &req.collection, + volume_id: req.volume_id, + compaction_revision: vol_info.compaction_revision, + stop_offset: 1024 * 1024, + dest_path: &vif_path, + ext: ".vif", + is_append: false, + ignore_source_not_found: true, + report_progress: false, + }, + &mut progress, ) .await?; if vif_modified_ts_ns > 0 { @@ -4123,16 +4129,18 @@ impl VolumeServer for VolumeGrpcService { // dat_file_size. The decoder infers the layout from the shard size // when .vif does not record it. // Write .dat file using block-interleaved reading from shards. - crate::storage::erasure_coding::ec_decoder::write_dat_file_from_shards_with_dirs( - &dat_dir, - &collection, - vid, - dat_file_size, - vif_dat_file_size, - data_shards, - &per_shard_dirs, - large_block_size as usize, - small_block_size as usize, + crate::storage::erasure_coding::ec_decoder::write_dat_file_from_shards( + &crate::storage::erasure_coding::ec_decoder::DatRebuild { + dat_dir: &dat_dir, + collection: &collection, + volume_id: vid, + dat_file_size, + encoded_dat_file_size: vif_dat_file_size, + data_shards, + shard_dirs: Some(&per_shard_dirs), + large_block_size: large_block_size as usize, + small_block_size: small_block_size as usize, + }, ) .map_err(|e| Status::internal(format!("WriteDatFile: {}", e)))?; @@ -5498,25 +5506,36 @@ async fn drain_copy_stream_to_file( } } -/// Copy a file from a remote volume server via CopyFile streaming RPC. -/// Returns the modified_ts_ns received from the source. -#[expect(clippy::too_many_arguments)] -async fn copy_file_from_source( - client: &mut volume_server_pb::volume_server_client::VolumeServerClient, +/// One file of a volume copy: what to ask the source for and where it lands. +#[derive(Clone, Copy)] +struct CopyFileSpec<'a> { is_ec_volume: bool, - collection: &str, + collection: &'a str, volume_id: u32, compaction_revision: u32, stop_offset: u64, - dest_path: &str, - ext: &str, + dest_path: &'a str, + ext: &'a str, is_append: bool, ignore_source_not_found: bool, - progress_tx: &tokio::sync::mpsc::Sender>, + /// Whether this file's bytes are reported back to the caller as progress. report_progress: bool, - next_report_target: &mut i64, +} + +/// Progress reporting and throttling shared by every file of one volume copy. +struct CopyProgress<'a> { + tx: &'a tokio::sync::mpsc::Sender>, + next_report_target: &'a mut i64, report_interval: i64, - throttler: &mut WriteThrottler, + throttler: &'a mut WriteThrottler, +} + +/// Copy a file from a remote volume server via CopyFile streaming RPC. +/// Returns the modified_ts_ns received from the source. +async fn copy_file_from_source( + client: &mut volume_server_pb::volume_server_client::VolumeServerClient, + spec: &CopyFileSpec<'_>, + progress: &mut CopyProgress<'_>, ) -> Result where T: tonic::client::GrpcService, @@ -5524,6 +5543,19 @@ where T::ResponseBody: http_body::Body + Send + 'static, ::Error: Into + Send, { + let CopyFileSpec { + is_ec_volume, + collection, + volume_id, + compaction_revision, + stop_offset, + dest_path, + ext, + is_append, + ignore_source_not_found, + report_progress, + } = *spec; + let progress_tx = progress.tx; let copy_req = volume_server_pb::CopyFileRequest { volume_id, ext: ext.to_string(), @@ -5617,11 +5649,11 @@ where // A throttled copy sleeps seconds at a time; wake for a departing // caller instead of finishing the nap first. tokio::select! { - _ = throttler.maybe_slowdown(resp.file_content.len() as i64) => {} + _ = progress.throttler.maybe_slowdown(resp.file_content.len() as i64) => {} _ = progress_tx.closed() => return Err(cancelled()), } - if report_progress && progressed_bytes > *next_report_target { + if report_progress && progressed_bytes > *progress.next_report_target { // Go aborts the transfer when this send fails // (volume_grpc_copy.go: `return false`); so do we. if progress_tx @@ -5634,7 +5666,7 @@ where { return Err(cancelled()); } - *next_report_target = progressed_bytes + report_interval; + *progress.next_report_target = progressed_bytes + progress.report_interval; } } } @@ -6035,13 +6067,9 @@ mod tests { let mut volume = crate::storage::volume::Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &crate::storage::volume::VolumeSpec::default(), ) .unwrap(); let mut needle = Needle { @@ -6208,12 +6236,12 @@ mod tests { store .add_volume( VolumeId(1), - collection, - None, - ttl, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection, + ttl, + ..Default::default() + }, ) .unwrap(); { @@ -6606,20 +6634,24 @@ mod tests { let mut throttler = WriteThrottler::new(0); let err = copy_file_from_source( &mut client, - false, - "", - 1, - u32::MAX, - dat_bytes.len() as u64, - &dest_path, - ".dat", - false, - true, - &tx, - true, - &mut next_report_target, - 128 * 1024 * 1024, - &mut throttler, + &CopyFileSpec { + is_ec_volume: false, + collection: "", + volume_id: 1, + compaction_revision: u32::MAX, + stop_offset: dat_bytes.len() as u64, + dest_path: &dest_path, + ext: ".dat", + is_append: false, + ignore_source_not_found: true, + report_progress: true, + }, + &mut CopyProgress { + tx: &tx, + next_report_target: &mut next_report_target, + report_interval: 128 * 1024 * 1024, + throttler: &mut throttler, + }, ) .await .expect_err("a copy whose caller is gone must not run to completion"); @@ -8041,13 +8073,9 @@ mod tests { let mut v = crate::storage::volume::Volume::new( src_s, src_s, - "", vid, NeedleMapKind::InMemory, - None, - None, - 0, - crate::storage::types::Version::current(), + &crate::storage::volume::VolumeSpec::default(), ) .unwrap(); for i in 1..=8u64 { diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 36003be96..61f36ec7a 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -1274,7 +1274,6 @@ async fn get_or_head_handler_inner( && let Some(resp) = try_expand_chunk_manifest( &state, &n, - &headers, &method, &path, &query, @@ -3231,11 +3230,9 @@ struct ChunkInfo { } /// Try to expand a chunk manifest needle. Returns None if manifest can't be parsed. -#[expect(clippy::too_many_arguments)] async fn try_expand_chunk_manifest( state: &Arc, n: &Needle, - _headers: &HeaderMap, method: &Method, path: &str, query: &ReadQueryParams, diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 2bd3eb28f..cd97cc041 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -1247,7 +1247,8 @@ mod tests { use crate::remote_storage::s3_tier::S3TierRegistry; use crate::security::{Guard, SigningKey}; use crate::storage::needle_map::NeedleMapKind; - use crate::storage::types::{DiskType, Version, VolumeId}; + use crate::storage::types::{DiskType, VolumeId}; + use crate::storage::volume::VolumeSpec; use std::sync::atomic::Ordering; use std::sync::RwLock; use std::time::{SystemTime, UNIX_EPOCH}; @@ -1360,12 +1361,11 @@ mod tests { store .add_volume( VolumeId(7), - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); @@ -1410,12 +1410,11 @@ mod tests { store .add_volume( VolumeId(7), - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); @@ -1476,12 +1475,11 @@ mod tests { store .add_volume( VolumeId(id), - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); } @@ -1530,12 +1528,11 @@ mod tests { store .add_volume( VolumeId(3), - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); @@ -1591,12 +1588,11 @@ mod tests { store .add_volume( VolumeId(3), - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); @@ -1634,12 +1630,11 @@ mod tests { store .add_volume( vid, - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); } @@ -1673,12 +1668,11 @@ mod tests { store .add_volume( VolumeId(17), - "heartbeat_metrics_case", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "heartbeat_metrics_case", + ..Default::default() + }, ) .unwrap(); store.locations[0] @@ -1776,12 +1770,11 @@ mod tests { store .add_volume( VolumeId(21), - collection, - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection, + ..Default::default() + }, ) .unwrap(); { @@ -1949,12 +1942,13 @@ mod tests { store .add_volume( VolumeId(41), - "expired_volume_case", - None, - Some(crate::storage::needle::ttl::TTL::read("20m").unwrap()), - 1024, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "expired_volume_case", + ttl: Some(crate::storage::needle::ttl::TTL::read("20m").unwrap()), + preallocate: 1024, + ..Default::default() + }, ) .unwrap(); let dat_path = { @@ -2005,12 +1999,11 @@ mod tests { store .add_volume( VolumeId(51), - "io_error_case", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "io_error_case", + ..Default::default() + }, ) .unwrap(); let (_, volume) = store.find_volume_mut(VolumeId(51)).unwrap(); @@ -2046,12 +2039,11 @@ mod tests { store .add_volume( VolumeId(71), - "remote_volume_case", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "remote_volume_case", + ..Default::default() + }, ) .unwrap(); let (_, volume) = store.find_volume_mut(VolumeId(71)).unwrap(); diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 0e8b27566..931d90916 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -186,15 +186,19 @@ pub async fn read_ec_shard_needle_distributed( } => { fetch_one_interval( state, - vid, - needle_id, - shard_id, - shard_offset, - size, - shard_locations, - data_shards, - parity_shards, - encode_ts_ns, + EcInterval { + vid, + needle_id, + shard_id, + shard_offset, + size, + expected_encode_ts_ns: encode_ts_ns, + }, + EcShardMap { + locations: shard_locations, + data_shards, + parity_shards, + }, ) .await } @@ -521,18 +525,15 @@ pub async fn scrub_ec_volume_distributed( } => { let sources: &[String] = locations.get(shard_id).map(Vec::as_slice).unwrap_or(&[]); - match read_remote_ec_shard_interval( - state, - sources, + let iv = EcInterval { vid, - id, - *shard_id, - *shard_offset, - *ssize, - snapshot.encode_ts_ns, - ) - .await - { + needle_id: id, + shard_id: *shard_id, + shard_offset: *shard_offset, + size: *ssize, + expected_encode_ts_ns: snapshot.encode_ts_ns, + }; + match read_remote_ec_shard_interval(state, sources, iv).await { // A deleted shard yields no bytes; zero-fill the interval so // the assembled needle reaches read_bytes -> SizeMismatch{0} // -> the delete-state suppression (mirrors Go's pre-zeroed buffer). @@ -560,15 +561,12 @@ pub async fn scrub_ec_volume_distributed( } match recover_one_remote_ec_shard_interval( state, - vid, - id, - *shard_id, - *shard_offset, - *ssize, - &locations, - data_shards, - total_shards - data_shards, - snapshot.encode_ts_ns, + iv, + EcShardMap { + locations: &locations, + data_shards, + parity_shards: total_shards - data_shards, + }, ) .await { @@ -965,37 +963,44 @@ fn format_location_as_server_address(loc: &master_pb::Location) -> String { raw.to_string() } -/// Try direct peer read; on failure, reconstruct via Reed-Solomon -/// from the other shards. Mirrors `readOneEcShardInterval`'s tail. -#[expect(clippy::too_many_arguments)] -async fn fetch_one_interval( - state: &Arc, +/// One shard-relative byte range of a needle on an EC volume, the unit the +/// peer-read and recovery paths work in. Mirrors the argument list of Go's +/// `readOneEcShardInterval`. +#[derive(Clone, Copy, Debug)] +struct EcInterval { vid: VolumeId, needle_id: NeedleId, + /// The shard the bytes live on, or the one to rebuild when recovering. shard_id: ShardId, shard_offset: i64, size: usize, - shard_locations: &HashMap>, + /// Encode run the caller expects the shard to belong to; 0 accepts any, + /// for peers that predate the identity check. + expected_encode_ts_ns: i64, +} + +/// Where the shards of one EC volume can be fetched from, and the volume's +/// Reed-Solomon shape, as recovery needs both together. +#[derive(Clone, Copy)] +struct EcShardMap<'a> { + locations: &'a HashMap>, data_shards: usize, parity_shards: usize, - expected_encode_ts_ns: i64, +} + +/// 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, + iv: EcInterval, + map: EcShardMap<'_>, ) -> io::Result<(Vec, bool)> { + let EcInterval { vid, shard_id, .. } = iv; // Direct peer read against the cached locations for this shard. - if let Some(sources) = shard_locations.get(&shard_id) + if let Some(sources) = map.locations.get(&shard_id) && !sources.is_empty() { - match read_remote_ec_shard_interval( - state, - sources, - vid, - needle_id, - shard_id, - shard_offset, - size, - expected_encode_ts_ns, - ) - .await - { + match read_remote_ec_shard_interval(state, sources, iv).await { // A deleted needle short-circuits: don't reconstruct (every shard // would report deleted), let the caller return "deleted". Ok((buf, is_deleted)) => return Ok((buf, is_deleted)), @@ -1016,46 +1021,17 @@ async fn fetch_one_interval( // 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, - expected_encode_ts_ns, - ) - .await + recover_one_remote_ec_shard_interval(state, iv, map).await } -#[expect(clippy::too_many_arguments)] async fn read_remote_ec_shard_interval( state: &Arc, sources: &[String], - vid: VolumeId, - needle_id: NeedleId, - shard_id: ShardId, - shard_offset: i64, - size: usize, - expected_encode_ts_ns: i64, + iv: EcInterval, ) -> io::Result<(Vec, bool)> { 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, - expected_encode_ts_ns, - ) - .await - { + match do_read_remote_ec_shard_interval(state, src, iv).await { Ok(res) => return Ok(res), Err(e) => last_err = Some(e), } @@ -1063,22 +1039,24 @@ async fn read_remote_ec_shard_interval( Err(last_err.unwrap_or_else(|| { io::Error::new( io::ErrorKind::NotFound, - format!("no source for ec shard {}.{}", vid.0, shard_id), + format!("no source for ec shard {}.{}", iv.vid.0, iv.shard_id), ) })) } -#[expect(clippy::too_many_arguments)] 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, - expected_encode_ts_ns: i64, + iv: EcInterval, ) -> io::Result<(Vec, bool)> { + let EcInterval { + vid, + needle_id, + shard_id, + shard_offset, + size, + expected_encode_ts_ns, + } = iv; 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()) @@ -1171,19 +1149,24 @@ async fn do_read_remote_ec_shard_interval( Ok((out, false)) } -#[expect(clippy::too_many_arguments)] 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, - expected_encode_ts_ns: i64, + iv: EcInterval, + map: EcShardMap<'_>, ) -> io::Result<(Vec, bool)> { + let EcInterval { + vid, + needle_id, + shard_id: shard_id_to_recover, + shard_offset, + size, + expected_encode_ts_ns, + } = iv; + let EcShardMap { + locations: shard_locations, + data_shards, + parity_shards, + } = map; let total_shards = data_shards + parity_shards; let rs = ReedSolomon::new(data_shards, parity_shards) .map_err(|e| io::Error::other(format!("reed-solomon init: {:?}", e)))?; @@ -1272,12 +1255,10 @@ async fn recover_one_remote_ec_shard_interval( let res = read_remote_ec_shard_interval( &state, &locs, - vid, - needle_id, - sid, - shard_offset, - size, - expected_encode_ts_ns, + EcInterval { + shard_id: sid, + ..iv + }, ) .await; (sid, res) diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index d4becf2b1..ea31595e2 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -20,10 +20,10 @@ use crate::storage::erasure_coding::ec_shard::{ }; use crate::storage::erasure_coding::ec_volume::EcVolume; use crate::storage::needle_map::NeedleMapKind; -use crate::storage::super_block::{ReplicaPlacement, SUPER_BLOCK_SIZE}; +use crate::storage::super_block::SUPER_BLOCK_SIZE; use crate::storage::types::*; use crate::storage::volume::{ - remove_volume_files, volume_file_name, VifVolumeInfo, Volume, VolumeError, + VifVolumeInfo, Volume, VolumeError, VolumeSpec, remove_volume_files, volume_file_name, }; /// A single disk location managing volumes in one directory. @@ -280,30 +280,33 @@ impl DiskLocation { let opened = Mutex::new(Vec::with_capacity(to_load.len())); std::thread::scope(|scope| { for _ in 0..workers { - scope.spawn(|| loop { - let i = next.fetch_add(1, Ordering::Relaxed); - let Some((vid, collections)) = to_load.get(i) else { - return; - }; - for collection in collections { - match Volume::new( - &self.directory, - &self.idx_directory, - collection, - *vid, - needle_map_kind, - None, // replica placement read from superblock - None, // TTL read from superblock - 0, // no preallocate on load - Version::current(), - ) { - Ok(mut v) => { - v.location_disk_space_low = self.is_disk_space_low.clone(); - opened.lock().unwrap().push((collection.clone(), *vid, v)); - break; - } - Err(e) => { - warn!(volume_id = vid.0, error = %e, "failed to load volume"); + scope.spawn(|| { + loop { + let i = next.fetch_add(1, Ordering::Relaxed); + let Some((vid, collections)) = to_load.get(i) else { + return; + }; + for collection in collections { + // Replica placement and TTL are read back from the + // superblock, and a load never preallocates. + match Volume::new( + &self.directory, + &self.idx_directory, + *vid, + needle_map_kind, + &VolumeSpec { + collection, + ..Default::default() + }, + ) { + Ok(mut v) => { + v.location_disk_space_low = self.is_disk_space_low.clone(); + opened.lock().unwrap().push((collection.clone(), *vid, v)); + break; + } + Err(e) => { + warn!(volume_id = vid.0, error = %e, "failed to load volume"); + } } } } @@ -547,31 +550,22 @@ impl DiskLocation { } /// Create a new volume in this location. - #[expect(clippy::too_many_arguments)] pub fn create_volume( &mut self, vid: VolumeId, - collection: &str, needle_map_kind: NeedleMapKind, - replica_placement: Option, - ttl: Option, - preallocate: u64, - version: Version, + spec: &VolumeSpec<'_>, ) -> Result<(), VolumeError> { let mut v = Volume::new( &self.directory, &self.idx_directory, - collection, vid, needle_map_kind, - replica_placement, - ttl, - preallocate, - version, + spec, )?; v.location_disk_space_low = self.is_disk_space_low.clone(); crate::metrics::VOLUME_GAUGE - .with_label_values(&[collection, "volume"]) + .with_label_values(&[spec.collection, "volume"]) .inc(); self.volumes.insert(vid, v); Ok(()) @@ -1504,16 +1498,8 @@ mod tests { ) .unwrap(); - loc.create_volume( - VolumeId(1), - "", - NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), - ) - .unwrap(); + loc.create_volume(VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default()) + .unwrap(); assert_eq!(loc.volumes_len(), 1); assert!(loc.find_volume(VolumeId(1)).is_some()); @@ -1537,24 +1523,15 @@ mod tests { Vec::new(), ) .unwrap(); - loc.create_volume( - VolumeId(1), - "", - NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), - ) - .unwrap(); + loc.create_volume(VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default()) + .unwrap(); loc.create_volume( VolumeId(2), - "test", NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection: "test", + ..Default::default() + }, ) .unwrap(); loc.close(); @@ -1597,12 +1574,11 @@ mod tests { .unwrap(); loc.create_volume( VolumeId(9), - "good", NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection: "good", + ..Default::default() + }, ) .unwrap(); loc.close(); @@ -1646,26 +1622,10 @@ mod tests { ) .unwrap(); - loc.create_volume( - VolumeId(1), - "", - NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), - ) - .unwrap(); - loc.create_volume( - VolumeId(2), - "", - NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), - ) - .unwrap(); + loc.create_volume(VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default()) + .unwrap(); + loc.create_volume(VolumeId(2), NeedleMapKind::InMemory, &VolumeSpec::default()) + .unwrap(); assert_eq!(loc.volumes_len(), 2); loc.delete_volume(VolumeId(1), false, false).unwrap(); @@ -1689,32 +1649,29 @@ mod tests { loc.create_volume( VolumeId(1), - "pics", NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); loc.create_volume( VolumeId(2), - "pics", NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); loc.create_volume( VolumeId(3), - "docs", NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection: "docs", + ..Default::default() + }, ) .unwrap(); assert_eq!(loc.volumes_len(), 3); diff --git a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs index 0a1f36ada..02619489b 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs @@ -67,73 +67,45 @@ pub fn find_dat_file_size_with_dirs( Ok(dat_size) } -/// Reconstruct a .dat file from EC data shards. -/// -/// Reads from .ec00-.ec09 and writes a new .dat file. All data shards -/// must live in `dir`. For the cross-disk reconciled layout where -/// shards are split across multiple data dirs of the same node, use -/// [`write_dat_file_from_shards_with_dirs`] instead. -#[expect(clippy::too_many_arguments)] -pub fn write_dat_file_from_shards( - dir: &str, - collection: &str, - volume_id: VolumeId, - dat_file_size: i64, - encoded_dat_file_size: i64, - data_shards: usize, - large_block_size: usize, - small_block_size: usize, -) -> io::Result<()> { - let dirs: Vec = (0..data_shards).map(|_| dir.to_string()).collect(); - write_dat_file_from_shards_with_dirs( - dir, - collection, - volume_id, - dat_file_size, - encoded_dat_file_size, - data_shards, - &dirs, - large_block_size, - small_block_size, - ) -} - -/// Reconstruct a .dat file from EC data shards, taking the source -/// directory for each shard separately. -/// -/// `dat_dir` is where the produced `.dat` is written. `shard_dirs[i]` -/// is the directory holding shard `i`. For the simple "all shards in -/// one dir" case both can be the same value. +/// What it takes to rebuild a volume's .dat from its EC data shards. /// /// Mirrors Go's `WriteDatFile(baseFileName, datFileSize, /// encodedDatFileSize, shardFileNames)` shape — Go passes per-shard /// paths so a reconciled volume with shards split across disks of the /// same volume server can still be decoded back to a regular .dat /// (seaweedfs/seaweedfs#9252). +#[derive(Clone, Copy, Debug)] +pub struct DatRebuild<'a> { + /// Where the produced `.dat` is written. + pub dat_dir: &'a str, + pub collection: &'a str, + pub volume_id: VolumeId, + /// The number of bytes to write, i.e. the live data extent from + /// [`find_dat_file_size`]. + pub dat_file_size: i64, + /// The .dat size at encode time, which fixed the shard block layout: + /// deletions can move the live extent below the large-block row + /// boundary, and deriving the layout from the shrunk extent would read + /// the shards in the wrong block order. Zero when the .vif does not + /// record the encode-time size; the layout is then inferred from the + /// shard size. + pub encoded_dat_file_size: i64, + pub data_shards: usize, + /// `shard_dirs[i]` is the directory holding shard `i`. `None` means every + /// data shard sits in `dat_dir`. + pub shard_dirs: Option<&'a [String]>, + /// The volume's shard block layout, e.g. `EcVolume::large_block_size()` + /// / `small_block_size()` from its .vif EC config. + pub large_block_size: usize, + pub small_block_size: usize, +} + +/// Reconstruct a .dat file from EC data shards. /// -/// `dat_file_size` is the number of bytes to write, i.e. the live data -/// extent from [`find_dat_file_size`]. `encoded_dat_file_size` is the -/// .dat size at encode time, which fixed the shard block layout: -/// deletions can move the live extent below the large-block row -/// boundary, and deriving the layout from the shrunk extent would read -/// the shards in the wrong block order. Pass zero when the .vif does -/// not record the encode-time size to infer the layout from the shard -/// size. `large_block_size`/`small_block_size` are the volume's shard -/// block layout, e.g. `EcVolume::large_block_size()` / -/// `small_block_size()` from its .vif EC config. -#[expect(clippy::too_many_arguments)] -pub fn write_dat_file_from_shards_with_dirs( - dat_dir: &str, - collection: &str, - volume_id: VolumeId, - dat_file_size: i64, - encoded_dat_file_size: i64, - data_shards: usize, - shard_dirs: &[String], - large_block_size: usize, - small_block_size: usize, -) -> io::Result<()> { - write_dat_file( +/// Reads from .ec00-.ec09 and writes a new .dat file, from one directory or +/// from the per-shard directories of a cross-disk reconciled volume. +pub fn write_dat_file_from_shards(spec: &DatRebuild<'_>) -> io::Result<()> { + let DatRebuild { dat_dir, collection, volume_id, @@ -143,21 +115,15 @@ pub fn write_dat_file_from_shards_with_dirs( shard_dirs, large_block_size, small_block_size, - ) -} - -#[expect(clippy::too_many_arguments)] -fn write_dat_file( - dat_dir: &str, - collection: &str, - volume_id: VolumeId, - dat_file_size: i64, - encoded_dat_file_size: i64, - data_shards: usize, - shard_dirs: &[String], - large_block_size: usize, - small_block_size: usize, -) -> io::Result<()> { + } = *spec; + let same_dir: Vec; + let shard_dirs: &[String] = match shard_dirs { + Some(dirs) => dirs, + None => { + same_dir = vec![dat_dir.to_string(); data_shards]; + &same_dir + } + }; if data_shards == 0 { return Err(io::Error::new( io::ErrorKind::InvalidInput, @@ -370,7 +336,7 @@ mod tests { use crate::storage::erasure_coding::ec_encoder; use crate::storage::needle::needle::Needle; use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; use tempfile::TempDir; #[test] @@ -382,13 +348,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -428,16 +390,17 @@ mod tests { std::fs::remove_file(format!("{}/1.idx", dir)).unwrap(); // Reconstruct from EC shards - write_dat_file_from_shards( - dir, - "", - VolumeId(1), - original_dat_size as i64, - original_dat_size as i64, + write_dat_file_from_shards(&DatRebuild { + dat_dir: dir, + collection: "", + volume_id: VolumeId(1), + dat_file_size: original_dat_size as i64, + encoded_dat_file_size: original_dat_size as i64, data_shards, - block_size as usize, - block_size as usize, - ) + shard_dirs: None, + large_block_size: block_size as usize, + small_block_size: block_size as usize, + }) .unwrap(); write_idx_file_from_ec_index(dir, "", VolumeId(1)).unwrap(); @@ -457,13 +420,9 @@ mod tests { let v2 = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -483,16 +442,17 @@ mod tests { let dir = tmp.path().to_str().unwrap(); // No shard files exist, so de-striping must fail and publish nothing: // neither the final .dat nor a partial .dat.tmp may remain. - let res = write_dat_file_from_shards( - dir, - "", - VolumeId(7), - 100, - 100, - 10, - ERASURE_CODING_LARGE_BLOCK_SIZE, - ERASURE_CODING_SMALL_BLOCK_SIZE, - ); + let res = write_dat_file_from_shards(&DatRebuild { + dat_dir: dir, + collection: "", + volume_id: VolumeId(7), + dat_file_size: 100, + encoded_dat_file_size: 100, + data_shards: 10, + shard_dirs: None, + large_block_size: ERASURE_CODING_LARGE_BLOCK_SIZE, + small_block_size: ERASURE_CODING_SMALL_BLOCK_SIZE, + }); assert!(res.is_err()); assert!(!std::path::Path::new(&format!("{}/7.dat", dir)).exists()); assert!(!std::path::Path::new(&format!("{}/7.dat.tmp", dir)).exists()); @@ -543,11 +503,13 @@ mod tests { &rs, &mut shards, &mut builders, - data_shards, - parity_shards, - SMALL, - LARGE, - SMALL, + ec_encoder::EcEncodeLayout { + data_shards, + parity_shards, + buffer_size: SMALL, + large_block_size: LARGE, + small_block_size: SMALL, + }, ) .unwrap(); for shard in &mut shards { @@ -565,7 +527,17 @@ mod tests { -> io::Result> { let out = format!("{}/{}", dir, sub); std::fs::create_dir_all(&out).unwrap(); - write_dat_file(&out, "", VolumeId(1), live, encoded, 10, shard_dirs, LARGE, SMALL)?; + write_dat_file_from_shards(&DatRebuild { + dat_dir: &out, + collection: "", + volume_id: VolumeId(1), + dat_file_size: live, + encoded_dat_file_size: encoded, + data_shards: 10, + shard_dirs: Some(shard_dirs), + large_block_size: LARGE, + small_block_size: SMALL, + })?; Ok(std::fs::read(format!("{}/1.dat", out)).unwrap()) }; @@ -635,11 +607,13 @@ mod tests { &rs, &mut shards, &mut builders, - data_shards, - parity_shards, - SMALL, - LARGE, - SMALL, + ec_encoder::EcEncodeLayout { + data_shards, + parity_shards, + buffer_size: SMALL, + large_block_size: LARGE, + small_block_size: SMALL, + }, ) .unwrap(); for shard in &mut shards { @@ -655,17 +629,17 @@ mod tests { std::fs::create_dir(&out_dir).unwrap(); let out = out_dir.to_str().unwrap(); let decode = |live_size: i64, encoded_size: i64| -> Vec { - write_dat_file( - out, - "", - VolumeId(1), - live_size, - encoded_size, + write_dat_file_from_shards(&DatRebuild { + dat_dir: out, + collection: "", + volume_id: VolumeId(1), + dat_file_size: live_size, + encoded_dat_file_size: encoded_size, data_shards, - &shard_dirs, - LARGE, - SMALL, - ) + shard_dirs: Some(&shard_dirs), + large_block_size: LARGE, + small_block_size: SMALL, + }) .unwrap(); let path = format!("{}/1.dat", out); let decoded = std::fs::read(&path).unwrap(); @@ -700,17 +674,19 @@ mod tests { assert_ne!(&original[..(large_row_size / 2) as usize], &control[..]); // the live extent can never exceed the encode-time size - assert!(write_dat_file( - out, - "", - VolumeId(1), - dat_size + 1, - dat_size, - data_shards, - &shard_dirs, - LARGE, - SMALL, - ) - .is_err()); + assert!( + write_dat_file_from_shards(&DatRebuild { + dat_dir: out, + collection: "", + volume_id: VolumeId(1), + dat_file_size: dat_size + 1, + encoded_dat_file_size: dat_size, + data_shards, + shard_dirs: Some(&shard_dirs), + large_block_size: LARGE, + small_block_size: SMALL, + }) + .is_err() + ); } } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs index 1c069c08d..a5d62fb08 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs @@ -77,11 +77,13 @@ pub fn write_ec_files( &rs, &mut shards, &mut builders, - data_shards, - parity_shards, - ENCODE_BUFFER_SIZE, - block_size as usize, - block_size as usize, + EcEncodeLayout { + data_shards, + parity_shards, + buffer_size: ENCODE_BUFFER_SIZE, + large_block_size: block_size as usize, + small_block_size: block_size as usize, + }, )?; // Close all shards @@ -702,30 +704,50 @@ fn read_from_data_shards( /// the uniform block is. const ENCODE_BUFFER_SIZE: usize = 256 * 1024; +/// Shape of one encode run: the Reed-Solomon split and the block sizes that +/// fix where every byte of the .dat lands in the shards. Mirrors Go's +/// `ECContext`. `buffer_size` must divide both block sizes. +#[derive(Clone, Copy, Debug)] +pub(crate) struct EcEncodeLayout { + pub(crate) data_shards: usize, + pub(crate) parity_shards: usize, + /// Bytes of each shard's block handled per sub-batch; bounds memory at + /// `total_shards * buffer_size` however large the blocks are. + pub(crate) buffer_size: usize, + pub(crate) large_block_size: usize, + pub(crate) small_block_size: usize, +} + /// Encode the .dat file data into shard files. /// /// Uses a two-phase approach matching Go's ec_encoder.go: /// 1. Process as many large blocks as possible /// 2. Process remaining data with small blocks -/// -/// `buffer_size` must divide both block sizes. -#[expect(clippy::too_many_arguments)] pub(crate) fn encode_dat_file( dat_file: &File, dat_size: i64, rs: &ReedSolomon, shards: &mut [EcVolumeShard], builders: &mut [ShardChecksumBuilder], - data_shards: usize, - parity_shards: usize, - buffer_size: usize, - large_block_size: usize, - small_block_size: usize, + layout: EcEncodeLayout, ) -> io::Result<()> { + let EcEncodeLayout { + data_shards, + parity_shards, + buffer_size, + large_block_size, + small_block_size, + } = layout; let total_shards = data_shards + parity_shards; - let mut buffers: Vec> = (0..total_shards) - .map(|_| vec![0u8; buffer_size]) - .collect(); + let mut buffers: Vec> = (0..total_shards).map(|_| vec![0u8; buffer_size]).collect(); + let mut run = EncodeRun { + dat_file, + rs, + buffers: &mut buffers, + shards, + builders, + data_shards, + }; let mut remaining = dat_size; let mut offset: u64 = 0; @@ -734,16 +756,7 @@ pub(crate) fn encode_dat_file( let large_row_size = large_block_size * data_shards; while remaining >= large_row_size as i64 { - encode_data( - dat_file, - offset, - large_block_size, - rs, - &mut buffers, - shards, - builders, - data_shards, - )?; + run.encode_row(offset, large_block_size)?; offset += large_row_size as u64; remaining -= large_row_size as i64; } @@ -753,16 +766,7 @@ pub(crate) fn encode_dat_file( while remaining > 0 { let to_process = remaining.min(small_row_size as i64); - encode_data( - dat_file, - offset, - small_block_size, - rs, - &mut buffers, - shards, - builders, - data_shards, - )?; + run.encode_row(offset, small_block_size)?; offset += to_process as u64; remaining -= to_process; } @@ -770,79 +774,66 @@ pub(crate) fn encode_dat_file( Ok(()) } -/// Encode one row of blocks, streaming it in ENCODE_BUFFER_SIZE sub-batches so -/// arbitrarily large blocks never require block-sized allocations. Mirrors -/// Go's encodeData. -#[expect(clippy::too_many_arguments)] -fn encode_data( - dat_file: &File, - row_offset: u64, - block_size: usize, - rs: &ReedSolomon, - buffers: &mut [Vec], - shards: &mut [EcVolumeShard], - builders: &mut [ShardChecksumBuilder], +/// Everything one encode run streams through: the source .dat, the codec, a +/// buffer per shard, and the per-shard file and checksum sinks. +struct EncodeRun<'a> { + dat_file: &'a File, + rs: &'a ReedSolomon, + buffers: &'a mut [Vec], + shards: &'a mut [EcVolumeShard], + builders: &'a mut [ShardChecksumBuilder], data_shards: usize, -) -> io::Result<()> { - let buffer_size = buffers[0].len(); - if !block_size.is_multiple_of(buffer_size) { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!( - "unexpected block size {} buffer size {}", - block_size, buffer_size - ), - )); - } - let batch_count = block_size / buffer_size; - for b in 0..batch_count { - encode_one_batch( - dat_file, - row_offset + (b * buffer_size) as u64, - block_size, - rs, - buffers, - shards, - builders, - data_shards, - )?; - } - Ok(()) } -/// Encode one sub-batch: the same buffer-sized slice of every shard's block in -/// this row. Mirrors Go's encodeDataOneBatch. -#[expect(clippy::too_many_arguments)] -fn encode_one_batch( - dat_file: &File, - offset: u64, - block_size: usize, - rs: &ReedSolomon, - buffers: &mut [Vec], - shards: &mut [EcVolumeShard], - builders: &mut [ShardChecksumBuilder], - data_shards: usize, -) -> io::Result<()> { - // Read data shards from the .dat file, zero-filling past EOF — the buffers - // are reused across batches, so the tail must be cleared explicitly. - for (i, buf) in buffers[..data_shards].iter_mut().enumerate() { - let read_offset = offset + (i * block_size) as u64; - let n = read_at_most(dat_file, buf, read_offset)?; - buf[n..].fill(0); +impl EncodeRun<'_> { + /// Encode one row of blocks, streaming it in ENCODE_BUFFER_SIZE sub-batches + /// so arbitrarily large blocks never require block-sized allocations. + /// Mirrors Go's encodeData. + fn encode_row(&mut self, row_offset: u64, block_size: usize) -> io::Result<()> { + let buffer_size = self.buffers[0].len(); + if !block_size.is_multiple_of(buffer_size) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!( + "unexpected block size {} buffer size {}", + block_size, buffer_size + ), + )); + } + let batch_count = block_size / buffer_size; + for b in 0..batch_count { + self.encode_one_batch(row_offset + (b * buffer_size) as u64, block_size)?; + } + Ok(()) } - // Encode parity shards - rs.encode(&mut *buffers) - .map_err(|e| io::Error::other(format!("reed-solomon encode: {:?}", e)))?; + /// Encode one sub-batch: the same buffer-sized slice of every shard's block + /// in this row. Mirrors Go's encodeDataOneBatch. + fn encode_one_batch(&mut self, offset: u64, block_size: usize) -> io::Result<()> { + // Read data shards from the .dat file, zero-filling past EOF — the + // buffers are reused across batches, so the tail must be cleared + // explicitly. + for (i, buf) in self.buffers[..self.data_shards].iter_mut().enumerate() { + let read_offset = offset + (i * block_size) as u64; + let n = read_at_most(self.dat_file, buf, read_offset)?; + buf[n..].fill(0); + } - // Write all shard buffers to files and feed the same bytes to each - // shard's bitrot checksum builder, keeping covered_size == on-disk length. - for (i, buf) in buffers.iter().enumerate() { - shards[i].write_all(buf)?; - builders[i].write(buf); + // Encode parity shards + self.rs + .encode(&mut *self.buffers) + .map_err(|e| io::Error::other(format!("reed-solomon encode: {:?}", e)))?; + + // Write all shard buffers to files and feed the same bytes to each + // shard's bitrot checksum builder, keeping covered_size == on-disk + // length. + for (i, buf) in self.buffers.iter().enumerate() { + self.shards[i].write_all(buf)?; + self.builders[i].write(buf); + } + + Ok(()) } - - Ok(()) } /// Read into `buf` at `offset` until it is full or EOF; returns bytes read. @@ -873,7 +864,7 @@ mod tests { use super::*; use crate::storage::needle::needle::Needle; use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; use tempfile::TempDir; #[test] @@ -885,13 +876,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -936,13 +923,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=n { @@ -1015,13 +998,9 @@ mod tests { let mut v = Volume::new( &dir, &dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=20 { @@ -1207,19 +1186,15 @@ mod tests { #[test] fn test_rebuild_ecx_file_uniform_layout() { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap().to_string(); let mut v = Volume::new( &dir, &dir, - "", VolumeId(2), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1u64..=12 { @@ -1257,19 +1232,15 @@ mod tests { #[test] fn test_rebuild_ecx_file_fails_on_truncated_shard() { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap().to_string(); let mut v = Volume::new( &dir, &dir, - "", VolumeId(3), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1u64..=12 { @@ -1367,13 +1338,9 @@ mod tests { let mut v = Volume::new( dat_dir, idx_dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -1451,13 +1418,9 @@ mod tests { let mut v = Volume::new( dat_dir, idx_dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -1489,13 +1452,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", vid, NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=8 { diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 18bb59099..e14e6008b 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -1721,20 +1721,19 @@ mod tests { #[test] fn test_destroy_removes_bitrot_sidecar() { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); let mut v = Volume::new( dir, dir, - "ec1c", VolumeId(2074), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection: "ec1c", + ..Default::default() + }, ) .unwrap(); for i in 1..=3 { @@ -1797,20 +1796,16 @@ mod tests { fn test_mount_loads_bitrot_sidecar() { use crate::storage::erasure_coding::ec_bitrot::BitrotStatus; use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=5 { @@ -1855,20 +1850,16 @@ mod tests { #[test] fn test_scrub_plans_are_self_contained_and_match_direct_call() { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=8 { @@ -1948,20 +1939,16 @@ mod tests { #[test] fn test_local_scrub_plan_reports_negative_size_ecx_row() { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=8 { @@ -2040,20 +2027,16 @@ mod tests { #[test] fn test_scrub_plans_survive_files_removed_after_snapshot() { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=8 { @@ -2139,20 +2122,16 @@ mod tests { #[test] fn test_checksum_scrub_clean_and_detects_corruption() { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=8 { @@ -2589,18 +2568,14 @@ mod tests { /// split-disk mount without needing N directories. fn split_runtimes(dir: &str, vid: VolumeId, subsets: &[&[u8]]) -> Vec { use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; let mut v = Volume::new( dir, dir, - "", vid, NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for i in 1..=8 { @@ -3307,7 +3282,7 @@ mod tests { mod uniform_layout_tests { use super::*; use crate::storage::needle_map::NeedleMapKind; - use crate::storage::volume::{VifEcShardConfig, VifVolumeInfo, Volume}; + use crate::storage::volume::{VifEcShardConfig, VifVolumeInfo, Volume, VolumeSpec}; use tempfile::TempDir; // Write ~26MB of needles so the uniform block size (3MB) diverges from the @@ -3324,13 +3299,9 @@ mod uniform_layout_tests { let mut v = Volume::new( dir, dir, - "", vid, NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); let mut expected: Vec<(NeedleId, Vec)> = Vec::new(); @@ -3383,11 +3354,13 @@ mod uniform_layout_tests { &rs, &mut shards, &mut builders, - 10, - 4, - 256 * 1024, - ERASURE_CODING_LARGE_BLOCK_SIZE, - ERASURE_CODING_SMALL_BLOCK_SIZE, + crate::storage::erasure_coding::ec_encoder::EcEncodeLayout { + data_shards: 10, + parity_shards: 4, + buffer_size: 256 * 1024, + large_block_size: ERASURE_CODING_LARGE_BLOCK_SIZE, + small_block_size: ERASURE_CODING_SMALL_BLOCK_SIZE, + }, ) .unwrap(); for shard in &mut shards { diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index b04299093..43e49fbc1 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -18,7 +18,7 @@ use crate::storage::needle::needle::Needle; use crate::storage::needle_map::NeedleMapKind; use crate::storage::super_block::ReplicaPlacement; use crate::storage::types::*; -use crate::storage::volume::{VifVolumeInfo, VolumeError}; +use crate::storage::volume::{VifVolumeInfo, VolumeError, VolumeSpec}; /// Top-level storage manager containing all disk locations and their volumes. pub struct Store { @@ -369,16 +369,11 @@ impl Store { } /// Create a new volume, placing it on the location with the most free space. - #[expect(clippy::too_many_arguments)] pub fn add_volume( &mut self, vid: VolumeId, - collection: &str, - replica_placement: Option, - ttl: Option, - preallocate: u64, disk_type: DiskType, - version: Version, + spec: &VolumeSpec<'_>, ) -> Result<(), VolumeError> { if self.find_volume(vid).is_some() { return Err(VolumeError::AlreadyExists); @@ -390,15 +385,7 @@ impl Store { ))) })?; - self.locations[loc_idx].create_volume( - vid, - collection, - self.needle_map_kind, - replica_placement, - ttl, - preallocate, - version, - ) + self.locations[loc_idx].create_volume(vid, self.needle_map_kind, spec) } /// Delete a volume from any location. When keep_remote_data is true the @@ -480,12 +467,11 @@ impl Store { } return loc.create_volume( vid, - collection, self.needle_map_kind, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection, + ..Default::default() + }, ); } } @@ -573,12 +559,11 @@ impl Store { // keep scanning (matches open_volumes / Go mountVolume). match loc.create_volume( vid, - collection, self.needle_map_kind, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection, + ..Default::default() + }, ) { Ok(()) => return Ok(()), Err(e) => { @@ -629,12 +614,11 @@ impl Store { let loc = &mut self.locations[loc_idx]; match loc.create_volume( vid, - &collection, self.needle_map_kind, - None, - None, - 0, - Version::current(), + &VolumeSpec { + collection: &collection, + ..Default::default() + }, ) { Ok(()) => return Ok(()), Err(e) => { @@ -1639,15 +1623,7 @@ mod tests { let mut store = make_test_store(&[dir]); store - .add_volume( - VolumeId(1), - "", - None, - None, - 0, - DiskType::HardDrive, - Version::current(), - ) + .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default()) .unwrap(); assert!(store.has_volume(VolumeId(1))); assert!(!store.has_volume(VolumeId(2))); @@ -1709,12 +1685,11 @@ mod tests { store .add_volume( VolumeId(7), - "coll", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "coll", + ..Default::default() + }, ) .unwrap(); // Write a needle so the volume has real data, then unmount it so the @@ -1756,12 +1731,11 @@ mod tests { store .add_volume( VolumeId(9), - "foo..bar", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "foo..bar", + ..Default::default() + }, ) .unwrap(); let mut n = Needle { @@ -1800,12 +1774,11 @@ mod tests { store .add_volume( VolumeId(11), - "coll", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "coll", + ..Default::default() + }, ) .unwrap(); let mut n = Needle { @@ -1856,12 +1829,11 @@ mod tests { store .add_volume( VolumeId(13), - "coll", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "coll", + ..Default::default() + }, ) .unwrap(); let mut n = Needle { @@ -1904,16 +1876,17 @@ mod tests { let mut store = make_test_store(&[dir0, dir1]); // Force add_volume onto disk 1 by marking disk 0 as low on space. - store.locations[0].is_disk_space_low.store(true, Ordering::Relaxed); + store.locations[0] + .is_disk_space_low + .store(true, Ordering::Relaxed); store .add_volume( VolumeId(15), - "coll", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "coll", + ..Default::default() + }, ) .unwrap(); let mut n = Needle { @@ -1965,15 +1938,7 @@ mod tests { let dir = tmp.path().to_str().unwrap(); let mut store = make_test_store(&[dir]); store - .add_volume( - VolumeId(1), - "", - None, - None, - 0, - DiskType::HardDrive, - Version::current(), - ) + .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default()) .unwrap(); // Write @@ -2039,15 +2004,7 @@ mod tests { ) .unwrap(); store - .add_volume( - VolumeId(1), - "", - None, - None, - 0, - DiskType::HardDrive, - Version::current(), - ) + .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default()) .unwrap(); let mut n = Needle { id: NeedleId(1), @@ -2110,26 +2067,10 @@ mod tests { // Add volumes — should go to location with fewest volumes store - .add_volume( - VolumeId(1), - "", - None, - None, - 0, - DiskType::HardDrive, - Version::current(), - ) + .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default()) .unwrap(); store - .add_volume( - VolumeId(2), - "", - None, - None, - 0, - DiskType::HardDrive, - Version::current(), - ) + .add_volume(VolumeId(2), DiskType::HardDrive, &VolumeSpec::default()) .unwrap(); assert_eq!(store.total_volume_count(), 2); @@ -2147,34 +2088,31 @@ mod tests { store .add_volume( VolumeId(1), - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); store .add_volume( VolumeId(2), - "pics", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "pics", + ..Default::default() + }, ) .unwrap(); store .add_volume( VolumeId(3), - "docs", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "docs", + ..Default::default() + }, ) .unwrap(); assert_eq!(store.total_volume_count(), 3); @@ -2210,23 +2148,21 @@ mod tests { store .add_volume( VolumeId(61), - "preallocate_case", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "preallocate_case", + ..Default::default() + }, ) .unwrap(); store .add_volume( VolumeId(62), - "preallocate_case", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "preallocate_case", + ..Default::default() + }, ) .unwrap(); for vid in [VolumeId(61), VolumeId(62)] { @@ -2327,12 +2263,11 @@ mod tests { store .add_volume( VolumeId(71), - "find_free_location_case", - None, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + collection: "find_free_location_case", + ..Default::default() + }, ) .unwrap(); diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index b4ba0a680..35558f41b 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -610,20 +610,48 @@ fn read_exact_at(file: &File, buf: &mut [u8], mut offset: u64) -> io::Result<()> Ok(()) } +/// What a volume is created with beyond its id, directories and index kind: +/// the tail of Go's `NewVolume` argument list. The default is an empty +/// collection with no replication, no TTL and no preallocation at the +/// current needle version, which is what most tests want. +#[derive(Clone, Copy, Debug)] +pub struct VolumeSpec<'a> { + pub collection: &'a str, + pub replica_placement: Option, + pub ttl: Option, + /// Bytes to reserve for the .dat up front; 0 grows on demand. + pub preallocate: u64, + pub version: Version, +} + +impl Default for VolumeSpec<'_> { + fn default() -> Self { + VolumeSpec { + collection: "", + replica_placement: None, + ttl: None, + preallocate: 0, + version: Version::current(), + } + } +} + impl Volume { /// Create and load a volume from disk. - #[expect(clippy::too_many_arguments)] pub fn new( dirname: &str, dir_idx: &str, - collection: &str, id: VolumeId, needle_map_kind: NeedleMapKind, - replica_placement: Option, - ttl: Option, - preallocate: u64, - version: Version, + spec: &VolumeSpec<'_>, ) -> Result { + let VolumeSpec { + collection, + replica_placement, + ttl, + preallocate, + version, + } = *spec; let mut v = Volume { id, dir: dirname.to_string(), @@ -4624,13 +4652,9 @@ mod tests { Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap() } @@ -5215,13 +5239,9 @@ mod tests { let v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); assert!( @@ -5370,13 +5390,12 @@ mod tests { Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - Some(ttl), - 0, - Version::current(), + &VolumeSpec { + ttl: Some(ttl), + ..Default::default() + }, ) .unwrap() } @@ -5409,13 +5428,9 @@ mod tests { Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap() } @@ -5455,13 +5470,12 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - VERSION_2, + &VolumeSpec { + version: VERSION_2, + ..Default::default() + }, ) .unwrap(); for i in 1..=3u64 { @@ -5799,13 +5813,9 @@ mod tests { let v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); assert_eq!(v.file_count(), 3); @@ -5826,13 +5836,9 @@ mod tests { Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::Redb, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap() }; @@ -5876,13 +5882,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::Redb, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); let rdb_path = std::path::PathBuf::from(v.file_name(".rdb")); @@ -5933,13 +5935,9 @@ mod tests { let mut v = Volume::new( data, data, - "", VolumeId(7), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); let payload = b"payload-across-relocate".to_vec(); @@ -5995,13 +5993,9 @@ mod tests { let mut v = Volume::new( data, data, - "", VolumeId(7), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); let payload = b"payload-beside-the-data".to_vec(); @@ -6021,13 +6015,9 @@ mod tests { let reopened = Volume::new( data, idx, - "", VolumeId(7), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); assert!( @@ -6643,13 +6633,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -6746,13 +6732,9 @@ mod tests { Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap() } @@ -6851,13 +6833,9 @@ mod tests { let v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); let mut probe = Needle { @@ -6944,13 +6922,9 @@ mod tests { let loaded = Volume::new( &dir, &dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ); set_mode(0o755); // restore before any assertion, so cleanup works @@ -7012,17 +6986,18 @@ mod tests { let loaded = Volume::new( &dir, &dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ); let marked = loaded.map(|mut v| { let r = v.set_writable(); - (v.no_write_or_delete, v.is_read_only(), v.nm.as_ref().unwrap().has_idx_writer(), r) + ( + v.no_write_or_delete, + v.is_read_only(), + v.nm.as_ref().unwrap().has_idx_writer(), + r, + ) }); // Restore before any assertion, so cleanup works. set_mode(tmp.path(), 0o755); @@ -7072,13 +7047,9 @@ mod tests { let mut v = Volume::new( &dir, &dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); assert_eq!( @@ -7103,13 +7074,9 @@ mod tests { let v = Volume::new( &dir, &dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for (id, want) in [(1u64, &b"first"[..]), (2u64, &b"second"[..])] { @@ -7163,13 +7130,9 @@ mod tests { let loaded = Volume::new( &dir, &dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ); set_mode(tmp.path(), 0o755); set_mode(&idx_path, 0o644); @@ -7255,13 +7218,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -7373,13 +7332,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); assert!( @@ -7418,13 +7373,9 @@ mod tests { let v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); let mut probe = Needle { @@ -7488,13 +7439,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); assert!(v.no_write_can_delete, "state must survive a restart"); @@ -7570,13 +7517,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); assert!(v.no_write_or_delete); @@ -7627,13 +7570,9 @@ mod tests { let v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -7667,13 +7606,9 @@ mod tests { let v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -7704,13 +7639,9 @@ mod tests { let result = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ); match result { @@ -7792,13 +7723,9 @@ mod tests { let v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -7891,13 +7818,9 @@ mod tests { let mut v = Volume::new( dat_dir, idx_dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); diff --git a/seaweed-volume/src/storage/volume_idx_rebuild.rs b/seaweed-volume/src/storage/volume_idx_rebuild.rs index d7fbb2188..36f09a037 100644 --- a/seaweed-volume/src/storage/volume_idx_rebuild.rs +++ b/seaweed-volume/src/storage/volume_idx_rebuild.rs @@ -109,7 +109,7 @@ mod tests { use crate::storage::needle::Needle; use crate::storage::needle_map::NeedleMapKind; use crate::storage::types::*; - use crate::storage::volume::Volume; + use crate::storage::volume::{Volume, VolumeSpec}; use std::fs; use std::path::Path; use tempfile::TempDir; @@ -145,13 +145,9 @@ mod tests { let mut v = Volume::new( data, old_idx, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); for id in 1..=3 { @@ -167,13 +163,9 @@ mod tests { let reopened = Volume::new( data, new_idx, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); @@ -207,13 +199,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); v.write_needle(&mut needle(1), true, false).unwrap(); @@ -233,13 +221,9 @@ mod tests { let reopened = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); drop(reopened); @@ -261,13 +245,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); v.write_needle(&mut needle(1), true, false).unwrap(); @@ -290,13 +270,9 @@ mod tests { let reopened = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); drop(reopened); @@ -318,13 +294,9 @@ mod tests { let mut v = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); v.write_needle(&mut needle(1), true, false).unwrap(); @@ -356,13 +328,9 @@ mod tests { let reopened = Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); drop(reopened); diff --git a/seaweed-volume/src/storage/volume_idx_repair.rs b/seaweed-volume/src/storage/volume_idx_repair.rs index d4dfef3fa..f57be9087 100644 --- a/seaweed-volume/src/storage/volume_idx_repair.rs +++ b/seaweed-volume/src/storage/volume_idx_repair.rs @@ -215,6 +215,7 @@ mod tests { use super::*; use crate::storage::needle::crc::CRC; use crate::storage::needle_map::NeedleMapKind; + use crate::storage::volume::VolumeSpec; use std::os::unix::fs::{FileExt, PermissionsExt}; use tempfile::TempDir; @@ -222,13 +223,9 @@ mod tests { Volume::new( dir, dir, - "", VolumeId(1), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap() } diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs index 7a2bea586..accb74bd1 100644 --- a/seaweed-volume/tests/http_integration.rs +++ b/seaweed-volume/tests/http_integration.rs @@ -17,7 +17,8 @@ use seaweed_volume::server::volume_server::{ }; use seaweed_volume::storage::needle_map::NeedleMapKind; use seaweed_volume::storage::store::Store; -use seaweed_volume::storage::types::{DiskType, Version, VolumeId}; +use seaweed_volume::storage::types::{DiskType, VolumeId}; +use seaweed_volume::storage::volume::VolumeSpec; use tempfile::TempDir; @@ -73,12 +74,11 @@ fn build_test_state( store .add_volume( VolumeId(1), - "", - replica_placement, - None, - 0, DiskType::HardDrive, - Version::current(), + &VolumeSpec { + replica_placement, + ..Default::default() + }, ) .expect("failed to create volume"); @@ -959,7 +959,7 @@ async fn chunk_manifest_expands_chunk_stored_on_ec_volume() { use seaweed_volume::storage::erasure_coding::ec_encoder::write_ec_files; use seaweed_volume::storage::needle::needle::{FileId, Needle}; use seaweed_volume::storage::types::{Cookie, NeedleId}; - use seaweed_volume::storage::volume::Volume; + use seaweed_volume::storage::volume::{Volume, VolumeSpec}; let (state, tmp) = test_state(); let dir = tmp.path().to_str().unwrap(); @@ -976,13 +976,9 @@ async fn chunk_manifest_expands_chunk_stored_on_ec_volume() { let mut v = Volume::new( dir, dir, - "", VolumeId(2), NeedleMapKind::InMemory, - None, - None, - 0, - Version::current(), + &VolumeSpec::default(), ) .unwrap(); let mut n = Needle {