volume server: route VolumeMarkReadonly to raft leader (#10120)

* volume server: route VolumeMarkReadonly to raft leader

After a master raft election, volume servers may still heartbeat a follower
while admin paths such as weed shell volume.mark call notifyMasterVolumeReadonly
via vs.GetMaster(). Followers reject VolumeMarkReadonly with NotLeader, which
breaks tiering and other mark-readonly workflows until the heartbeat loop
reconnects.

Resolve the leader through GetMasterConfiguration on configured -master peers
(same Leader field filer/master clients already use) before calling
VolumeMarkReadonly. When the leader differs from the heartbeat peer, update
currentMaster so the heartbeat loop converges faster.

Adds operation.LookupRaftLeaderMaster with unit tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix: address review feedback on volume.mark raft leader routing

Do not update currentMaster during leader lookup — heartbeat owns that
field and uses stream GetLeader() to reconnect. Try the heartbeat peer
first and only resolve the raft leader after a NotLeader rejection.
Add ctx.Err() early exit and quieter logging for context cancellation.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(operation): thread the lookup timeout ctx into connection invalidation

The 5s timeout drove only the RPC; WithMasterServerClient saw the
unbounded outer ctx, so a self-inflicted timeout (slow GetMasterConfiguration
during an election) was treated as a stale channel and tore down the shared
master connection. Pass the timeout ctx into the helper so its own expiry
leaves ctx.Err() set and spares the connection.

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/operation/lookup_raft_leader.go

@@ -0,0 +1,54 @@
+package operation
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+)
+
+// LookupRaftLeaderMaster returns the current raft leader by querying configured
+// master peers. Followers answer GetMasterConfiguration with the leader address;
+// topology mutations such as VolumeMarkReadonly must run on that node.
+//
+// The volume server's heartbeat peer (GetMaster) may still be a follower after an
+// election until the heartbeat loop reconnects.
+func LookupRaftLeaderMaster(ctx context.Context, masters []pb.ServerAddress, grpcDialOption grpc.DialOption) (pb.ServerAddress, error) {
+	if len(masters) == 0 {
+		return "", fmt.Errorf("no master peers configured")
+	}
+	var lastErr error
+	for _, peer := range masters {
+		if err := ctx.Err(); err != nil {
+			return "", err
+		}
+		// The same timeout ctx drives both the RPC and WithMasterServerClient's
+		// connection-invalidation decision, so a self-inflicted 5s timeout leaves
+		// callCtx.Err() != nil and does not poison the shared master connection.
+		callCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		var leader pb.ServerAddress
+		err := WithMasterServerClient(callCtx, false, peer, grpcDialOption, func(client master_pb.SeaweedClient) error {
+			resp, err := client.GetMasterConfiguration(callCtx, &master_pb.GetMasterConfigurationRequest{})
+			if err != nil {
+				return err
+			}
+			if resp.Leader == "" {
+				return fmt.Errorf("master %s returned empty leader", peer)
+			}
+			leader = pb.ServerAddress(resp.Leader)
+			return nil
+		})
+		cancel()
+		if err == nil && leader != "" {
+			return leader, nil
+		}
+		lastErr = err
+	}
+	if lastErr != nil {
+		return "", fmt.Errorf("lookup raft leader from %d master peer(s): %w", len(masters), lastErr)
+	}
+	return "", fmt.Errorf("lookup raft leader: no leader from %d master peer(s)", len(masters))
+}

weed/operation/lookup_raft_leader_test.go

@@ -0,0 +1,99 @@
+package operation
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type fakeLeaderConfigServer struct {
+	master_pb.UnimplementedSeaweedServer
+	leader string
+}
+
+func (s *fakeLeaderConfigServer) GetMasterConfiguration(_ context.Context, _ *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) {
+	return &master_pb.GetMasterConfigurationResponse{Leader: s.leader}, nil
+}
+
+func startFakeMasterServer(t *testing.T, srv master_pb.SeaweedServer) pb.ServerAddress {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	grpcServer := grpc.NewServer()
+	master_pb.RegisterSeaweedServer(grpcServer, srv)
+	go func() { _ = grpcServer.Serve(lis) }()
+	t.Cleanup(grpcServer.GracefulStop)
+
+	_, port, _ := net.SplitHostPort(lis.Addr().String())
+	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port))
+}
+
+func TestLookupRaftLeaderMasterFromFollower(t *testing.T) {
+	follower := startFakeMasterServer(t, &fakeLeaderConfigServer{
+		leader: "leader.example.com:9333.19333",
+	})
+
+	leader, err := LookupRaftLeaderMaster(
+		context.Background(),
+		[]pb.ServerAddress{follower},
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		t.Fatalf("lookup: %v", err)
+	}
+	if got, want := leader.ToHttpAddress(), "leader.example.com:9333"; got != want {
+		t.Fatalf("leader = %q, want %q", got, want)
+	}
+}
+
+func TestLookupRaftLeaderMasterTriesNextPeer(t *testing.T) {
+	down := pb.ServerAddress("127.0.0.1:1")
+	peer := startFakeMasterServer(t, &fakeLeaderConfigServer{
+		leader: "leader.example.com:9333.19333",
+	})
+
+	leader, err := LookupRaftLeaderMaster(
+		context.Background(),
+		[]pb.ServerAddress{down, peer},
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		t.Fatalf("lookup: %v", err)
+	}
+	if got, want := leader.ToHttpAddress(), "leader.example.com:9333"; got != want {
+		t.Fatalf("leader = %q, want %q", got, want)
+	}
+}
+
+func TestLookupRaftLeaderMasterNoPeers(t *testing.T) {
+	_, err := LookupRaftLeaderMaster(context.Background(), nil, nil)
+	if err == nil {
+		t.Fatal("expected error for empty peers")
+	}
+}
+
+func TestLookupRaftLeaderMasterCanceledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err := LookupRaftLeaderMaster(
+		ctx,
+		[]pb.ServerAddress{"127.0.0.1:1"},
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err == nil {
+		t.Fatal("expected error for canceled context")
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context.Canceled, got %v", err)
+	}
+}

weed/server/volume_grpc_admin.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"google.golang.org/grpc/codes"
@@ -279,9 +280,27 @@ func (vs *VolumeServer) makeVolumeWritable(ctx context.Context, v *storage.Volum
 	return nil
 }
 
+func isNotLeaderErr(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "Not current leader")
+}
+
 func (vs *VolumeServer) notifyMasterVolumeReadonly(ctx context.Context, v *storage.Volume, isReadOnly bool) error {
-	if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(ctx), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
-		_, err := client.VolumeMarkReadonly(context.Background(), &master_pb.VolumeMarkReadonlyRequest{
+	master := vs.GetMaster(ctx)
+	err := vs.volumeMarkReadonlyOnMaster(ctx, master, v, isReadOnly)
+	if err != nil && isNotLeaderErr(err) {
+		leader, lookupErr := vs.lookupRaftLeaderMaster(ctx)
+		if lookupErr != nil {
+			return fmt.Errorf("heartbeat master %s rejected mark-readonly and leader lookup failed: %w", master, lookupErr)
+		}
+		master = leader
+		err = vs.volumeMarkReadonlyOnMaster(ctx, master, v, isReadOnly)
+	}
+	return err
+}
+
+func (vs *VolumeServer) volumeMarkReadonlyOnMaster(ctx context.Context, master pb.ServerAddress, v *storage.Volume, isReadOnly bool) error {
+	if grpcErr := pb.WithMasterClient(ctx, false, master, vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		_, err := client.VolumeMarkReadonly(ctx, &master_pb.VolumeMarkReadonlyRequest{
 			Ip:               vs.store.Ip,
 			Port:             uint32(vs.store.Port),
 			VolumeId:         uint32(v.Id),
@@ -296,8 +315,8 @@ func (vs *VolumeServer) notifyMasterVolumeReadonly(ctx context.Context, v *stora
 		}
 		return nil
 	}); grpcErr != nil {
-		glog.V(0).Infof("connect to %s: %v", vs.GetMaster(context.Background()), grpcErr)
-		return fmt.Errorf("grpc VolumeMarkReadonly with master %s: %v", vs.GetMaster(context.Background()), grpcErr)
+		glog.V(0).Infof("connect to %s: %v", master, grpcErr)
+		return fmt.Errorf("grpc VolumeMarkReadonly with master %s: %v", master, grpcErr)
 	}
 	return nil
 }

weed/server/volume_grpc_client_to_master.go

@@ -1,6 +1,8 @@
 package weed_server
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"os"
 	"strings"
@@ -10,14 +12,14 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 
-	"context"
-
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -27,6 +29,40 @@ func (vs *VolumeServer) GetMaster(ctx context.Context) pb.ServerAddress {
 	return vs.getCurrentMaster()
 }
 
+// lookupRaftLeaderMaster resolves the raft leader for topology mutations via
+// GetMasterConfiguration on configured peers. It does not update currentMaster;
+// the heartbeat loop owns that field and reconnects when the stream reports a
+// different leader than the connected peer.
+func (vs *VolumeServer) lookupRaftLeaderMaster(ctx context.Context) (pb.ServerAddress, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	leader, err := operation.LookupRaftLeaderMaster(ctx, vs.SeedMasterNodes, vs.grpcDialOption)
+	if err != nil {
+		if isContextDoneErr(err) {
+			glog.V(1).Infof("volume server %s:%d: raft leader lookup: %v", vs.store.Ip, vs.store.Port, err)
+		} else {
+			glog.V(0).Infof("volume server %s:%d: raft leader lookup: %v", vs.store.Ip, vs.store.Port, err)
+		}
+		return "", err
+	}
+	current := vs.getCurrentMaster()
+	if !leader.Equals(current) {
+		glog.V(1).Infof("volume server %s:%d: raft leader %v (heartbeat peer %v)", vs.store.Ip, vs.store.Port, leader, current)
+	}
+	return leader, nil
+}
+
+func isContextDoneErr(err error) bool {
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	if st, ok := status.FromError(err); ok {
+		return st.Code() == codes.Canceled || st.Code() == codes.DeadlineExceeded
+	}
+	return false
+}
+
 // getCurrentMaster returns vs.currentMaster under a read lock so callers
 // (e.g. Ping admission) do not race with the heartbeat goroutine that
 // rewrites it on leader changes.
@@ -262,10 +298,10 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 		case first := <-vs.store.NewEcShardsChan:
 			shards := util.DrainChannel(vs.store.NewEcShardsChan, first)
 			deltaBeat := &master_pb.Heartbeat{
-				Ip:           ip,
-				Port:         port,
-				DataCenter:   dataCenter,
-				Rack:         rack,
+				Ip:          ip,
+				Port:        port,
+				DataCenter:  dataCenter,
+				Rack:        rack,
 				NewEcShards: shards,
 			}
 			for _, s := range shards {
@@ -295,10 +331,10 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 		case first := <-vs.store.DeletedEcShardsChan:
 			shards := util.DrainChannel(vs.store.DeletedEcShardsChan, first)
 			deltaBeat := &master_pb.Heartbeat{
-				Ip:               ip,
-				Port:             port,
-				DataCenter:       dataCenter,
-				Rack:             rack,
+				Ip:              ip,
+				Port:            port,
+				DataCenter:      dataCenter,
+				Rack:            rack,
 				DeletedEcShards: shards,
 			}
 			for _, s := range shards {

weed/server/volume_grpc_client_to_master_test.go

@@ -0,0 +1,62 @@
+package weed_server
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func TestLookupRaftLeaderMasterDoesNotUpdateCurrentMaster(t *testing.T) {
+	follower := startFakeMasterServerForLeaderLookup(t, &fakeLeaderConfigServer{
+		leader: "leader.example.com:9333.19333",
+	})
+	heartbeatPeer := pb.ServerAddress("10.0.0.1:9333")
+	vs := &VolumeServer{
+		SeedMasterNodes: []pb.ServerAddress{follower},
+		grpcDialOption:  grpc.WithTransportCredentials(insecure.NewCredentials()),
+		store:           &storage.Store{Ip: "127.0.0.1", Port: 8080},
+	}
+	vs.setCurrentMaster(heartbeatPeer)
+
+	leader, err := vs.lookupRaftLeaderMaster(context.Background())
+	if err != nil {
+		t.Fatalf("lookup: %v", err)
+	}
+	if got, want := leader.ToHttpAddress(), "leader.example.com:9333"; got != want {
+		t.Fatalf("leader = %q, want %q", got, want)
+	}
+	if got := vs.getCurrentMaster(); got != heartbeatPeer {
+		t.Fatalf("currentMaster = %v, want unchanged %v", got, heartbeatPeer)
+	}
+}
+
+type fakeLeaderConfigServer struct {
+	master_pb.UnimplementedSeaweedServer
+	leader string
+}
+
+func (s *fakeLeaderConfigServer) GetMasterConfiguration(_ context.Context, _ *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) {
+	return &master_pb.GetMasterConfigurationResponse{Leader: s.leader}, nil
+}
+
+func startFakeMasterServerForLeaderLookup(t *testing.T, srv master_pb.SeaweedServer) pb.ServerAddress {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	grpcServer := grpc.NewServer()
+	master_pb.RegisterSeaweedServer(grpcServer, srv)
+	go func() { _ = grpcServer.Serve(lis) }()
+	t.Cleanup(grpcServer.GracefulStop)
+
+	_, port, _ := net.SplitHostPort(lis.Addr().String())
+	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port))
+}