shell: keep volume.tier.compact replicas on the same remote object (#11783)

* shell: keep volume.tier.compact replicas on the same remote object The compact cycle ran on a single replica: it deleted the shared remote object up front and never touched the siblings, leaving their .vif/.idx pointing at a removed object. It also remounted the working replica into a writable local volume mid-operation. Freeze all replicas first, keep the shared object while siblings still reference it, replicate the new metadata to every replica, and let the last tiered replica retire the old object. * storage: surface remote object delete failures from volume destroy Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * shell: keep tier.compact replicas frozen until every copy lands A delete between the source unfreeze and a sibling's index copy could land past the copied snapshot, leaving the replica serving deleted data. Unfreeze all replicas only after every resync completes; a remote volume's readonly mark is inert, so the freeze adds no extra cost. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * storage: abort volume destroy when remote delete fails destroyLocked removed local files even when the backend delete errored, leaving a registered replica with no .dat/.idx and a leaked remote object. Run the remote delete before local cleanup and bail out on failure so the volume stays fully usable and the delete can be retried. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: mirror destroy ordering for remote delete failures Propagate a failed remote-object delete instead of warning past it, and reinsert the volume into the disk location registry so a still-intact replica is not unregistered. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * shell: recover partial tier.compact freezes and report held replicas A failed freeze left already-marked replicas readonly with nothing changed, and later stage failures returned without saying the replicas stay frozen or how to resume. Roll back the partial mark when freezing fails, note in every mid-flight error that replicas remain readonly and rerun is the recovery path, and report how many replicas a partial unfreeze released. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * shell: retire abandoned remote objects during tier.compact resync A resync overwrote each replica's .vif, orphaning its previous remote object; an interrupted run could leave replicas on different keys, and only the last tiered replica's object was deleted. Read every replica's remote file list up front, count holders per object, and delete-before- copy whichever replica is an object's last holder, so shared and stranded objects alike are retired instead of leaked. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: keep collection members registered when destroy fails --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-11-12 23:09:33 +01:00 · 2026-10-16 19:45:17 -07:00
1 parent d3a97f76bd
commit 5b4a707c43
@@ -9,10 +9,12 @@ import (
 	"google.golang.org/grpc"

 	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/operation/volume_move"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 )

 func init() {
@@ -112,7 +114,7 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri

 	var failedCount int
 	for _, rv := range remoteVolumes {
-		if err = doVolumeTierCompact(commandEnv, writer, rv, *garbageThreshold, *concurrency); err != nil {
+		if err = doVolumeTierCompact(commandEnv, writer, topologyInfo, rv, *garbageThreshold, *concurrency); err != nil {
 			fmt.Fprintf(writer, "error compacting volume %d: %v\n", rv.vid, err)
 			failedCount++
 		}
@@ -199,11 +201,49 @@ func collectRemoteVolumesWithInfo(topoInfo *master_pb.TopologyInfo, collectionPa
 	return result, nil
 }

-func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolumeInfo, garbageThreshold float64, concurrency int) error {
+// collectVolumeTierCompactLocations returns the volume's tiered replicas first
+// followed by any local ones. Every tiered replica references the same remote
+// object, so the source is always a tiered replica and the last tiered replica
+// doubles as the one that can retire the old object after the resync.
+func collectVolumeTierCompactLocations(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collection string) (tiered, local []wdclient.Location) {
+	eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
+		for _, disk := range dn.DiskInfos {
+			for _, vi := range disk.VolumeInfos {
+				if needle.VolumeId(vi.Id) != vid || vi.Collection != collection {
+					continue
+				}
+				loc := wdclient.Location{
+					Url:        dn.Id,
+					PublicUrl:  dn.Id,
+					GrpcPort:   int(dn.GrpcPort),
+					DataCenter: string(dc),
+				}
+				if vi.RemoteStorageName != "" {
+					tiered = append(tiered, loc)
+				} else {
+					local = append(local, loc)
+				}
+			}
+		}
+	})
+	return
+}
+
+func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, topoInfo *master_pb.TopologyInfo, rv remoteVolumeInfo, garbageThreshold float64, concurrency int) error {
 	grpcDialOption := commandEnv.option.GrpcDialOption
+	ctx := context.Background()
+
+	tieredLocations, localLocations := collectVolumeTierCompactLocations(topoInfo, rv.vid, rv.collection)
+	if len(tieredLocations) == 0 {
+		return fmt.Errorf("volume %d has no remote replica", rv.vid)
+	}
+	source := tieredLocations[0].ServerAddress()
+	// all replicas of the volume: tiered ones share one remote object, and any
+	// local copy is re-pointed at the new object during the resync below
+	locations := append(append([]wdclient.Location{}, tieredLocations...), localLocations...)

 	// step 1: check garbage level
-	garbageRatio, err := checkVolumeGarbage(grpcDialOption, rv.vid, rv.serverAddress)
+	garbageRatio, err := checkVolumeGarbage(grpcDialOption, rv.vid, source)
 	if err != nil {
 		return fmt.Errorf("check garbage for volume %d: %v", rv.vid, err)
 	}
@@ -216,43 +256,209 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu

 	fmt.Fprintf(writer, "volume %d garbage ratio %.4f, starting compaction...\n", rv.vid, garbageRatio)

-	// step 2: download .dat from remote to local
-	// this deletes the remote file and reloads the volume as local, then re-uploads below
-	fmt.Fprintf(writer, "  downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName)
-	err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false, concurrency)
+	// step 2: freeze every replica. The mark is persisted because the download
+	// remounts the volume into a fresh in-memory state, and a local .dat with no
+	// persisted readonly flag would load writable while its siblings stay readonly.
+	frozen, err := markVolumeReplicasWritable(ctx, grpcDialOption, rv.vid, locations, false, true)
 	if err != nil {
-		return fmt.Errorf("download volume %d from remote: %v", rv.vid, err)
+		// nothing changed yet: thaw the replicas already marked
+		if _, thawErr := markVolumeReplicasWritable(ctx, grpcDialOption, rv.vid, frozen, true, false); thawErr != nil {
+			return fmt.Errorf("mark volume %d readonly: %v (thawing the %d marked replicas also failed: %v)",
+				rv.vid, err, len(frozen), thawErr)
+		}
+		return fmt.Errorf("mark volume %d readonly: %v", rv.vid, err)
+	}
+	// Past this point a failure keeps every replica frozen on purpose: a replica
+	// mid-transition may hold divergent data, and thawing it would expose that.
+	// Re-running volume.tier.compact completes the resync and releases the marks.
+	stillFrozen := func(stageErr error) error {
+		return fmt.Errorf("%v (volume %d replicas stay readonly; rerun volume.tier.compact to resume)", stageErr, rv.vid)
 	}

-	// step 3: compact the local volume
-	fmt.Fprintf(writer, "  compacting volume %d...\n", rv.vid)
-	err = compactVolumeOnServer(grpcDialOption, writer, rv.vid, rv.serverAddress)
+	// step 3: download .dat from remote to local; the shared object stays while
+	// sibling replicas still reference it
+	fmt.Fprintf(writer, "  downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName)
+	err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, source, len(tieredLocations) > 1, concurrency)
 	if err != nil {
-		// compaction failed, but volume is now local without remote reference
-		// upload the uncompacted volume back to restore cloud tier state
-		fmt.Fprintf(writer, "  compaction failed: %v\n", err)
+		return stillFrozen(fmt.Errorf("download volume %d from remote: %v", rv.vid, err))
+	}
+
+	// step 4: compact the local volume; on failure re-upload what is there so
+	// the replica returns to the tier and the resync below can still proceed
+	compactErr := compactVolumeOnServer(grpcDialOption, writer, rv.vid, source)
+	if compactErr != nil {
+		fmt.Fprintf(writer, "  compaction failed: %v\n", compactErr)
 		fmt.Fprintf(writer, "  re-uploading volume %d to %s without compaction...\n", rv.vid, rv.remoteStorageName)
-		uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency)
-		if uploadErr != nil {
-			return fmt.Errorf("compaction failed (%v) and re-upload also failed (%v), volume %d remains local",
-				err, uploadErr, rv.vid)
+	}
+
+	// step 5: upload the (possibly uncompacted) volume back to remote under a new key
+	fmt.Fprintf(writer, "  uploading volume %d to %s...\n", rv.vid, rv.remoteStorageName)
+	uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, source, rv.remoteStorageName, false, concurrency)
+	if uploadErr != nil {
+		if compactErr != nil {
+			return stillFrozen(fmt.Errorf("compaction failed (%v) and re-upload also failed (%v), volume %d remains local",
+				compactErr, uploadErr, rv.vid))
+		}
+		return stillFrozen(fmt.Errorf("upload compacted volume %d to %s: %v (volume remains local with compacted data)",
+			rv.vid, rv.remoteStorageName, uploadErr))
+	}
+
+	// step 6: push the new .idx/.vif to every sibling so all replicas reference
+	// the new remote object; the last tiered replica goes last so the old shared
+	// object still has a holder until then. Replicas stay frozen until every
+	// copy lands, so a delete cannot slip past a sibling's index snapshot; a
+	// copied readonly mark is inert on a remote-backed volume anyway.
+	syncErr := syncVolumeTierCompactReplicas(ctx, grpcDialOption, writer, rv, tieredLocations, localLocations)
+	if compactErr != nil {
+		if syncErr != nil {
+			return stillFrozen(fmt.Errorf("compaction failed (%v), volume %d re-uploaded without compaction but resync failed: %v",
+				compactErr, rv.vid, syncErr))
 		}
 		return fmt.Errorf("compaction failed (%v), volume %d re-uploaded to %s without compaction",
-			err, rv.vid, rv.remoteStorageName)
+			compactErr, rv.vid, rv.remoteStorageName)
+	}
+	if syncErr != nil {
+		return stillFrozen(syncErr)
 	}

-	// step 4: upload compacted volume back to remote
-	fmt.Fprintf(writer, "  uploading compacted volume %d to %s...\n", rv.vid, rv.remoteStorageName)
-	err = uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency)
+	// step 7: unfreeze every replica, clearing the persisted readonly mark on
+	// each. Remote volumes refuse writes regardless; this re-enables deletes.
+	thawed, err := markVolumeReplicasWritable(ctx, grpcDialOption, rv.vid, locations, true, false)
 	if err != nil {
-		return fmt.Errorf("upload compacted volume %d to %s: %v (volume remains local with compacted data)",
-			rv.vid, rv.remoteStorageName, err)
+		return fmt.Errorf("unfreeze volume %d replicas: %v (%d of %d released)", rv.vid, err, len(thawed), len(locations))
 	}

 	fmt.Fprintf(writer, "volume %d compacted and uploaded to %s successfully\n", rv.vid, rv.remoteStorageName)
 	return nil
 }

+// planTierCompactResync orders the sibling resyncs. The last tiered replica is
+// held out for last: the resync retires a remote object only once its final
+// holder is re-pointed, so processing the old object's expected last holder at
+// the end keeps it alive for every sibling still referencing it. A
+// single-replica volume has no such holder: its download already removed the
+// object.
+func planTierCompactResync(tieredLocations, localLocations []wdclient.Location) (order, deleter []wdclient.Location) {
+	rest := append(append([]wdclient.Location{}, tieredLocations[1:]...), localLocations...)
+	if len(tieredLocations) > 1 {
+		last := tieredLocations[len(tieredLocations)-1]
+		for _, loc := range rest {
+			if loc.ServerAddress() == last.ServerAddress() {
+				deleter = append(deleter, loc)
+				continue
+			}
+			order = append(order, loc)
+		}
+		return
+	}
+	return rest, nil
+}
+
+func syncVolumeTierCompactReplicas(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, rv remoteVolumeInfo, tieredLocations, localLocations []wdclient.Location) error {
+	source := tieredLocations[0].ServerAddress()
+	mover := volume_move.NewMover(grpcDialOption)
+
+	resync := func(loc wdclient.Location) error {
+		fmt.Fprintf(writer, "  replicating volume %d metadata from %s to %s\n", rv.vid, tieredLocations[0].Url, loc.Url)
+		return replicateVolumeToServer(ctx, grpcDialOption, writer, rv.vid, source, loc.ServerAddress(), "", 0)
+	}
+
+	order, deleter := planTierCompactResync(tieredLocations, localLocations)
+
+	// Every replica's current remote objects, so a resynced-over replica can
+	// retire an object once it is the last holder — covers the shared object
+	// and any object a previous interrupted run left referenced. The source
+	// counts as a holder of the new object so a replica already sharing it is
+	// never mistaken for the last holder.
+	allLocations := append(append([]wdclient.Location{}, tieredLocations...), localLocations...)
+	holders, readErr := readRemoteObjectHolders(ctx, grpcDialOption, rv.vid, allLocations)
+	if readErr != nil {
+		// Fall back to the explicit last-replica deleter when a replica's
+		// remote files cannot be read.
+		for _, loc := range order {
+			if err := resync(loc); err != nil {
+				return fmt.Errorf("replicate volume %d from %s to %s: %v", rv.vid, tieredLocations[0].Url, loc.Url, err)
+			}
+		}
+		if len(deleter) == 0 {
+			return nil
+		}
+		fmt.Fprintf(writer, "  removing old remote copy of volume %d on %s\n", rv.vid, deleter[0].Url)
+		if err := mover.DeleteVolume(ctx, rv.vid, deleter[0].ServerAddress(), false, false, false); err != nil {
+			return fmt.Errorf("delete old copy of volume %d on %s: %v", rv.vid, deleter[0].Url, err)
+		}
+		return resync(deleter[0])
+	}
+
+	for _, loc := range append(append([]wdclient.Location{}, order...), deleter...) {
+		files := holders.files[loc.ServerAddress()]
+		soleHolder := len(files) > 0
+		for _, f := range files {
+			if holders.count[remoteFileKey(f)] > 1 {
+				soleHolder = false
+				break
+			}
+		}
+		if soleHolder {
+			// This replica is the only holder of its remote objects: removing
+			// its copy retires them, and the resync restores the replica on the
+			// new object.
+			fmt.Fprintf(writer, "  removing old remote copy of volume %d on %s\n", rv.vid, loc.Url)
+			if err := mover.DeleteVolume(ctx, rv.vid, loc.ServerAddress(), false, false, false); err != nil {
+				return fmt.Errorf("delete old copy of volume %d on %s: %v", rv.vid, loc.Url, err)
+			}
+		}
+		for _, f := range files {
+			holders.count[remoteFileKey(f)]--
+		}
+		if err := resync(loc); err != nil {
+			return fmt.Errorf("replicate volume %d from %s to %s: %v", rv.vid, tieredLocations[0].Url, loc.Url, err)
+		}
+	}
+	return nil
+}
+
+// remoteObjectHolders reads every replica's remote file list and counts how
+// many replicas reference each remote object.
+type remoteObjectHolders struct {
+	files map[pb.ServerAddress][]*volume_server_pb.RemoteFile
+	count map[string]int
+}
+
+func remoteFileKey(f *volume_server_pb.RemoteFile) string {
+	return f.BackendType + "/" + f.BackendId + "/" + f.Key
+}
+
+func readRemoteObjectHolders(ctx context.Context, grpcDialOption grpc.DialOption, vid needle.VolumeId, locations []wdclient.Location) (*remoteObjectHolders, error) {
+	h := &remoteObjectHolders{
+		files: make(map[pb.ServerAddress][]*volume_server_pb.RemoteFile),
+		count: make(map[string]int),
+	}
+	for _, loc := range locations {
+		addr := loc.ServerAddress()
+		err := operation.WithVolumeServerClient(false, addr, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
+			resp, reqErr := client.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
+				VolumeId: uint32(vid),
+			})
+			if reqErr != nil {
+				return reqErr
+			}
+			if resp == nil || resp.VolumeInfo == nil {
+				return fmt.Errorf("volume %d status has no remote file info", vid)
+			}
+			h.files[addr] = resp.VolumeInfo.GetFiles()
+			return nil
+		})
+		if err != nil {
+			return nil, err
+		}
+		for _, f := range h.files[addr] {
+			h.count[remoteFileKey(f)]++
+		}
+	}
+	return h, nil
+}
+
 func checkVolumeGarbage(grpcDialOption grpc.DialOption, vid needle.VolumeId, server pb.ServerAddress) (float64, error) {
 	var garbageRatio float64
 	err := operation.WithVolumeServerClient(false, server, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {