fix(ec): honor disk_id in ReceiveFile so EC shards respect admin placement (#9184) (#9185)

* test(volume_server): reproduce #9184 EC ReceiveFile disk-placement bug

The plugin-worker EC task sends shards via ReceiveFile, which picks
Locations[0] as the target directory regardless of the admin planner's
TargetDisk assignment. ReceiveFileInfo has no disk_id field, so there
is no wire channel to honor the plan.

Adds StartSingleVolumeClusterWithDataDirs to the integration framework
so tests can launch a volume server with N data directories. The new
repro asserts the current (buggy) behavior: sending three distinct EC
shards via ReceiveFile leaves all three files in dir[0] and the other
dirs empty. When the fix adds disk_id to ReceiveFileInfo, this
assertion must flip to verify the planned placement is respected.

* fix(ec): honor disk_id in ReceiveFile so EC shards respect admin placement

Before this change, VolumeServer.ReceiveFile for EC shards always
selected the first HDD location (Locations[0]). The plugin-worker EC
task had no way to pass the admin planner's per-shard disk
assignment — ReceiveFileInfo carried no disk_id field — so every
received EC shard piled onto a single disk per destination server.
On multi-disk servers this caused uneven load (one disk absorbing all
EC shard I/O), frequent ENOSPC retries, and a growing EC backlog
under sustained ingest (see issue #9184).

Changes:
- proto: add disk_id to ReceiveFileInfo, mirroring
  VolumeEcShardsCopyRequest.disk_id.
- worker: DistributeEcShards tracks the planner-assigned disk per
  shard; sendShardFileToDestination forwards that disk id. Metadata
  files (ecx/ecj/vif) inherit the disk of the first data shard
  targeting the same node so they land next to the shards.
- server: ReceiveFile honors disk_id when > 0 with bounds
  validation; disk_id=0 (unset) falls back to the same
  auto-selection pattern as VolumeEcShardsCopy (prefer disk that
  already has shards for this volume, then any HDD with free space,
  then any location with free space).

Tests updated:
- TestReceiveFileEcShardHonorsDiskID asserts three shards sent with
  disk_id={1,2,0} land on data dirs 1, 2, and 0 respectively.
- TestReceiveFileEcShardRejectsInvalidDiskID pins the out-of-range
  disk_id rejection path.

* fix(volume-rust): honor disk_id in ReceiveFile for EC shards

Mirror the Go-side change: when disk_id > 0 place the EC shard on the
requested disk; when unset, auto-select with the same preference order
as volume_ec_shards_copy (disk already holding shards, then any HDD,
then any disk).

* fix(volume): compare disk_id as uint32 to avoid 32-bit overflow

On 32-bit Go builds `int(fileInfo.DiskId) >= len(Locations)` can wrap a
high-bit uint32 to a negative int, bypassing the bounds check before the
index operation. Compare in the uint32 domain instead.

* test(ec): fail invalid-disk_id test on transport error

Previously a transport-level error from CloseAndRecv silently passed the
test by returning early, masking any real gRPC failure. Fail loudly so
only the structured ReceiveFileResponse rejection path counts as a pass.

* docs(test): explain why DiskId=0 auto-selects dir 0 in EC placement test

Documents the load-bearing assumption that shards are never mounted in
this test, so loc.FindEcVolume always returns false and auto-select
falls through to the first HDD. Saves future readers from re-deriving
the expected directory for the DiskId=0 case.

* fix(test): preserve baseDir/volume path for single-dir clusters

StartSingleVolumeClusterWithDataDirs started naming the data directory
volume0 even in the dataDirCount=1 case, which broke Scrub tests that
reach into baseDir/volume via CorruptDatFile / CorruptEcShardFile /
CorruptEcxFile. Keep the legacy name for single-dir clusters; only use
the indexed "volumeN" layout when multiple disks are requested.

test/volume_server/framework/cluster.go

@@ -54,12 +54,25 @@ type Cluster struct {
 	masterCmd *exec.Cmd
 	volumeCmd *exec.Cmd
 
+	volumeDataDirs []string
+
 	cleanupOnce sync.Once
 }
 
 // StartSingleVolumeCluster boots one master and one volume server.
 func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster {
+	return StartSingleVolumeClusterWithDataDirs(t, profile, 1)
+}
+
+// StartSingleVolumeClusterWithDataDirs boots one master and one volume server
+// with dataDirCount separate data directories (passed to -dir as a comma list).
+// Each directory becomes its own DiskLocation on the volume server, letting
+// tests exercise multi-disk EC placement paths.
+func StartSingleVolumeClusterWithDataDirs(t testing.TB, profile matrix.Profile, dataDirCount int) *Cluster {
 	t.Helper()
+	if dataDirCount < 1 {
+		t.Fatalf("dataDirCount must be >= 1, got %d", dataDirCount)
+	}
 
 	weedBinary, err := FindOrBuildWeedBinary()
 	if err != nil {
@@ -74,8 +87,19 @@ func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster {
 	configDir := filepath.Join(baseDir, "config")
 	logsDir := filepath.Join(baseDir, "logs")
 	masterDataDir := filepath.Join(baseDir, "master")
-	volumeDataDir := filepath.Join(baseDir, "volume")
-	for _, dir := range []string{configDir, logsDir, masterDataDir, volumeDataDir} {
+	volumeDataDirs := make([]string, dataDirCount)
+	// Single-dir layout stays at baseDir/volume so existing fixtures
+	// (CorruptDatFile etc.) that hardcode that path keep working. Only
+	// multi-dir clusters get the "volumeN" layout.
+	for i := 0; i < dataDirCount; i++ {
+		if dataDirCount == 1 {
+			volumeDataDirs[i] = filepath.Join(baseDir, "volume")
+		} else {
+			volumeDataDirs[i] = filepath.Join(baseDir, fmt.Sprintf("volume%d", i))
+		}
+	}
+	setupDirs := append([]string{configDir, logsDir, masterDataDir}, volumeDataDirs...)
+	for _, dir := range setupDirs {
 		if mkErr := os.MkdirAll(dir, 0o755); mkErr != nil {
 			t.Fatalf("create %s: %v", dir, mkErr)
 		}
@@ -120,11 +144,12 @@ func StartSingleVolumeCluster(t testing.TB, profile matrix.Profile) *Cluster {
 		t.Fatalf("wait for master readiness: %v\nmaster log tail:\n%s", err, masterLog)
 	}
 
-	if err = c.startVolume(volumeDataDir); err != nil {
+	if err = c.startVolume(volumeDataDirs); err != nil {
 		masterLog := c.tailLog("master.log")
 		c.Stop()
 		t.Fatalf("start volume: %v\nmaster log tail:\n%s", err, masterLog)
 	}
+	c.volumeDataDirs = volumeDataDirs
 	if err = c.waitForHTTP(c.VolumeAdminURL() + "/status"); err != nil {
 		volumeLog := c.tailLog("volume.log")
 		c.Stop()
@@ -184,12 +209,16 @@ func (c *Cluster) startMaster(dataDir string) error {
 	return c.masterCmd.Start()
 }
 
-func (c *Cluster) startVolume(dataDir string) error {
+func (c *Cluster) startVolume(dataDirs []string) error {
 	logFile, err := os.Create(filepath.Join(c.logsDir, "volume.log"))
 	if err != nil {
 		return err
 	}
 
+	maxPerDir := make([]string, len(dataDirs))
+	for i := range dataDirs {
+		maxPerDir[i] = "16"
+	}
 	args := []string{
 		"-config_dir=" + c.configDir,
 		"volume",
@@ -197,8 +226,8 @@ func (c *Cluster) startVolume(dataDir string) error {
 		"-port=" + strconv.Itoa(c.volumePort),
 		"-port.grpc=" + strconv.Itoa(c.volumeGrpcPort),
 		"-port.public=" + strconv.Itoa(c.volumePubPort),
-		"-dir=" + dataDir,
-		"-max=16",
+		"-dir=" + strings.Join(dataDirs, ","),
+		"-max=" + strings.Join(maxPerDir, ","),
 		"-master=127.0.0.1:" + strconv.Itoa(c.masterPort),
 		"-readMode=" + c.profile.ReadMode,
 		"-concurrentUploadLimitMB=" + strconv.Itoa(c.profile.ConcurrentUploadLimitMB),
@@ -415,3 +444,10 @@ func (c *Cluster) VolumePublicURL() string {
 func (c *Cluster) BaseDir() string {
 	return c.baseDir
 }
+
+// VolumeDataDirs returns the data directories the volume server was started with.
+// Index 0 corresponds to DiskLocation 0, index 1 to DiskLocation 1, and so on.
+// Tests can scan these directories to verify where files physically landed.
+func (c *Cluster) VolumeDataDirs() []string {
+	return append([]string(nil), c.volumeDataDirs...)
+}

test/volume_server/grpc/ec_receive_disk_placement_test.go

@@ -0,0 +1,178 @@
+package volume_server_grpc_test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
+	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+)
+
+// Sends EC shards with disk_id={1, 2, 0} and verifies each lands on the
+// requested disk (0 → auto-select → disk 0 for a fresh volume).
+func TestReceiveFileEcShardHonorsDiskID(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	const dataDirCount = 3
+	clusterHarness := framework.StartSingleVolumeClusterWithDataDirs(t, matrix.P1(), dataDirCount)
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	dataDirs := clusterHarness.VolumeDataDirs()
+	if len(dataDirs) != dataDirCount {
+		t.Fatalf("expected %d data dirs, got %d: %v", dataDirCount, len(dataDirs), dataDirs)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	const volumeID = uint32(9184)
+	const collection = "ec-disk-placement"
+
+	// DiskId=0 triggers auto-select. Since this test never calls
+	// VolumeEcShardsMount, loc.FindEcVolume returns false on every disk, so
+	// the "already holds shards" preference is skipped and auto-select falls
+	// through to the first HDD (dir 0). If shards get mounted between uploads
+	// in a future revision, the expected dir for DiskId=0 becomes the first
+	// disk that already holds a shard.
+	shards := []struct {
+		shardID        uint32
+		requestedDisk  uint32
+		expectedDirIdx int
+	}{
+		{shardID: 0, requestedDisk: 1, expectedDirIdx: 1},
+		{shardID: 5, requestedDisk: 2, expectedDirIdx: 2},
+		{shardID: 13, requestedDisk: 0, expectedDirIdx: 0},
+	}
+	shardExt := func(id uint32) string { return fmt.Sprintf(".ec%02d", id) }
+
+	for _, s := range shards {
+		payload := []byte(fmt.Sprintf("ec-shard-payload-%02d", s.shardID))
+		stream, err := grpcClient.ReceiveFile(ctx)
+		if err != nil {
+			t.Fatalf("ReceiveFile stream create for shard %d: %v", s.shardID, err)
+		}
+		if err = stream.Send(&volume_server_pb.ReceiveFileRequest{
+			Data: &volume_server_pb.ReceiveFileRequest_Info{
+				Info: &volume_server_pb.ReceiveFileInfo{
+					VolumeId:   volumeID,
+					Ext:        shardExt(s.shardID),
+					Collection: collection,
+					IsEcVolume: true,
+					ShardId:    s.shardID,
+					FileSize:   uint64(len(payload)),
+					DiskId:     s.requestedDisk,
+				},
+			},
+		}); err != nil {
+			t.Fatalf("ReceiveFile send info for shard %d: %v", s.shardID, err)
+		}
+		if err = stream.Send(&volume_server_pb.ReceiveFileRequest{
+			Data: &volume_server_pb.ReceiveFileRequest_FileContent{FileContent: payload},
+		}); err != nil {
+			t.Fatalf("ReceiveFile send content for shard %d: %v", s.shardID, err)
+		}
+		resp, err := stream.CloseAndRecv()
+		if err != nil {
+			t.Fatalf("ReceiveFile close for shard %d: %v", s.shardID, err)
+		}
+		if resp.GetError() != "" {
+			t.Fatalf("ReceiveFile shard %d error response: %s", s.shardID, resp.GetError())
+		}
+		if resp.GetBytesWritten() != uint64(len(payload)) {
+			t.Fatalf("ReceiveFile shard %d bytes_written mismatch: got %d want %d",
+				s.shardID, resp.GetBytesWritten(), len(payload))
+		}
+	}
+
+	// Scan every data dir and record which shard files live where.
+	perDirShards := make(map[int][]uint32)
+	for dirIdx, dir := range dataDirs {
+		for _, s := range shards {
+			shardPath := filepath.Join(dir, fmt.Sprintf("%s_%d%s", collection, volumeID, shardExt(s.shardID)))
+			if _, err := os.Stat(shardPath); err == nil {
+				perDirShards[dirIdx] = append(perDirShards[dirIdx], s.shardID)
+			} else if !os.IsNotExist(err) {
+				t.Fatalf("unexpected stat error for %s: %v", shardPath, err)
+			}
+		}
+	}
+
+	for _, s := range shards {
+		found := false
+		for _, got := range perDirShards[s.expectedDirIdx] {
+			if got == s.shardID {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Fatalf("shard %d (requestedDisk=%d) expected on dir[%d], full layout: %v",
+				s.shardID, s.requestedDisk, s.expectedDirIdx, perDirShards)
+		}
+		for dirIdx, ids := range perDirShards {
+			if dirIdx == s.expectedDirIdx {
+				continue
+			}
+			for _, id := range ids {
+				if id == s.shardID {
+					t.Fatalf("shard %d leaked onto dir[%d]: full layout: %v", s.shardID, dirIdx, perDirShards)
+				}
+			}
+		}
+	}
+
+	t.Logf("disk_id honored: shard placement = %v", perDirShards)
+}
+
+func TestReceiveFileEcShardRejectsInvalidDiskID(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	const dataDirCount = 2
+	clusterHarness := framework.StartSingleVolumeClusterWithDataDirs(t, matrix.P1(), dataDirCount)
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	const volumeID = uint32(91840)
+	const collection = "ec-invalid-disk"
+	payload := []byte("invalid-disk-id-payload")
+
+	stream, err := grpcClient.ReceiveFile(ctx)
+	if err != nil {
+		t.Fatalf("ReceiveFile stream create: %v", err)
+	}
+	if err = stream.Send(&volume_server_pb.ReceiveFileRequest{
+		Data: &volume_server_pb.ReceiveFileRequest_Info{
+			Info: &volume_server_pb.ReceiveFileInfo{
+				VolumeId:   volumeID,
+				Ext:        ".ec00",
+				Collection: collection,
+				IsEcVolume: true,
+				ShardId:    0,
+				FileSize:   uint64(len(payload)),
+				DiskId:     99,
+			},
+		},
+	}); err != nil {
+		t.Fatalf("ReceiveFile send info: %v", err)
+	}
+	resp, err := stream.CloseAndRecv()
+	if err != nil {
+		t.Fatalf("ReceiveFile close: %v", err)
+	}
+	if resp.GetError() == "" {
+		t.Fatalf("expected invalid disk_id rejection, got success response: %+v", resp)
+	}
+}