worker/ec: retry the pinned volume copy after a mid-copy compaction (#11929)

* worker/ec: retry the pinned volume copy after a mid-copy compaction The EC task pins .idx/.dat copies to the CompactionRevision read from ReadVolumeFileStatus so the pair comes from one volume generation. When vacuum compaction lands between that read and either CopyFile, the server rejects the copy with "volume N is compacted" and the whole task fails — on clusters with queued maintenance this makes most EC tasks die to a race they cannot avoid. On a compaction mismatch the copy now restarts under a freshly read revision, up to a bounded number of attempts; other errors still fail immediately. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * worker/ec: read the retried copy's expiry from its own file status A mid-copy compaction makes the source reload the volume, and the fresh ReadVolumeFileStatus can carry a different expireAtSec than the pre-compaction read. Keeping the attempt-0 expiry could let the EC volume outlive its source TTL. * worker/ec: cover the compaction retry with a stub volume server * worker/ec: also observe revised .idx bytes in the compaction retry test * worker/ec: read the fake source under its mutex in the retry test --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-23 22:40:39 +02:00 · 2026-10-22 21:32:57 -07:00
1 parent f2d5c27f65
commit 7154732b84
@@ -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) {
@@ -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)
+}