From 5b4a707c437fddf0d7e3f73c2a0e2c44064757b0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 16 Oct 2026 19:45:17 -0700 Subject: [PATCH] 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> --- seaweed-volume/src/storage/disk_location.rs | 27 +- seaweed-volume/src/storage/volume.rs | 13 +- weed/shell/command_volume_move.go | 11 +- weed/shell/command_volume_tier_compact.go | 254 ++++++++++++++++-- .../shell/command_volume_tier_compact_test.go | 87 ++++++ weed/shell/command_volume_tier_move.go | 4 +- weed/shell/command_volume_tier_upload.go | 2 +- weed/storage/remote_tier_integration_test.go | 40 ++- weed/storage/volume_write.go | 16 +- 9 files changed, 401 insertions(+), 53 deletions(-) diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 86925d2f8..733a1d033 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -620,11 +620,20 @@ impl DiskLocation { return Err(v.compacting_error()); } if let Some(mut v) = self.volumes.remove(&vid) { - crate::metrics::VOLUME_GAUGE - .with_label_values(&[&v.collection, "volume"]) - .dec(); - v.destroy(only_empty, only_garbage, keep_remote_data)?; - Ok(()) + match v.destroy(only_empty, only_garbage, keep_remote_data) { + Ok(()) => { + crate::metrics::VOLUME_GAUGE + .with_label_values(&[&v.collection, "volume"]) + .dec(); + Ok(()) + } + Err(e) => { + // A failed destroy left the local files in place, so the + // replica stays registered rather than becoming a phantom. + self.volumes.insert(vid, v); + Err(e) + } + } } else { Err(VolumeError::NotFound) } @@ -641,11 +650,13 @@ impl DiskLocation { for vid in vids { if let Some(mut v) = self.volumes.remove(&vid) { - crate::metrics::VOLUME_GAUGE - .with_label_values(&[&v.collection, "volume"]) - .dec(); if let Err(e) = v.destroy(false, false, false) { warn!(volume_id = vid.0, error = %e, "delete collection: failed to destroy volume"); + self.volumes.insert(vid, v); + } else { + crate::metrics::VOLUME_GAUGE + .with_label_values(&[&v.collection, "volume"]) + .dec(); } } } diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index e91721ec6..b8d5e7910 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -5006,13 +5006,12 @@ impl Volume { .get(&storage_name); if let Some(backend) = backend { if let Err(e) = backend.delete_file_blocking(&storage_key) { - warn!( - volume_id = self.id.0, - storage_name, - storage_key, - error = %e, - "failed to delete remote tier file during destroy" - ); + // Stop short of local cleanup: a failed remote delete must + // not strand a registered replica with no local files. + return Err(VolumeError::Io(io::Error::other(format!( + "volume {} remote object {}/{} delete failed: {}", + self.id, storage_name, storage_key, e + )))); } } else { warn!( diff --git a/weed/shell/command_volume_move.go b/weed/shell/command_volume_move.go index d8c74beb5..3486f413a 100644 --- a/weed/shell/command_volume_move.go +++ b/weed/shell/command_volume_move.go @@ -149,13 +149,16 @@ func markVolumeReplicaWritable(ctx context.Context, grpcDialOption grpc.DialOpti return markVolumeWritable(ctx, grpcDialOption, volumeId, location.ServerAddress(), writable, persist) } -func markVolumeReplicasWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, locations []wdclient.Location, writable, persist bool) error { +// markVolumeReplicasWritable marks each replica and returns the ones that +// took the mark, so a caller can roll back or report a partial operation. +func markVolumeReplicasWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, locations []wdclient.Location, writable, persist bool) (marked []wdclient.Location, err error) { for _, location := range locations { - if err := markVolumeReplicaWritable(ctx, grpcDialOption, volumeId, location, writable, persist); err != nil { - return err + if err = markVolumeReplicaWritable(ctx, grpcDialOption, volumeId, location, writable, persist); err != nil { + return } + marked = append(marked, location) } - return nil + return } // replicateVolumeToServer copies a volume from sourceAddress to targetAddress via the VolumeCopy gRPC stream. diff --git a/weed/shell/command_volume_tier_compact.go b/weed/shell/command_volume_tier_compact.go index fd55958b5..80c3d2865 100644 --- a/weed/shell/command_volume_tier_compact.go +++ b/weed/shell/command_volume_tier_compact.go @@ -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 { diff --git a/weed/shell/command_volume_tier_compact_test.go b/weed/shell/command_volume_tier_compact_test.go index 1de7802e3..67177b997 100644 --- a/weed/shell/command_volume_tier_compact_test.go +++ b/weed/shell/command_volume_tier_compact_test.go @@ -5,6 +5,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/wdclient" "github.com/stretchr/testify/assert" ) @@ -253,3 +254,89 @@ func TestCollectRemoteVolumesWithInfoInvalidPattern(t *testing.T) { _, err := collectRemoteVolumesWithInfo(topo, "[invalid") assert.Error(t, err) } + +func buildTestTopologyWithReplicas() *master_pb.TopologyInfo { + mkDn := func(id string, volumes ...*master_pb.VolumeInformationMessage) *master_pb.DataNodeInfo { + return &master_pb.DataNodeInfo{ + Id: id, + GrpcPort: 18080, + DiskInfos: map[string]*master_pb.DiskInfo{ + "": {VolumeInfos: volumes}, + }, + } + } + remote := func(vid uint32) *master_pb.VolumeInformationMessage { + return &master_pb.VolumeInformationMessage{ + Id: vid, + Collection: "col1", + RemoteStorageName: "s3.default", + RemoteStorageKey: "col1/vid.dat", + } + } + local := func(vid uint32) *master_pb.VolumeInformationMessage { + return &master_pb.VolumeInformationMessage{Id: vid, Collection: "col1"} + } + return &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{ + { + RackInfos: []*master_pb.RackInfo{ + { + DataNodeInfos: []*master_pb.DataNodeInfo{ + mkDn("n1:8080", remote(1)), + mkDn("n2:8080", remote(1), local(2)), + mkDn("n3:8080", remote(1)), + }, + }, + }, + }, + }, + } +} + +func TestCollectVolumeTierCompactLocations(t *testing.T) { + topo := buildTestTopologyWithReplicas() + + tiered, local := collectVolumeTierCompactLocations(topo, needle.VolumeId(1), "col1") + assert.Equal(t, 3, len(tiered)) + assert.Equal(t, 0, len(local)) + + tiered, local = collectVolumeTierCompactLocations(topo, needle.VolumeId(2), "col1") + assert.Equal(t, 0, len(tiered)) + assert.Equal(t, 1, len(local)) + + // a different collection or volume id matches nothing + tiered, local = collectVolumeTierCompactLocations(topo, needle.VolumeId(1), "other") + assert.Equal(t, 0, len(tiered)) + assert.Equal(t, 0, len(local)) +} + +func TestPlanTierCompactResync(t *testing.T) { + loc := func(url string) wdclient.Location { + return wdclient.Location{Url: url, GrpcPort: 1} + } + urls := func(locs []wdclient.Location) (out []string) { + for _, l := range locs { + out = append(out, l.Url) + } + return + } + + // single replica: nothing to resync, nothing to delete (its download + // already removed the object) + order, deleter := planTierCompactResync([]wdclient.Location{loc("a")}, nil) + assert.Empty(t, order) + assert.Empty(t, deleter) + + // two tiered replicas: the sibling is the last holder and goes last + order, deleter = planTierCompactResync([]wdclient.Location{loc("a"), loc("b")}, nil) + assert.Empty(t, order) + assert.Equal(t, []string{"b"}, urls(deleter)) + + // three tiered replicas plus one local copy: middle replicas resync in + // order, the last tiered replica deletes and is resynced last + order, deleter = planTierCompactResync( + []wdclient.Location{loc("a"), loc("b"), loc("c")}, + []wdclient.Location{loc("d")}) + assert.Equal(t, []string{"b", "d"}, urls(order)) + assert.Equal(t, []string{"c"}, urls(deleter)) +} diff --git a/weed/shell/command_volume_tier_move.go b/weed/shell/command_volume_tier_move.go index 4725c4470..0d0ac97be 100644 --- a/weed/shell/command_volume_tier_move.go +++ b/weed/shell/command_volume_tier_move.go @@ -322,7 +322,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i } // mark all replicas as read only - if err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, false, false); err != nil { + if _, err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, false, false); err != nil { return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err) } newAddress := pb.NewServerAddressFromDataNode(dst.dataNode) @@ -337,7 +337,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i // — reopening the replicas beside that copy would fork the volume. if !errors.Is(moveErr, volume_move.ErrSourceKeptReadonly) { // mark all replicas as writable - if err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, true, false); err != nil { + if _, err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, true, false); err != nil { glog.Errorf("mark volume %d as writable on %s: %v", vid, locations[0].Url, err) } } diff --git a/weed/shell/command_volume_tier_upload.go b/weed/shell/command_volume_tier_upload.go index df21d99dc..304676d8c 100644 --- a/weed/shell/command_volume_tier_upload.go +++ b/weed/shell/command_volume_tier_upload.go @@ -153,7 +153,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str return fmt.Errorf("volume %d not found in collection %s", vid, collection) } - err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, existingLocations, false, false) + _, err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, existingLocations, false, false) if err != nil { return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, existingLocations[0].Url, err) } diff --git a/weed/storage/remote_tier_integration_test.go b/weed/storage/remote_tier_integration_test.go index d6356efb8..84ca72755 100644 --- a/weed/storage/remote_tier_integration_test.go +++ b/weed/storage/remote_tier_integration_test.go @@ -36,6 +36,7 @@ type localDirBackend struct { mu sync.Mutex deletes []string // history of DeleteFile keys, for assertions + failDel bool // when set, DeleteFile reports an error without deleting } func newLocalDirBackend(t *testing.T) *localDirBackend { @@ -97,11 +98,20 @@ func (b *localDirBackend) DownloadFile(fileName string, key string, fn func(prog func (b *localDirBackend) DeleteFile(key string) error { b.mu.Lock() + defer b.mu.Unlock() b.deletes = append(b.deletes, key) - b.mu.Unlock() + if b.failDel { + return fmt.Errorf("injected remote delete failure for %s", key) + } return os.Remove(filepath.Join(b.root, key)) } +func (b *localDirBackend) failDeletes(fail bool) { + b.mu.Lock() + defer b.mu.Unlock() + b.failDel = fail +} + func (b *localDirBackend) deleteHistory() []string { b.mu.Lock() defer b.mu.Unlock() @@ -369,6 +379,34 @@ func TestRemoteTier_RealDelete_RemovesRemoteObject(t *testing.T) { require.Equal(t, []string{key}, b.deleteHistory()) } +// TestRemoteTier_FailedRemoteDelete_KeepsLocalVolume locks in the failure +// ordering: a backend delete error must abort Destroy before local files are +// removed, leaving the registered replica fully intact and retryable. +func TestRemoteTier_FailedRemoteDelete_KeepsLocalVolume(t *testing.T) { + b := newLocalDirBackend(t) + registerTestBackend(t, b) + + dir := t.TempDir() + const vid = needle.VolumeId(34) + _, key := tierUpVolume(t, dir, vid, b) + require.True(t, b.objectExists(key)) + + v := reloadVolume(t, dir, vid) + require.True(t, v.HasRemoteFile()) + + b.failDeletes(true) + err := v.Destroy(false, false, false) + require.Error(t, err, "Destroy must surface the remote delete failure") + + require.True(t, util.FileExists(filepath.Join(dir, "34.idx")), "local .idx must survive a failed remote delete") + require.True(t, util.FileExists(filepath.Join(dir, "34.vif")), "local .vif must survive a failed remote delete") + require.True(t, b.objectExists(key), "remote object must survive a failed remote delete") + + b.failDeletes(false) + require.NoError(t, v.Destroy(false, false, false), "retry must succeed once the backend recovers") + require.False(t, b.objectExists(key)) +} + // TestRemoteTier_Vacuum_DoesNotDeleteRemote runs the compact paths against a // remote-tier volume and asserts the safety property: regardless of whether // compact succeeds, errors, or no-ops, it must not delete the cloud object. diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 0054b4f1a..6d657dfca 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -156,15 +156,19 @@ func (v *Volume) destroyLocked(onlyEmpty bool, onlyGarbage bool, keepRemoteData err = fmt.Errorf("volume %d is compacting", v.Id) return } - v.stopWorker() - if !keepRemoteData { - storageName, storageKey := v.RemoteStorageNameKey() - if v.HasRemoteFile() && storageName != "" && storageKey != "" { - if backendStorage, found := backend.BackendStorages[storageName]; found { - backendStorage.DeleteFile(storageKey) + storageName, storageKey := v.RemoteStorageNameKey() + if !keepRemoteData && v.HasRemoteFile() && storageName != "" && storageKey != "" { + if backendStorage, found := backend.BackendStorages[storageName]; found { + if remoteErr := backendStorage.DeleteFile(storageKey); remoteErr != nil { + // Stop short of local cleanup: a failed remote delete must + // not strand a registered replica with no local files. + v.isCompactionInProgress.Store(false) + err = fmt.Errorf("volume %d remote object %s/%s delete failed: %w", v.Id, storageName, storageKey, remoteErr) + return } } } + v.stopWorker() // A regular volume and an EC volume for the same id share .vif. When // EC artefacts coexist on this disk (e.g. shards distributed onto a source // replica before it is deleted), keep the .vif so removing the regular