fix(ec): generate .ecx before EC shards to prevent data inconsistency (#8972)

* fix(ec): generate .ecx before EC shards to prevent data inconsistency

In VolumeEcShardsGenerate, the .ecx index was generated from .idx AFTER
the EC shards were generated from .dat. If any write occurred between
these two steps (e.g. WriteNeedleBlob during replica sync, which bypasses
the read-only check), the .ecx would contain entries pointing to data
that doesn't exist in the EC shards, causing "shard too short" and
"size mismatch" errors on subsequent reads and scrubs.

Fix by generating .ecx FIRST, then snapshotting datFileSize, then
encoding EC shards. If a write sneaks in after .ecx generation, the
EC shards contain more data than .ecx references — which is harmless
(the extra data is simply not indexed).

Also snapshot datFileSize before EC encoding to ensure the .vif
reflects the same .dat state that .ecx was generated from.

Add TestEcConsistency_WritesBetweenEncodeAndEcx that reproduces the
race condition by appending data between EC encoding and .ecx generation.

* fix: pass actual offset to ReadBytes, improve test quality

- Pass offset.ToActualOffset() to ReadBytes instead of 0 to preserve
  correct error metrics and error messages within ReadBytes
- Handle Stat() error in assembleFromIntervalsAllowError
- Rename TestEcConsistency_DatFileGrowsDuringEncoding to
  TestEcConsistency_ExactLargeRowEncoding (test verifies fixed-size
  encoding, not concurrent growth)
- Update test comment to clarify it reproduces the old buggy sequence
- Fix verification loop to advance by readSize for full data coverage

* fix(ec): add dat/idx consistency check in worker EC encoding

The erasure_coding worker copies .dat and .idx as separate network
transfers. If a write lands on the source between these copies, the
.idx may have entries pointing past the end of .dat, leading to EC
volumes with .ecx entries that reference non-existent shard data.

Add verifyDatIdxConsistency() that walks the .idx and verifies no
entry's offset+size exceeds the .dat file size. This fails the EC
task early with a clear error instead of silently producing corrupt
EC volumes.

* test(ec): add integration test verifying .ecx/.ecd consistency

TestEcIndexConsistencyAfterEncode uploads multiple needles of varying
sizes (14B to 256KB), EC-encodes the volume, mounts data shards, then
reads every needle back via the EC read path and verifies payload
correctness. This catches any inconsistency between .ecx index entries
and EC shard data.

* fix(test): account for needle overhead in test volume fixture

WriteTestVolumeFiles created a .dat of exactly datSize bytes but the
.idx entry claimed a needle of that same size. GetActualSize adds
header + checksum + timestamp overhead, so the consistency check
correctly rejects this as the needle extends past the .dat file.

Fix by sizing the .dat to GetActualSize(datSize) so the .idx entry
is consistent with the .dat contents.

* fix(test): remove flaky shard ID assertion in EC scrub test

When shard 0 is truncated on disk after mount, the volume server may
detect corruption via parity mismatches (shards 10-13) rather than a
direct read failure on shard 0, depending on OS caching/mmap behavior.
Replace the brittle shard-0-specific check with a volume ID validation.

* fix(test): close upload response bodies and tighten file count assertion

Wrap UploadBytes calls with ReadAllAndClose to prevent connection/fd
leaks during test execution. Also tighten TotalFiles check from >= 1
to == 1 since ecSetup uploads exactly one file.

test/plugin_workers/volume_fixtures.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
@@ -21,7 +22,12 @@ func WriteTestVolumeFiles(t *testing.T, baseDir string, volumeID uint32, datSize
 	datPath := filepath.Join(baseDir, volumeFilename(volumeID, ".dat"))
 	idxPath := filepath.Join(baseDir, volumeFilename(volumeID, ".idx"))
 
-	data := make([]byte, datSize)
+	// The idx entry records the needle's data size. The actual on-disk size
+	// includes header + checksum + timestamp (GetActualSize). The .dat must
+	// be large enough to hold the full needle.
+	needleDataSize := types.Size(datSize)
+	actualSize := needle.GetActualSize(needleDataSize, needle.Version3)
+	data := make([]byte, actualSize)
 	rng := rand.New(rand.NewSource(99))
 	_, _ = rng.Read(data)
 	if err := os.WriteFile(datPath, data, 0644); err != nil {
@@ -35,7 +41,7 @@ func WriteTestVolumeFiles(t *testing.T, baseDir string, volumeID uint32, datSize
 
 	types.NeedleIdToBytes(entry[:idEnd], types.NeedleId(1))
 	types.OffsetToBytes(entry[idEnd:offsetEnd], types.ToOffset(0))
-	types.SizeToBytes(entry[offsetEnd:sizeEnd], types.Size(datSize))
+	types.SizeToBytes(entry[offsetEnd:sizeEnd], needleDataSize)
 
 	if err := os.WriteFile(idxPath, entry, 0644); err != nil {
 		t.Fatalf("write idx file: %v", err)

test/volume_server/grpc/erasure_coding_test.go

@@ -752,6 +752,87 @@ func TestEcShardsCopyFromPeerSuccess(t *testing.T) {
 	}
 }
 
+// TestEcIndexConsistencyAfterEncode verifies that every needle indexed in .ecx
+// can be read back correctly from EC shards after VolumeEcShardsGenerate.
+// This catches the race condition fixed in this PR where .ecx could reference
+// data not present in EC shards.
+func TestEcIndexConsistencyAfterEncode(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
+	conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
+	defer conn.Close()
+
+	const volumeID = uint32(130)
+	framework.AllocateVolume(t, grpcClient, volumeID, "")
+
+	httpClient := framework.NewHTTPClient()
+
+	// Upload multiple needles of varying sizes
+	type testNeedle struct {
+		fid     string
+		payload []byte
+	}
+	needles := []testNeedle{
+		{framework.NewFileID(volumeID, 1001, 0xAABB0001), []byte("small-needle-1")},
+		{framework.NewFileID(volumeID, 1002, 0xAABB0002), make([]byte, 1024)},     // 1KB
+		{framework.NewFileID(volumeID, 1003, 0xAABB0003), make([]byte, 64*1024)},   // 64KB
+		{framework.NewFileID(volumeID, 1004, 0xAABB0004), make([]byte, 256*1024)},  // 256KB
+		{framework.NewFileID(volumeID, 1005, 0xAABB0005), []byte("small-needle-2")},
+	}
+
+	// Fill larger payloads with recognizable data
+	for i := range needles {
+		for j := range needles[i].payload {
+			needles[i].payload[j] = byte(i*37 + j%251)
+		}
+	}
+
+	for _, n := range needles {
+		resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), n.fid, n.payload)
+		_ = framework.ReadAllAndClose(t, resp)
+		if resp.StatusCode != http.StatusCreated {
+			t.Fatalf("upload %s expected 201, got %d", n.fid, resp.StatusCode)
+		}
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// EC encode
+	_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
+	}
+
+	// Mount all data shards so reads go through the EC path
+	_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
+		VolumeId:   volumeID,
+		Collection: "",
+		ShardIds:   []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
+	})
+	if err != nil {
+		t.Fatalf("VolumeEcShardsMount failed: %v", err)
+	}
+
+	// Read every needle back from EC shards and verify payload
+	for _, n := range needles {
+		readResp := framework.ReadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), n.fid)
+		readBody := framework.ReadAllAndClose(t, readResp)
+		if readResp.StatusCode != http.StatusOK {
+			t.Fatalf("EC read %s expected 200, got %d", n.fid, readResp.StatusCode)
+		}
+		if string(readBody) != string(n.payload) {
+			t.Fatalf("EC read %s payload mismatch: got %d bytes, want %d bytes", n.fid, len(readBody), len(n.payload))
+		}
+	}
+}
+
 func TestEcShardsCopyFailsWhenSourceUnavailable(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test in short mode")

weed/server/volume_grpc_erasure_coding.go

@@ -86,16 +86,30 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_
 		os.Remove(v.IndexFileName() + ".ecx")
 	}()
 
+	// IMPORTANT: Generate .ecx BEFORE EC shards to prevent a race condition.
+	// If .ecx were generated after EC shards, any write (e.g. from WriteNeedleBlob
+	// during replica sync) between the two steps would add entries to .idx that
+	// end up in .ecx but whose data is NOT in the EC shards — causing "shard too
+	// short" and "size mismatch" errors on reads.
+	//
+	// By generating .ecx first, it reflects the .idx state at or before the .dat
+	// is read for EC encoding. If a write sneaks in after .ecx but before/during
+	// EC encoding, the shards contain MORE data than .ecx references, which is
+	// harmless (the extra data is simply not indexed).
+
+	// write .ecx file from the current .idx
+	if err := erasure_coding.WriteSortedFileFromIdx(v.IndexFileName(), ".ecx"); err != nil {
+		return nil, fmt.Errorf("WriteSortedFileFromIdx %s: %v", v.IndexFileName(), err)
+	}
+
+	// snapshot .dat file size before encoding — must match what .ecx references
+	datSize, _, _ := v.FileStat()
+
 	// write .ec00 ~ .ec[TotalShards-1] files using context
 	if err := erasure_coding.WriteEcFilesWithContext(baseFileName, ecCtx); err != nil {
 		return nil, fmt.Errorf("WriteEcFilesWithContext %s: %v", baseFileName, err)
 	}
 
-	// write .ecx file
-	if err := erasure_coding.WriteSortedFileFromIdx(v.IndexFileName(), ".ecx"); err != nil {
-		return nil, fmt.Errorf("WriteSortedFileFromIdx %s: %v", v.IndexFileName(), err)
-	}
-
 	// write .vif files
 	var expireAtSec uint64
 	if v.Ttl != nil {
@@ -106,8 +120,6 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_
 	}
 	volumeInfo := &volume_server_pb.VolumeInfo{Version: uint32(v.Version())}
 	volumeInfo.ExpireAtSec = expireAtSec
-
-	datSize, _, _ := v.FileStat()
 	volumeInfo.DatFileSize = int64(datSize)
 
 	// Validate EC configuration before saving to .vif

weed/storage/erasure_coding/ec_consistency_test.go

@@ -0,0 +1,224 @@
+package erasure_coding
+
+import (
+	"bytes"
+	"crypto/rand"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// TestEcConsistency_WritesBetweenEncodeAndEcx reproduces a race condition that
+// existed in VolumeEcShardsGenerate before the fix in this PR.
+//
+// Previously, the order was:
+//   1. WriteEcFilesWithContext(baseFileName, ecCtx)        — EC shards from .dat
+//   2. WriteSortedFileFromIdx(v.IndexFileName(), ".ecx")   — .ecx from .idx
+//
+// If a write appended data to .dat/.idx between steps 1 and 2, the .ecx would
+// have entries pointing to data that doesn't exist in the EC shards.
+//
+// The fix reverses the order (write .ecx first, then generate EC shards), so
+// that .ecx is always a subset of what the EC shards contain.
+//
+// This test simulates the old buggy sequence to validate that the problem is real.
+func TestEcConsistency_WritesBetweenEncodeAndEcx(t *testing.T) {
+	dir := t.TempDir()
+	baseFileName := dir + "/consistency"
+
+	ctx := NewDefaultECContext("", 0)
+
+	// Phase 1: Create initial .dat and .idx with known data
+	datSize := int64(largeBlockSize*DataShardsCount + smallBlockSize*DataShardsCount*3) // 1 large row + 3 small rows
+	originalData := make([]byte, datSize)
+	rand.Read(originalData)
+
+	err := os.WriteFile(baseFileName+".dat", originalData, 0644)
+	require.NoError(t, err)
+
+	// Create a minimal .idx with one entry pointing to the data
+	createTestIdx(t, baseFileName+".idx", []idxEntry{
+		{id: 1, offset: 0, size: types.Size(datSize)},
+	})
+
+	// Phase 2: EC encode — generates .ec00-.ec13 from current .dat
+	err = generateEcFiles(baseFileName, int(smallBlockSize), largeBlockSize, smallBlockSize, ctx)
+	require.NoError(t, err, "EC encoding")
+
+	// Phase 3: SIMULATE a write between EC encoding and .ecx generation
+	// (reproducing the old buggy order where .ecx was generated after EC shards)
+	extraData := make([]byte, 5000)
+	rand.Read(extraData)
+
+	f, err := os.OpenFile(baseFileName+".dat", os.O_WRONLY|os.O_APPEND, 0644)
+	require.NoError(t, err)
+	_, err = f.Write(extraData)
+	require.NoError(t, err)
+	f.Close()
+
+	// Update .idx with the new entry
+	createTestIdx(t, baseFileName+".idx", []idxEntry{
+		{id: 1, offset: 0, size: types.Size(datSize)},
+		{id: 2, offset: datSize, size: types.Size(len(extraData))},
+	})
+
+	// Phase 4: Generate .ecx from the UPDATED .idx (as the old buggy code did)
+	err = WriteSortedFileFromIdx(baseFileName, ".ecx")
+	require.NoError(t, err, "WriteSortedFileFromIdx")
+
+	// Phase 5: Now try to read needle 2 via EC shards — it should fail
+	// because the EC shards were generated from the OLD .dat (without the extra data)
+	ecFiles, err := openEcFiles(baseFileName, true, ctx)
+	require.NoError(t, err)
+	defer closeEcFiles(ecFiles)
+
+	ecStat, err := ecFiles[0].Stat()
+	require.NoError(t, err)
+	shardSize := ecStat.Size()
+
+	// Read needle 2 (the one added after EC encoding) using LocateData.
+	// Use shardSize-1 to simulate the ecdFileSize fallback path used by
+	// LocateEcShardNeedleInterval when datFileSize is unavailable.
+	actualSize := needle.GetActualSize(types.Size(len(extraData)), needle.Version3)
+	intervals := LocateData(largeBlockSize, smallBlockSize, shardSize-1, datSize, types.Size(actualSize))
+
+	t.Logf("Trying to read needle 2 at offset %d size %d from EC shards (shardSize=%d)", datSize, actualSize, shardSize)
+	t.Logf("Intervals: %+v", intervals)
+
+	// Try to read — this will either fail with an error (offset out of bounds)
+	// or return garbage data (the padded zeros from EC encoding)
+	ecData, readErr := assembleFromIntervalsAllowError(ecFiles, intervals, largeBlockSize, smallBlockSize)
+
+	if readErr != nil {
+		t.Logf("CONFIRMED: Read error for needle written after EC encoding: %v", readErr)
+	} else {
+		// If we got data, it should be zeros (padding) or garbage, not the actual extraData
+		isAllZeros := true
+		for _, b := range ecData {
+			if b != 0 {
+				isAllZeros = false
+				break
+			}
+		}
+		if isAllZeros {
+			t.Logf("CONFIRMED: Read returned zero-padded data (EC shards don't have the needle)")
+		} else if !bytes.Equal(ecData[:len(extraData)], extraData) {
+			t.Logf("CONFIRMED: Read returned wrong data (EC shards don't have the needle)")
+		} else {
+			t.Error("UNEXPECTED: Read returned correct data — needle should NOT be in EC shards")
+		}
+	}
+
+	// Phase 6: Verify a small read from the original data still works.
+	// Use the correct shardDatSize (from the original datSize, not the modified one)
+	// to avoid the fallback heuristic issues.
+	shardDatSize := datSize / int64(DataShardsCount)
+	readSize := types.Size(smallBlockSize)
+	intervals1 := LocateData(largeBlockSize, smallBlockSize, shardDatSize, 0, readSize)
+	ecData1, err := assembleFromIntervalsAllowError(ecFiles, intervals1, largeBlockSize, smallBlockSize)
+	require.NoError(t, err, "reading original data from EC shards")
+
+	assert.True(t, bytes.Equal(originalData[:readSize], ecData1),
+		"Original data at offset 0 should match EC shard data")
+	t.Logf("Original data reads correctly from EC shards")
+}
+
+// TestEcConsistency_ExactLargeRowEncoding verifies that generateEcFiles correctly
+// encodes a .dat file whose size is exactly one large row (DataShardsCount *
+// largeBlockSize), producing shards of exactly largeBlockSize each, and that
+// every chunk of the encoded data can be read back correctly via LocateData.
+func TestEcConsistency_ExactLargeRowEncoding(t *testing.T) {
+	dir := t.TempDir()
+	baseFileName := dir + "/exact"
+	ctx := NewDefaultECContext("", 0)
+
+	datSize := int64(largeBlockSize * DataShardsCount) // exactly 1 large row
+	data := make([]byte, datSize)
+	rand.Read(data)
+	err := os.WriteFile(baseFileName+".dat", data, 0644)
+	require.NoError(t, err)
+
+	// EC encode
+	err = generateEcFiles(baseFileName, int(smallBlockSize), largeBlockSize, smallBlockSize, ctx)
+	require.NoError(t, err)
+
+	// Check shard sizes — each shard should be exactly largeBlockSize
+	ecFiles, err := openEcFiles(baseFileName, true, ctx)
+	require.NoError(t, err)
+	defer closeEcFiles(ecFiles)
+
+	for i := 0; i < ctx.DataShards; i++ {
+		stat, err := ecFiles[i].Stat()
+		require.NoError(t, err, "stat shard %d", i)
+		assert.Equal(t, int64(largeBlockSize), stat.Size(),
+			"data shard %d should be exactly largeBlockSize", i)
+	}
+
+	// Verify data reads correctly at every smallBlockSize offset via LocateData
+	shardDatSize := datSize / int64(ctx.DataShards)
+	readSize := types.Size(smallBlockSize)
+	for offset := int64(0); offset+int64(readSize) <= datSize; offset += int64(readSize) {
+		intervals := LocateData(largeBlockSize, smallBlockSize, shardDatSize, offset, readSize)
+		ecData, err := assembleFromIntervalsAllowError(ecFiles, intervals, largeBlockSize, smallBlockSize)
+		require.NoError(t, err, "reading at offset %d", offset)
+		expected := data[offset : offset+int64(readSize)]
+		assert.True(t, bytes.Equal(expected, ecData),
+			"data mismatch at offset %d", offset)
+	}
+}
+
+type idxEntry struct {
+	id     types.NeedleId
+	offset int64
+	size   types.Size
+}
+
+func createTestIdx(t *testing.T, filename string, entries []idxEntry) {
+	t.Helper()
+	f, err := os.Create(filename)
+	require.NoError(t, err)
+	defer f.Close()
+
+	buf := make([]byte, types.NeedleMapEntrySize)
+	for _, e := range entries {
+		types.NeedleIdToBytes(buf[:types.NeedleIdSize], e.id)
+		types.OffsetToBytes(buf[types.NeedleIdSize:types.NeedleIdSize+types.OffsetSize], types.ToOffset(e.offset))
+		types.SizeToBytes(buf[types.NeedleIdSize+types.OffsetSize:], e.size)
+		_, err := f.Write(buf)
+		require.NoError(t, err)
+	}
+}
+
+func assembleFromIntervalsAllowError(ecFiles []*os.File, intervals []Interval, large, small int64) ([]byte, error) {
+	var data []byte
+	for _, interval := range intervals {
+		shardId, shardOffset := interval.ToShardIdAndOffset(large, small)
+		if int(shardId) >= len(ecFiles) {
+			return nil, fmt.Errorf("shard %d out of range (have %d files)", shardId, len(ecFiles))
+		}
+		stat, err := ecFiles[shardId].Stat()
+		if err != nil {
+			return nil, fmt.Errorf("stat shard %d: %v", shardId, err)
+		}
+		if shardOffset+int64(interval.Size) > stat.Size() {
+			return nil, fmt.Errorf("read past end of shard %d: offset %d + size %d > fileSize %d",
+				shardId, shardOffset, interval.Size, stat.Size())
+		}
+		chunk := make([]byte, interval.Size)
+		n, err := ecFiles[shardId].ReadAt(chunk, shardOffset)
+		if err != nil {
+			return nil, fmt.Errorf("read shard %d offset %d: %v", shardId, shardOffset, err)
+		}
+		if n != int(interval.Size) {
+			return nil, fmt.Errorf("short read shard %d: got %d want %d", shardId, n, interval.Size)
+		}
+		data = append(data, chunk...)
+	}
+	return data, nil
+}

weed/storage/store_ec.go

@@ -167,7 +167,7 @@ func (s *Store) ReadEcShardNeedle(vid needle.VolumeId, n *needle.Needle, onReadS
 				return 0, ErrorDeleted
 			}
 
-			err = n.ReadBytes(bytes, 0, size, localEcVolume.Version)
+			err = n.ReadBytes(bytes, offset.ToActualOffset(), size, localEcVolume.Version)
 			if err != nil {
 				return 0, fmt.Errorf("ec volume %d needle %s offset %d size %d: %w", vid, n.String(), offset.ToActualOffset(), size, err)
 			}

weed/worker/tasks/erasure_coding/ec_task.go

@@ -15,7 +15,9 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/idx"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	storagetypes "github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types"
 	"github.com/seaweedfs/seaweedfs/weed/worker/types/base"
@@ -150,10 +152,16 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	}
 
 	// Step 2: Copy volume files to worker
+	// The .idx and .dat are copied as separate network transfers, with .idx
+	// copied first. If a write lands on the source after the .idx copy, the
+	// .dat will include extra data not referenced by .idx (harmless).
+	// verifyDatIdxConsistency() in generateEcShardsLocally catches the reverse
+	// case where .idx references data past .dat.
 	t.ReportProgressWithStage(25.0, "Copying volume files to worker")
 	t.GetLogger().Info("Copying volume files to worker")
 	localFiles, err := t.copyVolumeFilesToWorker(ctx, taskWorkDir)
 	if err != nil {
+		t.rollbackReadonly(ctx)
 		return fmt.Errorf("failed to copy volume files: %v", err)
 	}
 
@@ -162,6 +170,7 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP
 	t.GetLogger().Info("Generating EC shards locally")
 	shardFiles, err := t.generateEcShardsLocally(localFiles, taskWorkDir)
 	if err != nil {
+		t.rollbackReadonly(ctx)
 		return fmt.Errorf("failed to generate EC shards: %v", err)
 	}
 
@@ -259,7 +268,35 @@ func (t *ErasureCodingTask) markVolumeReadonly(ctx context.Context) error {
 		})
 }
 
-// copyVolumeFilesToWorker copies .dat and .idx files from source server to local worker
+// rollbackReadonly is a best-effort rollback of markVolumeReadonly, used when the
+// EC task fails before any shards are distributed. Logs but does not return errors.
+// Uses a fresh context with timeout since the caller's ctx may already be cancelled.
+func (t *ErasureCodingTask) rollbackReadonly(_ context.Context) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := t.markVolumeWritable(ctx); err != nil {
+		glog.Warningf("failed to restore volume %d to writable after EC task failure: %v", t.volumeID, err)
+	} else {
+		glog.V(0).Infof("restored volume %d to writable after EC task failure", t.volumeID)
+	}
+}
+
+// markVolumeWritable restores the volume to writable on the source server.
+func (t *ErasureCodingTask) markVolumeWritable(ctx context.Context) error {
+	return operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
+		func(client volume_server_pb.VolumeServerClient) error {
+			_, err := client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{
+				VolumeId: t.volumeID,
+			})
+			return err
+		})
+}
+
+// copyVolumeFilesToWorker copies .idx and .dat files from source server to local worker.
+// The .idx is copied first, then .dat. Both copies are capped to the sizes reported by
+// ReadVolumeFileStatus. If a write lands after .idx is copied, .dat may include extra
+// data not referenced by .idx (harmless). The reverse (idx referencing data past .dat)
+// is caught by verifyDatIdxConsistency in generateEcShardsLocally.
 func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir string) (map[string]string, error) {
 	localFiles := make(map[string]string)
 
@@ -277,31 +314,14 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir
 		"idx_file_size_bytes": fileStatus.GetIdxFileSize(),
 	}).Info("Starting volume file copy from source server")
 
-	// Copy .dat file
-	datFile := filepath.Join(workDir, fmt.Sprintf("%d.dat", t.volumeID))
-	if err := t.copyFileFromSource(ctx, ".dat", datFile, fileStatus.GetCompactionRevision(), fileStatus.GetDatFileSize()); err != nil {
-		return nil, fmt.Errorf("failed to copy .dat file: %v", err)
-	}
-	localFiles["dat"] = datFile
-
-	// Log .dat file size
-	if info, err := os.Stat(datFile); err == nil {
-		t.GetLogger().WithFields(map[string]interface{}{
-			"file_type":  ".dat",
-			"file_path":  datFile,
-			"size_bytes": info.Size(),
-			"size_mb":    float64(info.Size()) / (1024 * 1024),
-		}).Info("Volume data file copied successfully")
-	}
-
-	// Copy .idx file
+	// Copy .idx file FIRST — if a write lands on the source after this copy,
+	// the .dat copy will include the new data but .idx won't reference it.
 	idxFile := filepath.Join(workDir, fmt.Sprintf("%d.idx", t.volumeID))
 	if err := t.copyFileFromSource(ctx, ".idx", idxFile, fileStatus.GetCompactionRevision(), fileStatus.GetIdxFileSize()); err != nil {
 		return nil, fmt.Errorf("failed to copy .idx file: %v", err)
 	}
 	localFiles["idx"] = idxFile
 
-	// Log .idx file size
 	if info, err := os.Stat(idxFile); err == nil {
 		t.GetLogger().WithFields(map[string]interface{}{
 			"file_type":  ".idx",
@@ -311,6 +331,22 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir
 		}).Info("Volume index file copied successfully")
 	}
 
+	// Copy .dat file SECOND — guaranteed to have at least as much data as .idx references.
+	datFile := filepath.Join(workDir, fmt.Sprintf("%d.dat", t.volumeID))
+	if err := t.copyFileFromSource(ctx, ".dat", datFile, fileStatus.GetCompactionRevision(), fileStatus.GetDatFileSize()); err != nil {
+		return nil, fmt.Errorf("failed to copy .dat file: %v", err)
+	}
+	localFiles["dat"] = datFile
+
+	if info, err := os.Stat(datFile); err == nil {
+		t.GetLogger().WithFields(map[string]interface{}{
+			"file_type":  ".dat",
+			"file_path":  datFile,
+			"size_bytes": info.Size(),
+			"size_mb":    float64(info.Size()) / (1024 * 1024),
+		}).Info("Volume data file copied successfully")
+	}
+
 	return localFiles, nil
 }
 
@@ -401,16 +437,23 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string
 
 	glog.V(1).Infof("Generating EC shards from local files: dat=%s, idx=%s", datFile, idxFile)
 
+	// Verify .dat and .idx are consistent before EC encoding.
+	// Since they were copied as separate network transfers, the .idx may have
+	// entries pointing past the end of .dat if a write landed between the copies.
+	if err := verifyDatIdxConsistency(datFile, idxFile); err != nil {
+		return nil, fmt.Errorf("dat/idx consistency check failed: %v", err)
+	}
+
+	// Generate .ecx file from .idx BEFORE EC shards to prevent inconsistency.
+	if err := erasure_coding.WriteSortedFileFromIdx(baseName, ".ecx"); err != nil {
+		return nil, fmt.Errorf("failed to generate .ecx file: %v", err)
+	}
+
 	// Generate EC shard files (.ec00 ~ .ec13)
 	if err := erasure_coding.WriteEcFiles(baseName); err != nil {
 		return nil, fmt.Errorf("failed to generate EC shard files: %v", err)
 	}
 
-	// Generate .ecx file from .idx (use baseName, not full idx path)
-	if err := erasure_coding.WriteSortedFileFromIdx(baseName, ".ecx"); err != nil {
-		return nil, fmt.Errorf("failed to generate .ecx file: %v", err)
-	}
-
 	// Collect generated shard file paths and log details
 	var generatedShards []string
 	var totalShardSize int64
@@ -588,3 +631,62 @@ func (t *ErasureCodingTask) getReplicas() []string {
 	}
 	return replicas
 }
+
+// verifyDatIdxConsistency checks that all .idx entries reference data within the
+// .dat file. Since .dat and .idx are copied as separate network transfers, the
+// .idx may have entries from writes that landed after the .dat was copied.
+func verifyDatIdxConsistency(datFile, idxFile string) error {
+	datInfo, err := os.Stat(datFile)
+	if err != nil {
+		return fmt.Errorf("stat dat file: %v", err)
+	}
+	datSize := datInfo.Size()
+
+	// Read volume version from superblock to compute actual needle sizes
+	df, err := os.Open(datFile)
+	if err != nil {
+		return fmt.Errorf("open dat file: %v", err)
+	}
+	defer df.Close()
+
+	versionBytes := make([]byte, 1)
+	if _, err := df.ReadAt(versionBytes, 0); err != nil {
+		return fmt.Errorf("read version byte: %v", err)
+	}
+	version := needle.Version(versionBytes[0])
+
+	idxF, err := os.Open(idxFile)
+	if err != nil {
+		return fmt.Errorf("open idx file: %v", err)
+	}
+	defer idxF.Close()
+
+	var maxEnd int64
+	var maxEndNeedleId storagetypes.NeedleId
+	var entryCount int64
+	err = idx.WalkIndexFile(idxF, 0, func(key storagetypes.NeedleId, offset storagetypes.Offset, size storagetypes.Size) error {
+		entryCount++
+		if size.IsDeleted() {
+			return nil
+		}
+		end := offset.ToActualOffset() + needle.GetActualSize(size, version)
+		if end > maxEnd {
+			maxEnd = end
+			maxEndNeedleId = key
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("walk idx file: %v", err)
+	}
+
+	if maxEnd > datSize {
+		return fmt.Errorf(
+			"idx references data beyond dat file: needle %d ends at offset %d but dat file is only %d bytes (%d entries total)",
+			maxEndNeedleId, maxEnd, datSize, entryCount,
+		)
+	}
+
+	glog.V(1).Infof("dat/idx consistency check passed: %d entries, max offset %d, dat size %d", entryCount, maxEnd, datSize)
+	return nil
+}