volume_move: keep the live-move source frozen across volume reloads (#12597)

* volume_move: persist the move's source freeze across reloads

A live move froze the source in memory only, so an unmount/mount or a
volume server restart mid-move reopened it for writes that the move then
deletes with the source. Persist the freeze when the move froze a
writable source, and reopen the target right after the copy since the
copied .vif carries the freeze. An already-readonly source keeps an
in-memory mark, and a failed move still restores the source writable.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: live move keeps the source frozen across a reload, on Go and Rust volume servers

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: require 409 from the reloaded frozen source

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume_move: reopen an interrupted move's source only without a target copy

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

test/volume_server/grpc/move_persist_freeze_test.go

@@ -0,0 +1,137 @@
+package volume_server_grpc_test
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
+	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/operation/volume_move"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func moverServerAddress(cluster framework.MultiCluster, index int) pb.ServerAddress {
+	return pb.ServerAddress(cluster.VolumeAdminAddress(index) + "." + strings.Split(cluster.VolumeGRPCAddress(index), ":")[1])
+}
+
+func reloadVolume(t *testing.T, client volume_server_pb.VolumeServerClient, volumeID uint32) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if _, err := client.VolumeUnmount(ctx, &volume_server_pb.VolumeUnmountRequest{VolumeId: volumeID}); err != nil {
+		t.Fatalf("unmount volume %d: %v", volumeID, err)
+	}
+	if _, err := client.VolumeMount(ctx, &volume_server_pb.VolumeMountRequest{VolumeId: volumeID}); err != nil {
+		t.Fatalf("mount volume %d: %v", volumeID, err)
+	}
+}
+
+func uploadStatus(t *testing.T, httpClient *http.Client, volumeURL string, volumeID uint32, key uint64, data string) int {
+	t.Helper()
+	resp := framework.UploadBytes(t, httpClient, volumeURL, framework.NewFileID(volumeID, key, 0x5EED0000+uint32(key)), []byte(data))
+	framework.ReadAllAndClose(t, resp)
+	return resp.StatusCode
+}
+
+func TestLiveMoveFreezeSurvivesSourceReload(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	cluster := framework.StartMultiVolumeClusterAuto(t, matrix.P1(), 2)
+	sourceConn, sourceClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
+	defer sourceConn.Close()
+	targetConn, targetClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1))
+	defer targetConn.Close()
+
+	const volumeID = uint32(977)
+	framework.AllocateVolume(t, sourceClient, volumeID, "")
+	httpClient := framework.NewHTTPClient()
+	if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 1, "before-move"); code != http.StatusCreated {
+		t.Fatalf("initial upload: status %d", code)
+	}
+
+	var reloadedSourceWrite, targetWrite int
+	mover := volume_move.NewMover(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	err := mover.LiveMoveVolume(ctx, needle.VolumeId(volumeID), moverServerAddress(cluster, 0), moverServerAddress(cluster, 1), volume_move.VolumeMoveOptions{
+		IdleTimeout: time.Second,
+		Progress: func(percent float64, stage string) {
+			switch percent {
+			case 20:
+				// frozen, not yet copied: a reload must not reopen the source
+				reloadVolume(t, sourceClient, volumeID)
+				reloadedSourceWrite = uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 2, "lost-after-move")
+			case 70:
+				// copied, not yet tailed: the copy must not inherit the freeze
+				targetWrite = uploadStatus(t, httpClient, cluster.VolumeAdminURL(1), volumeID, 3, "on-target")
+			}
+		},
+	})
+	if err != nil {
+		t.Fatalf("LiveMoveVolume: %v", err)
+	}
+	if reloadedSourceWrite != http.StatusConflict {
+		t.Fatalf("reloaded source write: status %d, want %d", reloadedSourceWrite, http.StatusConflict)
+	}
+	if targetWrite != http.StatusCreated {
+		t.Fatalf("copied target refused a write before the tail: status %d", targetWrite)
+	}
+
+	// The writable mark on the target is persisted too.
+	reloadVolume(t, targetClient, volumeID)
+	if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(1), volumeID, 4, "after-reload"); code != http.StatusCreated {
+		t.Fatalf("moved volume refused a write after a reload: status %d", code)
+	}
+	for key, want := range map[uint64]string{1: "before-move", 3: "on-target", 4: "after-reload"} {
+		resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), framework.NewFileID(volumeID, key, 0x5EED0000+uint32(key)))
+		body := framework.ReadAllAndClose(t, resp)
+		if resp.StatusCode != http.StatusOK || string(body) != want {
+			t.Fatalf("read key %d from target: status %d body %q, want %q", key, resp.StatusCode, body, want)
+		}
+	}
+}
+
+func TestLiveMoveAbortClearsPersistedFreeze(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	cluster := framework.StartMultiVolumeClusterAuto(t, matrix.P1(), 2)
+	sourceConn, sourceClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0))
+	defer sourceConn.Close()
+
+	const volumeID = uint32(978)
+	framework.AllocateVolume(t, sourceClient, volumeID, "")
+	httpClient := framework.NewHTTPClient()
+	if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 1, "before-move"); code != http.StatusCreated {
+		t.Fatalf("initial upload: status %d", code)
+	}
+
+	mover := volume_move.NewMover(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	err := mover.LiveMoveVolume(ctx, needle.VolumeId(volumeID), moverServerAddress(cluster, 0), moverServerAddress(cluster, 1), volume_move.VolumeMoveOptions{
+		Progress: func(percent float64, stage string) {
+			if percent == 20 {
+				cancel()
+			}
+		},
+	})
+	if err == nil {
+		t.Fatalf("LiveMoveVolume succeeded with a cancelled context")
+	}
+
+	reloadVolume(t, sourceClient, volumeID)
+	if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 2, "after-abort"); code != http.StatusCreated {
+		t.Fatalf("aborted move left the source readonly after a reload: status %d", code)
+	}
+}

weed/operation/volume_move/fake_client_test.go

@@ -40,9 +40,10 @@ type fakeCluster struct {
 
 	lastAppendAtNs uint64
 
-	copyReqs   []*volume_server_pb.VolumeCopyRequest
-	deleteReqs []*volume_server_pb.VolumeDeleteRequest
-	ecCopyReqs []*volume_server_pb.VolumeEcShardsCopyRequest
+	copyReqs     []*volume_server_pb.VolumeCopyRequest
+	readonlyReqs []*volume_server_pb.VolumeMarkReadonlyRequest
+	deleteReqs   []*volume_server_pb.VolumeDeleteRequest
+	ecCopyReqs   []*volume_server_pb.VolumeEcShardsCopyRequest
 }
 
 func newFakeCluster() *fakeCluster {
@@ -94,6 +95,9 @@ func (f *fakeClient) VolumeMarkReadonly(ctx context.Context, req *volume_server_
 	if err := f.cluster.record(f.addr, "VolumeMarkReadonly"); err != nil {
 		return nil, err
 	}
+	f.cluster.mu.Lock()
+	f.cluster.readonlyReqs = append(f.cluster.readonlyReqs, req)
+	f.cluster.mu.Unlock()
 	return &volume_server_pb.VolumeMarkReadonlyResponse{}, nil
 }
 

weed/operation/volume_move/volume_move.go

@@ -65,7 +65,7 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so
 	}
 
 	opts.Progress(10, fmt.Sprintf("marking volume %d readonly on %s", volumeId, source))
-	sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, true)
+	sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, true, true)
 	// The source stays authoritative until its delete succeeds; on any earlier
 	// failure undo the freeze this move added and clean up the target copy.
 	// Installed before the error check: the marking RPC itself can fail after
@@ -130,8 +130,9 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so
 				m.restoreVolumeWritable(volumeId, source)
 			} else {
 				// The hard freeze added on top of a readonly-reporting status is
-				// kept: it is indistinguishable from an operator's mark. It is
-				// not persisted, so a volume server restart clears it.
+				// kept: it is indistinguishable from an operator's mark. Only a
+				// freeze of a writable volume is persisted, so a volume server
+				// restart clears this one.
 				fmt.Fprintf(opts.Writer, "volume %d on %s keeps the readonly mark the move added; volume.mark -writable clears it if the prior readonly state was transient\n", volumeId, source)
 			}
 		}
@@ -139,6 +140,9 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so
 	if err != nil {
 		return fmt.Errorf("mark volume %d readonly on %s: %v", volumeId, source, err)
 	}
+	if sourceWasWritable {
+		fmt.Fprintf(opts.Writer, "volume %d on %s stays readonly across restarts until the move ends; if the move is interrupted and %s has no copy of it, volume.mark -node %s -volumeId %d -writable reopens it, otherwise delete one copy before reopening the other\n", volumeId, source, target, source, volumeId)
+	}
 
 	targetHadVolume, targetStateKnown = m.probeVolume(ctx, volumeId, target)
 	if !sourceWasWritable {
@@ -168,6 +172,13 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so
 		return fmt.Errorf("copy volume %d from %s to %s: %v", volumeId, source, target, err)
 	}
 	copyCompleted = true
+	if sourceWasWritable {
+		// The copy carries the source's persisted freeze; the target must take
+		// the tailed writes and serve as the moved volume.
+		if err = m.MarkVolumeWritable(ctx, volumeId, target, true, false); err != nil {
+			return fmt.Errorf("mark volume %d writable on %s: %v", volumeId, target, err)
+		}
+	}
 
 	opts.Progress(70, fmt.Sprintf("tailing volume %d from %s to %s", volumeId, source, target))
 	fmt.Fprintf(opts.Writer, "tailing volume %d from %s to %s\n", volumeId, source, target)
@@ -273,7 +284,7 @@ func (m *Mover) CopyVolume(ctx context.Context, volumeId needle.VolumeId, source
 	// is (force=false): missed concurrent deletes only make the new replica
 	// trivially stale, and hard-marking here could pin a transiently readonly
 	// (e.g. low-disk) source readonly with nothing to gate on it.
-	sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, false)
+	sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, false, false)
 	defer func() {
 		if sourceWasWritable && (err != nil || restoreWritable) {
 			m.restoreVolumeWritable(volumeId, source)
@@ -293,7 +304,9 @@ func (m *Mover) CopyVolume(ctx context.Context, volumeId needle.VolumeId, source
 // readonly-but-can-delete flag, and neither blocks needle deletes, so only the
 // hard mark makes the volume immutable — required before deleting the source
 // of a move. Without force, a readonly-reporting volume is left untouched.
-func (m *Mover) ensureVolumeReadonly(ctx context.Context, volumeId needle.VolumeId, server pb.ServerAddress, force bool) (wasWritable bool, err error) {
+// persist keeps the freeze of a writable volume across a volume reload or
+// server restart; a mark over an already-readonly volume is never persisted.
+func (m *Mover) ensureVolumeReadonly(ctx context.Context, volumeId needle.VolumeId, server pb.ServerAddress, force, persist bool) (wasWritable bool, err error) {
 	err = m.withClient(false, server, func(client volume_server_pb.VolumeServerClient) error {
 		resp, statusErr := client.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{
 			VolumeId: uint32(volumeId),
@@ -307,7 +320,7 @@ func (m *Mover) ensureVolumeReadonly(ctx context.Context, volumeId needle.Volume
 		}
 		_, readonlyErr := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{
 			VolumeId: uint32(volumeId),
-			Persist:  false,
+			Persist:  persist && wasWritable,
 		})
 		return readonlyErr
 	})
@@ -331,7 +344,8 @@ func (m *Mover) probeVolume(ctx context.Context, volumeId needle.VolumeId, serve
 	return false, false
 }
 
-// restoreVolumeWritable undoes a readonly mark after a failed move. It runs on
+// restoreVolumeWritable undoes a readonly mark after a failed move, including
+// a persisted one: the volume servers always persist a writable mark. It runs on
 // its own deadline so an abort via cancelled context still restores the source,
 // and only logs a failure — the caller is already returning the move error.
 func (m *Mover) restoreVolumeWritable(volumeId needle.VolumeId, server pb.ServerAddress) {

weed/operation/volume_move/volume_move_test.go

@@ -147,12 +147,19 @@ func TestLiveMoveVolumeSequence(t *testing.T) {
 		"src:8080 VolumeMarkReadonly",
 		"dst:8080 ReadVolumeFileStatus",
 		"dst:8080 VolumeCopy",
+		"dst:8080 VolumeMarkWritable",
 		"dst:8080 VolumeTailReceiver",
 		"src:8080 ReadVolumeFileStatus",
 		"dst:8080 ReadVolumeFileStatus",
 		"src:8080 VolumeDelete",
 	})
 
+	// The freeze survives a reload of the source; the copy inherits it, so the
+	// target is reopened before the tail writes to it.
+	if !cluster.readonlyReqs[0].Persist {
+		t.Errorf("freeze of a writable source not persisted: %+v", cluster.readonlyReqs[0])
+	}
+
 	copyReq := cluster.copyReqs[0]
 	if copyReq.DiskType != "ssd" || copyReq.IoBytePerSecond != 42 || copyReq.SourceDataNode != string(srcAddr) {
 		t.Errorf("copy request not propagated: %+v", copyReq)
@@ -631,3 +638,59 @@ func TestCopyVolumeRestoreWritable(t *testing.T) {
 		}
 	}
 }
+
+func TestLiveMoveVolumeReadonlySourceFreezeNotPersisted(t *testing.T) {
+	// A readonly-reporting source may be only transiently so (e.g. low disk);
+	// the hard mark over it must not outlive a restart, and the copy keeps
+	// whatever state the source had.
+	cluster := newFakeCluster()
+	cluster.readonly[string(srcAddr)] = true
+	cluster.status[string(srcAddr)] = volumeStatus(1000, 100, 10)
+	cluster.status[string(dstAddr)] = volumeStatus(1000, 100, 10)
+	cluster.statusFailures[string(dstAddr)] = 1
+
+	if err := cluster.mover().LiveMoveVolume(context.Background(), 7, srcAddr, dstAddr, VolumeMoveOptions{}); err != nil {
+		t.Fatalf("LiveMoveVolume: %v", err)
+	}
+	if cluster.readonlyReqs[0].Persist {
+		t.Errorf("freeze of an already-readonly source persisted: %+v", cluster.readonlyReqs[0])
+	}
+	for _, call := range cluster.callList() {
+		if call == "dst:8080 VolumeMarkWritable" {
+			t.Fatalf("copy of a readonly source made writable: %v", cluster.callList())
+		}
+	}
+}
+
+func TestLiveMoveVolumeTargetReopenFailureRestores(t *testing.T) {
+	// A target left with the copied freeze would refuse the tailed writes;
+	// abort, drop the copy and reopen the source.
+	cluster := newFakeCluster()
+	cluster.status[string(srcAddr)] = volumeStatus(1000, 100, 10)
+	cluster.errs["dst:8080 VolumeMarkWritable"] = errors.New("target unreachable")
+
+	err := cluster.mover().LiveMoveVolume(context.Background(), 7, srcAddr, dstAddr, VolumeMoveOptions{})
+	if err == nil || !strings.Contains(err.Error(), "mark volume 7 writable on dst:8080") {
+		t.Fatalf("expected target reopen error, got: %v", err)
+	}
+	calls := cluster.callList()
+	for _, call := range calls {
+		if call == "dst:8080 VolumeTailReceiver" {
+			t.Fatalf("tailed into a target that is still frozen: %v", calls)
+		}
+	}
+	if fmt.Sprint(calls[len(calls)-2:]) != fmt.Sprint([]string{"dst:8080 VolumeDelete", "src:8080 VolumeMarkWritable"}) {
+		t.Fatalf("expected target cleanup then source restore, got: %v", calls)
+	}
+}
+
+func TestCopyVolumeFreezeNotPersisted(t *testing.T) {
+	// A copy leaves the source serving, so its freeze must not outlive a crash.
+	cluster := newFakeCluster()
+	if _, err := cluster.mover().CopyVolume(context.Background(), 7, srcAddr, dstAddr, "", 0, false, nil); err != nil {
+		t.Fatalf("CopyVolume: %v", err)
+	}
+	if cluster.readonlyReqs[0].Persist {
+		t.Errorf("copy freeze persisted: %+v", cluster.readonlyReqs[0])
+	}
+}