From f1ed270942d303dc55283d2abdaf595525b0dad4 Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Sun, 20 Sep 2026 23:30:40 +0300 Subject: [PATCH] rust volume: one S3 tier registry instead of two kept in sync by hand (#11357) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * rust volume: one S3 tier registry instead of two kept in sync by hand `VolumeServerState.s3_tier_registry` and `global_s3_tier_registry()` held the same S3 tier backends. `apply_storage_backends` — the only production writer — registered every backend into both, and each half of the tiering code then read a different one: the gRPC tier-move handlers resolved the backend from the per-server field, while `Volume`'s remote mount and destroy paths resolved it from the global registry, because a `Volume` has no handle to the server state. Two registries that must agree, kept in agreement by a duplicated `register_s3_backend` call and a comment in a test constructor explaining the hand-sync. Delete the field and let both tier-move handlers resolve from the global registry, so `apply_storage_backends` registers once and no longer needs the server state at all. Injecting a registry handle through `VolumeSpec` instead was considered and rejected here: it would touch every `Volume` constructor for no functional gain, and the process-wide registry is what `Volume` already uses. Behaviour is unchanged: the same names were registered in both registries, so every lookup resolves exactly as before. The tier-down test now registers its backend only in the global registry — before this change it fails with `remote storage s3.tier_down_delete not found from supported: []`. The tier-up handler had no test at all, so it gets a cheap probe: register a backend only in the global registry, ask for that destination, and check the call gets past the lookup — the response is dropped straight away, so the transfer sees a departed caller and never opens a connection. Co-Authored-By: Claude Fable 5.1 * rust volume: await the tier-up probe terminal error instead of racing it Dropping the response left it to chance whether the detached transfer saw the closed channel before its initial check; if it won that race it went on to attempt the multipart upload with no one waiting on the outcome. Hold the stream and read until the dead endpoint fails the upload — the terminal error proves the task ran and finished, so no background network work outlives the test. --------- Co-authored-by: Claude Fable 5.1 Co-authored-by: Chris Lu --- seaweed-volume/src/main.rs | 3 - seaweed-volume/src/server/grpc_server.rs | 85 ++++++++++++++++------ seaweed-volume/src/server/heartbeat.rs | 85 +++++++++------------- seaweed-volume/src/server/volume_server.rs | 2 - seaweed-volume/src/server/write_queue.rs | 3 - seaweed-volume/tests/http_integration.rs | 3 - 6 files changed, 100 insertions(+), 81 deletions(-) diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index 4ac7d0bfb..844b5c15b 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -343,9 +343,6 @@ async fn run( pre_stop_seconds: config.pre_stop_seconds, volume_state_notify: tokio::sync::Notify::new(), write_queue: std::sync::OnceLock::new(), - s3_tier_registry: std::sync::RwLock::new( - seaweed_volume::remote_storage::s3_tier::S3TierRegistry::new(), - ), read_mode: config.read_mode, allow_untrusted_remote_endpoints: config.allow_untrusted_remote_endpoints, master_url, diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 0ea817846..ef0403540 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -4343,7 +4343,9 @@ impl VolumeServer for VolumeGrpcService { // Look up the S3 tier backend let backend = { - let registry = self.state.s3_tier_registry.read().unwrap(); + let registry = crate::remote_storage::s3_tier::global_s3_tier_registry() + .read() + .unwrap(); registry.get(&req.destination_backend_name).ok_or_else(|| { let keys = registry.names(); Status::not_found(format!( @@ -4549,7 +4551,9 @@ impl VolumeServer for VolumeGrpcService { // Look up the S3 tier backend let backend = { - let registry = self.state.s3_tier_registry.read().unwrap(); + let registry = crate::remote_storage::s3_tier::global_s3_tier_registry() + .read() + .unwrap(); registry.get(&storage_name).ok_or_else(|| { let keys = registry.names(); Status::not_found(format!( @@ -6375,16 +6379,6 @@ mod tests { pre_stop_seconds: 0, volume_state_notify: tokio::sync::Notify::new(), write_queue: std::sync::OnceLock::new(), - s3_tier_registry: std::sync::RwLock::new({ - // The tier-down handler resolves the backend from the per-server - // registry, so register it here too (reads use the global one). - let mut reg = crate::remote_storage::s3_tier::S3TierRegistry::new(); - reg.register( - format!("s3.{}", backend_id), - S3TierBackend::new(&tier_config), - ); - reg - }), read_mode: crate::config::ReadMode::Local, allow_untrusted_remote_endpoints: false, master_url: String::new(), @@ -6496,9 +6490,6 @@ mod tests { pre_stop_seconds: 0, volume_state_notify: tokio::sync::Notify::new(), write_queue: std::sync::OnceLock::new(), - s3_tier_registry: std::sync::RwLock::new( - crate::remote_storage::s3_tier::S3TierRegistry::new(), - ), read_mode: crate::config::ReadMode::Local, allow_untrusted_remote_endpoints: allow_untrusted, master_url: String::new(), @@ -6808,6 +6799,64 @@ mod tests { .remove("s3.tier_down_keep"); } + // The tier-up handler has no end-to-end test — exercising it needs a fake + // S3 that accepts multipart uploads — so this probes only the part that + // changed: the destination is resolved from the process-wide registry, now + // the only one. A backend registered nowhere else has to get past that + // lookup. The stream stays open until the spawned transfer reports its + // terminal error, so the task cannot race a dropped receiver or outlive + // the test. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_tier_move_to_remote_resolves_the_destination_from_the_global_registry() { + let (service, _tmp) = make_local_service_with_volume("", None); + { + let mut registry = global_s3_tier_registry().write().unwrap(); + registry.register( + "s3.tier_up_probe".to_string(), + S3TierBackend::new(&S3TierConfig { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + region: "us-east-1".to_string(), + bucket: "bucket-a".to_string(), + // Nothing listens here; the upload fails instead of hanging. + endpoint: "http://127.0.0.1:1".to_string(), + storage_class: "STANDARD".to_string(), + force_path_style: true, + }), + ); + } + + let mut stream = service + .volume_tier_move_dat_to_remote(Request::new( + volume_server_pb::VolumeTierMoveDatToRemoteRequest { + volume_id: 1, + collection: String::new(), + destination_backend_name: "s3.tier_up_probe".to_string(), + keep_local_dat_file: true, + }, + )) + .await + .expect("tier-up must resolve its destination from the global registry") + .into_inner(); + + // The dead endpoint fails the multipart upload; the terminal error is + // also what proves the spawned task ran to completion rather than + // leaving background network work behind. + let terminal = stream.next().await; + global_s3_tier_registry() + .write() + .unwrap() + .remove("s3.tier_up_probe"); + + match terminal { + Some(Err(status)) => { + assert_eq!(status.code(), tonic::Code::Internal, "{status:?}"); + assert!(status.message().contains("s3.tier_up_probe"), "{status:?}"); + } + other => panic!("the dead endpoint must fail the upload, got {other:?}"), + } + } + /// Build a local service whose volume has a `.dat` large enough to span /// several 2MB copy chunks, so the streaming copy paths are exercised /// across multiple messages rather than a single buffer. @@ -7330,9 +7379,6 @@ mod tests { pre_stop_seconds: 0, volume_state_notify: tokio::sync::Notify::new(), write_queue: std::sync::OnceLock::new(), - s3_tier_registry: std::sync::RwLock::new( - crate::remote_storage::s3_tier::S3TierRegistry::new(), - ), read_mode: crate::config::ReadMode::Local, allow_untrusted_remote_endpoints: false, master_url: master_urls.first().cloned().unwrap_or_default(), @@ -8184,9 +8230,6 @@ mod tests { pre_stop_seconds: 0, volume_state_notify: tokio::sync::Notify::new(), write_queue: std::sync::OnceLock::new(), - s3_tier_registry: std::sync::RwLock::new( - crate::remote_storage::s3_tier::S3TierRegistry::new(), - ), read_mode: crate::config::ReadMode::Local, allow_untrusted_remote_endpoints: false, master_url: String::new(), diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index cce60ccac..8a5879697 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -222,7 +222,7 @@ async fn check_with_master(config: &HeartbeatConfig, state: &Arc, - /// Registry of S3 tier backends for tiered storage operations. - pub s3_tier_registry: std::sync::RwLock, /// Read mode: local, proxy, or redirect for non-local volumes. pub read_mode: ReadMode, /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation, diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs index 689db005c..528007325 100644 --- a/seaweed-volume/src/server/write_queue.rs +++ b/seaweed-volume/src/server/write_queue.rs @@ -207,9 +207,6 @@ mod tests { pre_stop_seconds: 0, volume_state_notify: tokio::sync::Notify::new(), write_queue: std::sync::OnceLock::new(), - s3_tier_registry: std::sync::RwLock::new( - crate::remote_storage::s3_tier::S3TierRegistry::new(), - ), read_mode: crate::config::ReadMode::Local, allow_untrusted_remote_endpoints: false, master_url: String::new(), diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs index a622d045a..53f358b8a 100644 --- a/seaweed-volume/tests/http_integration.rs +++ b/seaweed-volume/tests/http_integration.rs @@ -112,9 +112,6 @@ fn build_test_state( pre_stop_seconds: 0, volume_state_notify: tokio::sync::Notify::new(), write_queue: std::sync::OnceLock::new(), - s3_tier_registry: std::sync::RwLock::new( - seaweed_volume::remote_storage::s3_tier::S3TierRegistry::new(), - ), read_mode: seaweed_volume::config::ReadMode::Local, allow_untrusted_remote_endpoints: false, master_url,