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

@@ -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
 }

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)
 		}
 	}

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
 		}