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,