diff --git a/test/plugin_workers/volume_fixtures.go b/test/plugin_workers/volume_fixtures.go index 64b5cfcfa..fe4529778 100644 --- a/test/plugin_workers/volume_fixtures.go +++ b/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) diff --git a/test/volume_server/grpc/erasure_coding_test.go b/test/volume_server/grpc/erasure_coding_test.go index f5852c6f3..45b1b1458 100644 --- a/test/volume_server/grpc/erasure_coding_test.go +++ b/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") diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index dbca04587..b3f10efd3 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/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 diff --git a/weed/storage/erasure_coding/ec_consistency_test.go b/weed/storage/erasure_coding/ec_consistency_test.go new file mode 100644 index 000000000..84f4d0638 --- /dev/null +++ b/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 +} diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index accc7e6a7..0c1cc91b5 100644 --- a/weed/storage/store_ec.go +++ b/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) } diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index bf0c84731..7bdc4f741 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/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 +}