master: scope the startup capacity shed to a truly empty topology (#11058)

* master: scope the startup capacity shed to a truly empty topology

The retryable "no volume server capacity registered yet" shed checked
capacity for the requested disk type, so a cluster serving only other
media -- where that capacity will never register -- shed every assign
until the client's deadline instead of failing fast. An unsteered write
to such a cluster hung for its full HTTP deadline and surfaced "context
deadline exceeded" in place of "No writable volumes". Shed only while
no disk type has any registered capacity, and name the unserved medium
in the fast failure.

Claude-Session: https://claude.ai/code/session_01TF7FQghfDkpdoZgakTMX4R

* master: name the unserved medium for every fail-fast caller

The diagnostic sat in the growth-initiator block, so a follower joining
an in-flight growth and a growth-disabled master failed the same way
with only the generic pick error. Wrap at the fail-fast break instead,
which every caller reaches, and cover all three paths in the test.

Claude-Session: https://claude.ai/code/session_01TF7FQghfDkpdoZgakTMX4R

weed/server/master_grpc_server_assign.go

@@ -136,13 +136,24 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
 			}
 			if shouldGrow {
 				if ms.Topo.AvailableSpaceFor(option) <= 0 {
-					if ms.Topo.CapacityFor(option) > 0 {
+					// Fail fast whenever any capacity is registered: full for
-						break // out of space: surface the real error, not a retryable shed
+					// this medium, or a medium no volume server serves — a
+					// state a heartbeat won't change, so a retryable shed
+					// would loop until the client's deadline. Shed retryably
+					// only while nothing at all has registered, a just-started
+					// cluster whose volume servers have not heartbeated, so
+					// the first write rides out the startup window instead of
+					// failing outright.
+					if ms.Topo.CapacityForAnyDisk() > 0 {
+						if ms.Topo.CapacityFor(option) <= 0 {
+							// Wrapped here, not beside the "no free volumes left"
+							// wrap above, so followers and growth-disabled
+							// masters name the unserved medium too — the
+							// initiator block is skipped for both.
+							lastErr = fmt.Errorf("%s and no volume server carries disk type %q for %s", err.Error(), option.DiskType.ReadableString(), option.String())
+						}
+						break // surface the real error, not a retryable shed
 					}
-					// No capacity registered for this disk type yet, typically a
-					// just-started cluster whose volume servers have not
-					// heartbeated. Shed retryably so the first write rides out
-					// the startup window instead of failing outright.
 					return nil, status.Errorf(codes.ResourceExhausted, "no volume server capacity registered yet for %s", option.String())
 				}
 				// Only the initiator waits, and only while the growth it triggered

weed/server/master_grpc_server_assign_test.go

@@ -228,3 +228,48 @@ func TestAssignShedsRetryablyBeforeCapacityRegisters(t *testing.T) {
 	assert.Equal(t, codes.ResourceExhausted, st.Code())
 	assert.Less(t, elapsed, 2*time.Second)
 }
+
+// A cluster serving only other media is not one still starting up: capacity for
+// the requested disk type will never register, so the startup shed would loop
+// until the client's deadline. Assign must fail fast with the real error and
+// name the unserved medium — for the growth initiator, for a follower whose
+// growth is already in flight, and when growth is disabled outright.
+func TestAssignFailsFastWhenDiskTypeUnserved(t *testing.T) {
+	tests := []struct {
+		name  string
+		setup func(t *testing.T, ms *MasterServer, req *master_pb.AssignRequest)
+	}{
+		{"initiator", func(t *testing.T, ms *MasterServer, req *master_pb.AssignRequest) {}},
+		{"follower joins growth in flight", func(t *testing.T, ms *MasterServer, req *master_pb.AssignRequest) {
+			markGrowthInFlight(t, ms.Topo, req)
+		}},
+		{"growth disabled", func(t *testing.T, ms *MasterServer, req *master_pb.AssignRequest) {
+			ms.option.VolumeGrowthDisabled = true
+		}},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ms := newLeaderMaster()
+			// ssd capacity registered, but the request asks for the default (hdd).
+			ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+				GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"ssd": 1})
+
+			req := &master_pb.AssignRequest{Count: 1, Replication: "000", Collection: "fresh"}
+			tc.setup(t, ms, req)
+
+			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.NotEqual(t, codes.ResourceExhausted, st.Code())
+			}
+			assert.Contains(t, err.Error(), topology.NoWritableVolumes)
+			assert.Contains(t, err.Error(), `no volume server carries disk type "hdd"`)
+			assert.Less(t, elapsed, 2*time.Second)
+		})
+	}
+}

weed/topology/node.go

@@ -320,6 +320,18 @@ func (n *NodeImpl) CapacityFor(option *VolumeGrowOption) int64 {
 	return atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount)
 }
 
+// CapacityForAnyDisk is the total registered volume slots across every disk
+// type. CapacityFor answers zero both while a cluster is still starting and
+// when it never serves the option's medium; this tells the two apart.
+func (n *NodeImpl) CapacityForAnyDisk() (total int64) {
+	n.diskUsages.RLock()
+	defer n.diskUsages.RUnlock()
+	for _, t := range n.diskUsages.usages {
+		total += atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount)
+	}
+	return
+}
+
 // AvailableSpaceForReservation returns available space considering existing reservations
 func (n *NodeImpl) AvailableSpaceForReservation(option *VolumeGrowOption) int64 {
 	baseAvailable := n.AvailableSpaceFor(option)