From 7f83227560702213c6bd5af2581c1aa415c905e6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 18:47:30 -0800 Subject: [PATCH] shell: block deletes on the tier download copy source A readonly-can-delete source took deletes after VolumeCopy read its index, so the replaced replica lost them and served the files again. Hard-mark the source readonly while replicas are replaced, then restore the writable mark or reload the persisted state. Co-Authored-By: Claude Opus 5.5 --- weed/shell/command_volume_tier_download.go | 59 ++++++++++++++----- .../command_volume_tier_download_test.go | 57 +++++++++++++++++- 2 files changed, 100 insertions(+), 16 deletions(-) diff --git a/weed/shell/command_volume_tier_download.go b/weed/shell/command_volume_tier_download.go index ea5103ea2c..5ac11cc04e 100644 --- a/weed/shell/command_volume_tier_download.go +++ b/weed/shell/command_volume_tier_download.go @@ -179,6 +179,12 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i } } + restore, err := freezeTierDownloadSource(grpcDialOption, vid, source) + if err != nil { + return fmt.Errorf("block deletes on volume %d at %s: %v", vid, source, err) + } + defer restore(writer) + // Copy over all remote-backed replicas but one, re-reading the topology // so a replica that was offline or created meanwhile is replaced too. The // last one is deleted together with the remote object and then replaced. @@ -219,6 +225,43 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i } } +// freezeTierDownloadSource blocks deletes on the copy source: a delete applied +// after VolumeCopy read the source index would be lost on the replaced replica. +// restore brings back the writable mark, or reloads the persisted state. +func freezeTierDownloadSource(grpcDialOption grpc.DialOption, vid needle.VolumeId, source pb.ServerAddress) (restore func(io.Writer), err error) { + var wasWritable bool + err = operation.WithVolumeServerClient(false, source, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + resp, statusErr := client.VolumeStatus(context.Background(), &volume_server_pb.VolumeStatusRequest{VolumeId: uint32(vid)}) + if statusErr != nil { + return statusErr + } + wasWritable = !resp.IsReadOnly + _, markErr := client.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid)}) + return markErr + }) + restore = func(writer io.Writer) { + restoreErr := operation.WithVolumeServerClient(false, source, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + if wasWritable { + _, markErr := client.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)}) + return markErr + } + return remountVolume(client, vid) + }) + if restoreErr != nil { + fmt.Fprintf(writer, "restore volume %d state on %s: %v\n", vid, source, restoreErr) + } + } + return +} + +func remountVolume(client volume_server_pb.VolumeServerClient, vid needle.VolumeId) error { + if _, err := client.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{VolumeId: uint32(vid)}); err != nil { + return err + } + _, err := client.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{VolumeId: uint32(vid)}) + return err +} + func replicateTierDownloadedVolume(grpcDialOption grpc.DialOption, writer io.Writer, vid needle.VolumeId, source pb.ServerAddress, target tierDownloadReplica) error { fmt.Fprintf(writer, "replicate downloaded volume %d from %s to %s\n", vid, source, target.server) if _, err := copyVolume(context.Background(), grpcDialOption, writer, vid, source, target.server, types.ToDiskType(target.diskType).ReadableString(), 0, true); err != nil { @@ -284,21 +327,7 @@ func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, return downloadErr } - _, unmountErr := volumeServerClient.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{ - VolumeId: uint32(volumeId), - }) - if unmountErr != nil { - return unmountErr - } - - _, mountErr := volumeServerClient.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{ - VolumeId: uint32(volumeId), - }) - if mountErr != nil { - return mountErr - } - - return nil + return remountVolume(volumeServerClient, volumeId) }) return err diff --git a/weed/shell/command_volume_tier_download_test.go b/weed/shell/command_volume_tier_download_test.go index c08b54fc5b..b1aa83b3ee 100644 --- a/weed/shell/command_volume_tier_download_test.go +++ b/weed/shell/command_volume_tier_download_test.go @@ -38,6 +38,9 @@ type tierDownloadCluster struct { // hiddenOnRead hides servers from the n-th topology read (0-based) hiddenOnRead map[int]map[pb.ServerAddress]bool reads int + frozen map[pb.ServerAddress]bool + writable map[pb.ServerAddress]bool + remounts map[pb.ServerAddress]int } type tierDownloadTestServer struct { @@ -78,11 +81,36 @@ func (s *tierDownloadTestServer) VolumeUnmount(context.Context, *volume_server_p } func (s *tierDownloadTestServer) VolumeMount(context.Context, *volume_server_pb.VolumeMountRequest) (*volume_server_pb.VolumeMountResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + c.frozen[s.addr] = false + c.remounts[s.addr]++ return &volume_server_pb.VolumeMountResponse{}, nil } func (s *tierDownloadTestServer) VolumeStatus(context.Context, *volume_server_pb.VolumeStatusRequest) (*volume_server_pb.VolumeStatusResponse, error) { - return &volume_server_pb.VolumeStatusResponse{IsReadOnly: true}, nil + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + return &volume_server_pb.VolumeStatusResponse{IsReadOnly: !c.writable[s.addr] || c.frozen[s.addr]}, nil +} + +func (s *tierDownloadTestServer) VolumeMarkReadonly(_ context.Context, req *volume_server_pb.VolumeMarkReadonlyRequest) (*volume_server_pb.VolumeMarkReadonlyResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + c.frozen[s.addr] = !req.CanDelete + return &volume_server_pb.VolumeMarkReadonlyResponse{}, nil +} + +func (s *tierDownloadTestServer) VolumeMarkWritable(context.Context, *volume_server_pb.VolumeMarkWritableRequest) (*volume_server_pb.VolumeMarkWritableResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + c.frozen[s.addr] = false + c.writable[s.addr] = true + return &volume_server_pb.VolumeMarkWritableResponse{}, nil } func (s *tierDownloadTestServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, _ volume_server_pb.VolumeServer_VolumeCopyServer) error { @@ -97,6 +125,9 @@ func (s *tierDownloadTestServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequ if got != want { return fmt.Errorf("copy volume %d onto disk type %q, replica is on %q", req.VolumeId, got, want) } + if !c.frozen[pb.ServerAddress(req.SourceDataNode)] { + return fmt.Errorf("volume %d on %s still accepts deletes during the copy", req.VolumeId, req.SourceDataNode) + } switch c.replicas[pb.ServerAddress(req.SourceDataNode)] { case tierReplicaLocal: c.replicas[s.addr] = tierReplicaLocal @@ -127,6 +158,9 @@ func newTierDownloadCluster(t *testing.T, states ...string) *tierDownloadCluster diskTypes: make(map[pb.ServerAddress]string), objectExists: true, hiddenOnRead: make(map[int]map[pb.ServerAddress]bool), + frozen: make(map[pb.ServerAddress]bool), + writable: make(map[pb.ServerAddress]bool), + remounts: make(map[pb.ServerAddress]int), } for _, state := range states { lis, err := net.Listen("tcp", "127.0.0.1:0") @@ -277,3 +311,24 @@ func TestTierDownloadCopiesFromSsdSourceOntoHddReplica(t *testing.T) { } c.assertAllLocal(t) } + +func TestTierDownloadRestoresSourceAfterBlockingDeletes(t *testing.T) { + for _, writable := range []bool{false, true} { + c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote) + source := c.servers[0] + c.writable[source] = writable + if err := c.run(t); err != nil { + t.Fatalf("tier download (writable=%v): %v", writable, err) + } + c.assertAllLocal(t) + if c.frozen[source] { + t.Errorf("writable=%v: source still blocks deletes after the command", writable) + } + if c.writable[source] != writable { + t.Errorf("writable=%v: source writable state changed to %v", writable, c.writable[source]) + } + if !writable && c.remounts[source] != 1 { + t.Errorf("readonly source should reload its persisted state once, got %d remounts", c.remounts[source]) + } + } +}