diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index f2838ffc81..b9df6e3ed3 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -455,39 +455,60 @@ func (t *ErasureCodingTask) rollbackReadonly(_ context.Context) { } } +// ecCopyAttempts bounds how often the pinned file pair copy restarts after a +// concurrent compaction bumps the source's CompactionRevision mid-copy. +const ecCopyAttempts = 3 + // 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. +// ReadVolumeFileStatus and pinned to its CompactionRevision so they come from the same +// volume generation. A concurrent vacuum bumps that revision mid-copy, so a compaction +// failure restarts the pair under a freshly read revision. func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir string) (map[string]string, error) { localFiles := make(map[string]string) - fileStatus, err := t.readSourceVolumeFileStatus(ctx) - if err != nil { - return nil, fmt.Errorf("failed to read source volume file status: %w", err) - } - t.expireAtSec = fileStatus.GetExpireAtSec() - if t.expireAtSec == 0 { - // Volume servers that predate the top-level field still carry the - // expiry in the nested .vif once SaveVolumeInfo has stamped it. - t.expireAtSec = fileStatus.GetVolumeInfo().GetExpireAtSec() - } + var copyErr error + for attempt := 0; attempt < ecCopyAttempts; attempt++ { + fileStatus, err := t.readSourceVolumeFileStatus(ctx) + if err != nil { + return nil, fmt.Errorf("failed to read source volume file status: %w", err) + } + t.expireAtSec = fileStatus.GetExpireAtSec() + if t.expireAtSec == 0 { + // Volume servers that predate the top-level field still carry the + // expiry in the nested .vif once SaveVolumeInfo has stamped it. + t.expireAtSec = fileStatus.GetVolumeInfo().GetExpireAtSec() + } - t.GetLogger().WithFields(map[string]interface{}{ - "volume_id": t.volumeID, - "source": t.server, - "working_dir": workDir, - "compaction_revision": fileStatus.GetCompactionRevision(), - "dat_file_size_bytes": fileStatus.GetDatFileSize(), - "idx_file_size_bytes": fileStatus.GetIdxFileSize(), - }).Info("Starting volume file copy from source server") + t.GetLogger().WithFields(map[string]interface{}{ + "volume_id": t.volumeID, + "source": t.server, + "working_dir": workDir, + "compaction_revision": fileStatus.GetCompactionRevision(), + "dat_file_size_bytes": fileStatus.GetDatFileSize(), + "idx_file_size_bytes": fileStatus.GetIdxFileSize(), + }).Info("Starting volume file copy from source server") - // 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. + if copyErr = t.copyVolumeFilePair(ctx, workDir, fileStatus, localFiles); copyErr == nil { + return localFiles, nil + } + if !isVolumeCompactedError(copyErr) || ctx.Err() != nil { + return nil, copyErr + } + glog.Warningf("volume %d on %s compacted during EC file copy, restarting with fresh file status: %v", t.volumeID, t.server, copyErr) + } + return nil, copyErr +} + +// copyVolumeFilePair copies .idx first, then .dat, pinned to the compaction +// revision in fileStatus. 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) copyVolumeFilePair(ctx context.Context, workDir string, fileStatus *volume_server_pb.ReadVolumeFileStatusResponse, localFiles map[string]string) error { 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: %w", err) + return fmt.Errorf("failed to copy .idx file: %w", err) } localFiles["idx"] = idxFile @@ -503,7 +524,7 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir // 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: %w", err) + return fmt.Errorf("failed to copy .dat file: %w", err) } localFiles["dat"] = datFile @@ -516,7 +537,13 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir }).Info("Volume data file copied successfully") } - return localFiles, nil + return nil +} + +// isVolumeCompactedError matches the failure CopyFile returns when the source +// volume's CompactionRevision moved past the pinned one mid-copy. +func isVolumeCompactedError(err error) bool { + return err != nil && strings.Contains(err.Error(), "is compacted") } func (t *ErasureCodingTask) readSourceVolumeFileStatus(ctx context.Context) (*volume_server_pb.ReadVolumeFileStatusResponse, error) { diff --git a/weed/worker/tasks/erasure_coding/ec_task_copy_retry_test.go b/weed/worker/tasks/erasure_coding/ec_task_copy_retry_test.go new file mode 100644 index 0000000000..76294f6b14 --- /dev/null +++ b/weed/worker/tasks/erasure_coding/ec_task_copy_retry_test.go @@ -0,0 +1,166 @@ +package erasure_coding + +import ( + "context" + "fmt" + "net" + "os" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// fakeCopySource is a stub VolumeServer. Rejecting a pinned .dat copy advances +// the compaction revision and rewrites the volume contents, so a mid-copy +// compaction is replayed deterministically. +type fakeCopySource struct { + volume_server_pb.UnimplementedVolumeServerServer + + mu sync.Mutex + revision uint32 + datBytes []byte + idxBytes []byte + expireAtSec uint64 + rejectDat int + failErr error + statusCalls int + copyCalls int +} + +func (f *fakeCopySource) snapshot() (revision uint32, expireAtSec uint64, statusCalls, copyCalls int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.revision, f.expireAtSec, f.statusCalls, f.copyCalls +} + +func (f *fakeCopySource) ReadVolumeFileStatus(context.Context, *volume_server_pb.ReadVolumeFileStatusRequest) (*volume_server_pb.ReadVolumeFileStatusResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.statusCalls++ + return &volume_server_pb.ReadVolumeFileStatusResponse{ + CompactionRevision: f.revision, + DatFileSize: uint64(len(f.datBytes)), + IdxFileSize: uint64(len(f.idxBytes)), + ExpireAtSec: f.expireAtSec, + }, nil +} + +func (f *fakeCopySource) CopyFile(req *volume_server_pb.CopyFileRequest, stream grpc.ServerStreamingServer[volume_server_pb.CopyFileResponse]) error { + f.mu.Lock() + f.copyCalls++ + reject := req.GetExt() == ".dat" && f.rejectDat > 0 + if reject { + f.rejectDat-- + f.revision++ + f.datBytes = append(f.datBytes, 0xEC) + f.idxBytes = append(f.idxBytes, 0xEC) + f.expireAtSec += 100 + } + failErr := f.failErr + revision := f.revision + var content []byte + switch req.GetExt() { + case ".idx": + content = f.idxBytes + case ".dat": + content = f.datBytes + } + f.mu.Unlock() + + if failErr != nil { + return failErr + } + if reject || req.GetCompactionRevision() != revision { + return fmt.Errorf("volume %d is compacted", req.GetVolumeId()) + } + content = content[:req.GetStopOffset()] + for len(content) > 0 { + n := min(len(content), 4) + if err := stream.Send(&volume_server_pb.CopyFileResponse{FileContent: content[:n]}); err != nil { + return err + } + content = content[n:] + } + return nil +} + +func startFakeCopySource(t *testing.T, src *fakeCopySource) string { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + grpcServer := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(grpcServer, src) + go grpcServer.Serve(lis) + t.Cleanup(grpcServer.Stop) + // server addresses carry httpPort.grpcPort; the gRPC dialer uses the part + // after the dot, so report the listening port that way + _, port, err := net.SplitHostPort(lis.Addr().String()) + require.NoError(t, err) + return "127.0.0.1:0." + port +} + +func newCopyRetryTask(addr string) *ErasureCodingTask { + return NewErasureCodingTask("copy-retry", addr, 951, "", grpc.WithTransportCredentials(insecure.NewCredentials())) +} + +func TestCopyVolumeFilesToWorkerRetriesAfterMidCopyCompaction(t *testing.T) { + src := &fakeCopySource{ + revision: 7, + idxBytes: []byte("idx-data-v7"), + datBytes: []byte("dat-payload-v7"), + expireAtSec: 1000, + rejectDat: 1, + } + task := newCopyRetryTask(startFakeCopySource(t, src)) + + localFiles, err := task.copyVolumeFilesToWorker(context.Background(), t.TempDir()) + require.NoError(t, err) + + datContent, err := os.ReadFile(localFiles["dat"]) + require.NoError(t, err) + require.Equal(t, []byte("dat-payload-v7\xEC"), datContent) + idxContent, err := os.ReadFile(localFiles["idx"]) + require.NoError(t, err) + require.Equal(t, []byte("idx-data-v7\xEC"), idxContent) + revision, expireAtSec, statusCalls, copyCalls := src.snapshot() + require.Equal(t, uint32(8), revision) + require.Equal(t, 2, statusCalls) + require.Equal(t, 4, copyCalls) + require.Equal(t, expireAtSec, task.expireAtSec) +} + +func TestCopyVolumeFilesToWorkerStopsAfterRetryLimit(t *testing.T) { + src := &fakeCopySource{ + revision: 1, + idxBytes: []byte("i"), + datBytes: []byte("d"), + rejectDat: 10, + } + task := newCopyRetryTask(startFakeCopySource(t, src)) + + _, err := task.copyVolumeFilesToWorker(context.Background(), t.TempDir()) + require.ErrorContains(t, err, "is compacted") + _, _, statusCalls, copyCalls := src.snapshot() + require.Equal(t, ecCopyAttempts, statusCalls) + require.Equal(t, 2*ecCopyAttempts, copyCalls) +} + +func TestCopyVolumeFilesToWorkerReturnsNonCompactionErrors(t *testing.T) { + src := &fakeCopySource{ + revision: 1, + idxBytes: []byte("i"), + datBytes: []byte("d"), + failErr: fmt.Errorf("boom"), + } + task := newCopyRetryTask(startFakeCopySource(t, src)) + + _, err := task.copyVolumeFilesToWorker(context.Background(), t.TempDir()) + require.ErrorContains(t, err, "boom") + _, _, statusCalls, copyCalls := src.snapshot() + require.Equal(t, 1, statusCalls) + require.Equal(t, 1, copyCalls) +}