mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-20 15:04:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

rust-volume: fold the 8–15-argument functions into parameter structs (#11328)

Browse Source 
* 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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Eliah Rusin
2026-09-15 09:04:16 -07:00
committed by GitHub
co-authored by Claude Fable 5.1
parent 49a680dd64
commit 517f60e875
13 changed files with 727 additions and 1045 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+128 -100
View File
@@ -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<T>(
client: &mut volume_server_pb::volume_server_client::VolumeServerClient<T>,
/// 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<Result<volume_server_pb::VolumeCopyResponse, Status>>,
/// 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<Result<volume_server_pb::VolumeCopyResponse, Status>>,
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<T>(
client: &mut volume_server_pb::volume_server_client::VolumeServerClient<T>,
spec: &CopyFileSpec<'_>,
progress: &mut CopyProgress<'_>,
) -> Result<i64, Status>
where
T: tonic::client::GrpcService<tonic::body::Body>,
@@ -5524,6 +5543,19 @@ where
T::ResponseBody: http_body::Body<Data = bytes::Bytes> + Send + 'static,
<T::ResponseBody as http_body::Body>::Error: Into<tonic::codegen::StdError> + 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 {
seaweed-volume/src/server/handlers.rs
-3
View File
@@ -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<VolumeServerState>,
n: &Needle,
_headers: &HeaderMap,
method: &Method,
path: &str,
query: &ReadQueryParams,
seaweed-volume/src/server/heartbeat.rs
+48 -56
View File
@@ -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();
seaweed-volume/src/server/store_ec.rs
+86 -105
View File
@@ -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<VolumeServerState>,
/// 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<ShardId, Vec<String>>,
/// 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<ShardId, Vec<String>>,
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<VolumeServerState>,
iv: EcInterval,
map: EcShardMap<'_>,
) -> io::Result<(Vec<u8>, 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<VolumeServerState>,
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<u8>, bool)> {
let mut last_err: Option<io::Error> = 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<VolumeServerState>,
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<u8>, 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<VolumeServerState>,
vid: VolumeId,
needle_id: NeedleId,
shard_id_to_recover: ShardId,
shard_offset: i64,
size: usize,
shard_locations: &HashMap<ShardId, Vec<String>>,
data_shards: usize,
parity_shards: usize,
expected_encode_ts_ns: i64,
iv: EcInterval,
map: EcShardMap<'_>,
) -> io::Result<(Vec<u8>, 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)
seaweed-volume/src/storage/disk_location.rs
+60 -103
View File
@@ -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<ReplicaPlacement>,
ttl: Option<crate::storage::needle::ttl::TTL>,
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);
seaweed-volume/src/storage/erasure_coding/ec_decoder.rs
+114 -138
View File
@@ -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<String> = (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<String>;
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<Vec<u8>> {
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<u8> {
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()
);
}
}
seaweed-volume/src/storage/erasure_coding/ec_encoder.rs
+104 -145
View File
@@ -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<Vec<u8>> = (0..total_shards)
.map(|_| vec![0u8; buffer_size])
.collect();
let mut buffers: Vec<Vec<u8>> = (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<u8>],
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<u8>],
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<u8>],
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 {
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+26 -53
View File
@@ -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<EcVolume> {
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<u8>)> = 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 {
seaweed-volume/src/storage/store.rs
+67 -132
View File
@@ -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<ReplicaPlacement>,
ttl: Option<crate::storage::needle::ttl::TTL>,
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();
seaweed-volume/src/storage/volume.rs
+75 -152
View File
@@ -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<ReplicaPlacement>,
pub ttl: Option<crate::storage::needle::ttl::TTL>,
/// 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<ReplicaPlacement>,
ttl: Option<crate::storage::needle::ttl::TTL>,
preallocate: u64,
version: Version,
spec: &VolumeSpec<'_>,
) -> Result<Self, VolumeError> {
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();
seaweed-volume/src/storage/volume_idx_rebuild.rs
+9 -41
View File
@@ -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);
seaweed-volume/src/storage/volume_idx_repair.rs
+2 -5
View File
@@ -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()
}
seaweed-volume/tests/http_integration.rs
+8 -12
View File
@@ -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 {