From d4d8e097dd72592bb3e37973ce0fde8eab836478 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 10 Aug 2026 12:35:40 -0700 Subject: [PATCH] shell: volume.move cleans up when aborted after the copy phase (#10704) * shell: volume.move restores source writability when aborted after the copy phase * shell: volume.move removes the incomplete target copy when aborted before the source delete * shell: give each abort cleanup RPC its own timeout --- weed/shell/command_volume_copy.go | 2 +- weed/shell/command_volume_merge.go | 2 +- weed/shell/command_volume_move.go | 29 +++++++++++++++++++++++++++-- 3 files changed, 29 insertions(+), 4 deletions(-) diff --git a/weed/shell/command_volume_copy.go b/weed/shell/command_volume_copy.go index 6fe486751..e60447d0a 100644 --- a/weed/shell/command_volume_copy.go +++ b/weed/shell/command_volume_copy.go @@ -61,6 +61,6 @@ func (c *commandVolumeCopy) Do(args []string, commandEnv *CommandEnv, writer io. return fmt.Errorf("source and target volume servers are the same!") } - _, err = copyVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, "", 0, true) + _, _, err = copyVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, "", 0, true) return } diff --git a/weed/shell/command_volume_merge.go b/weed/shell/command_volume_merge.go index 5605ae632..92887bfed 100644 --- a/weed/shell/command_volume_merge.go +++ b/weed/shell/command_volume_merge.go @@ -187,7 +187,7 @@ func (c *commandVolumeMerge) Do(args []string, commandEnv *CommandEnv, writer io for i, replica := range replicas { sourceServer := pb.NewServerAddressFromDataNode(replica.location.dataNode) - if _, err = copyVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, volumeId, targetServer, sourceServer, "", 0, false); err != nil { + if _, _, err = copyVolume(context.Background(), commandEnv.option.GrpcDialOption, writer, volumeId, targetServer, sourceServer, "", 0, false); err != nil { return fmt.Errorf("rebuild replica %d/%d on %s from merged volume %d: %w; merged copy kept on %s, re-run to finish", i+1, len(replicas), sourceServer, volumeId, err, targetServer) } } diff --git a/weed/shell/command_volume_move.go b/weed/shell/command_volume_move.go index 1dd788c5e..485425f7a 100644 --- a/weed/shell/command_volume_move.go +++ b/weed/shell/command_volume_move.go @@ -99,11 +99,34 @@ func (c *commandVolumeMove) Do(args []string, commandEnv *CommandEnv, writer io. func LiveMoveVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, idleTimeout time.Duration, diskType string, ioBytePerSecond int64, skipTailError bool) (err error) { log.Printf("copying volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer) - lastAppendAtNs, err := copyVolume(ctx, grpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, diskType, ioBytePerSecond, false) + lastAppendAtNs, leftReadonly, err := copyVolume(ctx, grpcDialOption, writer, volumeId, sourceVolumeServer, targetVolumeServer, diskType, ioBytePerSecond, false) if err != nil { return fmt.Errorf("copy volume %d from %s to %s: %v", volumeId, sourceVolumeServer, targetVolumeServer, err) } + // A move aborted after the copy must restore the source writability, or + // the source volume is left permanently readonly. + var sourceDeleteStarted bool + defer func() { + if err == nil || !leftReadonly { + return + } + if !sourceDeleteStarted { + // The target copy may be missing tailed entries; remove it so the + // restored source stays the only replica. + deleteCtx, deleteCancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer deleteCancel() + if dErr := deleteVolume(deleteCtx, grpcDialOption, volumeId, targetVolumeServer, false, true); dErr != nil { + log.Printf("failed to delete the incomplete copy of volume %d on %s: %v", volumeId, targetVolumeServer, dErr) + } + } + restoreCtx, restoreCancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer restoreCancel() + if wErr := markVolumeWritable(restoreCtx, grpcDialOption, volumeId, sourceVolumeServer, true, false); wErr != nil { + log.Printf("failed to restore volume %d writable on %s: %v", volumeId, sourceVolumeServer, wErr) + } + }() + log.Printf("tailing volume %d from %s to %s", volumeId, sourceVolumeServer, targetVolumeServer) if err = tailVolume(ctx, grpcDialOption, volumeId, sourceVolumeServer, targetVolumeServer, lastAppendAtNs, idleTimeout); err != nil { if skipTailError { @@ -114,6 +137,7 @@ func LiveMoveVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer } log.Printf("deleting volume %d from %s", volumeId, sourceVolumeServer) + sourceDeleteStarted = true if err = deleteVolume(ctx, grpcDialOption, volumeId, sourceVolumeServer, false, true); err != nil { return fmt.Errorf("delete volume %d from %s: %v", volumeId, sourceVolumeServer, err) } @@ -122,7 +146,7 @@ func LiveMoveVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer return nil } -func copyVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, diskType string, ioBytePerSecond int64, restoreWritable bool) (lastAppendAtNs uint64, err error) { +func copyVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceVolumeServer, targetVolumeServer pb.ServerAddress, diskType string, ioBytePerSecond int64, restoreWritable bool) (lastAppendAtNs uint64, leftReadonly bool, err error) { // check to see if the volume is already read-only and if its not then we need // to mark it as read-only and then before we return we need to undo what we @@ -133,6 +157,7 @@ func copyVolume(ctx context.Context, grpcDialOption grpc.DialOption, writer io.W return } if !restoreWritable && err == nil { + leftReadonly = true return }