rust volume: one S3 tier registry instead of two kept in sync by hand (#11357)

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

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,

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(),

seaweed-volume/src/server/heartbeat.rs

@@ -222,7 +222,7 @@ async fn check_with_master(config: &HeartbeatConfig, state: &Arc<VolumeServerSta
                     if changed {
                         state.metrics_notify.notify_waiters();
                     }
-                    apply_storage_backends(state, &resp.storage_backends);
+                    apply_storage_backends(&resp.storage_backends);
                     info!(
                         "Got master configuration from {}: metrics_address={}, metrics_interval={}s",
                         master_addr, resp.metrics_address, resp.metrics_interval_seconds
@@ -674,16 +674,15 @@ fn apply_metrics_push_settings(
     true
 }
 
-fn apply_storage_backends(
-    state: &VolumeServerState,
-    storage_backends: &[master_pb::StorageBackend],
-) {
+/// Registers the master's S3 storage backends in the process-wide tier
+/// registry, the single place both the tier-move handlers and `Volume` itself
+/// resolve a backend from.
+fn apply_storage_backends(storage_backends: &[master_pb::StorageBackend]) {
     if storage_backends.is_empty() {
         return;
     }
 
-    let mut registry = state.s3_tier_registry.write().unwrap();
-    let mut global_registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
+    let mut registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
         .write()
         .unwrap();
     for backend in storage_backends {
@@ -714,7 +713,6 @@ fn apply_storage_backends(
             backend.id.as_str()
         };
         register_s3_backend(&mut registry, backend, backend_id, &config);
-        register_s3_backend(&mut global_registry, backend, backend_id, &config);
     }
 }
 
@@ -1250,7 +1248,6 @@ mod tests {
         READ_ONLY_LABEL_NO_WRITE_CAN_DELETE, READ_ONLY_LABEL_NO_WRITE_OR_DELETE,
         READ_ONLY_VOLUME_GAUGE,
     };
-    use crate::remote_storage::s3_tier::S3TierRegistry;
     use crate::security::{Guard, SigningKey};
     use crate::storage::needle_map::NeedleMapKind;
     use crate::storage::types::{DiskType, VolumeId};
@@ -1302,7 +1299,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(S3TierRegistry::new()),
             read_mode: ReadMode::Local,
             allow_untrusted_remote_endpoints: false,
             master_url: String::new(),
@@ -2072,65 +2068,56 @@ mod tests {
         assert_eq!(heartbeat.volumes[0].remote_storage_key, "volumes/71.dat");
     }
 
+    // Not hermetic, and cannot be made so cheaply: `register_s3_backend` skips
+    // a name that is already registered, so had another test put `s3` or
+    // `s3.default` in the process-wide registry first, this would pass without
+    // proving that this call registered anything. Removing them afterwards is
+    // no better — unlike the tier tests' unique ids, the bare `s3` alias is the
+    // production one. Nothing else in the tree registers those two names.
     #[test]
     fn test_apply_storage_backends_registers_s3_default_aliases() {
-        let state = test_state_with_store(Store::new(NeedleMapKind::InMemory));
         // Do not call clear() on the global registry — other tests may be
         // running concurrently.  Just register our entries and verify them.
 
-        apply_storage_backends(
-            &state,
-            &[master_pb::StorageBackend {
-                r#type: "s3".to_string(),
-                id: "default".to_string(),
-                properties: std::collections::HashMap::from([
-                    ("aws_access_key_id".to_string(), "access".to_string()),
-                    ("aws_secret_access_key".to_string(), "secret".to_string()),
-                    ("bucket".to_string(), "bucket-a".to_string()),
-                    ("region".to_string(), "us-west-2".to_string()),
-                    ("endpoint".to_string(), "http://127.0.0.1:8333".to_string()),
-                    ("storage_class".to_string(), "STANDARD".to_string()),
-                    ("force_path_style".to_string(), "false".to_string()),
-                ]),
-            }],
-        );
+        apply_storage_backends(&[master_pb::StorageBackend {
+            r#type: "s3".to_string(),
+            id: "default".to_string(),
+            properties: std::collections::HashMap::from([
+                ("aws_access_key_id".to_string(), "access".to_string()),
+                ("aws_secret_access_key".to_string(), "secret".to_string()),
+                ("bucket".to_string(), "bucket-a".to_string()),
+                ("region".to_string(), "us-west-2".to_string()),
+                ("endpoint".to_string(), "http://127.0.0.1:8333".to_string()),
+                ("storage_class".to_string(), "STANDARD".to_string()),
+                ("force_path_style".to_string(), "false".to_string()),
+            ]),
+        }]);
 
-        let registry = state.s3_tier_registry.read().unwrap();
-        assert!(registry.get("s3.default").is_some());
-        assert!(registry.get("s3").is_some());
-        let global_registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
+        let registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
             .read()
             .unwrap();
-        assert!(global_registry.get("s3.default").is_some());
-        assert!(global_registry.get("s3").is_some());
+        assert!(registry.get("s3.default").is_some());
+        assert!(registry.get("s3").is_some());
     }
 
     #[test]
     fn test_apply_storage_backends_ignores_unsupported_types() {
-        let state = test_state_with_store(Store::new(NeedleMapKind::InMemory));
         // Do not call clear() on the global registry — other tests may be
         // running concurrently.
 
-        apply_storage_backends(
-            &state,
-            &[master_pb::StorageBackend {
-                r#type: "rclone".to_string(),
-                id: "default".to_string(),
-                properties: std::collections::HashMap::new(),
-            }],
-        );
+        apply_storage_backends(&[master_pb::StorageBackend {
+            r#type: "rclone".to_string(),
+            id: "default".to_string(),
+            properties: std::collections::HashMap::new(),
+        }]);
 
-        // The per-state registry is freshly created and should have no entries
-        // since "rclone" is unsupported.
-        let registry = state.s3_tier_registry.read().unwrap();
-        assert!(registry.names().is_empty());
         // Only check that the unsupported type was not added to the global
         // registry.  Other tests may have their own entries present.
-        let global_registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
+        let registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
             .read()
             .unwrap();
-        assert!(global_registry.get("rclone.default").is_none());
-        assert!(global_registry.get("rclone").is_none());
+        assert!(registry.get("rclone.default").is_none());
+        assert!(registry.get("rclone").is_none());
     }
 
     #[test]

seaweed-volume/src/server/volume_server.rs

@@ -73,8 +73,6 @@ pub struct VolumeServerState {
     pub volume_state_notify: tokio::sync::Notify,
     /// Optional batched write queue for improved throughput under load.
     pub write_queue: std::sync::OnceLock<WriteQueue>,
-    /// Registry of S3 tier backends for tiered storage operations.
-    pub s3_tier_registry: std::sync::RwLock<crate::remote_storage::s3_tier::S3TierRegistry>,
     /// Read mode: local, proxy, or redirect for non-local volumes.
     pub read_mode: ReadMode,
     /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation,

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(),

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,