fix(master): shed assign load when volume growth is already in flight (#10121)

Under a herd of concurrent assigns with no writable volume, Assign spun
PickForWrite for the full 10s timeout, pinning a goroutine per request and
starving the master of the cycles it needs to process growth and answer
heartbeats. When growth is the relevant remedy and already in flight, stop
spinning: if free space exists, shed with a fast retryable error so clients
back off and retry once growth lands; if the cluster is out of space, fail fast
with the real out-of-space error instead of masking it as retryable.

The gRPC shed uses ResourceExhausted, not Unavailable: operation.Assign retries
it, but the client connection layer doesn't treat it as a dead channel, so a
per-request shed across a herd doesn't tear down the shared master connection
and cancel every other in-flight assign. The HTTP dirAssignHandler sheds with
503 + Retry-After.

weed/operation/assign_file_id.go

@@ -74,7 +74,9 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO
 					return false
 				}
 				switch st.Code() {
-				case codes.Unavailable:
+				case codes.Unavailable, codes.ResourceExhausted:
+					// ResourceExhausted: the master is shedding because volume growth
+					// is in flight; retry so we pick up the new volume once it lands.
 					return true
 				case codes.Canceled, codes.DeadlineExceeded:
 					// A stale cached gRPC channel (e.g., master restart behind

weed/pb/grpc_client_server.go

@@ -391,6 +391,13 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool {
 		switch code {
 		case codes.Unavailable, codes.Aborted, codes.Internal:
 			return true
+		case codes.ResourceExhausted:
+			// Server-side backpressure on a healthy channel (e.g. the master's
+			// assign shed while volume growth is in flight), not a dead connection.
+			// Invalidating would Close() the shared conn and cancel every other
+			// in-flight RPC with "the client connection is closing"; keep it and let
+			// the caller retry.
+			return false
 		case codes.Canceled, codes.DeadlineExceeded:
 			// Ambiguous: this fires both when a stale cached channel rejects
 			// RPCs (e.g. a peer restart behind a k8s Service VIP), where we must

weed/pb/grpc_client_server_test.go

@@ -66,6 +66,22 @@ func TestShouldInvalidateConnection_StaleChannelStillInvalidates(t *testing.T) {
 	}
 }
 
+// TestShouldInvalidateConnection_ResourceExhaustedIsPerRequest ensures the
+// master's growth-in-progress assign shed (codes.ResourceExhausted) does NOT
+// tear down the shared cached ClientConn. The shed fires per-request across a
+// herd of concurrent assigns; invalidating on it would cancel every other
+// in-flight assign with "the client connection is closing" — the cascade in
+// seaweedfs#10118. It is retried by the caller without touching the channel.
+func TestShouldInvalidateConnection_ResourceExhaustedIsPerRequest(t *testing.T) {
+	shed := status.Error(codes.ResourceExhausted, "no writable volumes for x, volume growth in progress")
+	if shouldInvalidateConnection(context.Background(), shed) {
+		t.Fatalf("ResourceExhausted shed must not invalidate the shared connection")
+	}
+	if shouldInvalidateConnection(nil, shed) {
+		t.Fatalf("ResourceExhausted shed must not invalidate the shared connection (nil ctx)")
+	}
+}
+
 // TestShouldInvalidateConnection_GenuineInternalStillInvalidates ensures the
 // marshal-error carve-out does not swallow real server-side Internal errors,
 // which previously caused — and should continue to cause — connection

weed/server/master_grpc_server_assign.go

@@ -30,9 +30,10 @@ func (ms *MasterServer) StreamAssign(server master_pb.Seaweed_StreamAssignServer
 		}
 		resp, err := ms.Assign(context.Background(), req)
 		if err != nil {
-			// Return transient errors (e.g. warmup) as in-band error responses
-			// instead of killing the stream, so pooled connections survive.
-			if st, ok := status.FromError(err); ok && st.Code() == codes.Unavailable {
+			// Return transient errors (warmup, growth-in-progress shed) as in-band
+			// error responses instead of killing the stream, so pooled connections
+			// survive.
+			if st, ok := status.FromError(err); ok && (st.Code() == codes.Unavailable || st.Code() == codes.ResourceExhausted) {
 				glog.V(1).Infof("StreamAssign transient error: %v", err)
 				resp = &master_pb.AssignResponse{Error: st.Message()}
 			} else {
@@ -130,6 +131,17 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
 			if (req.DataCenter != "" || req.Rack != "") && strings.Contains(err.Error(), topology.NoWritableVolumes) {
 				break
 			}
+			// Growth is the remedy and already in flight; don't pin a goroutine
+			// spinning out the timeout under an assign herd.
+			if shouldGrow && vl.HasGrowRequest() {
+				if ms.Topo.AvailableSpaceFor(option) <= 0 {
+					break // out of space: surface the real error, not a retryable shed
+				}
+				// ResourceExhausted, not Unavailable: clients retry it (assign_file_id.go)
+				// but the gRPC layer doesn't treat it as a dead channel, so the shed
+				// doesn't tear down the shared master connection mid-herd.
+				return nil, status.Errorf(codes.ResourceExhausted, "no writable volumes for %s, volume growth in progress", option.String())
+			}
 			time.Sleep(200 * time.Millisecond)
 			continue
 		}

weed/server/master_grpc_server_assign_test.go

@@ -0,0 +1,97 @@
+package weed_server
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/raft"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+// leaderRaftServer answers only State(); the embedded nil interface panics if any
+// other method is called.
+type leaderRaftServer struct {
+	raft.Server
+}
+
+func (leaderRaftServer) State() string { return raft.Leader }
+
+func newLeaderMaster() *MasterServer {
+	topo := topology.NewTopology("test", sequence.NewMemorySequencer(), 1<<20, 5, false)
+	topo.RaftServer = leaderRaftServer{}
+	return &MasterServer{
+		Topo:                    topo,
+		option:                  &MasterOption{},
+		volumeGrowthRequestChan: make(chan *topology.VolumeGrowRequest, 1<<6),
+	}
+}
+
+// markGrowthInFlight flags growth on the same VolumeLayout the handler resolves,
+// by mirroring its exact lookup.
+func markGrowthInFlight(t *testing.T, topo *topology.Topology, req *master_pb.AssignRequest) {
+	t.Helper()
+	rp, err := super_block.NewReplicaPlacementFromString(req.Replication)
+	require.NoError(t, err)
+	ttl, err := needle.ReadTTL(req.Ttl)
+	require.NoError(t, err)
+	topo.GetVolumeLayout(req.Collection, rp, ttl, types.ToDiskType(req.DiskType)).AddGrowRequest()
+}
+
+// With free space but no writable volume and growth already in flight, Assign
+// sheds with a retryable ResourceExhausted immediately instead of spinning to
+// timeout. ResourceExhausted (not Unavailable) so the shed isn't mistaken for a
+// dead channel and doesn't tear down the shared master connection.
+func TestAssignShedsLoadWhenGrowthInFlight(t *testing.T) {
+	ms := newLeaderMaster()
+	// Free volume slots but no writable volume yet, so growth is the remedy.
+	ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100})
+
+	req := &master_pb.AssignRequest{Count: 1, Replication: "000"}
+	markGrowthInFlight(t, ms.Topo, req)
+
+	start := time.Now()
+	resp, err := ms.Assign(context.Background(), req)
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	require.Nil(t, resp)
+	st, ok := status.FromError(err)
+	require.True(t, ok)
+	assert.Equal(t, codes.ResourceExhausted, st.Code())
+	// Shed immediately rather than spinning out the 10s retry budget.
+	assert.Less(t, elapsed, 2*time.Second)
+	// Growth was already pending, so we must not enqueue another grow request.
+	assert.Len(t, ms.volumeGrowthRequestChan, 0)
+}
+
+// Out of space, Assign fails fast with the real error rather than masking it as
+// a retryable "growth in progress".
+func TestAssignFailsFastWhenOutOfSpace(t *testing.T) {
+	ms := newLeaderMaster() // no data nodes -> no free space
+
+	req := &master_pb.AssignRequest{Count: 1, Replication: "000"}
+
+	start := time.Now()
+	resp, err := ms.Assign(context.Background(), req)
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	require.Nil(t, resp)
+	if st, ok := status.FromError(err); ok {
+		assert.NotEqual(t, codes.Unavailable, st.Code())
+	}
+	assert.Contains(t, err.Error(), "no free volumes left")
+	assert.Less(t, elapsed, 2*time.Second)
+}

weed/server/master_server_handlers.go

@@ -187,6 +187,17 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request)
 		if err != nil {
 			stats.MasterPickForWriteErrorCounter.Inc()
 			lastErr = err
+			// See Assign: shed instead of spinning when growth is already in flight.
+			if shouldGrow && vl.HasGrowRequest() {
+				if ms.Topo.AvailableSpaceFor(option) <= 0 {
+					break // out of space: surface the real error (406 below)
+				}
+				w.Header().Set("Retry-After", "1")
+				writeJsonQuiet(w, r, http.StatusServiceUnavailable, operation.AssignResult{
+					Error: fmt.Sprintf("no writable volumes for %s, volume growth in progress", option.String()),
+				})
+				return
+			}
 			time.Sleep(200 * time.Millisecond)
 			continue
 		} else {