From 166af06a2b3c87c5055662d87ff88abed1e620b1 Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Tue, 15 Sep 2026 19:29:22 +0300 Subject: [PATCH] rust: cargo fmt both crates, with a commented-out fmt --check CI step (#11329) * rust: migrate seaweed-volume and seaweed-worker to tonic 0.14 / prost 0.14 tonic 0.14 boxes the contents of tonic::Status, which is what made every RPC path trip clippy's result_large_err; the allow for that lint goes in the next commit. The prost codec moved out of tonic into tonic-prost and tonic-prost-build, so both build scripts now call tonic_prost_build::configure() and both crates depend on tonic-prost for the generated code. The `tls` feature was split into a per-backend feature; `tls-aws-lc` is the same backend both crates already install through rustls::crypto::aws_lc_rs. tonic 0.14 depends on axum 0.8 and tower 0.5, which would have left a second axum and a second tower in each tree next to the 0.7 / 0.4 the crates named themselves. Bumping them keeps one copy of each: axum 0.8 only changes the path-parameter syntax for the routes here (`/:vid` -> `/{vid}`, `/*path` -> `/{*path}`), tower 0.5 needs the `util` feature named explicitly for ServiceExt::oneshot (it used to arrive through tonic's feature unification), and tower-http 0.6 is the matching release. Lock files move only through cargo's own resolution for the new versions; no other dependency was refreshed. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust: drop the result_large_err allow now that tonic::Status is boxed tonic 0.14 stores Status behind a Box, so Result<_, Status> is no longer a large-Err type and clippy has nothing to say about it. Both crates pass `cargo clippy --all-targets -- -D warnings` without the allow (seaweed-volume in both feature sets), so the policy entry and its comment go. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: drop the unused headers argument of try_expand_chunk_manifest The parameter was already named `_headers`; nothing in the body reads it. With it gone the function is under clippy's argument threshold and the expect goes. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: pass EC peer reads an EcInterval instead of ten arguments fetch_one_interval, read_remote_ec_shard_interval, do_read_remote_ec_shard_interval and recover_one_remote_ec_shard_interval all took the same (vid, needle_id, shard_id, shard_offset, size, expected_encode_ts_ns) tuple, and the two that reconstruct also took the location map with the data/parity counts. Those are now EcInterval (Copy) and EcShardMap (a borrow of the map plus the counts). The fan-out inside recovery builds its per-shard request with `EcInterval { shard_id: sid, ..iv }`, which is the one place the old argument list was easy to get wrong. Bodies destructure at the top, so the code below the signatures is unchanged. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: give the EC encoder an EcEncodeLayout and an EncodeRun encode_dat_file took the Reed-Solomon shape and three block sizes as five loose integers; they are now one Copy struct, EcEncodeLayout, which is what Go calls ECContext. The per-row and per-batch helpers took the same six sinks and the offsets; they become methods on EncodeRun, which owns the borrows for one run, so each call names only the offset and block size that vary. The byte-level work is unchanged. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: describe a .dat rebuild with DatRebuild instead of nine arguments write_dat_file_from_shards, its _with_dirs twin and the private write_dat_file were three layers over one nine-argument signature. One public function now takes a DatRebuild, whose shard_dirs is None when every shard sits beside the .dat and Some(dirs) for the cross-disk reconciled layout. The field docs carry what the function doc used to say about the encode-time size and the block layout. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: split copy_file_from_source's fifteen arguments into two structs CopyFileSpec is the per-file request (what to ask the source for, where it lands, whether its bytes count as progress); CopyProgress is the sender, throttler and report state that all three files of one VolumeCopy share, held by &mut across the calls. The three production call sites now read as the .dat/.idx/.vif literals they are, instead of positional trues and falses. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-volume: create volumes from a VolumeSpec Volume::new, DiskLocation::create_volume and Store::add_volume each took the same five-value tail of Go's NewVolume argument list: collection, replica placement, TTL, preallocation and needle version. That tail is now VolumeSpec, a Copy struct whose Default is what almost every test wanted anyway (empty collection, no replication, no TTL, no preallocation, current version), so most of the 104 call sites shrink to `&VolumeSpec::default()` or name the one field they set. The id, directories, index kind and disk type stay positional because they differ at every site. Two imports that only test modules use moved into those modules, and DiskLocation no longer imports ReplicaPlacement. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU * rust-worker: run cargo fmt Layout only; no token in the workspace changes. Co-Authored-By: Claude Fable 5.1 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 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 Claude-Session: https://claude.ai/code/session_01CjZY429aVU74SLDmo1wiuU --------- Co-authored-by: Claude Fable 5.1 Co-authored-by: Chris Lu --- .../workflows/rust-volume-server-tests.yml | 5 + .github/workflows/rust-worker-tests.yml | 5 + seaweed-volume/src/config.rs | 4 +- seaweed-volume/src/main.rs | 57 ++-- seaweed-volume/src/metrics.rs | 6 +- .../src/remote_storage/endpoint_guard.rs | 292 +++++++++++------- seaweed-volume/src/remote_storage/s3.rs | 2 +- seaweed-volume/src/remote_storage/s3_tier.rs | 2 +- seaweed-volume/src/security.rs | 10 +- seaweed-volume/src/security/tls.rs | 6 +- seaweed-volume/src/server/debug.rs | 4 +- seaweed-volume/src/server/grpc_client.rs | 4 +- seaweed-volume/src/server/grpc_server.rs | 35 +-- seaweed-volume/src/server/handlers.rs | 32 +- seaweed-volume/src/server/heartbeat.rs | 16 +- seaweed-volume/src/server/store_ec.rs | 14 +- seaweed-volume/src/server/volume_server.rs | 15 +- seaweed-volume/src/server/write_queue.rs | 2 +- seaweed-volume/src/storage/disk_location.rs | 126 ++++++-- .../src/storage/erasure_coding/ec_bitrot.rs | 25 +- .../src/storage/erasure_coding/ec_decoder.rs | 15 +- .../src/storage/erasure_coding/ec_encoder.rs | 19 +- .../src/storage/erasure_coding/ec_volume.rs | 4 +- .../src/storage/erasure_coding/mod.rs | 4 +- seaweed-volume/src/storage/idx/mod.rs | 8 +- seaweed-volume/src/storage/needle/mod.rs | 5 +- seaweed-volume/src/storage/needle/needle.rs | 8 +- seaweed-volume/src/storage/needle/ttl.rs | 64 +++- seaweed-volume/src/storage/needle_map.rs | 83 +++-- .../src/storage/needle_map/idx_metric.rs | 4 +- .../src/storage/needle_map/sorted_file.rs | 19 +- seaweed-volume/src/storage/store.rs | 87 +++--- seaweed-volume/src/storage/store_ec_mirror.rs | 6 +- .../src/storage/store_ec_reconcile.rs | 101 ++++-- seaweed-volume/src/storage/volume.rs | 230 ++++++++------ .../src/storage/volume_idx_rebuild.rs | 4 +- .../src/storage/volume_idx_repair.rs | 4 +- seaweed-volume/tests/admin_auth_coverage.rs | 81 ++++- seaweed-volume/tests/http_integration.rs | 4 +- seaweed-worker/crates/core/src/config_form.rs | 2 +- seaweed-worker/crates/core/src/metrics.rs | 39 ++- seaweed-worker/crates/core/src/senders.rs | 4 +- seaweed-worker/crates/core/src/stream.rs | 10 +- .../crates/lance/src/catalog/mod.rs | 2 +- seaweed-worker/crates/lance/src/dataset.rs | 2 +- .../crates/lance/src/jobs/cleanup.rs | 6 +- .../crates/lance/src/jobs/compact.rs | 8 +- .../crates/lance/src/jobs/indices.rs | 4 +- seaweed-worker/crates/lance/src/jobs/mod.rs | 2 +- seaweed-worker/crates/lance/src/jobs/sort.rs | 14 +- .../crates/lance/tests/common/mod.rs | 4 +- .../crates/lance/tests/compaction.rs | 8 +- .../crates/lance/tests/lifecycle.rs | 4 +- seaweed-worker/crates/lance/tests/sort.rs | 10 +- seaweed-worker/crates/sort/src/lib.rs | 2 +- 55 files changed, 970 insertions(+), 563 deletions(-) diff --git a/.github/workflows/rust-volume-server-tests.yml b/.github/workflows/rust-volume-server-tests.yml index 1c1d3358a..93ff56a3c 100644 --- a/.github/workflows/rust-volume-server-tests.yml +++ b/.github/workflows/rust-volume-server-tests.yml @@ -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 diff --git a/.github/workflows/rust-worker-tests.yml b/.github/workflows/rust-worker-tests.yml index 871c52340..ae08c06fb 100644 --- a/.github/workflows/rust-worker-tests.yml +++ b/.github/workflows/rust-worker-tests.yml @@ -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. diff --git a/seaweed-volume/src/config.rs b/seaweed-volume/src/config.rs index 77ca6f974..fa0c2f1a6 100644 --- a/seaweed-volume/src/config.rs +++ b/seaweed-volume/src/config.rs @@ -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 { diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index fbd5ca124..4ac7d0bfb 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -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] diff --git a/seaweed-volume/src/metrics.rs b/seaweed-volume/src/metrics.rs index fba531216..f4774286f 100644 --- a/seaweed-volume/src/metrics.rs +++ b/seaweed-volume/src/metrics.rs @@ -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); diff --git a/seaweed-volume/src/remote_storage/endpoint_guard.rs b/seaweed-volume/src/remote_storage/endpoint_guard.rs index 60087a49d..82daf88c7 100644 --- a/seaweed-volume/src/remote_storage/endpoint_guard.rs +++ b/seaweed-volume/src/remote_storage/endpoint_guard.rs @@ -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 { 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()); diff --git a/seaweed-volume/src/remote_storage/s3.rs b/seaweed-volume/src/remote_storage/s3.rs index bac5485ae..8b86e64e4 100644 --- a/seaweed-volume/src/remote_storage/s3.rs +++ b/seaweed-volume/src/remote_storage/s3.rs @@ -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}; diff --git a/seaweed-volume/src/remote_storage/s3_tier.rs b/seaweed-volume/src/remote_storage/s3_tier.rs index 61be1a339..570a093d5 100644 --- a/seaweed-volume/src/remote_storage/s3_tier.rs +++ b/seaweed-volume/src/remote_storage/s3_tier.rs @@ -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; diff --git a/seaweed-volume/src/security.rs b/seaweed-volume/src/security.rs index 1be223f47..de4168a6f 100644 --- a/seaweed-volume/src/security.rs +++ b/seaweed-volume/src/security.rs @@ -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); diff --git a/seaweed-volume/src/security/tls.rs b/seaweed-volume/src/security/tls.rs index 21de7a646..42a0fba32 100644 --- a/seaweed-volume/src/security/tls.rs +++ b/seaweed-volume/src/security/tls.rs @@ -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; diff --git a/seaweed-volume/src/server/debug.rs b/seaweed-volume/src/server/debug.rs index dd1b69cf1..63bdc6301 100644 --- a/seaweed-volume/src/server/debug.rs +++ b/seaweed-volume/src/server/debug.rs @@ -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; diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index 114840d22..4a55992ed 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/seaweed-volume/src/server/grpc_client.rs @@ -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) }; diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index e0446d9d0..214462314 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -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 = Pin> + 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, ) { - 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); diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 61f36ec7a..641981935 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -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> { - 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", diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index cd97cc041..b57240272 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -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 { diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 931d90916..20bc6557f 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -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 diff --git a/seaweed-volume/src/server/volume_server.rs b/seaweed-volume/src/server/volume_server.rs index 9e59605c2..2d42b9633 100644 --- a/seaweed-volume/src/server/volume_server.rs +++ b/seaweed-volume/src/server/volume_server.rs @@ -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::(), grpc_port.parse::()) { - 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"); } diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs index 7c8e84285..689db005c 100644 --- a/seaweed-volume/src/server/write_queue.rs +++ b/seaweed-volume/src/server/write_queue.rs @@ -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); diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index ea31595e2..5fc699ebf 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -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 = 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), diff --git a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs index fc28a7d57..af44045ed 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs @@ -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(¬found, 0, 10, 4), BitrotStatus::Off); // Integrity failure => Invalid. - let bad: Result = - Err(BitrotLoadError::BadMagic(0)); + let bad: Result = Err(BitrotLoadError::BadMagic(0)); assert_eq!(resolve_status(&bad, 0, 10, 4), BitrotStatus::Invalid); // Generation mismatch => Off. diff --git a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs index 02619489b..bcc8afd8b 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs @@ -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; diff --git a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs index a5d62fb08..0d82615bb 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs @@ -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 diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index e14e6008b..aaec531ba 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -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); diff --git a/seaweed-volume/src/storage/erasure_coding/mod.rs b/seaweed-volume/src/storage/erasure_coding/mod.rs index fd1dab928..c6cf4d8f3 100644 --- a/seaweed-volume/src/storage/erasure_coding/mod.rs +++ b/seaweed-volume/src/storage/erasure_coding/mod.rs @@ -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; diff --git a/seaweed-volume/src/storage/idx/mod.rs b/seaweed-volume/src/storage/idx/mod.rs index 2cae9bec6..0a7182c3a 100644 --- a/seaweed-volume/src/storage/idx/mod.rs +++ b/seaweed-volume/src/storage/idx/mod.rs @@ -57,7 +57,7 @@ pub fn check_index_file( 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] diff --git a/seaweed-volume/src/storage/needle/mod.rs b/seaweed-volume/src/storage/needle/mod.rs index 5a6277bff..c009dddf8 100644 --- a/seaweed-volume/src/storage/needle/mod.rs +++ b/seaweed-volume/src/storage/needle/mod.rs @@ -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; diff --git a/seaweed-volume/src/storage/needle/needle.rs b/seaweed-volume/src/storage/needle/needle.rs index 5bb730978..df4205ad1 100644 --- a/seaweed-volume/src/storage/needle/needle.rs +++ b/seaweed-volume/src/storage/needle/needle.rs @@ -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, diff --git a/seaweed-volume/src/storage/needle/ttl.rs b/seaweed-volume/src/storage/needle/ttl.rs index 5d67c367b..408169c53 100644 --- a/seaweed-volume/src/storage/needle/ttl.rs +++ b/seaweed-volume/src/storage/needle/ttl.rs @@ -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 + } + ); } } diff --git a/seaweed-volume/src/storage/needle_map.rs b/seaweed-volume/src/storage/needle_map.rs index 4ca6300cb..e279ccfdd 100644 --- a/seaweed-volume/src/storage/needle_map.rs +++ b/seaweed-volume/src/storage/needle_map.rs @@ -801,23 +801,26 @@ impl RedbNeedleMap { } #[cfg(feature = "redb-experimental-cursor")] { - let mut cursor = table - .upper_bound_mut(Bound::::Unbounded) - .map_err(|e| { - io::Error::new( - io::ErrorKind::Other, - format!("redb upper_bound_mut: {}", e), - ) - })?; + let mut cursor = + table + .upper_bound_mut(Bound::::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 diff --git a/seaweed-volume/src/storage/needle_map/idx_metric.rs b/seaweed-volume/src/storage/needle_map/idx_metric.rs index 67cd9c368..39ba03dc6 100644 --- a/seaweed-volume/src/storage/needle_map/idx_metric.rs +++ b/seaweed-volume/src/storage/needle_map/idx_metric.rs @@ -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}" diff --git a/seaweed-volume/src/storage/needle_map/sorted_file.rs b/seaweed-volume/src/storage/needle_map/sorted_file.rs index 4d60d5d4e..6f1f9b4a6 100644 --- a/seaweed-volume/src/storage/needle_map/sorted_file.rs +++ b/seaweed-volume/src/storage/needle_map/sorted_file.rs @@ -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 ); } diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 43e49fbc1..f3d6a1b63 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -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(); diff --git a/seaweed-volume/src/storage/store_ec_mirror.rs b/seaweed-volume/src/storage/store_ec_mirror.rs index 77b3af56c..1e53e4a34 100644 --- a/seaweed-volume/src/storage/store_ec_mirror.rs +++ b/seaweed-volume/src/storage/store_ec_mirror.rs @@ -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> 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 } diff --git a/seaweed-volume/src/storage/store_ec_reconcile.rs b/seaweed-volume/src/storage/store_ec_reconcile.rs index 9c791f0ee..7057761ab 100644 --- a/seaweed-volume/src/storage/store_ec_reconcile.rs +++ b/seaweed-volume/src/storage/store_ec_reconcile.rs @@ -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 diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 35558f41b..dadb10e32 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -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> { 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) -> (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 diff --git a/seaweed-volume/src/storage/volume_idx_rebuild.rs b/seaweed-volume/src/storage/volume_idx_rebuild.rs index 36f09a037..68109ba07 100644 --- a/seaweed-volume/src/storage/volume_idx_rebuild.rs +++ b/seaweed-volume/src/storage/volume_idx_rebuild.rs @@ -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}; diff --git a/seaweed-volume/src/storage/volume_idx_repair.rs b/seaweed-volume/src/storage/volume_idx_repair.rs index f57be9087..459987200 100644 --- a/seaweed-volume/src/storage/volume_idx_repair.rs +++ b/seaweed-volume/src/storage/volume_idx_repair.rs @@ -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, Vec); diff --git a/seaweed-volume/tests/admin_auth_coverage.rs b/seaweed-volume/tests/admin_auth_coverage.rs index 08e31824d..612b8c72f 100644 --- a/seaweed-volume/tests/admin_auth_coverage.rs +++ b/seaweed-volume/tests/admin_auth_coverage.rs @@ -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") + ); } diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs index accb74bd1..b0cbd91d4 100644 --- a/seaweed-volume/tests/http_integration.rs +++ b/seaweed-volume/tests/http_integration.rs @@ -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; diff --git a/seaweed-worker/crates/core/src/config_form.rs b/seaweed-worker/crates/core/src/config_form.rs index 02235cd66..ceb7db70b 100644 --- a/seaweed-worker/crates/core/src/config_form.rs +++ b/seaweed-worker/crates/core/src/config_form.rs @@ -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 { diff --git a/seaweed-worker/crates/core/src/metrics.rs b/seaweed-worker/crates/core/src/metrics.rs index f3bb9fae7..e6b57a10f 100644 --- a/seaweed-worker/crates/core/src/metrics.rs +++ b/seaweed-worker/crates/core/src/metrics.rs @@ -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 diff --git a/seaweed-worker/crates/core/src/senders.rs b/seaweed-worker/crates/core/src/senders.rs index 39c5bff7e..bd1b8d1ed 100644 --- a/seaweed-worker/crates/core/src/senders.rs +++ b/seaweed-worker/crates/core/src/senders.rs @@ -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. diff --git a/seaweed-worker/crates/core/src/stream.rs b/seaweed-worker/crates/core/src/stream.rs index 9dc7770f2..67fe8fa83 100644 --- a/seaweed-worker/crates/core/src/stream.rs +++ b/seaweed-worker/crates/core/src/stream.rs @@ -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}; diff --git a/seaweed-worker/crates/lance/src/catalog/mod.rs b/seaweed-worker/crates/lance/src/catalog/mod.rs index 002a74c36..142de4eb5 100644 --- a/seaweed-worker/crates/lance/src/catalog/mod.rs +++ b/seaweed-worker/crates/lance/src/catalog/mod.rs @@ -7,4 +7,4 @@ pub mod namespace; -pub use namespace::{parse_id, NamespaceClient, TableDescription}; +pub use namespace::{NamespaceClient, TableDescription, parse_id}; diff --git a/seaweed-worker/crates/lance/src/dataset.rs b/seaweed-worker/crates/lance/src/dataset.rs index 74effcc92..4b65c3fb3 100644 --- a/seaweed-worker/crates/lance/src/dataset.rs +++ b/seaweed-worker/crates/lance/src/dataset.rs @@ -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}; diff --git a/seaweed-worker/crates/lance/src/jobs/cleanup.rs b/seaweed-worker/crates/lance/src/jobs/cleanup.rs index 84351e8b2..608187ad4 100644 --- a/seaweed-worker/crates/lance/src/jobs/cleanup.rs +++ b/seaweed-worker/crates/lance/src/jobs/cleanup.rs @@ -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}; diff --git a/seaweed-worker/crates/lance/src/jobs/compact.rs b/seaweed-worker/crates/lance/src/jobs/compact.rs index e20bd3b6a..2cad96963 100644 --- a/seaweed-worker/crates/lance/src/jobs/compact.rs +++ b/seaweed-worker/crates/lance/src/jobs/compact.rs @@ -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"; diff --git a/seaweed-worker/crates/lance/src/jobs/indices.rs b/seaweed-worker/crates/lance/src/jobs/indices.rs index a3c23c61c..927748b9b 100644 --- a/seaweed-worker/crates/lance/src/jobs/indices.rs +++ b/seaweed-worker/crates/lance/src/jobs/indices.rs @@ -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}; diff --git a/seaweed-worker/crates/lance/src/jobs/mod.rs b/seaweed-worker/crates/lance/src/jobs/mod.rs index 064a4af7d..b5502a774 100644 --- a/seaweed-worker/crates/lance/src/jobs/mod.rs +++ b/seaweed-worker/crates/lance/src/jobs/mod.rs @@ -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; diff --git a/seaweed-worker/crates/lance/src/jobs/sort.rs b/seaweed-worker/crates/lance/src/jobs/sort.rs index 4273b7ff2..0bbf9b063 100644 --- a/seaweed-worker/crates/lance/src/jobs/sort.rs +++ b/seaweed-worker/crates/lance/src/jobs/sort.rs @@ -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" - )) + )); } } diff --git a/seaweed-worker/crates/lance/tests/common/mod.rs b/seaweed-worker/crates/lance/tests/common/mod.rs index 31f603a5b..10ffb3db8 100644 --- a/seaweed-worker/crates/lance/tests/common/mod.rs +++ b/seaweed-worker/crates/lance/tests/common/mod.rs @@ -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}; diff --git a/seaweed-worker/crates/lance/tests/compaction.rs b/seaweed-worker/crates/lance/tests/compaction.rs index 5de4cb379..307e3d437 100644 --- a/seaweed-worker/crates/lance/tests/compaction.rs +++ b/seaweed-worker/crates/lance/tests/compaction.rs @@ -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( diff --git a/seaweed-worker/crates/lance/tests/lifecycle.rs b/seaweed-worker/crates/lance/tests/lifecycle.rs index e0d25ba31..9b5253fff 100644 --- a/seaweed-worker/crates/lance/tests/lifecycle.rs +++ b/seaweed-worker/crates/lance/tests/lifecycle.rs @@ -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 { std::env::var("WEED_LANCE_TABLE") diff --git a/seaweed-worker/crates/lance/tests/sort.rs b/seaweed-worker/crates/lance/tests/sort.rs index 22fe28ef9..9b76d5feb 100644 --- a/seaweed-worker/crates/lance/tests/sort.rs +++ b/seaweed-worker/crates/lance/tests/sort.rs @@ -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. diff --git a/seaweed-worker/crates/sort/src/lib.rs b/seaweed-worker/crates/sort/src/lib.rs index de248cb3f..8f6fc6a88 100644 --- a/seaweed-worker/crates/sort/src/lib.rs +++ b/seaweed-worker/crates/sort/src/lib.rs @@ -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;