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