diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 1da5e908a8..bd1c9011de 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/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( diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index cb563fe4bf..8a1c4acf9e 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -268,24 +268,34 @@ async fn refresh_leader_from_peers( state: &Arc, 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 { + 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() }