volume: mark volumes readonly on the leader, retrying the named one (rust)

notify_master_volume_readonly fell back to the seed when no leader was
published, which the peer answers with "Not current leader". It now
uses only the confirmed leader, skips the call in masterless mode, and
on NotLeader asks the seed masters who leads and retries there — the
same path Go's notifyMasterVolumeReadonly takes. The peer leader query
is shared with refresh_leader_from_peers as lookup_leader_from_peers.

seaweed-volume/src/server/grpc_server.rs

@@ -53,6 +53,12 @@ fn needle_not_found(needle_id: NeedleId) -> Status {
     Status::not_found(format!("needle not found {}", needle_id.0))
 }
 
+/// Mirrors Go's `isNotLeaderErr`: the master returns raft's NotLeaderError,
+/// which arrives as an Unknown-status message.
+fn is_not_leader_status(e: &Status) -> bool {
+    e.message().contains("Not current leader")
+}
+
 /// Deadline from the caller's `grpc-timeout` header (TimeoutValue +
 /// H|M|S|m|u|n unit, per the gRPC-over-HTTP2 spec). Tonic's GrpcTimeout layer
 /// drops an over-deadline handler future anyway; parsing it here lets a queued
@@ -344,22 +350,47 @@ impl VolumeGrpcService {
         is_readonly: bool,
     ) -> Result<(), Status> {
         // VolumeMarkReadonly mutates raft-replicated master topology, so it must
-        // reach the leader. Prefer the live leader the heartbeat is talking to
-        // (current_master_url), falling back to the static seed before the first
-        // heartbeat — mirrors store_ec.rs and Go's vs.GetMaster(). Sending it to
-        // the static seed fails "not current leader" after any master failover.
-        let master_url = {
-            let live = self.state.current_master_url.read().await.clone();
-            if !live.is_empty() {
-                live
-            } else {
-                self.state.master_url.clone()
-            }
-        };
+        // reach the leader the heartbeat confirmed. On "Not current leader",
+        // ask the seed masters who leads and retry there — mirrors Go's
+        // notifyMasterVolumeReadonly.
+        let master_url = self.state.current_master().await;
         if master_url.is_empty() {
-            return Ok(());
+            if !self.state.has_master {
+                return Ok(());
+            }
+            return Err(Status::unavailable("no master leader known"));
         }
-        let grpc_addr = parse_grpc_address(&master_url).map_err(|e| {
+        match self
+            .volume_mark_readonly_on_master(&master_url, info, is_readonly)
+            .await
+        {
+            Err(e) if is_not_leader_status(&e) => {
+                let leader = super::heartbeat::lookup_leader_from_peers(
+                    &self.state.master_urls,
+                    self.state.outgoing_grpc_tls.as_ref(),
+                )
+                .await
+                .ok_or_else(|| {
+                    Status::unavailable(format!(
+                        "master {} rejected mark-readonly and no peer named a leader",
+                        master_url
+                    ))
+                })?;
+                let leader = super::volume_server::to_http_address(&leader).into_owned();
+                self.volume_mark_readonly_on_master(&leader, info, is_readonly)
+                    .await
+            }
+            result => result,
+        }
+    }
+
+    async fn volume_mark_readonly_on_master(
+        &self,
+        master_url: &str,
+        info: &MasterVolumeInfo,
+        is_readonly: bool,
+    ) -> Result<(), Status> {
+        let grpc_addr = parse_grpc_address(master_url).map_err(|e| {
             Status::internal(format!("invalid master address {}: {}", master_url, e))
         })?;
         let channel = connect_channel(

seaweed-volume/src/server/heartbeat.rs

@@ -268,24 +268,34 @@ async fn refresh_leader_from_peers(
     state: &Arc<VolumeServerState>,
     timeout: Duration,
 ) {
-    let lookup = async {
-        for master_addr in &config.master_addresses {
-            let grpc_addr = to_grpc_address(master_addr);
-            if let Ok(resp) =
-                try_get_master_configuration(&grpc_addr, state.outgoing_grpc_tls.as_ref()).await
-                && !resp.leader.is_empty()
-            {
-                return Some(resp.leader);
-            }
-        }
-        None
-    };
+    let lookup = lookup_leader_from_peers(
+        &config.master_addresses,
+        state.outgoing_grpc_tls.as_ref(),
+    );
     match tokio::time::timeout(timeout, lookup).await {
         Ok(Some(leader)) => publish_current_master(state, &leader).await,
         _ => state.current_master_url.write().await.clear(),
     }
 }
 
+/// Ask each seed master for the raft leader and return the first answer, or
+/// None when none replies with one. Mirrors Go's
+/// `operation.LookupRaftLeaderMaster`.
+pub async fn lookup_leader_from_peers(
+    master_addrs: &[String],
+    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
+) -> Option<String> {
+    for master_addr in master_addrs {
+        let grpc_addr = to_grpc_address(master_addr);
+        if let Ok(resp) = try_get_master_configuration(&grpc_addr, tls).await
+            && !resp.leader.is_empty()
+        {
+            return Some(resp.leader);
+        }
+    }
+    None
+}
+
 fn is_stopping(state: &VolumeServerState) -> bool {
     *state.is_stopping.read().unwrap()
 }