From 517f60e875781540cd9a50f67c7e303084b8ab61 Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Tue, 15 Sep 2026 19:04:16 +0300 Subject: [PATCH] =?UTF-8?q?rust-volume:=20fold=20the=208=E2=80=9315-argume?= =?UTF-8?q?nt=20functions=20into=20parameter=20structs=20(#11328)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * rust: migrate seaweed-volume and seaweed-worker to tonic 0.14 / prost 0.14 tonic 0.14 boxes the contents of tonic::Status, which is what made every RPC path trip clippy's result_large_err; the allow for that lint goes in the next commit. The prost codec moved out of tonic into tonic-prost and tonic-prost-build, so both build scripts now call tonic_prost_build::configure() and both crates depend on tonic-prost for the generated code. The `tls` feature was split into a per-backend feature; `tls-aws-lc` is the same backend both crates already install through rustls::crypto::aws_lc_rs. tonic 0.14 depends on axum 0.8 and tower 0.5, which would have left a second axum and a second tower in each tree next to the 0.7 / 0.4 the crates named themselves. Bumping them keeps one copy of each: axum 0.8 only changes the path-parameter syntax for the routes here (`/:vid` -> `/{vid}`, `/*path` -> `/{*path}`), tower 0.5 needs the `util` feature named explicitly for ServiceExt::oneshot (it used to arrive through tonic's feature unification), and tower-http 0.6 is the matching release. Lock files move only through cargo's own resolution for the new versions; no other dependency was refreshed. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust: drop the result_large_err allow now that tonic::Status is boxed tonic 0.14 stores Status behind a Box, so Result<_, Status> is no longer a large-Err type and clippy has nothing to say about it. Both crates pass `cargo clippy --all-targets -- -D warnings` without the allow (seaweed-volume in both feature sets), so the policy entry and its comment go. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: drop the unused headers argument of try_expand_chunk_manifest The parameter was already named `_headers`; nothing in the body reads it. With it gone the function is under clippy's argument threshold and the expect goes. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: pass EC peer reads an EcInterval instead of ten arguments fetch_one_interval, read_remote_ec_shard_interval, do_read_remote_ec_shard_interval and recover_one_remote_ec_shard_interval all took the same (vid, needle_id, shard_id, shard_offset, size, expected_encode_ts_ns) tuple, and the two that reconstruct also took the location map with the data/parity counts. Those are now EcInterval (Copy) and EcShardMap (a borrow of the map plus the counts). The fan-out inside recovery builds its per-shard request with `EcInterval { shard_id: sid, ..iv }`, which is the one place the old argument list was easy to get wrong. Bodies destructure at the top, so the code below the signatures is unchanged. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: give the EC encoder an EcEncodeLayout and an EncodeRun encode_dat_file took the Reed-Solomon shape and three block sizes as five loose integers; they are now one Copy struct, EcEncodeLayout, which is what Go calls ECContext. The per-row and per-batch helpers took the same six sinks and the offsets; they become methods on EncodeRun, which owns the borrows for one run, so each call names only the offset and block size that vary. The byte-level work is unchanged. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: describe a .dat rebuild with DatRebuild instead of nine arguments write_dat_file_from_shards, its _with_dirs twin and the private write_dat_file were three layers over one nine-argument signature. One public function now takes a DatRebuild, whose shard_dirs is None when every shard sits beside the .dat and Some(dirs) for the cross-disk reconciled layout. The field docs carry what the function doc used to say about the encode-time size and the block layout. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: split copy_file_from_source's fifteen arguments into two structs CopyFileSpec is the per-file request (what to ask the source for, where it lands, whether its bytes count as progress); CopyProgress is the sender, throttler and report state that all three files of one VolumeCopy share, held by &mut across the calls. The three production call sites now read as the .dat/.idx/.vif literals they are, instead of positional trues and falses. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: create volumes from a VolumeSpec Volume::new, DiskLocation::create_volume and Store::add_volume each took the same five-value tail of Go's NewVolume argument list: collection, replica placement, TTL, preallocation and needle version. That tail is now VolumeSpec, a Copy struct whose Default is what almost every test wanted anyway (empty collection, no replication, no TTL, no preallocation, current version), so most of the 104 call sites shrink to `&VolumeSpec::default()` or name the one field they set. The id, directories, index kind and disk type stay positional because they differ at every site. Two imports that only test modules use moved into those modules, and DiskLocation no longer imports ReplicaPlacement. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU --------- Co-authored-by: Claude Fable 5.1 --- seaweed-volume/src/server/grpc_server.rs | 228 +++++++++------- seaweed-volume/src/server/handlers.rs | 3 - seaweed-volume/src/server/heartbeat.rs | 104 ++++---- seaweed-volume/src/server/store_ec.rs | 191 ++++++------- seaweed-volume/src/storage/disk_location.rs | 163 +++++------ .../src/storage/erasure_coding/ec_decoder.rs | 252 ++++++++---------- .../src/storage/erasure_coding/ec_encoder.rs | 249 ++++++++--------- .../src/storage/erasure_coding/ec_volume.rs | 79 ++---- seaweed-volume/src/storage/store.rs | 199 +++++--------- seaweed-volume/src/storage/volume.rs | 227 ++++++---------- .../src/storage/volume_idx_rebuild.rs | 50 +--- .../src/storage/volume_idx_repair.rs | 7 +- seaweed-volume/tests/http_integration.rs | 20 +- 13 files changed, 727 insertions(+), 1045 deletions(-) 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 {