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: cargo fmt both crates, with a commented-out fmt --check CI step (#11329)

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

* rust-worker: run cargo fmt

Layout only; no token in the workspace changes.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU

* rust-volume: run cargo fmt

Layout only; no token in the crate changes. Every earlier Rust PR here
formatted only the blocks it touched so as not to drown its diff in
this one, and this commit is that debt paid in a single place. rustfmt
needed two passes to settle one block in handlers.rs; the committed
form is the fixed point, so `cargo fmt --check` is clean.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU

* ci: add a commented-out cargo fmt --check step to both Rust workflows

Same shape as the commented clippy step from #11312: the check is
written out so that making formatting a gate is a one-line uncomment,
and whether to do that stays a maintainer call.

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>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
This commit is contained in:
Eliah Rusin
2026-09-15 09:29:22 -07:00
committed by GitHub
co-authored by Claude Fable 5.1 Chris Lu
parent 517f60e875
commit 166af06a2b
55 changed files with 970 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/rust-volume-server-tests.yml
+5
View File
@@ -65,6 +65,11 @@ jobs:
# - name: Clippy
# run: cd seaweed-volume && cargo clippy --all-targets -- -D warnings
# The crate is rustfmt-clean as of the PR that added this step.
# Uncomment to keep it that way.
# - name: Check formatting
# run: cd seaweed-volume && cargo fmt --check
- name: Run Rust unit tests
run: cd seaweed-volume && cargo test
.github/workflows/rust-worker-tests.yml
+5
View File
@@ -79,6 +79,11 @@ jobs:
# - name: Clippy
# run: cd seaweed-worker && cargo clippy --workspace --all-targets -- -D warnings
# The workspace is rustfmt-clean as of the PR that added this step.
# Uncomment to keep it that way.
# - name: Check formatting
# run: cd seaweed-worker && cargo fmt --all --check
# The tests that need a live gateway skip themselves without one, the way
# the Go integration tests skip without Docker; the lifecycle suite in
# test/s3tables/lifecycle is what runs them against a real cluster.
seaweed-volume/src/config.rs
+3 -1
View File
@@ -987,7 +987,9 @@ pub fn parse_security_config(path: &str) -> SecurityConfig {
},
Section::JwtSigning => match key {
"key" => cfg.jwt_signing_key = value.as_bytes().to_vec(),
"expires_after_seconds" => cfg.jwt_signing_expires = value.parse().unwrap_or(10),
"expires_after_seconds" => {
cfg.jwt_signing_expires = value.parse().unwrap_or(10)
}
_ => {}
},
Section::HttpsClient => match key {
seaweed-volume/src/main.rs
+30 -27
View File
@@ -6,8 +6,8 @@ use seaweed_volume::config::{self, VolumeServerConfig};
use seaweed_volume::metrics;
use seaweed_volume::pb::volume_server_pb::volume_server_server::VolumeServerServer;
use seaweed_volume::security::tls::{
build_rustls_server_config, build_rustls_server_config_with_grpc_client_auth,
install_default_crypto_provider, GrpcClientAuthPolicy, TlsPolicy,
GrpcClientAuthPolicy, TlsPolicy, build_rustls_server_config,
build_rustls_server_config_with_grpc_client_auth, install_default_crypto_provider,
};
use seaweed_volume::security::{Guard, SigningKey};
#[cfg(unix)]
@@ -18,7 +18,7 @@ use seaweed_volume::server::grpc_server::VolumeGrpcService;
use seaweed_volume::server::profiling::CpuProfileSession;
use seaweed_volume::server::request_id::GrpcRequestIdLayer;
use seaweed_volume::server::volume_server::{
build_metrics_router, RuntimeMetricsConfig, VolumeServerState,
RuntimeMetricsConfig, VolumeServerState, build_metrics_router,
};
use seaweed_volume::server::write_queue::WriteQueue;
use seaweed_volume::storage::store::Store;
@@ -671,8 +671,7 @@ async fn run(
})
.await
} else {
let incoming =
tokio_stream::wrappers::TcpListenerStream::new(grpc_listener);
let incoming = tokio_stream::wrappers::TcpListenerStream::new(grpc_listener);
info!("gRPC server listening on {}", grpc_local_addr);
build_grpc_server_builder()
.layer(GrpcRequestIdLayer)
@@ -1058,15 +1057,17 @@ mod tests {
#[test]
fn test_grpc_server_tls_returns_none_when_files_are_missing() {
assert!(build_grpc_server_tls_acceptor(
"/missing/server.crt",
"/missing/server.key",
"/missing/ca.crt",
&TlsPolicy::default(),
"",
&[],
)
.is_none());
assert!(
build_grpc_server_tls_acceptor(
"/missing/server.crt",
"/missing/server.key",
"/missing/ca.crt",
&TlsPolicy::default(),
"",
&[],
)
.is_none()
);
}
#[test]
@@ -1088,19 +1089,21 @@ mod tests {
"-----BEGIN CERTIFICATE-----\nZmFrZQ==\n-----END CERTIFICATE-----\n",
);
assert!(build_grpc_server_tls_acceptor(
&cert,
&key,
&ca,
&TlsPolicy {
min_version: "TLS 1.0".to_string(),
max_version: "TLS 1.1".to_string(),
cipher_suites: String::new(),
},
"",
&[],
)
.is_none());
assert!(
build_grpc_server_tls_acceptor(
&cert,
&key,
&ca,
&TlsPolicy {
min_version: "TLS 1.0".to_string(),
max_version: "TLS 1.1".to_string(),
cipher_suites: String::new(),
},
"",
&[],
)
.is_none()
);
}
#[test]
seaweed-volume/src/metrics.rs
+4 -2
View File
@@ -525,7 +525,7 @@ pub async fn push_metrics_once(
#[cfg(test)]
mod tests {
use super::*;
use axum::{routing::put, Router};
use axum::{Router, routing::put};
use std::sync::{Arc, Mutex};
#[test]
@@ -597,7 +597,9 @@ mod tests {
register_metrics();
VOLUME_GAUGE.with_label_values(&["pics", "volume"]).set(2.0);
VOLUME_GAUGE.with_label_values(&["pics", "ec_shards"]).set(3.0);
VOLUME_GAUGE
.with_label_values(&["pics", "ec_shards"])
.set(3.0);
READ_ONLY_VOLUME_GAUGE
.with_label_values(&["pics", "volume"])
.set(1.0);
seaweed-volume/src/remote_storage/endpoint_guard.rs
+186 -106
View File
@@ -119,7 +119,11 @@ pub fn check_blocked_ip(endpoint: &str, ip: IpAddr) -> Result<(), String> {
/// reachable for callers whose target legitimately sits on an internal network
/// (peer volume servers), while still blocking loopback, link-local (IMDS) and
/// unspecified. Mirrors Go's `checkBlockedIPPolicy`.
pub fn check_blocked_ip_policy(endpoint: &str, ip: IpAddr, allow_private: bool) -> Result<(), String> {
pub fn check_blocked_ip_policy(
endpoint: &str,
ip: IpAddr,
allow_private: bool,
) -> Result<(), String> {
// Normalize IPv4-mapped IPv6 (`::ffff:a.b.c.d`) to its IPv4 form so the
// IPv4 deny rules apply. The OS routes these to the embedded IPv4 address,
// so without this `::ffff:127.0.0.1` / `::ffff:169.254.169.254` would slip
@@ -231,7 +235,7 @@ fn precheck_endpoint(endpoint: &str) -> Result<HostCheck, String> {
return Err(format!(
"remote endpoint {:?} has a malformed IPv6 host",
endpoint
))
));
}
}
} else {
@@ -307,7 +311,10 @@ pub async fn validate_replica_target(target: &str) -> Result<(), String> {
return Err("replica target is empty".to_string());
}
if trimmed.contains("://") || trimmed.contains(['/', '?', '#', '@', '\\']) {
return Err(format!("replica target {:?} must be a bare host:port", target));
return Err(format!(
"replica target {:?} must be a bare host:port",
target
));
}
// Require an explicit host:port, handling `[IPv6]:port`. A bracketless IPv6
@@ -316,12 +323,22 @@ pub async fn validate_replica_target(target: &str) -> Result<(), String> {
let host = if let Some(rest) = trimmed.strip_prefix('[') {
match rest.split_once(']') {
Some((h, port)) if port.starts_with(':') && port.len() > 1 => h,
_ => return Err(format!("replica target {:?} must be a bare host:port", target)),
_ => {
return Err(format!(
"replica target {:?} must be a bare host:port",
target
));
}
}
} else {
match trimmed.rsplit_once(':') {
Some((h, port)) if !port.is_empty() && !h.contains(':') => h,
_ => return Err(format!("replica target {:?} must be a bare host:port", target)),
_ => {
return Err(format!(
"replica target {:?} must be a bare host:port",
target
));
}
}
};
@@ -340,7 +357,10 @@ pub async fn validate_replica_target(target: &str) -> Result<(), String> {
let addrs = resolve_host(host).await?;
if addrs.is_empty() {
return Err(format!("resolve replica target host {:?}: no addresses", host));
return Err(format!(
"resolve replica target host {:?}: no addresses",
host
));
}
for ip in addrs {
check_blocked_ip_policy(target, ip, true)?;
@@ -378,22 +398,30 @@ mod tests {
#[test]
fn rejects_empty_and_bad_scheme() {
assert!(precheck_endpoint("").unwrap_err().contains("empty"));
assert!(precheck_endpoint("ftp://example.com/")
.unwrap_err()
.contains("http or https"));
assert!(precheck_endpoint("example.com/")
.unwrap_err()
.contains("http or https"));
assert!(
precheck_endpoint("ftp://example.com/")
.unwrap_err()
.contains("http or https")
);
assert!(
precheck_endpoint("example.com/")
.unwrap_err()
.contains("http or https")
);
}
#[test]
fn rejects_imds_hostnames() {
assert!(precheck_endpoint("http://metadata.google.internal/")
.unwrap_err()
.contains("metadata service"));
assert!(precheck_endpoint("http://metadata/")
.unwrap_err()
.contains("metadata service"));
assert!(
precheck_endpoint("http://metadata.google.internal/")
.unwrap_err()
.contains("metadata service")
);
assert!(
precheck_endpoint("http://metadata/")
.unwrap_err()
.contains("metadata service")
);
}
#[test]
@@ -413,27 +441,41 @@ mod tests {
#[test]
fn check_blocked_ip_matches_resolved_categories() {
// Mirror Go's "host resolves to X" cases at the address level.
assert!(check_blocked_ip("e", ip("127.0.0.1"))
.unwrap_err()
.contains("loopback"));
assert!(check_blocked_ip("e", ip("169.254.10.20"))
.unwrap_err()
.contains("link-local"));
assert!(check_blocked_ip("e", ip("10.1.2.3"))
.unwrap_err()
.contains("private"));
assert!(check_blocked_ip("e", ip("172.20.0.5"))
.unwrap_err()
.contains("private"));
assert!(check_blocked_ip("e", ip("192.168.1.1"))
.unwrap_err()
.contains("private"));
assert!(check_blocked_ip("e", ip("100.64.0.42"))
.unwrap_err()
.contains("CGNAT"));
assert!(check_blocked_ip("e", ip("fc00::1"))
.unwrap_err()
.contains("private"));
assert!(
check_blocked_ip("e", ip("127.0.0.1"))
.unwrap_err()
.contains("loopback")
);
assert!(
check_blocked_ip("e", ip("169.254.10.20"))
.unwrap_err()
.contains("link-local")
);
assert!(
check_blocked_ip("e", ip("10.1.2.3"))
.unwrap_err()
.contains("private")
);
assert!(
check_blocked_ip("e", ip("172.20.0.5"))
.unwrap_err()
.contains("private")
);
assert!(
check_blocked_ip("e", ip("192.168.1.1"))
.unwrap_err()
.contains("private")
);
assert!(
check_blocked_ip("e", ip("100.64.0.42"))
.unwrap_err()
.contains("CGNAT")
);
assert!(
check_blocked_ip("e", ip("fc00::1"))
.unwrap_err()
.contains("private")
);
assert!(check_blocked_ip("e", ip("52.216.10.10")).is_ok());
assert!(check_blocked_ip("e", ip("2606:4700:4700::1111")).is_ok());
}
@@ -476,33 +518,45 @@ mod tests {
assert!(check_blocked_ip("e", ip("2001::f7f7:f7f7")).is_ok());
assert!(check_blocked_ip("e", ip("::808:808")).is_ok());
// Bracketed transition literal via the full endpoint path.
assert!(precheck_endpoint("http://[64:ff9b::a9fe:a9fe]/")
.unwrap_err()
.contains("metadata"));
assert!(
precheck_endpoint("http://[64:ff9b::a9fe:a9fe]/")
.unwrap_err()
.contains("metadata")
);
}
#[test]
fn rejects_ipv4_mapped_ipv6() {
// IPv4-mapped IPv6 must be unmapped so the IPv4 rules catch it.
assert!(check_blocked_ip("e", ip("::ffff:127.0.0.1"))
.unwrap_err()
.contains("loopback"));
assert!(check_blocked_ip("e", ip("::ffff:169.254.169.254"))
.unwrap_err()
.contains("metadata"));
assert!(check_blocked_ip("e", ip("::ffff:10.0.0.1"))
.unwrap_err()
.contains("private"));
assert!(
check_blocked_ip("e", ip("::ffff:127.0.0.1"))
.unwrap_err()
.contains("loopback")
);
assert!(
check_blocked_ip("e", ip("::ffff:169.254.169.254"))
.unwrap_err()
.contains("metadata")
);
assert!(
check_blocked_ip("e", ip("::ffff:10.0.0.1"))
.unwrap_err()
.contains("private")
);
// A mapped public address still passes, and genuine IPv6 loopback is
// still caught by the V6 path.
assert!(check_blocked_ip("e", ip("::ffff:52.216.10.10")).is_ok());
assert!(check_blocked_ip("e", ip("::1"))
.unwrap_err()
.contains("loopback"));
assert!(
check_blocked_ip("e", ip("::1"))
.unwrap_err()
.contains("loopback")
);
// Bracketed mapped literal via the full endpoint path.
assert!(precheck_endpoint("http://[::ffff:127.0.0.1]/")
.unwrap_err()
.contains("loopback"));
assert!(
precheck_endpoint("http://[::ffff:127.0.0.1]/")
.unwrap_err()
.contains("loopback")
);
}
#[test]
@@ -512,60 +566,86 @@ mod tests {
assert!(check_blocked_ip_policy("e", ip("192.168.1.5"), true).is_ok());
assert!(check_blocked_ip_policy("e", ip("100.64.0.42"), true).is_ok());
// Loopback / IMDS / unspecified stay blocked even when private is allowed.
assert!(check_blocked_ip_policy("e", ip("127.0.0.1"), true)
.unwrap_err()
.contains("loopback"));
assert!(check_blocked_ip_policy("e", ip("169.254.169.254"), true)
.unwrap_err()
.contains("metadata"));
assert!(check_blocked_ip_policy("e", ip("0.0.0.0"), true)
.unwrap_err()
.contains("unspecified"));
assert!(
check_blocked_ip_policy("e", ip("127.0.0.1"), true)
.unwrap_err()
.contains("loopback")
);
assert!(
check_blocked_ip_policy("e", ip("169.254.169.254"), true)
.unwrap_err()
.contains("metadata")
);
assert!(
check_blocked_ip_policy("e", ip("0.0.0.0"), true)
.unwrap_err()
.contains("unspecified")
);
}
#[tokio::test]
async fn validate_replica_target_rejects_and_allows() {
// A path plus a trailing ?a= would otherwise swallow ?type=replicate.
assert!(validate_replica_target("127.0.0.1:7000/status/x/?a=")
.await
.unwrap_err()
.contains("bare host:port"));
assert!(validate_replica_target("http://10.0.0.7:8080")
.await
.unwrap_err()
.contains("bare host:port"));
assert!(validate_replica_target("user@10.0.0.7:8080")
.await
.unwrap_err()
.contains("bare host:port"));
assert!(validate_replica_target("10.0.0.7")
.await
.unwrap_err()
.contains("bare host:port"));
assert!(validate_replica_target("peer.example.com")
.await
.unwrap_err()
.contains("bare host:port"));
assert!(validate_replica_target("127.0.0.1:8080")
.await
.unwrap_err()
.contains("loopback"));
assert!(validate_replica_target("[::1]:8080")
.await
.unwrap_err()
.contains("loopback"));
assert!(validate_replica_target("169.254.169.254:80")
.await
.unwrap_err()
.contains("metadata"));
assert!(validate_replica_target("metadata:80")
.await
.unwrap_err()
.contains("metadata"));
assert!(validate_replica_target("")
.await
.unwrap_err()
.contains("empty"));
assert!(
validate_replica_target("127.0.0.1:7000/status/x/?a=")
.await
.unwrap_err()
.contains("bare host:port")
);
assert!(
validate_replica_target("http://10.0.0.7:8080")
.await
.unwrap_err()
.contains("bare host:port")
);
assert!(
validate_replica_target("user@10.0.0.7:8080")
.await
.unwrap_err()
.contains("bare host:port")
);
assert!(
validate_replica_target("10.0.0.7")
.await
.unwrap_err()
.contains("bare host:port")
);
assert!(
validate_replica_target("peer.example.com")
.await
.unwrap_err()
.contains("bare host:port")
);
assert!(
validate_replica_target("127.0.0.1:8080")
.await
.unwrap_err()
.contains("loopback")
);
assert!(
validate_replica_target("[::1]:8080")
.await
.unwrap_err()
.contains("loopback")
);
assert!(
validate_replica_target("169.254.169.254:80")
.await
.unwrap_err()
.contains("metadata")
);
assert!(
validate_replica_target("metadata:80")
.await
.unwrap_err()
.contains("metadata")
);
assert!(
validate_replica_target("")
.await
.unwrap_err()
.contains("empty")
);
// Legitimate peer volume servers on private networks pass.
assert!(validate_replica_target("10.0.0.7:8080").await.is_ok());
assert!(validate_replica_target("192.168.1.5:8080").await.is_ok());
seaweed-volume/src/remote_storage/s3.rs
+1 -1
View File
@@ -2,9 +2,9 @@
//!
//! Works with AWS S3, MinIO, SeaweedFS S3, and all S3-compatible providers.
use aws_sdk_s3::Client;
use aws_sdk_s3::config::{BehaviorVersion, Credentials, Region};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::Client;
use super::{RemoteEntry, RemoteStorageClient, RemoteStorageError};
use crate::pb::remote_pb::{RemoteConf, RemoteStorageLocation};
seaweed-volume/src/remote_storage/s3_tier.rs
+1 -1
View File
@@ -7,9 +7,9 @@ use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, OnceLock, RwLock};
use aws_sdk_s3::Client;
use aws_sdk_s3::config::{BehaviorVersion, Credentials, Region};
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use aws_sdk_s3::Client;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
use tokio::sync::Semaphore;
seaweed-volume/src/security.rs
+6 -4
View File
@@ -10,7 +10,7 @@ use std::collections::HashSet;
use std::net::IpAddr;
use std::time::{SystemTime, UNIX_EPOCH};
use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};
use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation, decode, encode};
use serde::{Deserialize, Serialize};
// ============================================================================
@@ -481,9 +481,11 @@ mod tests {
let token = gen_jwt(&key, 3600, "3,01637037d6").unwrap();
// Correct file ID
assert!(guard
.check_jwt_for_file(Some(&token), "3,01637037d6", true)
.is_ok());
assert!(
guard
.check_jwt_for_file(Some(&token), "3,01637037d6", true)
.is_ok()
);
// Wrong file ID
let err = guard.check_jwt_for_file(Some(&token), "4,deadbeef", true);
seaweed-volume/src/security/tls.rs
+3 -3
View File
@@ -3,12 +3,12 @@ use std::fmt;
use std::sync::Arc;
use rustls::client::danger::HandshakeSignatureValid;
use rustls::crypto::aws_lc_rs;
use rustls::crypto::CryptoProvider;
use rustls::crypto::aws_lc_rs;
use rustls::pki_types::UnixTime;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use rustls::server::danger::{ClientCertVerified, ClientCertVerifier};
use rustls::server::WebPkiClientVerifier;
use rustls::server::danger::{ClientCertVerified, ClientCertVerifier};
use rustls::{
CipherSuite, DigitallySignedStruct, DistinguishedName, RootCertStore, ServerConfig,
SignatureScheme, SupportedCipherSuite, SupportedProtocolVersion,
@@ -376,7 +376,7 @@ fn go_tls_version_for_supported(version: &SupportedProtocolVersion) -> GoTlsVers
#[cfg(test)]
mod tests {
use super::{build_supported_versions, common_name_is_allowed, parse_cipher_suites, TlsPolicy};
use super::{TlsPolicy, build_supported_versions, common_name_is_allowed, parse_cipher_suites};
use rustls::crypto::aws_lc_rs;
use std::collections::HashSet;
seaweed-volume/src/server/debug.rs
+2 -2
View File
@@ -1,9 +1,9 @@
use axum::Router;
use axum::body::Body;
use axum::extract::Query;
use axum::http::{header, StatusCode};
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::{any, get};
use axum::Router;
use pprof::protos::Message;
use serde::Deserialize;
seaweed-volume/src/server/grpc_client.rs
+3 -1
View File
@@ -40,7 +40,9 @@ pub fn load_outgoing_grpc_tls(
(&config.grpc_client_cert_file, &config.grpc_client_key_file)
} else {
if !config.grpc_client_cert_file.is_empty() || !config.grpc_client_key_file.is_empty() {
tracing::warn!("grpc.volume.client_cert and grpc.volume.client_key must both be set, falling back to grpc.volume.cert and grpc.volume.key");
tracing::warn!(
"grpc.volume.client_cert and grpc.volume.client_key must both be set, falling back to grpc.volume.cert and grpc.volume.key"
);
}
(&config.grpc_cert_file, &config.grpc_key_file)
};
seaweed-volume/src/server/grpc_server.rs
+17 -18
View File
@@ -5,8 +5,8 @@
//! EC operations are stubbed with appropriate error messages.
use std::pin::Pin;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tokio_stream::Stream;
use tonic::{Request, Response, Status, Streaming};
@@ -21,7 +21,7 @@ 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::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint};
use super::volume_server::VolumeServerState;
type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>;
@@ -1398,7 +1398,7 @@ impl VolumeServer for VolumeGrpcService {
return Err(Status::not_found(format!(
"remote dat not loaded for volume id {}",
vid
)))
)));
}
}
} else {
@@ -3008,7 +3008,7 @@ impl VolumeServer for VolumeGrpcService {
// (V3) or checksum only (V2) + padding. Validate minimum
// footer length for the protocol version.
use crate::storage::types::{
Version, NEEDLE_CHECKSUM_SIZE, TIMESTAMP_SIZE, VERSION_3,
NEEDLE_CHECKSUM_SIZE, TIMESTAMP_SIZE, VERSION_3, Version,
};
let version = Version(resp_version as u8);
let min_footer = if version >= VERSION_3 {
@@ -3019,7 +3019,10 @@ impl VolumeServer for VolumeGrpcService {
if needle_body.len() < min_footer {
return Err(Status::invalid_argument(format!(
"tombstone needle {} body too short: got {} bytes, need >= {} for version {}",
n.id.0, needle_body.len(), min_footer, resp_version
n.id.0,
needle_body.len(),
min_footer,
resp_version
)));
}
}
@@ -4983,7 +4986,7 @@ impl VolumeServer for VolumeGrpcService {
return Err(Status::invalid_argument(format!(
"unsupported volume scrub mode {}",
mode
)))
)));
}
}
@@ -5012,7 +5015,7 @@ impl VolumeServer for VolumeGrpcService {
return Err(Status::invalid_argument(format!(
"unsupported EC volume scrub mode {}",
mode
)))
)));
}
}
@@ -5312,7 +5315,7 @@ impl VolumeServer for VolumeGrpcService {
return Err(Status::internal(format!(
"ping {} {}: {}",
req.target_type, req.target, e
)))
)));
}
}
} else if req.target_type == "master" {
@@ -5323,7 +5326,7 @@ impl VolumeServer for VolumeGrpcService {
return Err(Status::internal(format!(
"ping {} {}: {}",
req.target_type, req.target, e
)))
)));
}
}
} else if req.target_type == "filer" {
@@ -5333,7 +5336,7 @@ impl VolumeServer for VolumeGrpcService {
return Err(Status::internal(format!(
"ping {} {}: {}",
req.target_type, req.target, e
)))
)));
}
}
} else {
@@ -5757,11 +5760,7 @@ fn find_last_append_at_ns(idx_path: &str, dat_path: &str, version: u32) -> Optio
let ts = u64::from_be_bytes([
tail[4], tail[5], tail[6], tail[7], tail[8], tail[9], tail[10], tail[11],
]);
if ts > 0 {
Some(ts)
} else {
None
}
if ts > 0 { Some(ts) } else { None }
}
/// Get disk usage (total, free) in bytes for the given path.
@@ -5789,7 +5788,7 @@ fn get_disk_usage(path: &str) -> (u64, u64) {
mod tests {
use super::*;
use crate::config::MinFreeSpace;
use crate::remote_storage::s3_tier::{global_s3_tier_registry, S3TierBackend, S3TierConfig};
use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig, global_s3_tier_registry};
use crate::security::{Guard, SigningKey};
use crate::storage::needle_map::NeedleMapKind;
use crate::storage::store::Store;
@@ -5958,9 +5957,9 @@ mod tests {
tokio::sync::oneshot::Sender<()>,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
) {
use axum::http::{header, HeaderMap, HeaderValue, Method, StatusCode};
use axum::routing::any;
use axum::Router;
use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, header};
use axum::routing::any;
use std::sync::atomic::{AtomicUsize, Ordering};
let body = Arc::new(body);
seaweed-volume/src/server/handlers.rs
+18 -14
View File
@@ -6,17 +6,17 @@
use std::collections::HashMap;
use std::future::Future;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use axum::body::Body;
use axum::extract::{Path, Query, State};
use axum::http::{header, HeaderMap, Method, Request, StatusCode};
use axum::http::{HeaderMap, Method, Request, StatusCode, header};
use axum::response::{IntoResponse, Response};
use serde::{Deserialize, Serialize};
use super::grpc_client::{build_grpc_endpoint, GRPC_MAX_MESSAGE_SIZE};
use super::volume_server::{normalize_outgoing_http_url, to_http_address, VolumeServerState};
use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint};
use super::volume_server::{VolumeServerState, normalize_outgoing_http_url, to_http_address};
use crate::config::ReadMode;
use crate::metrics;
use crate::pb::volume_server_pb;
@@ -1505,7 +1505,7 @@ async fn get_or_head_handler_inner(
StatusCode::PAYLOAD_TOO_LARGE,
"compressed object exceeds decompression limit",
)
.into_response()
.into_response();
}
Err(GunzipError::Decode) => {} // not valid gzip; keep raw bytes
}
@@ -1531,7 +1531,7 @@ async fn get_or_head_handler_inner(
StatusCode::PAYLOAD_TOO_LARGE,
"compressed object exceeds decompression limit",
)
.into_response()
.into_response();
}
Err(GunzipError::Decode) => {} // not valid gzip; keep raw bytes
}
@@ -1810,7 +1810,7 @@ fn handle_range_request_from_source(
StatusCode::INTERNAL_SERVER_ERROR,
format!("range read error: {}", err),
)
.into_response()
.into_response();
}
};
headers.insert(
@@ -1840,7 +1840,7 @@ fn handle_range_request_from_source(
StatusCode::INTERNAL_SERVER_ERROR,
format!("range read error: {}", err),
)
.into_response()
.into_response();
}
};
if i == 0 {
@@ -2118,7 +2118,11 @@ pub async fn post_handler(
let (vid, needle_id, cookie) = match parse_url_path(&path) {
Some(parsed) => parsed,
None => {
return json_error_with_query(StatusCode::BAD_REQUEST, "invalid URL path", Some(&query))
return json_error_with_query(
StatusCode::BAD_REQUEST,
"invalid URL path",
Some(&query),
);
}
};
@@ -2728,7 +2732,7 @@ pub async fn delete_handler(
StatusCode::BAD_REQUEST,
"invalid URL path",
Some(&del_query),
)
);
}
};
@@ -3249,7 +3253,7 @@ async fn try_expand_chunk_manifest(
"compressed manifest exceeds decompression limit",
)
.into_response(),
)
);
}
Err(GunzipError::Decode) => return None,
}
@@ -3296,7 +3300,7 @@ async fn try_expand_chunk_manifest(
format!("read chunk {}: {}", chunk.fid, e),
)
.into_response(),
)
);
}
};
let offset = chunk.offset as usize;
@@ -3797,8 +3801,8 @@ fn is_compressible_file_type(ext: &str, mtype: &str) -> bool {
/// Try to gzip data. Returns None on error.
fn try_gzip_data(data: &[u8]) -> Option<Vec<u8>> {
use flate2::write::GzEncoder;
use flate2::Compression;
use flate2::write::GzEncoder;
use std::io::Write;
let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
encoder.write_all(data).ok()?;
@@ -4240,7 +4244,7 @@ mod tests {
/// replicated write to fail.
#[tokio::test]
async fn test_lookup_volume_strips_grpc_port_from_master_url() {
use axum::{routing::get, Router};
use axum::{Router, routing::get};
let app = Router::new().route(
"/dir/lookup",
seaweed-volume/src/server/heartbeat.rs
+11 -5
View File
@@ -5,14 +5,14 @@
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
use super::grpc_client::{build_grpc_endpoint, GRPC_MAX_MESSAGE_SIZE};
use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint};
use super::volume_server::VolumeServerState;
use crate::pb::master_pb;
use crate::pb::master_pb::seaweed_client::SeaweedClient;
@@ -119,7 +119,9 @@ pub async fn run_heartbeat_with_state(
if err_msg.contains(DUPLICATE_UUID_RETRY_MESSAGE) {
if duplicate_retry_count >= MAX_DUPLICATE_UUID_RETRIES {
error!("Shut down Volume Server due to persistent duplicate volume directories after 3 retries");
error!(
"Shut down Volume Server due to persistent duplicate volume directories after 3 retries"
);
error!(
"Please check if another volume server is using the same directory"
);
@@ -981,7 +983,11 @@ fn build_heartbeat_with_ec_status(
> DISK_CHECK_INTERVAL_NS
{
if !Path::new(&vol.file_name(".dat")).exists() {
warn!("Volume {}: data file {} missing (held open as deleted FD) - not reporting to master", vol.id.0, vol.file_name(".dat"));
warn!(
"Volume {}: data file {} missing (held open as deleted FD) - not reporting to master",
vol.id.0,
vol.file_name(".dat")
);
continue;
}
vol.last_disk_check_ns.store(now_ns, Ordering::Relaxed);
@@ -1249,8 +1255,8 @@ mod tests {
use crate::storage::needle_map::NeedleMapKind;
use crate::storage::types::{DiskType, VolumeId};
use crate::storage::volume::VolumeSpec;
use std::sync::atomic::Ordering;
use std::sync::RwLock;
use std::sync::atomic::Ordering;
use std::time::{SystemTime, UNIX_EPOCH};
fn test_config() -> HeartbeatConfig {
seaweed-volume/src/server/store_ec.rs
+7 -7
View File
@@ -38,15 +38,15 @@ use reed_solomon_erasure::galois_8::ReedSolomon;
use tokio::sync::Semaphore;
use tonic::Request;
use crate::pb::master_pb::{self, seaweed_client::SeaweedClient, LookupEcVolumeRequest};
use crate::pb::master_pb::{self, LookupEcVolumeRequest, seaweed_client::SeaweedClient};
use crate::pb::volume_server_pb::{
volume_server_client::VolumeServerClient, CopyFileRequest, VolumeEcShardReadRequest,
CopyFileRequest, VolumeEcShardReadRequest, volume_server_client::VolumeServerClient,
};
use crate::server::grpc_client::{build_grpc_endpoint, parse_grpc_address, GRPC_MAX_MESSAGE_SIZE};
use crate::server::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint, parse_grpc_address};
use crate::server::request_id::outgoing_request_id_interceptor;
use crate::server::volume_server::{to_http_address, VolumeServerState};
use crate::server::volume_server::{VolumeServerState, to_http_address};
use crate::storage::erasure_coding::ec_shard::ShardId;
use crate::storage::needle::needle::{get_actual_size, Needle, NeedleError};
use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size};
use crate::storage::store_ec_reconcile::EcVolumeMissingIndex;
use crate::storage::types::*;
use crate::storage::volume::volume_file_name;
@@ -300,7 +300,7 @@ pub async fn scrub_ec_volume_distributed(
0,
Vec::new(),
vec![format!("EC volume id {} not found", vid.0)],
)
);
}
};
// full scan means verifying the index as well
@@ -672,7 +672,7 @@ fn scrub_snapshot_under_lock(
return Err(io::Error::new(
io::ErrorKind::NotFound,
format!("EC volume {} not found (unmounted mid-scan)", vid.0),
))
));
}
};
// The volume was torn down and remounted as a DIFFERENT encode run between
seaweed-volume/src/server/volume_server.rs
+8 -7
View File
@@ -14,12 +14,12 @@ use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, Ordering};
use std::sync::{Arc, RwLock};
use axum::{
extract::{connect_info::ConnectInfo, Request, State},
http::{header, HeaderValue, Method, StatusCode},
Router,
extract::{Request, State, connect_info::ConnectInfo},
http::{HeaderValue, Method, StatusCode, header},
middleware::{self, Next},
response::{IntoResponse, Response},
routing::{any, get},
Router,
};
use crate::config::ReadMode;
@@ -200,9 +200,7 @@ pub fn to_http_address(addr: &str) -> std::borrow::Cow<'_, str> {
// rather than being silently rewritten. Mirrors the validation already
// done in `to_grpc_address` for the inverse direction.
if let (Ok(_), Ok(_)) = (http_port.parse::<u16>(), grpc_port.parse::<u16>()) {
return std::borrow::Cow::Owned(
addr[..ports_sep_index + 1 + dot_idx].to_string(),
);
return std::borrow::Cow::Owned(addr[..ports_sep_index + 1 + dot_idx].to_string());
}
}
std::borrow::Cow::Borrowed(addr)
@@ -514,7 +512,10 @@ mod tests {
// "host:abc.def"), and silently rewriting it would just hide the bug.
assert_eq!(to_http_address("host:abc.def"), "host:abc.def");
assert_eq!(to_http_address("host:9333.notaport"), "host:9333.notaport");
assert_eq!(to_http_address("host:notaport.19333"), "host:notaport.19333");
assert_eq!(
to_http_address("host:notaport.19333"),
"host:notaport.19333"
);
// Out-of-range ports must not be silently truncated either.
assert_eq!(to_http_address("host:99999.19333"), "host:99999.19333");
}
seaweed-volume/src/server/write_queue.rs
+1 -1
View File
@@ -178,8 +178,8 @@ mod tests {
use crate::server::volume_server::RuntimeMetricsConfig;
use crate::storage::needle_map::NeedleMapKind;
use crate::storage::store::Store;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32};
use std::sync::RwLock;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32};
let store = Store::new(NeedleMapKind::InMemory);
let guard = Guard::new(&[], SigningKey(vec![]), 0, SigningKey(vec![]), 0);
seaweed-volume/src/storage/disk_location.rs
+93 -33
View File
@@ -15,8 +15,8 @@ use tracing::warn;
use crate::config::MinFreeSpace;
use crate::storage::erasure_coding::ec_bitrot::remove_bitrot_sidecars;
use crate::storage::erasure_coding::ec_shard::{
EcVolumeShard, DATA_SHARDS_COUNT, ERASURE_CODING_LARGE_BLOCK_SIZE,
ERASURE_CODING_SMALL_BLOCK_SIZE,
DATA_SHARDS_COUNT, ERASURE_CODING_LARGE_BLOCK_SIZE, ERASURE_CODING_SMALL_BLOCK_SIZE,
EcVolumeShard,
};
use crate::storage::erasure_coding::ec_volume::EcVolume;
use crate::storage::needle_map::NeedleMapKind;
@@ -205,7 +205,6 @@ impl DiskLocation {
continue;
}
// Load existing data only; never create a phantom `.dat`. A lone
// `.vif`/`.idx` (e.g. an EC sidecar whose `.ecx` is on a sibling
// disk) would otherwise have Volume::new write an 8-byte stub that
@@ -377,8 +376,10 @@ impl DiskLocation {
let mut expected_shard_size: Option<i64> = None;
let dat_exists = match fs::metadata(&dat_path) {
Ok(meta) if meta.len() > SUPER_BLOCK_SIZE as u64 => {
expected_shard_size =
Some(calculate_expected_shard_size(meta.len() as i64, data_shards));
expected_shard_size = Some(calculate_expected_shard_size(
meta.len() as i64,
data_shards,
));
true
}
Ok(_) => false,
@@ -402,7 +403,13 @@ impl DiskLocation {
if size != prev {
// Inconsistent sizes signal corruption or mixed
// generations; not trusted for deletion -> keep.
warn!(volume_id = vid.0, shard = i, size, expected = prev, "EC shard size mismatch; keeping shards");
warn!(
volume_id = vid.0,
shard = i,
size,
expected = prev,
"EC shard size mismatch; keeping shards"
);
return true;
}
} else {
@@ -669,8 +676,7 @@ impl DiskLocation {
pub fn free_volume_count(&self) -> i32 {
use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
let max = self.max_volume_count.load(Ordering::Relaxed);
let free_count = (max as i64 - self.volumes.len() as i64)
* DATA_SHARDS_COUNT as i64
let free_count = (max as i64 - self.volumes.len() as i64) * DATA_SHARDS_COUNT as i64
- self.ec_shard_count() as i64;
let effective_free = free_count / DATA_SHARDS_COUNT as i64;
if effective_free > 0 {
@@ -845,14 +851,10 @@ impl DiskLocation {
// propagate the error to the caller.
let created = !self.ec_volumes.contains_key(&vid);
if created {
let ec_vol = EcVolume::new(&dir, idx_dir, collection, vid)
.map_err(VolumeError::Io)?;
let ec_vol = EcVolume::new(&dir, idx_dir, collection, vid).map_err(VolumeError::Io)?;
self.ec_volumes.insert(vid, ec_vol);
}
let ec_vol = self
.ec_volumes
.get_mut(&vid)
.expect("just inserted above");
let ec_vol = self.ec_volumes.get_mut(&vid).expect("just inserted above");
// When the orchestrator supplied a source disk type on the Mount
// RPC, override the EC volume's disk type so heartbeats report
// under the source volume's disk type (#9423). When the caller
@@ -1057,8 +1059,7 @@ impl DiskLocation {
// delete on a load error.
warn!(
volume_id = vid.0,
"Failed to load EC shards: {}; keeping files for retry",
e,
"Failed to load EC shards: {}; keeping files for retry", e,
);
self.unmount_ec_shards(vid, &shard_ids);
}
@@ -1176,7 +1177,12 @@ fn rm_if_present(path: String) -> io::Result<()> {
}
}
fn ec_data_shards_from_vif(directory: &str, idx_directory: &str, collection: &str, vid: VolumeId) -> usize {
fn ec_data_shards_from_vif(
directory: &str,
idx_directory: &str,
collection: &str,
vid: VolumeId,
) -> usize {
for dir in [directory, idx_directory] {
let vif = format!("{}.vif", volume_file_name(dir, collection, vid));
if let Some(ds) = fs::read_to_string(&vif)
@@ -1307,7 +1313,10 @@ fn remove_empty_ec_dat_stub(volume_name: &str, idx_name: &str, vid: VolumeId) ->
return false;
}
warn!(volume_id = vid.0, "removing leftover empty .dat stub for EC volume");
warn!(
volume_id = vid.0,
"removing leftover empty .dat stub for EC volume"
);
let _ = fs::remove_file(&dat_path);
let _ = fs::remove_file(format!("{}.idx", idx_name));
true
@@ -1352,7 +1361,11 @@ mod tests {
}),
..Default::default()
};
std::fs::write(format!("{}.vif", ibase), serde_json::to_string(&vif).unwrap()).unwrap();
std::fs::write(
format!("{}.vif", ibase),
serde_json::to_string(&vif).unwrap(),
)
.unwrap();
assert!(
remove_empty_ec_dat_stub(&vbase, &ibase, VolumeId(42)),
@@ -1368,16 +1381,30 @@ mod tests {
fn test_validate_ec_volume_partial_dat_next_to_full_shards_keeps() {
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
let loc = DiskLocation::new(dir, dir, 10, DiskType::HardDrive, MinFreeSpace::Percent(1.0), Vec::new()).unwrap();
let loc = DiskLocation::new(
dir,
dir,
10,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
let base = volume_file_name(dir, "", VolumeId(70));
let ds = crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
let full = calculate_expected_shard_size(30 * 1024 * 1024, ds);
for i in 0..ds {
std::fs::File::create(format!("{}.ec{:02}", base, i)).unwrap().set_len(full as u64).unwrap();
std::fs::File::create(format!("{}.ec{:02}", base, i))
.unwrap()
.set_len(full as u64)
.unwrap();
}
// Partial .dat: bigger than a superblock so it is not swept as a stub,
// but smaller than what these shards encode.
std::fs::File::create(format!("{}.dat", base)).unwrap().set_len(5 * 1024 * 1024).unwrap();
std::fs::File::create(format!("{}.dat", base))
.unwrap()
.set_len(5 * 1024 * 1024)
.unwrap();
assert!(
loc.validate_ec_volume("", VolumeId(70)),
"full-size shards beside a smaller (stale/partial) .dat must be kept",
@@ -1391,15 +1418,29 @@ mod tests {
fn test_validate_ec_volume_interrupted_encode_reclaims() {
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
let loc = DiskLocation::new(dir, dir, 10, DiskType::HardDrive, MinFreeSpace::Percent(1.0), Vec::new()).unwrap();
let loc = DiskLocation::new(
dir,
dir,
10,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
let base = volume_file_name(dir, "", VolumeId(71));
let ds = crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
let dat_size = 30 * 1024 * 1024i64;
std::fs::File::create(format!("{}.dat", base)).unwrap().set_len(dat_size as u64).unwrap();
std::fs::File::create(format!("{}.dat", base))
.unwrap()
.set_len(dat_size as u64)
.unwrap();
let partial = calculate_expected_shard_size(dat_size, ds) / 3;
assert!(partial > 0);
for i in 0..ds {
std::fs::File::create(format!("{}.ec{:02}", base, i)).unwrap().set_len(partial as u64).unwrap();
std::fs::File::create(format!("{}.ec{:02}", base, i))
.unwrap()
.set_len(partial as u64)
.unwrap();
}
assert!(
!loc.validate_ec_volume("", VolumeId(71)),
@@ -1444,7 +1485,11 @@ mod tests {
}),
..Default::default()
};
std::fs::write(format!("{}.vif", dbase), serde_json::to_string(&with_gen).unwrap()).unwrap();
std::fs::write(
format!("{}.vif", dbase),
serde_json::to_string(&with_gen).unwrap(),
)
.unwrap();
assert_eq!(loc.ec_generation_ts_ns("", vid), Some(4242));
// A .vif with no EC config reads as generation 0 (recovered/pre-upgrade live volume).
@@ -1453,12 +1498,20 @@ mod tests {
version: 3,
..Default::default()
};
std::fs::write(format!("{}.vif", dbase), serde_json::to_string(&no_cfg).unwrap()).unwrap();
std::fs::write(
format!("{}.vif", dbase),
serde_json::to_string(&no_cfg).unwrap(),
)
.unwrap();
assert_eq!(loc.ec_generation_ts_ns("", vid), Some(0));
// idx-dir fallback: only the idx dir holds the .vif.
std::fs::remove_file(format!("{}.vif", dbase)).unwrap();
std::fs::write(format!("{}.vif", ibase), serde_json::to_string(&with_gen).unwrap()).unwrap();
std::fs::write(
format!("{}.vif", ibase),
serde_json::to_string(&with_gen).unwrap(),
)
.unwrap();
assert_eq!(loc.ec_generation_ts_ns("", vid), Some(4242));
}
@@ -1734,7 +1787,8 @@ mod tests {
// mount_ec_shards with source_disk_type="ssd" — simulating the
// VolumeEcShardsMount RPC path.
std::fs::write(format!("{}/pics_7.ec00", dir), b"ec-shard").unwrap();
loc.mount_ec_shards(VolumeId(7), "pics", &[0], "ssd").unwrap();
loc.mount_ec_shards(VolumeId(7), "pics", &[0], "ssd")
.unwrap();
{
let ec_vol = loc.find_ec_volume(VolumeId(7)).expect("ec volume mounted");
assert_eq!(
@@ -1751,7 +1805,9 @@ mod tests {
std::fs::write(format!("{}/pics_7.ec01", dir), b"ec-shard").unwrap();
loc.mount_ec_shards(VolumeId(7), "pics", &[1], "").unwrap();
{
let ec_vol = loc.find_ec_volume(VolumeId(7)).expect("ec volume still mounted");
let ec_vol = loc
.find_ec_volume(VolumeId(7))
.expect("ec volume still mounted");
assert_eq!(
ec_vol.disk_type,
DiskType::Ssd,
@@ -1833,7 +1889,8 @@ mod tests {
let gauge = crate::metrics::VOLUME_GAUGE.with_label_values(&["dupmount", "ec_shards"]);
let before = gauge.get();
loc.mount_ec_shards(VolumeId(11), "dupmount", &[0], "").unwrap();
loc.mount_ec_shards(VolumeId(11), "dupmount", &[0], "")
.unwrap();
loc.mount_ec_shards(VolumeId(11), "dupmount", &[0], "")
.expect("a duplicate mount must succeed as a no-op");
@@ -1909,8 +1966,11 @@ mod tests {
let path = format!("{}/{}_{}.ec{:02}", dir, collection, vid.0, sid);
std::fs::write(&path, b"shard data nonempty").unwrap();
}
std::fs::write(format!("{}/{}_{}.ecx", dir, collection, vid.0), vec![0u8; 20])
.unwrap();
std::fs::write(
format!("{}/{}_{}.ecx", dir, collection, vid.0),
vec![0u8; 20],
)
.unwrap();
std::fs::write(format!("{}/{}_{}.ecj", dir, collection, vid.0), b"").unwrap();
std::fs::write(
format!("{}/{}_{}.vif", dir, collection, vid.0),
seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs
+14 -11
View File
@@ -627,7 +627,10 @@ mod tests {
save_bitrot_sidecar(path, &prot).unwrap();
let bytes = std::fs::read(path).unwrap();
let hex: String = bytes.iter().map(|b| format!("{:02x}", b)).collect();
assert_eq!(hex, CANONICAL_HEX, "Rust .ecsum bytes drifted from the Go canonical form");
assert_eq!(
hex, CANONICAL_HEX,
"Rust .ecsum bytes drifted from the Go canonical form"
);
let _ = std::fs::remove_file(path);
}
@@ -662,7 +665,11 @@ mod tests {
format!("{}.ecsum.v1", base),
format!("{}.ecsum.v7", base),
] {
assert!(!std::path::Path::new(&p).exists(), "{} should be removed", p);
assert!(
!std::path::Path::new(&p).exists(),
"{} should be removed",
p
);
}
assert!(std::path::Path::new(&keep_shard).exists());
assert!(std::path::Path::new(&keep_other_vid).exists());
@@ -681,7 +688,9 @@ mod tests {
assert!(!is_pow2_multiple_of_1mib(1 << 19)); // 512 KiB, too small
assert!(!is_pow2_multiple_of_1mib(3 << 20)); // 3 MiB, not pow2
assert!(!is_pow2_multiple_of_1mib(128 * 1024 * 1024)); // pow2 but > MAX_BITROT_BLOCK_SIZE
assert!(!is_pow2_multiple_of_1mib(DEFAULT_BITROT_BLOCK_SIZE as u32 + 1));
assert!(!is_pow2_multiple_of_1mib(
DEFAULT_BITROT_BLOCK_SIZE as u32 + 1
));
}
#[test]
@@ -735,12 +744,7 @@ mod tests {
#[test]
fn test_save_load_roundtrip() {
let tmp = tempfile::TempDir::new().unwrap();
let path = tmp
.path()
.join("vol.ecsum")
.to_str()
.unwrap()
.to_string();
let path = tmp.path().join("vol.ecsum").to_str().unwrap().to_string();
let mut builder = ShardChecksumBuilder::new(DEFAULT_BITROT_BLOCK_SIZE as i64);
builder.write(b"hello world");
@@ -901,8 +905,7 @@ mod tests {
assert_eq!(resolve_status(&notfound, 0, 10, 4), BitrotStatus::Off);
// Integrity failure => Invalid.
let bad: Result<EcBitrotProtection, BitrotLoadError> =
Err(BitrotLoadError::BadMagic(0));
let bad: Result<EcBitrotProtection, BitrotLoadError> = Err(BitrotLoadError::BadMagic(0));
assert_eq!(resolve_status(&bad, 0, 10, 4), BitrotStatus::Invalid);
// Generation mismatch => Off.
seaweed-volume/src/storage/erasure_coding/ec_decoder.rs
+10 -5
View File
@@ -458,14 +458,13 @@ mod tests {
assert!(!std::path::Path::new(&format!("{}/7.dat.tmp", dir)).exists());
}
// Decoding when .vif does not record the encode-time size: the layout is
// inferred from the shard size, except when that is an exact large-block
// multiple and the live extent reaches the ambiguous region.
#[test]
fn test_write_dat_file_fallback_layout() {
use crate::storage::erasure_coding::ec_bitrot::{
ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE,
DEFAULT_BITROT_BLOCK_SIZE, ShardChecksumBuilder,
};
use reed_solomon_erasure::galois_8::ReedSolomon;
@@ -551,14 +550,20 @@ mod tests {
// each shard exactly one large block, indistinguishable from one large row
let (dir, shard_dirs, _) = encode("ambig1", large_row_size - 1);
let err = decode_to(&dir, "out", large_row_size / 2, 0, &shard_dirs).unwrap_err();
assert!(err.to_string().contains("does not identify the block layout"));
assert!(
err.to_string()
.contains("does not identify the block layout")
);
// two-row equivalent: decoding within the agreed prefix still works
let (dir, shard_dirs, original) = encode("ambig2", 2 * large_row_size - 1);
let decoded = decode_to(&dir, "outa", large_row_size, 0, &shard_dirs).unwrap();
assert_eq!(&original[..large_row_size as usize], &decoded[..]);
let err = decode_to(&dir, "outb", large_row_size + 1, 0, &shard_dirs).unwrap_err();
assert!(err.to_string().contains("does not identify the block layout"));
assert!(
err.to_string()
.contains("does not identify the block layout")
);
}
// Decoding after deletions moved the live extent below the large-block row
@@ -567,7 +572,7 @@ mod tests {
#[test]
fn test_write_dat_file_after_tail_deletion() {
use crate::storage::erasure_coding::ec_bitrot::{
ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE,
DEFAULT_BITROT_BLOCK_SIZE, ShardChecksumBuilder,
};
use reed_solomon_erasure::galois_8::ReedSolomon;
seaweed-volume/src/storage/erasure_coding/ec_encoder.rs
+12 -7
View File
@@ -10,11 +10,9 @@ use std::io::{Read, Seek, SeekFrom};
use reed_solomon_erasure::galois_8::ReedSolomon;
use crate::pb::volume_server_pb::{
ChecksumAlgorithm, EcBitrotProtection, EcShardChecksums,
};
use crate::pb::volume_server_pb::{ChecksumAlgorithm, EcBitrotProtection, EcShardChecksums};
use crate::storage::erasure_coding::ec_bitrot::{
self, ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE,
self, DEFAULT_BITROT_BLOCK_SIZE, ShardChecksumBuilder,
};
use crate::storage::erasure_coding::ec_shard::*;
use crate::storage::idx;
@@ -585,7 +583,8 @@ pub fn rebuild_ecx_file(
}
let cookie = Cookie::from_bytes(&header_buf[..COOKIE_SIZE]);
let needle_id = NeedleId::from_bytes(&header_buf[COOKIE_SIZE..COOKIE_SIZE + NEEDLE_ID_SIZE]);
let needle_id =
NeedleId::from_bytes(&header_buf[COOKIE_SIZE..COOKIE_SIZE + NEEDLE_ID_SIZE]);
let size = Size::from_bytes(&header_buf[COOKIE_SIZE + NEEDLE_ID_SIZE..header_size]);
// Validate: stop if we hit zero cookie+id (end of data)
@@ -1032,7 +1031,10 @@ mod tests {
let victim = format!("{}/1.ec03", dir);
let full = std::fs::metadata(&victim).unwrap().len();
assert!(full > 0, "encoded shard should be non-empty");
let f = std::fs::OpenOptions::new().write(true).open(&victim).unwrap();
let f = std::fs::OpenOptions::new()
.write(true)
.open(&victim)
.unwrap();
f.set_len(full / 2).unwrap();
drop(f);
@@ -1224,7 +1226,10 @@ mod tests {
rebuild_ecx_file(&dir, "", VolumeId(2), 10, block_size, 0, &[]).unwrap();
let rebuilt = std::fs::read(&ecx_path).unwrap();
assert_eq!(canonical, rebuilt, "rebuilt .ecx must match the encode-time .ecx");
assert_eq!(
canonical, rebuilt,
"rebuilt .ecx must match the encode-time .ecx"
);
}
// A truncated data shard must FAIL the .ecx rebuild, not publish the
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+2 -2
View File
@@ -12,7 +12,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
use crate::pb::master_pb;
use crate::storage::erasure_coding::ec_locate;
use crate::storage::erasure_coding::ec_shard::*;
use crate::storage::needle::needle::{get_actual_size, Needle, NeedleError};
use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size};
use crate::storage::types::*;
use crate::storage::volume_open::open_volume_file;
@@ -3328,7 +3328,7 @@ mod uniform_layout_tests {
// Legacy fixture: two-tier encode plus a .vif without a block
// size, the state every pre-upgrade EC volume is in.
use crate::storage::erasure_coding::ec_bitrot::{
ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE,
DEFAULT_BITROT_BLOCK_SIZE, ShardChecksumBuilder,
};
use reed_solomon_erasure::galois_8::ReedSolomon;
let base = crate::storage::volume::volume_file_name(dir, "", vid);
seaweed-volume/src/storage/erasure_coding/mod.rs
+2 -2
View File
@@ -11,7 +11,7 @@ pub mod ec_shard;
pub mod ec_volume;
pub use ec_shard::{
EcVolumeShard, ShardId, DATA_SHARDS_COUNT, MAX_SHARD_COUNT, MIN_TOTAL_DISKS,
PARITY_SHARDS_COUNT, TOTAL_SHARDS_COUNT,
DATA_SHARDS_COUNT, EcVolumeShard, MAX_SHARD_COUNT, MIN_TOTAL_DISKS, PARITY_SHARDS_COUNT,
ShardId, TOTAL_SHARDS_COUNT,
};
pub use ec_volume::EcVolume;
seaweed-volume/src/storage/idx/mod.rs
+6 -2
View File
@@ -57,7 +57,7 @@ pub fn check_index_file<R: Read + Seek>(
errs.push(format!("walk index file: {}", e));
}
entries.sort_by(|a, b| a.2.cmp(&b.2).then(a.3 .0.cmp(&b.3 .0)));
entries.sort_by(|a, b| a.2.cmp(&b.2).then(a.3.0.cmp(&b.3.0)));
// Offset-0 logical tombstones (remote-tier deletes) occupy no physical extent,
// so they cannot overlap anything — exclude them from the overlap check. They
@@ -213,7 +213,11 @@ mod tests {
let size = data.len() as i64;
let (count, errs) = check_index_file(&mut Cursor::new(data), size, Version(3));
assert_eq!(count, 2, "tombstone row is still counted: {:?}", errs);
assert!(errs.is_empty(), "offset-0 tombstone must not overlap: {:?}", errs);
assert!(
errs.is_empty(),
"offset-0 tombstone must not overlap: {:?}",
errs
);
}
#[test]
seaweed-volume/src/storage/needle/mod.rs
+4 -1
View File
@@ -1,5 +1,8 @@
pub mod crc;
#[expect(clippy::module_inception, reason = "needle/needle.rs mirrors the Go package layout")]
#[expect(
clippy::module_inception,
reason = "needle/needle.rs mirrors the Go package layout"
)]
pub mod needle;
pub mod ttl;
seaweed-volume/src/storage/needle/needle.rs
+5 -3
View File
@@ -198,8 +198,8 @@ impl Needle {
/// the data payload from disk at all, matching Go's `ReadNeedleMeta`.
pub fn read_paged_meta(
&mut self,
header_bytes: &[u8], // first 20 bytes: NEEDLE_HEADER_SIZE + DATA_SIZE_SIZE
meta_bytes: &[u8], // tail: non-data body metadata + checksum + timestamp + padding
header_bytes: &[u8], // first 20 bytes: NEEDLE_HEADER_SIZE + DATA_SIZE_SIZE
meta_bytes: &[u8], // tail: non-data body metadata + checksum + timestamp + padding
offset: i64,
expected_size: Size,
version: Version,
@@ -770,7 +770,9 @@ pub fn parse_needle_id_cookie(s: &str) -> Result<(NeedleId, Cookie), String> {
#[derive(Debug, thiserror::Error)]
pub enum NeedleError {
#[error("size mismatch at offset {offset}: found id={id} size={found:?}, expected size={expected:?}")]
#[error(
"size mismatch at offset {offset}: found id={id} size={found:?}, expected size={expected:?}"
)]
SizeMismatch {
offset: i64,
id: NeedleId,
seaweed-volume/src/storage/needle/ttl.rs
+56 -8
View File
@@ -258,7 +258,13 @@ mod tests {
// 24h normalizes to 1d via fitTtlCount
let ttl = TTL::read("24h").unwrap();
assert_eq!(ttl.to_seconds(), 86400);
assert_eq!(ttl, TTL { count: 1, unit: TTL_UNIT_DAY });
assert_eq!(
ttl,
TTL {
count: 1,
unit: TTL_UNIT_DAY
}
);
}
#[test]
@@ -304,12 +310,24 @@ mod tests {
fn test_ttl_overflow_normalizes() {
// Go's ReadTTL calls fitTtlCount: 300m = 18000s = 5h (exact fit)
let ttl = TTL::read("300m").unwrap();
assert_eq!(ttl, TTL { count: 5, unit: TTL_UNIT_HOUR });
assert_eq!(
ttl,
TTL {
count: 5,
unit: TTL_UNIT_HOUR
}
);
// 256h = 921600s. Doesn't fit in hours (256 >= 256), doesn't fit exact in days.
// Second pass: 921600/86400 = 10 (truncated) < 256 -> 10d
let ttl = TTL::read("256h").unwrap();
assert_eq!(ttl, TTL { count: 10, unit: TTL_UNIT_DAY });
assert_eq!(
ttl,
TTL {
count: 10,
unit: TTL_UNIT_DAY
}
);
}
#[test]
@@ -317,19 +335,49 @@ mod tests {
// Go's ReadTTL calls fitTtlCount which normalizes to coarsest unit.
// 120m -> 2h, 7d -> 1w, 24h -> 1d.
let ttl = TTL::read("120m").unwrap();
assert_eq!(ttl, TTL { count: 2, unit: TTL_UNIT_HOUR });
assert_eq!(
ttl,
TTL {
count: 2,
unit: TTL_UNIT_HOUR
}
);
let ttl = TTL::read("7d").unwrap();
assert_eq!(ttl, TTL { count: 1, unit: TTL_UNIT_WEEK });
assert_eq!(
ttl,
TTL {
count: 1,
unit: TTL_UNIT_WEEK
}
);
let ttl = TTL::read("24h").unwrap();
assert_eq!(ttl, TTL { count: 1, unit: TTL_UNIT_DAY });
assert_eq!(
ttl,
TTL {
count: 1,
unit: TTL_UNIT_DAY
}
);
// Values that don't simplify stay as-is
let ttl = TTL::read("5d").unwrap();
assert_eq!(ttl, TTL { count: 5, unit: TTL_UNIT_DAY });
assert_eq!(
ttl,
TTL {
count: 5,
unit: TTL_UNIT_DAY
}
);
let ttl = TTL::read("3m").unwrap();
assert_eq!(ttl, TTL { count: 3, unit: TTL_UNIT_MINUTE });
assert_eq!(
ttl,
TTL {
count: 3,
unit: TTL_UNIT_MINUTE
}
);
}
}
seaweed-volume/src/storage/needle_map.rs
+51 -32
View File
@@ -801,23 +801,26 @@ impl RedbNeedleMap {
}
#[cfg(feature = "redb-experimental-cursor")]
{
let mut cursor = table
.upper_bound_mut(Bound::<u64>::Unbounded)
.map_err(|e| {
io::Error::new(
io::ErrorKind::Other,
format!("redb upper_bound_mut: {}", e),
)
})?;
let mut cursor =
table
.upper_bound_mut(Bound::<u64>::Unbounded)
.map_err(|e| {
io::Error::new(
io::ErrorKind::Other,
format!("redb upper_bound_mut: {}", e),
)
})?;
for (key, nv) in &entries {
let key_u64: u64 = (*key).into();
let packed = pack_needle_value(nv);
cursor.insert_before(key_u64, packed.as_slice()).map_err(|e| {
io::Error::new(
io::ErrorKind::Other,
format!("redb insert_before: {}", e),
)
})?;
cursor
.insert_before(key_u64, packed.as_slice())
.map_err(|e| {
io::Error::new(
io::ErrorKind::Other,
format!("redb insert_before: {}", e),
)
})?;
}
cursor.close().map_err(|e| {
io::Error::new(io::ErrorKind::Other, format!("redb cursor close: {}", e))
@@ -1114,18 +1117,12 @@ impl RedbNeedleMap {
let read_file = std::fs::OpenOptions::new()
.read(true)
.open(&idx_path)
.map_err(|e| {
io::Error::other(format!("reopen: open .idx {}: {}", idx_path, e))
})?;
.map_err(|e| io::Error::other(format!("reopen: open .idx {}: {}", idx_path, e)))?;
let actual_idx_size = read_file.metadata()?.len();
let mut reader = io::BufReader::new(read_file);
let reopened = Self::load_from_idx(
&self.rdb_path,
&mut reader,
self.version,
self.cache_bytes,
)?;
let reopened =
Self::load_from_idx(&self.rdb_path, &mut reader, self.version, self.cache_bytes)?;
// Preserve the append writer and the paths/version/cache; adopt the
// repaired database, metrics, and idx_file_offset from the reload.
@@ -2136,8 +2133,14 @@ mod tests {
// server opens one redb database per volume, so the process-wide
// ceiling is roughly (volumes x budget).
assert_eq!(NeedleMapKind::Redb.redb_cache_bytes(), 4 * 1024 * 1024);
assert_eq!(NeedleMapKind::RedbMedium.redb_cache_bytes(), 8 * 1024 * 1024);
assert_eq!(NeedleMapKind::RedbLarge.redb_cache_bytes(), 16 * 1024 * 1024);
assert_eq!(
NeedleMapKind::RedbMedium.redb_cache_bytes(),
8 * 1024 * 1024
);
assert_eq!(
NeedleMapKind::RedbLarge.redb_cache_bytes(),
16 * 1024 * 1024
);
}
#[test]
@@ -2176,8 +2179,12 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let (mut nm, db_path, _idx_path) = open_writable_redb(dir.path());
for i in 1..EXPECTED_INTERVAL {
nm.put(NeedleId(i), Offset::from_actual_offset((i * 8) as i64), Size(1))
.unwrap();
nm.put(
NeedleId(i),
Offset::from_actual_offset((i * 8) as i64),
Size(1),
)
.unwrap();
assert!(!nm.checkpoint_due(), "due after only {i} writes");
}
nm.put(
@@ -2187,7 +2194,11 @@ mod tests {
)
.unwrap();
assert!(nm.checkpoint_due());
assert_eq!(durable_idx_size(&db_path), None, "put() must not commit durably");
assert_eq!(
durable_idx_size(&db_path),
None,
"put() must not commit durably"
);
nm.checkpoint(true).unwrap();
assert!(!nm.checkpoint_due());
@@ -2215,8 +2226,12 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let (mut nm, db_path, idx_path) = open_writable_redb(dir.path());
for i in 1..=5u64 {
nm.put(NeedleId(i), Offset::from_actual_offset((i * 8) as i64), Size(1))
.unwrap();
nm.put(
NeedleId(i),
Offset::from_actual_offset((i * 8) as i64),
Size(1),
)
.unwrap();
}
nm.close();
drop(nm);
@@ -2240,8 +2255,12 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let (mut nm, db_path, idx_path) = open_writable_redb(dir.path());
for i in 1..=5u64 {
nm.put(NeedleId(i), Offset::from_actual_offset((i * 8) as i64), Size(1))
.unwrap();
nm.put(
NeedleId(i),
Offset::from_actual_offset((i * 8) as i64),
Size(1),
)
.unwrap();
}
// Drop without close(): redb makes the table durable on drop, but the
// recorded .idx size stays at its load-time value (0), so the reload
seaweed-volume/src/storage/needle_map/idx_metric.rs
+1 -3
View File
@@ -132,9 +132,7 @@ mod tests {
let mut seen = SeenKeys::new(10_000, FALSE_POSITIVE_RATE);
// Fresh keys may occasionally collide (that is the false-positive
// rate), but only rarely.
let fresh_reported_seen = (0..10_000u64)
.filter(|&key| seen.test_and_add(key))
.count();
let fresh_reported_seen = (0..10_000u64).filter(|&key| seen.test_and_add(key)).count();
assert!(
fresh_reported_seen < 50,
"fresh keys reported seen: {fresh_reported_seen}"
seaweed-volume/src/storage/needle_map/sorted_file.rs
+10 -9
View File
@@ -133,9 +133,7 @@ impl SortedFileNeedleMap {
}
let file = pooled_index_files()
.borrow(&self.db_file_name, false)
.map_err(|e| {
io::Error::new(e.kind(), format!("open {}: {}", self.db_file_name, e))
})?;
.map_err(|e| io::Error::new(e.kind(), format!("open {}: {}", self.db_file_name, e)))?;
match search_sorted_index(&file, self.db_file_size, key)? {
Some((_, offset, size)) => Ok(Some(NeedleValue { offset, size })),
None => Ok(None),
@@ -700,10 +698,11 @@ mod tests {
// without a reload — the same contract Go's Get has, where callers
// check size.is_deleted().
assert!(m.get(NeedleId(2)).unwrap().unwrap().size.is_deleted());
assert!(m
.delete(NeedleId(2), Offset::from_actual_offset(16))
.unwrap()
.is_none());
assert!(
m.delete(NeedleId(2), Offset::from_actual_offset(16))
.unwrap()
.is_none()
);
assert!(!m.get(NeedleId(1)).unwrap().unwrap().size.is_deleted());
}
@@ -999,7 +998,8 @@ mod tests {
// The retry is a no-op: no second tombstone, no double counting.
assert_eq!(
m.delete(NeedleId(1), Offset::from_actual_offset(8)).unwrap(),
m.delete(NeedleId(1), Offset::from_actual_offset(8))
.unwrap(),
None
);
assert_eq!(m.deleted_count(), deleted_before + 2);
@@ -1029,7 +1029,8 @@ mod tests {
);
// And a retry must not append a second tombstone for it.
assert_eq!(
m.delete(NeedleId(1), Offset::from_actual_offset(8)).unwrap(),
m.delete(NeedleId(1), Offset::from_actual_offset(8))
.unwrap(),
None
);
}
seaweed-volume/src/storage/store.rs
+50 -37
View File
@@ -6,8 +6,8 @@
use std::collections::HashSet;
use std::io;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use crate::config::MinFreeSpace;
use crate::pb::master_pb;
@@ -511,11 +511,8 @@ impl Store {
&& collection != "..";
if hint_safe {
for loc in &mut self.locations {
let base = crate::storage::volume::volume_file_name(
&loc.directory,
collection,
vid,
);
let base =
crate::storage::volume::volume_file_name(&loc.directory, collection, vid);
// Confirm a collection-named sidecar exists before using the
// hint. A lone .vif/.idx (e.g. an EC sidecar whose .ecx is on
// a sibling disk) must NOT mount here: create_volume would
@@ -594,9 +591,10 @@ impl Store {
&collection,
vid,
);
let has_remote = crate::storage::disk_location::vif_references_remote_file(
&format!("{}.vif", base_path),
) || crate::storage::disk_location::vif_references_remote_file(
let has_remote = crate::storage::disk_location::vif_references_remote_file(&format!(
"{}.vif",
base_path
)) || crate::storage::disk_location::vif_references_remote_file(
&format!("{}.vif", idx_base),
);
if dat_exists || has_remote {
@@ -628,10 +626,12 @@ impl Store {
}
}
}
Err(last_err.unwrap_or_else(|| VolumeError::Io(io::Error::new(
io::ErrorKind::NotFound,
format!("volume {} not found on disk", vid),
))))
Err(last_err.unwrap_or_else(|| {
VolumeError::Io(io::Error::new(
io::ErrorKind::NotFound,
format!("volume {} not found on disk", vid),
))
}))
}
fn find_volume_file_base(&self, vid: VolumeId) -> Option<(usize, String, String)> {
@@ -970,12 +970,7 @@ impl Store {
);
continue;
}
tracing::info!(
volume_id = vid.0,
shard_id,
disk_id,
"UnmountEcShards"
);
tracing::info!(volume_id = vid.0, shard_id, disk_id, "UnmountEcShards");
self.locations[disk_id].unmount_ec_shards(vid, &[shard_id]);
}
// Go returns nil if shard not found (no error)
@@ -1149,7 +1144,8 @@ impl Store {
expired_vids.push(*vid);
} else {
let (_, io_count, quarantined) = ec_vol.get_io_error_state();
if quarantined || io_count >= crate::storage::erasure_coding::ec_volume::IO_ERROR_TOLERANCE
if quarantined
|| io_count >= crate::storage::erasure_coding::ec_volume::IO_ERROR_TOLERANCE
{
io_quarantined_vids.push(*vid);
} else {
@@ -1259,8 +1255,9 @@ impl Store {
collection,
vid,
);
let _ =
crate::storage::erasure_coding::ec_bitrot::remove_bitrot_sidecars(&idx_base);
let _ = crate::storage::erasure_coding::ec_bitrot::remove_bitrot_sidecars(
&idx_base,
);
}
}
}
@@ -1662,10 +1659,7 @@ mod tests {
let dir = tmp.path().to_str().unwrap();
let mut store = make_test_store(&[dir]);
let escaped = format!(
"{}/../evil_5.dat",
dir
);
let escaped = format!("{}/../evil_5.dat", dir);
let err = store
.mount_volume_by_id(VolumeId(5), Some("../evil"))
.unwrap_err();
@@ -1706,9 +1700,7 @@ mod tests {
.unwrap();
assert!(store.unmount_volume(VolumeId(7)));
store
.mount_volume_by_id(VolumeId(7), Some("coll"))
.unwrap();
store.mount_volume_by_id(VolumeId(7), Some("coll")).unwrap();
assert!(store.find_volume(VolumeId(7)).is_some());
let mut got = Needle {
@@ -1745,7 +1737,9 @@ mod tests {
data_size: 4,
..Needle::default()
};
store.write_volume_needle(VolumeId(9), &mut n, false).unwrap();
store
.write_volume_needle(VolumeId(9), &mut n, false)
.unwrap();
assert!(store.unmount_volume(VolumeId(9)));
// The hint is accepted and mounts the volume.
@@ -1788,7 +1782,9 @@ mod tests {
data_size: 7,
..Needle::default()
};
store.write_volume_needle(VolumeId(11), &mut n, false).unwrap();
store
.write_volume_needle(VolumeId(11), &mut n, false)
.unwrap();
assert!(store.unmount_volume(VolumeId(11)));
// Simulate an interrupted copy: drop a .note marker.
@@ -1843,7 +1839,9 @@ mod tests {
data_size: 4,
..Needle::default()
};
store.write_volume_needle(VolumeId(13), &mut n, false).unwrap();
store
.write_volume_needle(VolumeId(13), &mut n, false)
.unwrap();
assert!(store.unmount_volume(VolumeId(13)));
// No hint: the fallback scan finds the sidecar on disk 0 first (skip,
@@ -1896,10 +1894,14 @@ mod tests {
data_size: 5,
..Needle::default()
};
store.write_volume_needle(VolumeId(15), &mut n, false).unwrap();
store
.write_volume_needle(VolumeId(15), &mut n, false)
.unwrap();
assert!(store.unmount_volume(VolumeId(15)));
// Clear the low-space flag so mount_volume_by_id considers disk 0.
store.locations[0].is_disk_space_low.store(false, Ordering::Relaxed);
store.locations[0]
.is_disk_space_low
.store(false, Ordering::Relaxed);
// disk 0: a .dat that exists but is unreadable (chmod 000). The guard
// sees dat_exists=true (metadata succeeds, not a dir), but
@@ -2492,7 +2494,10 @@ mod tests {
// Mount an EC shard on disk 1 so has_ec_volume returns true.
std::fs::write(
format!("{}/{}_{}.ec00", store.locations[1].directory, collection, vid.0),
format!(
"{}/{}_{}.ec00",
store.locations[1].directory, collection, vid.0
),
b"shard data",
)
.unwrap();
@@ -2505,7 +2510,12 @@ mod tests {
std::fs::write(format!("{}.ecx", base), vec![0u8; 20]).unwrap();
let got = store.find_ec_shard_target_location(collection, vid, 10, &[]);
assert_eq!(got, Some(1), "expected the mounted disk to win; got {:?}", got);
assert_eq!(
got,
Some(1),
"expected the mounted disk to win; got {:?}",
got
);
}
/// Cold-volume case: no mount, no `.ecx` anywhere on this server.
@@ -2559,7 +2569,10 @@ mod tests {
// free shard slots remaining; the old formula would have
// rounded that to 0.
std::fs::write(
format!("{}/{}_{}.ec00", store.locations[1].directory, collection, vid.0),
format!(
"{}/{}_{}.ec00",
store.locations[1].directory, collection, vid.0
),
b"shard data",
)
.unwrap();
seaweed-volume/src/storage/store_ec_mirror.rs
+2 -4
View File
@@ -8,7 +8,7 @@ use std::path::Path;
use tracing::{info, warn};
use crate::storage::disk_location::{parse_collection_volume_id_pub, DiskLocation};
use crate::storage::disk_location::{DiskLocation, parse_collection_volume_id_pub};
use crate::storage::store::Store;
use crate::storage::types::VolumeId;
@@ -286,9 +286,7 @@ fn collect_shard_disk_volumes(loc: &DiskLocation) -> HashMap<EcKey, Vec<String>>
let Some((collection, vid)) = parse_collection_volume_id_pub(base) else {
continue;
};
out.entry(EcKey { collection, vid })
.or_default()
.push(name);
out.entry(EcKey { collection, vid }).or_default().push(name);
}
out
}
seaweed-volume/src/storage/store_ec_reconcile.rs
+68 -33
View File
@@ -122,9 +122,7 @@ impl Store {
let use_local_idx = std::path::Path::new(&local_ecx).exists()
|| std::path::Path::new(&local_ecx_in_data).exists();
if !use_local_idx
&& owner.location == loc_idx
&& owner.idx_dir == loc.idx_directory
if !use_local_idx && owner.location == loc_idx && owner.idx_dir == loc.idx_directory
{
// Same-disk no-op: load_all_ec_shards already
// tried and logged the failure.
@@ -593,7 +591,13 @@ mod tests {
std::fs::write(&p, b"shard data nonempty").unwrap();
}
fn write_index_files(idx_dir: &str, collection: &str, vid: u32, data_shards: u32, parity_shards: u32) {
fn write_index_files(
idx_dir: &str,
collection: &str,
vid: u32,
data_shards: u32,
parity_shards: u32,
) {
// Minimal sealed .ecx (the loader only opens the file; it
// doesn't parse it during placement).
std::fs::write(
@@ -947,15 +951,24 @@ mod tests {
// dir1 owns the .ecx and so already has shard 1 mounted via
// its own load_all_ec_shards.
let ev1 = store.locations[1].find_ec_volume(VolumeId(vid));
assert!(ev1.is_some(), "baseline broken: dir1 should have mounted shard 1");
assert!(
ev1.is_some(),
"baseline broken: dir1 should have mounted shard 1"
);
// dir0's shards must be reconciled across to its own
// ec_volumes map, pointing at dir1's idx dir.
let ev0 = store.locations[0]
.find_ec_volume(VolumeId(vid))
.expect("dir0 should now have an EcVolume after reconcile");
assert!(ev0.has_shard(0), "shard 0 missing from dir0 after reconcile");
assert!(ev0.has_shard(12), "shard 12 missing from dir0 after reconcile");
assert!(
ev0.has_shard(0),
"shard 0 missing from dir0 after reconcile"
);
assert!(
ev0.has_shard(12),
"shard 12 missing from dir0 after reconcile"
);
}
/// PR 9244 review case: idx_directory is configured but the
@@ -1064,7 +1077,13 @@ mod tests {
assert!(store.locations[0].find_ec_volume(VolumeId(vid)).is_none());
// Shard files must still exist on disk for operator recovery.
for sid in [0u8, 12u8] {
let p = format!("{}/{}_{}.ec{:02}", dir0.to_str().unwrap(), collection, vid, sid);
let p = format!(
"{}/{}_{}.ec{:02}",
dir0.to_str().unwrap(),
collection,
vid,
sid
);
assert!(
std::path::Path::new(&p).exists(),
"orphan shard {} was destroyed",
@@ -1129,10 +1148,12 @@ mod tests {
assert!(ev1.has_shard(6), "dir1 shard missing");
// Nothing left to recover.
assert!(store
.collect_ec_volumes_missing_index()
.iter()
.all(|m| m.vid != VolumeId(vid)));
assert!(
store
.collect_ec_volumes_missing_index()
.iter()
.all(|m| m.vid != VolumeId(vid))
);
}
#[test]
@@ -1164,10 +1185,12 @@ mod tests {
.unwrap();
}
assert!(store
.collect_ec_volumes_missing_index()
.iter()
.all(|m| m.vid != VolumeId(vid)));
assert!(
store
.collect_ec_volumes_missing_index()
.iter()
.all(|m| m.vid != VolumeId(vid))
);
}
/// Helper: build a 2-disk store where reconcile produces the
@@ -1245,7 +1268,11 @@ mod tests {
let vid = VolumeId(7010);
let all = store.find_all_ec_volumes(vid);
assert_eq!(all.len(), 2, "expected one EcVolume per disk holding the vid");
assert_eq!(
all.len(),
2,
"expected one EcVolume per disk holding the vid"
);
// Disk 0 carries shards 0 and 12; disk 1 carries shard 1.
assert!(all[0].has_shard(0));
@@ -1266,7 +1293,7 @@ mod tests {
#[test]
fn test_scrub_plans_reach_every_disk_through_the_store() {
use crate::storage::erasure_coding::ec_volume::{
merge_ec_runtimes, EcChecksumScrubPlan, EcLocalScrubPlan,
EcChecksumScrubPlan, EcLocalScrubPlan, merge_ec_runtimes,
};
let (store, _tmp) = build_split_disk_store(7030);
@@ -1281,8 +1308,15 @@ mod tests {
let merged = merge_ec_runtimes(&runtimes).expect("two runtimes merge");
assert!(merged.slots[0].is_some(), "disk 0's shard 0 unreachable");
assert!(merged.slots[12].is_some(), "disk 0's shard 12 unreachable");
assert!(merged.slots[1].is_some(), "disk 1's shard 1 unreachable — the bug");
assert!(merged.skipped.is_empty(), "same generation: {:?}", merged.skipped);
assert!(
merged.slots[1].is_some(),
"disk 1's shard 1 unreachable — the bug"
);
assert!(
merged.skipped.is_empty(),
"same generation: {:?}",
merged.skipped
);
// Shard 1 is owned by the sibling runtime, not the anchor.
let (owner, _) = merged.slots[1].unwrap();
@@ -1389,17 +1423,22 @@ mod tests {
let (_ev, dirs) = store.collect_ec_shard_dirs(vid, max_shards).unwrap();
// Shards 0 and 12 → disk 0's directory.
assert_eq!(dirs[0].as_deref(), Some(store.locations[0].directory.as_str()));
assert_eq!(dirs[12].as_deref(), Some(store.locations[0].directory.as_str()));
assert_eq!(
dirs[0].as_deref(),
Some(store.locations[0].directory.as_str())
);
assert_eq!(
dirs[12].as_deref(),
Some(store.locations[0].directory.as_str())
);
// Shard 1 → disk 1's directory.
assert_eq!(dirs[1].as_deref(), Some(store.locations[1].directory.as_str()));
assert_eq!(
dirs[1].as_deref(),
Some(store.locations[1].directory.as_str())
);
// Unmounted shards → None.
for sid in [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 13] {
assert_eq!(
dirs[sid], None,
"shard {} unexpectedly reported a dir",
sid,
);
assert_eq!(dirs[sid], None, "shard {} unexpectedly reported a dir", sid,);
}
}
@@ -1708,11 +1747,7 @@ mod tests {
vec![0u8; 20],
)
.unwrap();
std::fs::write(
ec_dir.join(format!("{}_{}.ecj", collection, vid)),
b"",
)
.unwrap();
std::fs::write(ec_dir.join(format!("{}_{}.ecj", collection, vid)), b"").unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
seaweed-volume/src/storage/volume.rs
+137 -93
View File
@@ -12,18 +12,18 @@
use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Condvar, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use tracing::{error, info, warn};
use crate::storage::idx;
use crate::storage::needle::needle::{self, get_actual_size, Needle, NeedleError};
use crate::storage::needle::needle::{self, Needle, NeedleError, get_actual_size};
use crate::storage::needle_map::sorted_file::SortedFileNeedleMap;
use crate::storage::needle_map::{CompactNeedleMap, NeedleMap, NeedleMapKind, RedbNeedleMap};
use crate::storage::super_block::{ReplicaPlacement, SuperBlock, SUPER_BLOCK_SIZE};
use crate::storage::super_block::{ReplicaPlacement, SUPER_BLOCK_SIZE, SuperBlock};
use crate::storage::types::*;
use crate::storage::volume_open::open_volume_file;
@@ -112,8 +112,7 @@ pub fn is_storage_io_error(e: &io::Error) -> bool {
{
const ERROR_CRC: i32 = 23;
const ERROR_IO_DEVICE: i32 = 1117;
return e.raw_os_error() == Some(ERROR_CRC)
|| e.raw_os_error() == Some(ERROR_IO_DEVICE);
return e.raw_os_error() == Some(ERROR_CRC) || e.raw_os_error() == Some(ERROR_IO_DEVICE);
}
#[cfg(not(any(unix, windows)))]
{
@@ -239,9 +238,19 @@ struct OldVersionVifVolumeInfo {
pub replication: String,
#[serde(default, alias = "bytesOffset", alias = "BytesOffset")]
pub bytes_offset: u32,
#[serde(default, alias = "datFileSize", alias = "dat_file_size", with = "string_or_i64")]
#[serde(
default,
alias = "datFileSize",
alias = "dat_file_size",
with = "string_or_i64"
)]
pub dat_file_size: i64,
#[serde(default, alias = "destroyTime", alias = "DestroyTime", with = "string_or_u64")]
#[serde(
default,
alias = "destroyTime",
alias = "DestroyTime",
with = "string_or_u64"
)]
pub destroy_time: u64,
#[serde(default, alias = "readOnly", alias = "read_only")]
pub read_only: bool,
@@ -810,7 +819,8 @@ impl Volume {
self.dat_file = Some(file);
}
Err(e) if e.kind() == io::ErrorKind::PermissionDenied => {
self.dat_file = Some(open_volume_file(OpenOptions::new().read(true), &dat_path)?);
self.dat_file =
Some(open_volume_file(OpenOptions::new().read(true), &dat_path)?);
self.no_write_or_delete = true;
}
Err(e) => return Err(e.into()),
@@ -1368,7 +1378,12 @@ impl Volume {
return Ok(0);
}
match self.read_needle_data_at_unlocked(n, nv.offset.to_actual_offset(), read_size, read_option) {
match self.read_needle_data_at_unlocked(
n,
nv.offset.to_actual_offset(),
read_size,
read_option,
) {
Ok(()) => self.check_read_write_error(None),
Err(VolumeError::Io(ref e)) => {
self.check_read_write_error(Some(e));
@@ -1540,10 +1555,7 @@ impl Volume {
)));
}
let mut meta_buf = vec![0u8; meta_size as usize];
self.read_exact_at_backend(
&mut meta_buf,
(offset + NEEDLE_HEADER_SIZE as i64) as u64,
)?;
self.read_exact_at_backend(&mut meta_buf, (offset + NEEDLE_HEADER_SIZE as i64) as u64)?;
n.read_paged_meta(&header_buf, &meta_buf, offset, size, version)?;
} else {
// V2/V3: extract DataSize from bytes 16..20
@@ -1662,9 +1674,9 @@ impl Volume {
};
let source = match (self.dat_file.as_ref(), self.remote_dat_file.as_ref()) {
(Some(dat_file), _) => NeedleStreamSource::Local(
dat_file.try_clone().map_err(VolumeError::Io)?,
),
(Some(dat_file), _) => {
NeedleStreamSource::Local(dat_file.try_clone().map_err(VolumeError::Io)?)
}
(None, Some(remote_dat_file)) => NeedleStreamSource::Remote(remote_dat_file.clone()),
(None, None) => return Err(VolumeError::StreamingUnsupported),
};
@@ -2294,10 +2306,7 @@ impl Volume {
if idx_size % NEEDLE_MAP_ENTRY_SIZE as i64 != 0 {
return Err(VolumeError::Io(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"index file's size is {} bytes, maybe corrupted",
idx_size
),
format!("index file's size is {} bytes, maybe corrupted", idx_size),
)));
}
@@ -2542,7 +2551,12 @@ impl Volume {
.map_err(|_| size_mismatch_error(actual_offset, needle_id, needle_size, size))?;
let (_, alt_id, alt_size) = Needle::parse_header(&alt_header);
if alt_size != size {
return Err(size_mismatch_error(actual_offset, needle_id, needle_size, size));
return Err(size_mismatch_error(
actual_offset,
needle_id,
needle_size,
size,
));
}
checked_offset = alt_offset;
checked_id = alt_id;
@@ -2648,8 +2662,8 @@ impl Volume {
/// corruption. Returns the open file + size, or the messages to report.
fn open_index_for_scrub(&self) -> Result<(File, i64), Vec<String>> {
let idx_path = self.file_name(".idx");
let idx_file =
File::open(&idx_path).map_err(|e| vec![format!("open index file {}: {}", idx_path, e)])?;
let idx_file = File::open(&idx_path)
.map_err(|e| vec![format!("open index file {}: {}", idx_path, e)])?;
let idx_file_size = idx_file
.metadata()
.map_err(|e| vec![format!("stat index file {}: {}", idx_path, e)])?
@@ -2667,7 +2681,7 @@ impl Volume {
return Err(vec![format!(
"stat data file for volume {}: {}",
self.id.0, e
)])
)]);
}
}
}
@@ -2715,42 +2729,43 @@ impl Volume {
let mut count: u64 = 0;
let mut total_read: i64 = 0;
let walk = crate::storage::idx::walk_index_file(&mut idx_file, 0, |needle_id, offset, size| {
count += 1;
// A remote-tier delete records an offset-0 tombstone with no physical
// .dat bytes, so it must not contribute to total_read.
if offset.is_zero() && size.is_deleted() {
return Ok(());
}
// Deleted needles still occupy .dat space: count their size, don't read.
total_read += get_actual_size(size, version);
if size.is_deleted() {
return Ok(());
}
let actual_offset = offset.to_actual_offset();
if actual_offset < 0 || actual_offset as u64 >= dat_size {
broken.push(format!(
"needle {} offset {} out of range (dat_size={})",
needle_id.0, actual_offset, dat_size
));
return Ok(());
}
// Lock held above; read directly via the unlocked path (matches Go).
let mut n = Needle {
id: needle_id,
..Needle::default()
};
let mut read_option = ReadOption::default();
if let Err(e) =
self.read_needle_data_at_unlocked(&mut n, actual_offset, size, &mut read_option)
{
broken.push(format!(
"failed to read needle {} on volume {}: {}",
needle_id.0, self.id.0, e
));
}
Ok(())
});
let walk =
crate::storage::idx::walk_index_file(&mut idx_file, 0, |needle_id, offset, size| {
count += 1;
// A remote-tier delete records an offset-0 tombstone with no physical
// .dat bytes, so it must not contribute to total_read.
if offset.is_zero() && size.is_deleted() {
return Ok(());
}
// Deleted needles still occupy .dat space: count their size, don't read.
total_read += get_actual_size(size, version);
if size.is_deleted() {
return Ok(());
}
let actual_offset = offset.to_actual_offset();
if actual_offset < 0 || actual_offset as u64 >= dat_size {
broken.push(format!(
"needle {} offset {} out of range (dat_size={})",
needle_id.0, actual_offset, dat_size
));
return Ok(());
}
// Lock held above; read directly via the unlocked path (matches Go).
let mut n = Needle {
id: needle_id,
..Needle::default()
};
let mut read_option = ReadOption::default();
if let Err(e) =
self.read_needle_data_at_unlocked(&mut n, actual_offset, size, &mut read_option)
{
broken.push(format!(
"failed to read needle {} on volume {}: {}",
needle_id.0, self.id.0, e
));
}
Ok(())
});
if let Err(e) = walk {
broken.push(format!("walk index file: {}", e));
}
@@ -3435,8 +3450,7 @@ impl Volume {
// Patch appendAtNs timestamp into V3 blobs (matches Go WriteNeedleBlob L64-77)
let mut blob_buf;
let blob_to_write = if self.version() == VERSION_3 {
let ts_offset =
NEEDLE_HEADER_SIZE + size.0 as usize + NEEDLE_CHECKSUM_SIZE;
let ts_offset = NEEDLE_HEADER_SIZE + size.0 as usize + NEEDLE_CHECKSUM_SIZE;
if ts_offset + TIMESTAMP_SIZE > needle_blob.len() {
return Err(VolumeError::Io(io::Error::new(
io::ErrorKind::InvalidData,
@@ -3495,7 +3509,9 @@ impl Volume {
// This happens for .sdx converted back to normal .idx
// where deleted entry size is missing
let dat_file_size = self.dat_file_size().unwrap_or(0);
deleted = dat_file_size.saturating_sub(content).saturating_sub(SUPER_BLOCK_SIZE as u64);
deleted = dat_file_size
.saturating_sub(content)
.saturating_sub(SUPER_BLOCK_SIZE as u64);
file_size = dat_file_size;
}
@@ -4234,8 +4250,7 @@ impl Volume {
if has_ecx(&volume_file_name(&self.dir_idx, &self.collection, self.id)) {
return true;
}
self.dir != self.dir_idx
&& has_ecx(&volume_file_name(&self.dir, &self.collection, self.id))
self.dir != self.dir_idx && has_ecx(&volume_file_name(&self.dir, &self.collection, self.id))
}
/// Check if an I/O error is a storage-media failure and record it for
@@ -4407,11 +4422,7 @@ fn get_append_at_ns(last: u64) -> u64 {
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as u64;
if now <= last {
last + 1
} else {
now
}
if now <= last { last + 1 } else { now }
}
/// Remove all files associated with a volume.
@@ -4570,9 +4581,9 @@ mod tests {
use tempfile::TempDir;
fn spawn_fake_s3_server(body: Vec<u8>) -> (String, tokio::sync::oneshot::Sender<()>) {
use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
use axum::routing::any;
use axum::Router;
use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
use axum::routing::any;
let body = Arc::new(body);
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
@@ -5012,7 +5023,6 @@ mod tests {
assert_eq!(read_n.data, b"first-copy");
}
/// A durable write that dedups against an earlier non-durable one still has
/// to flush the index. The row that earlier write left behind can be sitting
/// in the page cache, so acking without flushing is the same false promise
@@ -5681,8 +5691,13 @@ mod tests {
.append(true)
.open(v.file_name(".idx"))
.unwrap();
crate::storage::idx::write_index_entry(&mut f, NeedleId(2), Offset::from_actual_offset(0), Size(-1))
.unwrap();
crate::storage::idx::write_index_entry(
&mut f,
NeedleId(2),
Offset::from_actual_offset(0),
Size(-1),
)
.unwrap();
f.sync_all().unwrap();
drop(f);
@@ -5960,7 +5975,10 @@ mod tests {
v.relocate_index_to(idx).unwrap();
assert!(Path::new(&idx_dir_idx).exists(), "index moved to the idx dir");
assert!(
Path::new(&idx_dir_idx).exists(),
"index moved to the idx dir"
);
assert!(
!Path::new(&data_idx).exists(),
"index gone from the data dir"
@@ -6180,7 +6198,10 @@ mod tests {
let bodies: Vec<&[u8]> = needles.iter().map(|n| n.data.as_slice()).collect();
assert_eq!(ids, vec![20, 10]);
assert_eq!(bodies, vec![b"second".as_slice(), b"first-overwrite".as_slice()]);
assert_eq!(
bodies,
vec![b"second".as_slice(), b"first-overwrite".as_slice()]
);
}
#[test]
@@ -6363,8 +6384,7 @@ mod tests {
let bad_key = NeedleId(9999);
let bad_offset = Offset::from_actual_offset((dat_size + 1024 * 1024) as i64);
let bad_size = Size(2048);
v.nm
.as_mut()
v.nm.as_mut()
.expect("needle map present")
.put(bad_key, bad_offset, bad_size)
.unwrap();
@@ -6433,8 +6453,7 @@ mod tests {
// Sanity: a fresh load walk over the healthy .idx puts max_needle_end
// somewhere inside the .dat.
let mut idx_reader = File::open(&idx_path).unwrap();
let healthy_nm =
CompactNeedleMap::load_from_idx(&mut idx_reader, version).unwrap();
let healthy_nm = CompactNeedleMap::load_from_idx(&mut idx_reader, version).unwrap();
let healthy_end = healthy_nm.max_needle_end();
assert!(
healthy_end > 0 && healthy_end <= dat_size,
@@ -6787,13 +6806,12 @@ mod tests {
"tier-up must install the sorted map without waiting for a restart"
);
// The index still answers, now off .sdx.
let nv = v
.nm
.as_ref()
.unwrap()
.get(NeedleId(1))
.unwrap()
.expect("needle 1 still resolves through the sorted index");
let nv =
v.nm.as_ref()
.unwrap()
.get(NeedleId(1))
.unwrap()
.expect("needle 1 still resolves through the sorted index");
assert!(!nv.size.is_deleted());
}
@@ -6811,7 +6829,10 @@ mod tests {
v.open_local_dat_backend().unwrap();
assert!(!v.has_remote_file);
assert!(!v.is_read_only(), "tier-down should publish a writable volume");
assert!(
!v.is_read_only(),
"tier-down should publish a writable volume"
);
assert!(
!matches!(v.nm, Some(NeedleMap::SortedFile(_))),
"tier-down must swap the read-only map out before writes are allowed"
@@ -7226,7 +7247,10 @@ mod tests {
assert!(v.has_remote_file);
let Some(NeedleMap::SortedFile(ref nm)) = v.nm else {
panic!("tiered volume should search the on-disk .sdx, got {:?}", v.nm.is_some());
panic!(
"tiered volume should search the on-disk .sdx, got {:?}",
v.nm.is_some()
);
};
let idx_path = nm.index_file_name().to_string();
let sdx_path = nm.db_file_name().to_string();
@@ -7247,7 +7271,16 @@ mod tests {
}
// Lookups still resolve, straight off .sdx.
assert!(!v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().unwrap().size.is_deleted());
assert!(
!v.nm
.as_ref()
.unwrap()
.get(NeedleId(3))
.unwrap()
.unwrap()
.size
.is_deleted()
);
assert!(v.nm.as_ref().unwrap().get(NeedleId(9)).unwrap().is_none());
// Deletes are still allowed on a tiered volume and land in .idx.
@@ -7265,7 +7298,15 @@ mod tests {
idx_before + NEEDLE_MAP_ENTRY_SIZE as u64,
"the tombstone must be appended to the .idx tail"
);
assert!(v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().unwrap().size.is_deleted());
assert!(
v.nm.as_ref()
.unwrap()
.get(NeedleId(3))
.unwrap()
.unwrap()
.size
.is_deleted()
);
if crate::storage::needle_map::file_pool::open_index_fds(tmp.path()).is_some() {
drop_pooled();
@@ -7383,7 +7424,10 @@ mod tests {
..Needle::default()
};
v.read_needle(&mut probe).unwrap();
assert_eq!(std::str::from_utf8(&probe.data).unwrap(), "after-mark-writable");
assert_eq!(
std::str::from_utf8(&probe.data).unwrap(),
"after-mark-writable"
);
}
// readOnlyCanDelete rejects writes, keeps accepting deletes, and survives
seaweed-volume/src/storage/volume_idx_rebuild.rs
+2 -2
View File
@@ -10,7 +10,7 @@ use crate::storage::needle::Needle;
use crate::storage::super_block::SuperBlock;
use crate::storage::types::*;
use crate::storage::volume::{
fsync_dir, needle_disk_end, scan_volume_file, Volume, VolumeError, VolumeFileVisitor,
Volume, VolumeError, VolumeFileVisitor, fsync_dir, needle_disk_end, scan_volume_file,
};
/// Writes one .idx row per .dat record, in .dat append order, which is the
@@ -105,8 +105,8 @@ impl Volume {
#[cfg(test)]
mod tests {
use crate::storage::needle::crc::CRC;
use crate::storage::needle::Needle;
use crate::storage::needle::crc::CRC;
use crate::storage::needle_map::NeedleMapKind;
use crate::storage::types::*;
use crate::storage::volume::{Volume, VolumeSpec};
seaweed-volume/src/storage/volume_idx_repair.rs
+2 -2
View File
@@ -9,10 +9,10 @@ use std::path::Path;
use tracing::info;
use crate::storage::idx;
use crate::storage::needle::needle::needle_body_length;
use crate::storage::needle::Needle;
use crate::storage::needle::needle::needle_body_length;
use crate::storage::types::*;
use crate::storage::volume::{fsync_dir, Volume, VolumeError};
use crate::storage::volume::{Volume, VolumeError, fsync_dir};
/// Needles found in the head of .dat, keyed by id, plus the ids in .dat order.
type DatHeadNeedles = (HashMap<NeedleId, (Offset, Size)>, Vec<NeedleId>);
seaweed-volume/tests/admin_auth_coverage.rs
+65 -16
View File
@@ -20,26 +20,71 @@ use std::collections::{HashMap, HashSet};
fn ungated_handlers() -> HashMap<&'static str, &'static str> {
[
// Cluster-internal: issued volume server -> volume server.
("copy_file", "replica sync and EC task pull whole files from a peer"),
("read_needle_blob", "replica sync, vacuum and EC rebuild read needles from a peer"),
("read_needle_meta", "replica sync compares needle metadata across peers"),
(
"copy_file",
"replica sync and EC task pull whole files from a peer",
),
(
"read_needle_blob",
"replica sync, vacuum and EC rebuild read needles from a peer",
),
(
"read_needle_meta",
"replica sync compares needle metadata across peers",
),
("write_needle_blob", "replica sync repairs a peer's needle"),
("receive_file", "EC shard distribution pushes shards to a peer"),
("read_volume_file_status", "the copy path queries the source volume server"),
("volume_ec_shard_read", "a volume server reads EC shards held by a peer"),
("volume_ec_blob_delete", "EC delete is fanned out to the shard holders"),
("volume_ec_shards_info", "EC verification polls shard holders"),
("volume_ec_shards_mount", "EC shard distribution mounts on the receiving peer"),
("volume_incremental_copy", "volume backup pulls increments from a peer"),
("volume_sync_status", "sync compares volume state across peers"),
("volume_tail_sender", "the tail source streams to the receiving peer"),
("volume_status", "replica sync and the master's vacuum loop poll volume status"),
(
"receive_file",
"EC shard distribution pushes shards to a peer",
),
(
"read_volume_file_status",
"the copy path queries the source volume server",
),
(
"volume_ec_shard_read",
"a volume server reads EC shards held by a peer",
),
(
"volume_ec_blob_delete",
"EC delete is fanned out to the shard holders",
),
(
"volume_ec_shards_info",
"EC verification polls shard holders",
),
(
"volume_ec_shards_mount",
"EC shard distribution mounts on the receiving peer",
),
(
"volume_incremental_copy",
"volume backup pulls increments from a peer",
),
(
"volume_sync_status",
"sync compares volume state across peers",
),
(
"volume_tail_sender",
"the tail source streams to the receiving peer",
),
(
"volume_status",
"replica sync and the master's vacuum loop poll volume status",
),
// Read-only or liveness: no state change.
("ping", "liveness probe"),
("get_state", "read-only volume server state"),
("query", "read-only data query"),
("vacuum_volume_check", "read-only garbage ratio; the vacuum steps that act on it are gated"),
("volume_server_status", "read-only status, the gRPC counterpart of the /status page"),
(
"vacuum_volume_check",
"read-only garbage ratio; the vacuum steps that act on it are gated",
),
(
"volume_server_status",
"read-only status, the gRPC counterpart of the /status page",
),
]
.into_iter()
.collect()
@@ -129,5 +174,9 @@ fn volume_server_admin_auth_coverage() {
}
}
assert!(problems.is_empty(), "admin-auth coverage gaps:\n{}", problems.join("\n"));
assert!(
problems.is_empty(),
"admin-auth coverage gaps:\n{}",
problems.join("\n")
);
}
seaweed-volume/tests/http_integration.rs
+2 -2
View File
@@ -12,8 +12,8 @@ use tower::ServiceExt; // for `oneshot`
use seaweed_volume::security::{Guard, SigningKey};
use seaweed_volume::server::volume_server::{
build_admin_router, build_admin_router_with_ui, build_metrics_router, build_public_router,
VolumeServerState,
VolumeServerState, build_admin_router, build_admin_router_with_ui, build_metrics_router,
build_public_router,
};
use seaweed_volume::storage::needle_map::NeedleMapKind;
use seaweed_volume::storage::store::Store;
seaweed-worker/crates/core/src/config_form.rs
+1 -1
View File
@@ -8,7 +8,7 @@
use std::collections::HashMap;
use crate::pb::{
config_value::Kind, ConfigField, ConfigFieldType, ConfigForm, ConfigSection, ConfigValue,
ConfigField, ConfigFieldType, ConfigForm, ConfigSection, ConfigValue, config_value::Kind,
};
pub fn int_value(value: i64) -> ConfigValue {
seaweed-worker/crates/core/src/metrics.rs
+24 -15
View File
@@ -270,10 +270,10 @@ impl Metrics {
/// logged rather than fatal: a worker that cannot publish metrics should still
/// do its work.
pub async fn serve(metrics: Metrics, addr: SocketAddr) -> Result<()> {
use axum::Router;
use axum::extract::State;
use axum::http::StatusCode;
use axum::routing::get;
use axum::Router;
let app = Router::new()
.route("/health", get(|| async { StatusCode::OK }))
@@ -355,28 +355,37 @@ mod tests {
metrics.stream_connected();
assert!(metrics.is_ready());
assert!(metrics
.gather()
.unwrap()
.contains("SeaweedFS_worker_connected 1"));
assert!(
metrics
.gather()
.unwrap()
.contains("SeaweedFS_worker_connected 1")
);
metrics.stream_ended("closed");
assert!(!metrics.is_ready());
assert!(metrics
.gather()
.unwrap()
.contains("SeaweedFS_worker_connected 0"));
assert!(metrics
.gather()
.unwrap()
.contains("SeaweedFS_worker_stream_events_total{event=\"closed\"} 1"));
assert!(
metrics
.gather()
.unwrap()
.contains("SeaweedFS_worker_connected 0")
);
assert!(
metrics
.gather()
.unwrap()
.contains("SeaweedFS_worker_stream_events_total{event=\"closed\"} 1")
);
}
#[test]
fn build_info_names_the_worker() {
let text = metrics().gather().expect("gather");
assert!(text
.contains("SeaweedFS_worker_build_info{version=\"0.1.0\",worker_id=\"worker-1\"} 1"));
assert!(
text.contains(
"SeaweedFS_worker_build_info{version=\"0.1.0\",worker_id=\"worker-1\"} 1"
)
);
}
// A format's own numbers land on the same registry, so one endpoint serves
seaweed-worker/crates/core/src/senders.rs
+2 -2
View File
@@ -2,8 +2,8 @@ use anyhow::Result;
use tokio::sync::mpsc;
use crate::pb::{
worker_to_admin_message::Body, ActivityEvent, DetectionComplete, DetectionProposals,
JobCompleted, JobProgressUpdate, WorkerObservations, WorkerToAdminMessage,
ActivityEvent, DetectionComplete, DetectionProposals, JobCompleted, JobProgressUpdate,
WorkerObservations, WorkerToAdminMessage, worker_to_admin_message::Body,
};
/// Replies to one detection request.
seaweed-worker/crates/core/src/stream.rs
+5 -5
View File
@@ -1,8 +1,8 @@
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{anyhow, Context, Result};
use tokio::sync::{mpsc, Semaphore};
use anyhow::{Context, Result, anyhow};
use tokio::sync::{Semaphore, mpsc};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
use tracing::{info, warn};
@@ -10,11 +10,11 @@ use tracing::{info, warn};
use crate::config::WorkerOptions;
use crate::metrics::Metrics;
use crate::pb::{
ConfigSchemaResponse, ExecuteJobRequest, JobCompleted, ObjectPreviewResponse, PreviewRow,
RequestObjectPreview, RunDetectionRequest, RunningWork, WorkerHeartbeat, WorkerHello,
admin_to_worker_message::Body as AdminBody,
plugin_control_service_client::PluginControlServiceClient,
worker_to_admin_message::Body as WorkerBody, ConfigSchemaResponse, ExecuteJobRequest,
JobCompleted, ObjectPreviewResponse, PreviewRow, RequestObjectPreview, RunDetectionRequest,
RunningWork, WorkerHeartbeat, WorkerHello,
worker_to_admin_message::Body as WorkerBody,
};
use crate::registry::Registry;
use crate::senders::{MeteredSender, StreamSender};
seaweed-worker/crates/lance/src/catalog/mod.rs
+1 -1
View File
@@ -7,4 +7,4 @@
pub mod namespace;
pub use namespace::{parse_id, NamespaceClient, TableDescription};
pub use namespace::{NamespaceClient, TableDescription, parse_id};
seaweed-worker/crates/lance/src/dataset.rs
+1 -1
View File
@@ -8,8 +8,8 @@
use std::collections::HashMap;
use anyhow::{Context, Result};
use lance::dataset::builder::DatasetBuilder;
use lance::dataset::Dataset;
use lance::dataset::builder::DatasetBuilder;
use crate::catalog::{NamespaceClient, TableDescription};
seaweed-worker/crates/lance/src/jobs/cleanup.rs
+3 -3
View File
@@ -1,9 +1,9 @@
use std::collections::HashMap;
use anyhow::{anyhow, Context, Result};
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use chrono::{Duration, Utc};
use lance::dataset::cleanup::{cleanup_old_versions, CleanupPolicy};
use lance::dataset::cleanup::{CleanupPolicy, cleanup_old_versions};
use seaweed_worker_core::config_form::{form, int_or, int_value, number_field};
use seaweed_worker_core::pb::{
ConfigValue, DetectionComplete, DetectionProposals, ExecuteJobRequest, JobCompleted,
@@ -13,7 +13,7 @@ use seaweed_worker_core::pb::{
use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler};
use tracing::warn;
use crate::catalog::{parse_id, NamespaceClient};
use crate::catalog::{NamespaceClient, parse_id};
use crate::dataset;
use crate::jobs::{clamp, string_list, table_id};
seaweed-worker/crates/lance/src/jobs/compact.rs
+4 -4
View File
@@ -1,8 +1,8 @@
use std::collections::HashMap;
use anyhow::{anyhow, Context, Result};
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use lance::dataset::optimize::{compact_files, CompactionOptions};
use lance::dataset::optimize::{CompactionOptions, compact_files};
use seaweed_worker_core::config_form::{form, int_or, int_value, number_field, string_value};
use seaweed_worker_core::pb::{
ConfigValue, DetectionComplete, DetectionProposals, ExecuteJobRequest, JobCompleted,
@@ -12,9 +12,9 @@ use seaweed_worker_core::pb::{
use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler};
use tracing::warn;
use crate::catalog::{parse_id, NamespaceClient};
use crate::catalog::{NamespaceClient, parse_id};
use crate::dataset;
use crate::jobs::{clamp, observation, string_list, table_id, FORMAT};
use crate::jobs::{FORMAT, clamp, observation, string_list, table_id};
pub const JOB_TYPE: &str = "lance_compact";
seaweed-worker/crates/lance/src/jobs/indices.rs
+2 -2
View File
@@ -1,6 +1,6 @@
use std::collections::HashMap;
use anyhow::{anyhow, Context, Result};
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use lance::index::DatasetIndexExt;
use lance_index::optimize::OptimizeOptions;
@@ -13,7 +13,7 @@ use seaweed_worker_core::pb::{
use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler};
use tracing::warn;
use crate::catalog::{parse_id, NamespaceClient};
use crate::catalog::{NamespaceClient, parse_id};
use crate::dataset::{self, OpenTable};
use crate::jobs::{clamp, string_list, table_id};
seaweed-worker/crates/lance/src/jobs/mod.rs
+1 -1
View File
@@ -14,8 +14,8 @@ pub mod sort;
use std::collections::HashMap;
use std::sync::Arc;
use seaweed_worker_core::pb::{config_value::Kind, ConfigValue, ObjectObservation, StringList};
use seaweed_worker_core::JobHandler;
use seaweed_worker_core::pb::{ConfigValue, ObjectObservation, StringList, config_value::Kind};
use crate::catalog::parse_id;
seaweed-worker/crates/lance/src/jobs/sort.rs
+7 -7
View File
@@ -1,14 +1,14 @@
use std::collections::HashMap;
use std::sync::Arc;
use anyhow::{anyhow, Context, Result};
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use lance::dataset::scanner::ColumnOrdering;
use lance::dataset::transaction::Operation;
use lance::dataset::write::{CommitBuilder, InsertBuilder};
use lance::dataset::{WriteDestination, WriteMode, WriteParams};
use lance::index::DatasetIndexExt;
use lance_datafusion::exec::{execute_plan, LanceExecutionOptions};
use lance_datafusion::exec::{LanceExecutionOptions, execute_plan};
use seaweed_worker_core::config_form::{form, int_or, int_value, string_or, string_value};
use seaweed_worker_core::pb::{
ConfigValue, DetectionComplete, DetectionProposals, ExecuteJobRequest, JobCompleted,
@@ -17,15 +17,15 @@ use seaweed_worker_core::pb::{
};
use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler};
use seaweed_worker_sort::{
config_fields, resolve, verdict, FragmentSummary, SortSpec, SortState,
CONFIG_MAX_ROWS_PER_FILE, CONFIG_MEMORY_BUDGET_MB, CONFIG_MIN_UNSORTED_ROWS,
CONFIG_SORT_FIELDS, DECLARED_FIELDS_KEY,
CONFIG_SORT_FIELDS, DECLARED_FIELDS_KEY, FragmentSummary, SortSpec, SortState, config_fields,
resolve, verdict,
};
use tracing::warn;
use crate::catalog::{parse_id, NamespaceClient};
use crate::catalog::{NamespaceClient, parse_id};
use crate::dataset;
use crate::jobs::{clamp, observation, string_list, table_id, FORMAT};
use crate::jobs::{FORMAT, clamp, observation, string_list, table_id};
pub const JOB_TYPE: &str = "lance_sort";
@@ -424,7 +424,7 @@ impl JobHandler for SortHandler {
other => {
return Err(anyhow!(
"a sorted rewrite produced {other} instead of an overwrite"
))
));
}
}
seaweed-worker/crates/lance/tests/common/mod.rs
+2 -2
View File
@@ -10,8 +10,8 @@ use std::sync::Mutex;
use anyhow::Result;
use seaweed_worker_core::pb::{
config_value::Kind, ActivityEvent, ConfigValue, DetectionComplete, DetectionProposals,
JobCompleted, JobProgressUpdate, JobProposal, ObjectObservation, WorkerObservations,
ActivityEvent, ConfigValue, DetectionComplete, DetectionProposals, JobCompleted,
JobProgressUpdate, JobProposal, ObjectObservation, WorkerObservations, config_value::Kind,
};
use seaweed_worker_core::{DetectionSender, ExecutionSender};
seaweed-worker/crates/lance/tests/compaction.rs
+4 -4
View File
@@ -8,7 +8,7 @@ use std::collections::HashMap;
use anyhow::Result;
use seaweed_worker_core::pb::{
config_value::Kind, ExecuteJobRequest, JobSpec, RunDetectionRequest,
ExecuteJobRequest, JobSpec, RunDetectionRequest, config_value::Kind,
};
use seaweed_worker_core::{JobHandler, PreviewProvider};
use weed_lance_worker::catalog::NamespaceClient;
@@ -18,7 +18,7 @@ use weed_lance_worker::jobs::indices::OptimizeIndicesHandler;
use weed_lance_worker::preview::LancePreview;
mod common;
use common::{fallback, int_config, namespace_url, Recorder};
use common::{Recorder, fallback, int_config, namespace_url};
/// These tests drive one live gateway and one shared catalog: `list_all_tables`
/// sweeps everything, so a table another test is writing shows up in this test's
@@ -121,10 +121,10 @@ async fn seed_table(
// The index is built after the first batch, so everything appended
// afterwards is a row it does not cover.
if with_index && i == 0 {
use lance::index::vector::VectorIndexParams;
use lance::index::DatasetIndexExt;
use lance_index::vector::{ivf::IvfBuildParams, pq::PQBuildParams};
use lance::index::vector::VectorIndexParams;
use lance_index::IndexType;
use lance_index::vector::{ivf::IvfBuildParams, pq::PQBuildParams};
let mut dataset = dataset;
let params = VectorIndexParams::with_ivf_pq_params(
seaweed-worker/crates/lance/tests/lifecycle.rs
+2 -2
View File
@@ -12,13 +12,13 @@
use std::collections::HashMap;
use seaweed_worker_core::pb::{ConfigValue, ExecuteJobRequest, JobSpec, RunDetectionRequest};
use seaweed_worker_core::JobHandler;
use seaweed_worker_core::pb::{ConfigValue, ExecuteJobRequest, JobSpec, RunDetectionRequest};
use weed_lance_worker::jobs::cleanup::{self, CleanupVersionsHandler};
use weed_lance_worker::jobs::compact::{CompactHandler, JOB_TYPE as COMPACT_JOB_TYPE};
mod common;
use common::{fallback, int_config, namespace_url, Recorder};
use common::{Recorder, fallback, int_config, namespace_url};
fn table() -> Option<String> {
std::env::var("WEED_LANCE_TABLE")
seaweed-worker/crates/lance/tests/sort.rs
+5 -5
View File
@@ -5,15 +5,15 @@
//! the commit is the half worth testing.
use anyhow::Result;
use seaweed_worker_core::pb::{
config_value::Kind, ConfigValue, ExecuteJobRequest, JobSpec, RunDetectionRequest,
};
use seaweed_worker_core::JobHandler;
use seaweed_worker_core::pb::{
ConfigValue, ExecuteJobRequest, JobSpec, RunDetectionRequest, config_value::Kind,
};
use weed_lance_worker::catalog::NamespaceClient;
use weed_lance_worker::jobs::sort::{SortHandler, JOB_TYPE};
use weed_lance_worker::jobs::sort::{JOB_TYPE, SortHandler};
mod common;
use common::{fallback, namespace_url, Recorder};
use common::{Recorder, fallback, namespace_url};
/// One live gateway and one shared catalog, and `list_all_tables` sweeps
/// everything, so these tests take a lock the way the compaction ones do.
seaweed-worker/crates/sort/src/lib.rs
+1 -1
View File
@@ -14,7 +14,7 @@
use std::collections::{HashMap, HashSet};
use std::fmt;
use anyhow::{bail, Context, Result};
use anyhow::{Context, Result, bail};
use seaweed_worker_core::config_form::{number_field, text_field};
use seaweed_worker_core::pb::ConfigField;