diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 2f20e58bb..02e0e808e 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1080,21 +1080,15 @@ impl VolumeServer for VolumeGrpcService { let req = request.into_inner(); let vid = VolumeId(req.volume_id); - // If volume already exists locally, delete it first - { + // A pre-existing local replica is NOT deleted up front. Deleting before + // the source is confirmed reachable destroys a healthy copy on a + // transient source outage (and, on retry, can lose the volume + // entirely). The delete is deferred until read_volume_file_status below + // proves the source holds the volume; readability alone is the gate. + let had_existing_volume = { let store = self.state.store.read().unwrap(); - if store.find_volume(vid).is_some() { - drop(store); - let mut store = self.state.store.write().unwrap(); - // keep remote data: the inbound copy carries a .vif that may - // point at the same cloud-tier object the existing volume - // references. - store.delete_volume(vid, false, true).map_err(|e| { - Status::internal(format!("failed to delete existing volume {}: {}", vid, e)) - })?; - self.state.volume_state_notify.notify_one(); - } - } + store.find_volume(vid).is_some() + }; // Parse source_data_node address: "ip:port.grpcPort" or "ip:port" (grpc = port + 10000) let source = &req.source_data_node; @@ -1135,6 +1129,19 @@ impl VolumeServer for VolumeGrpcService { .map_err(|e| Status::internal(format!("read volume file status failed, {}", e)))? .into_inner(); + // Source is reachable and holds the volume: only now is it safe to drop + // an existing local replica before overwriting its files. + if had_existing_volume { + let mut store = self.state.store.write().unwrap(); + // keep remote data: the inbound copy carries a .vif that may point + // at the same cloud-tier object the existing volume references. + store.delete_volume(vid, false, true).map_err(|e| { + Status::internal(format!("failed to delete existing volume {}: {}", vid, e)) + })?; + drop(store); + self.state.volume_state_notify.notify_one(); + } + let requested_disk_type = if !req.disk_type.is_empty() { DiskType::from_string(&req.disk_type) } else { @@ -1166,9 +1173,12 @@ impl VolumeServer for VolumeGrpcService { let idx_base_name = crate::storage::volume::volume_file_name(&idx_base, &vol_info.collection, vid); - // Write a .note file to indicate copy in progress + // Write a .note file to indicate copy in progress. A leftover note + // fails the volume load on restart, so a write failure must abort. let note_path = format!("{}.note", data_base_name); - let _ = std::fs::write(¬e_path, format!("copying from {}", source)); + std::fs::write(¬e_path, format!("copying from {}", source)).map_err(|e| { + Status::internal(format!("write .note for volume {}: {}", vid, e)) + })?; let has_remote_dat = vol_info .volume_info @@ -1299,8 +1309,16 @@ impl VolumeServer for VolumeGrpcService { let _ = set_file_mtime(&vif_path, vif_modified_ts_ns); } - // Remove the .note file - let _ = std::fs::remove_file(¬e_path); + // Remove the .note file. A leftover note fails the load on the + // next restart, so a removal failure must fail the copy. + if let Err(e) = std::fs::remove_file(¬e_path) { + if e.kind() != std::io::ErrorKind::NotFound { + return Err(Status::internal(format!( + "remove .note for volume {}: {}", + vid, e + ))); + } + } // Verify file sizes if !has_remote_dat { diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index d3c4a0179..ca5ea407c 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -125,19 +125,6 @@ impl DiskLocation { let volume_name = volume_file_name(&self.directory, &collection, vid); let idx_name = volume_file_name(&self.idx_directory, &collection, vid); - // Check for incomplete volume (.note file means a VolumeCopy was interrupted) - let note_path = format!("{}.note", volume_name); - if std::path::Path::new(¬e_path).exists() { - let note = fs::read_to_string(¬e_path).unwrap_or_default(); - warn!( - volume_id = vid.0, - "volume was not completed: {}, removing files", note - ); - remove_volume_files(&volume_name, false); - remove_volume_files(&idx_name, false); - continue; - } - // Sweep a leftover empty `.dat` stub (a phantom from the pre-fix // loader) before it loads as a phantom volume or blocks startup. if remove_empty_ec_dat_stub(&volume_name, &idx_name, vid) { @@ -181,6 +168,30 @@ impl DiskLocation { let _ = fs::remove_file(&cpx_path); } + // Check for an incomplete volume (.note means a VolumeCopy was + // interrupted). This runs BELOW the empty-stub sweep and EC + // validation: when an .ecx for this vid coexists on the disk, the + // regular and EC volumes share .vif, so removing the + // incomplete regular copy must keep the .vif (keep_vif=true) or it + // would strip the EC volume's info file. + let note_path = format!("{}.note", volume_name); + if std::path::Path::new(¬e_path).exists() { + let note = fs::read_to_string(¬e_path).unwrap_or_default(); + warn!( + volume_id = vid.0, + "volume was not completed: {}, removing files", note + ); + // Re-check .ecx now (not the pre-validation ecx_exists): the + // invalid-EC cleanup above may have removed it, in which case + // the .vif is no longer shared and must not be preserved. + let keep_vif = std::path::Path::new(&ecx_path).exists() + || (self.idx_directory != self.directory + && std::path::Path::new(&format!("{}.ecx", volume_name)).exists()); + remove_volume_files(&volume_name, keep_vif); + remove_volume_files(&idx_name, keep_vif); + continue; + } + // Skip if already loaded (e.g., from a previous call) if self.volumes.contains_key(&vid) { continue; diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index bb815be4c..0ad57e2bf 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -369,6 +369,16 @@ impl Store { let vif_path = format!("{}.vif", base); if std::path::Path::new(&dat_path).exists() || std::path::Path::new(&vif_path).exists() { + // A persisting .note means the copy that produced these files + // never completed; mounting it would expose a truncated volume. + // Fail the mount so the caller (VolumeCopy) treats it as an error. + let note_path = format!("{}.note", base); + if std::path::Path::new(¬e_path).exists() { + return Err(VolumeError::Io(io::Error::new( + io::ErrorKind::Other, + format!("volume {} copy incomplete: .note still present", vid), + ))); + } return loc.create_volume( vid, collection, diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 89202d56d..0d372982f 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -31,20 +31,13 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return err } - v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) - if v != nil { - - glog.V(0).Infof("volume %d already exists. deleted before copying...", req.VolumeId) - - // keep remote data: the inbound copy carries a .vif that may point at - // the same cloud-tier object the existing volume references. - err := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), false, true) - if err != nil { - return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, err) - } - - glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) - } + // A pre-existing local replica is NOT deleted up front. Deleting before the + // source is confirmed reachable destroys a healthy copy on a transient + // source outage (and, on retry, can lose the volume entirely). The delete is + // deferred until ReadVolumeFileStatus below proves the source holds the + // volume; readability alone is the gate (size/count comparisons invert after + // divergent vacuum/compaction and would block valid re-replication). + hasExistingVolume := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) != nil // the master will not start compaction for read-only volumes, so it is safe to just copy files directly // copy .dat and .idx files @@ -65,6 +58,18 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return fmt.Errorf("read volume file status failed, %w", err) } + // Source is reachable and holds the volume: only now is it safe to drop + // an existing local replica before overwriting its files. + if hasExistingVolume { + glog.V(0).Infof("volume %d already exists. deleting before copying from %s...", req.VolumeId, req.SourceDataNode) + // keep remote data: the inbound copy carries a .vif that may point at + // the same cloud-tier object the existing volume references. + if delErr := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), false, true); delErr != nil { + return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, delErr) + } + glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) + } + diskType := volFileInfoResp.DiskType if req.DiskType != "" { diskType = req.DiskType @@ -81,7 +86,12 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre indexBaseFileName = storage.VolumeFileName(location.IdxDirectory, volFileInfoResp.Collection, int(req.VolumeId)) hasRemoteDatFile = volFileInfoResp.VolumeInfo != nil && len(volFileInfoResp.VolumeInfo.Files) > 0 - util.WriteFile(dataBaseFileName+".note", []byte(fmt.Sprintf("copying from %s", req.SourceDataNode)), 0755) + // The .note marks the copy as in-progress; a leftover note fails the + // volume load on restart, so a write failure must abort the copy. + if noteErr := util.WriteFile(dataBaseFileName+".note", []byte(fmt.Sprintf("copying from %s", req.SourceDataNode)), 0755); noteErr != nil { + err = noteErr + return fmt.Errorf("write .note for volume %d: %w", req.VolumeId, noteErr) + } defer func() { if err != nil { @@ -163,7 +173,12 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre os.Chtimes(dataBaseFileName+".vif", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) } - os.Remove(dataBaseFileName + ".note") + // A leftover .note fails the load on the next restart, so a removal + // failure must fail the copy rather than be silently swallowed. + if noteErr := os.Remove(dataBaseFileName + ".note"); noteErr != nil && !os.IsNotExist(noteErr) { + err = noteErr + return fmt.Errorf("remove .note for volume %d: %w", req.VolumeId, noteErr) + } return nil }) diff --git a/weed/server/volume_grpc_copy_verify_test.go b/weed/server/volume_grpc_copy_verify_test.go new file mode 100644 index 000000000..ef10832ab --- /dev/null +++ b/weed/server/volume_grpc_copy_verify_test.go @@ -0,0 +1,75 @@ +package weed_server + +import ( + "context" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// fakeVolumeCopyStream is a no-op VolumeServer_VolumeCopyServer; VolumeCopy +// errors out before sending anything in this test. +type fakeVolumeCopyStream struct { + grpc.ServerStream +} + +func (s *fakeVolumeCopyStream) Send(*volume_server_pb.VolumeCopyResponse) error { return nil } +func (s *fakeVolumeCopyStream) Context() context.Context { return context.Background() } +func (s *fakeVolumeCopyStream) SetHeader(metadata.MD) error { return nil } +func (s *fakeVolumeCopyStream) SendHeader(metadata.MD) error { return nil } +func (s *fakeVolumeCopyStream) SetTrailer(metadata.MD) {} +func (s *fakeVolumeCopyStream) SendMsg(any) error { return nil } +func (s *fakeVolumeCopyStream) RecvMsg(any) error { return nil } + +// TestVolumeCopy_KeepsExistingReplicaWhenSourceUnreachable verifies the +// verify-before-destroy invariant: a pre-existing healthy local replica must +// NOT be deleted when the source cannot be reached. The pre-fix code deleted +// the destination up front (and, on retry, could lose the volume entirely); +// the fix defers the delete until the source ReadVolumeFileStatus succeeds. +func TestVolumeCopy_KeepsExistingReplicaWhenSourceUnreachable(t *testing.T) { + dir := t.TempDir() + store := storage.NewStore( + grpc.WithTransportCredentials(insecure.NewCredentials()), + "127.0.0.1", 0, 0, "", "test-store", + []string{dir}, []int32{10}, []util.MinFreeSpace{{}}, + dir, storage.NeedleMapInMemory, + []types.DiskType{types.HardDriveType}, [][]string{nil}, + 0, stats.DiskIOProbeConfig{}, + ) + + const vid = needle.VolumeId(42) + if err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("AddVolume: %v", err) + } + if store.GetVolume(vid) == nil { + t.Fatalf("setup: volume %d should exist", vid) + } + + vs := &VolumeServer{ + store: store, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + + // 127.0.0.1:1 is unreachable, so ReadVolumeFileStatus on the source fails. + req := &volume_server_pb.VolumeCopyRequest{ + VolumeId: uint32(vid), + SourceDataNode: "127.0.0.1:1", + } + err := vs.VolumeCopy(req, &fakeVolumeCopyStream{}) + if err == nil { + t.Fatalf("VolumeCopy should fail when the source is unreachable") + } + + if store.GetVolume(vid) == nil { + t.Fatalf("existing replica %d was destroyed before the source was verified", vid) + } +} diff --git a/weed/shell/command_volume_check_disk.go b/weed/shell/command_volume_check_disk.go index af5b806a3..10e22588b 100644 --- a/weed/shell/command_volume_check_disk.go +++ b/weed/shell/command_volume_check_disk.go @@ -22,7 +22,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/server/constants" "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" - "github.com/seaweedfs/seaweedfs/weed/storage/volume_replica" "google.golang.org/grpc" ) @@ -44,6 +43,12 @@ type volumeCheckDisk struct { syncDeletions bool fixReadOnly bool nonRepairThreshold float64 + // resurrectMissingNeedles controls whether a needle present on the source + // but entirely absent on the target is pushed back. Default false: an + // absent needle is indistinguishable from a vacuumed delete, so the safe + // default never raises deleted data. No caller sets this true today; it is + // the seam for a future tombstone-aware repair path. + resurrectMissingNeedles bool ewg *ErrorWaitGroup } @@ -511,9 +516,8 @@ func (vcd *volumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_map.Me // hash join, can be more efficient var missingNeedles []needle_map.NeedleValue var partiallyDeletedNeedles []needle_map.NeedleValue + var skippedAbsentNeedles int var counter int - doCutoffOfLastNeedle := true - cutoffFromAtNs := uint64(vcd.now.UnixNano()) minuend.DescendingVisit(func(minuendValue needle_map.NeedleValue) error { counter++ @@ -521,27 +525,34 @@ func (vcd *volumeCheckDisk) doVolumeCheckDisk(minuend, subtrahend *needle_map.Me if minuendValue.Size.IsDeleted() { return nil } - if doCutoffOfLastNeedle { - if needleMeta, err := volume_replica.ReadNeedleMeta(vcd.grpcDialOption(), pb.NewServerAddressFromDataNode(source.location.dataNode), source.info.Id, minuendValue); err == nil { - // needles older than the cutoff time are not missing yet - if needleMeta.AppendAtNs > cutoffFromAtNs { - return nil - } - doCutoffOfLastNeedle = false - } + // A key present-and-live on the source but entirely absent on the + // target is ambiguous: either a genuine missing write, or a needle + // that was deleted on the target and then vacuumed away (its index + // entry, including any tombstone, is gone after vacuum). An + // individual needle's AppendAtNs has no monotonic relation to a + // vacuum watermark, so it cannot distinguish the two. Without + // positive proof the absence is a missing write (rather than a + // vacuumed delete), the safe default is to NOT resurrect: a real + // missing write may go unrepaired until a tombstone-aware path + // exists, but we never raise back data the operator deleted. + if !vcd.resurrectMissingNeedles { + skippedAbsentNeedles++ + return nil } missingNeedles = append(missingNeedles, minuendValue) } else { if minuendValue.Size.IsDeleted() && !subtrahendValue.Size.IsDeleted() { partiallyDeletedNeedles = append(partiallyDeletedNeedles, minuendValue) } - if doCutoffOfLastNeedle { - doCutoffOfLastNeedle = false - } } return nil }) + if skippedAbsentNeedles > 0 { + vcd.write("volume %d %s: not resurrecting %d needle(s) absent on %s (cannot prove they are missing writes vs vacuumed deletes)", + source.info.Id, source.location.dataNode.Id, skippedAbsentNeedles, target.location.dataNode.Id) + } + vcd.write("volume %d %s has %d entries, %s missed %d and partially deleted %d entries", source.info.Id, source.location.dataNode.Id, counter, target.location.dataNode.Id, len(missingNeedles), len(partiallyDeletedNeedles)) diff --git a/weed/shell/command_volume_check_disk_test.go b/weed/shell/command_volume_check_disk_test.go index ec958fbc4..8a31e5cdf 100644 --- a/weed/shell/command_volume_check_disk_test.go +++ b/weed/shell/command_volume_check_disk_test.go @@ -6,8 +6,66 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" ) +// TestDoVolumeCheckDiskDoesNotResurrectAbsentNeedle verifies that a needle +// present-and-live on the source but entirely absent on the target is NOT +// pushed back by default. Such an absence is indistinguishable from a needle +// that was deleted on the target and then vacuumed away, so resurrecting it +// would raise back data the operator deleted. +func TestDoVolumeCheckDiskDoesNotResurrectAbsentNeedle(t *testing.T) { + sourceDB, targetDB := needle_map.NewMemDb(), needle_map.NewMemDb() + defer sourceDB.Close() + defer targetDB.Close() + + // Source has a live needle; target lacks it (e.g. deleted+vacuumed there). + if err := sourceDB.Set(types.NeedleId(1001), types.ToOffset(8), types.Size(123)); err != nil { + t.Fatalf("seed source: %v", err) + } + + var buf bytes.Buffer + vcd := &volumeCheckDisk{ + commandEnv: &CommandEnv{ + option: &ShellOptions{ + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }, + }, + writer: &buf, + now: time.Now(), + applyChanges: false, // simulation: should not even read the source blob + nonRepairThreshold: 1, + // resurrectMissingNeedles left false: the safe default. + } + + // Source points at an unreachable address: with the fix no blob read is + // attempted, so this is never contacted. The pre-fix code queues the + // absent needle and tries to read it from here, surfacing as an error. + source := &VolumeReplica{ + location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "127.0.0.1:1"}}, + info: &master_pb.VolumeInformationMessage{Id: 7}, + } + target := &VolumeReplica{ + location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "127.0.0.1:2"}}, + info: &master_pb.VolumeInformationMessage{Id: 7}, + } + + // With the safe default, the absent needle is skipped before any network + // read, so this returns cleanly with no changes. On the pre-fix code the + // needle is queued and a source blob read is attempted, which has no server + // to reach and surfaces as an error (or a resurrection) instead. + hasChanges, err := vcd.doVolumeCheckDisk(sourceDB, targetDB, source, target) + if err != nil { + t.Fatalf("doVolumeCheckDisk returned error: %v", err) + } + if hasChanges { + t.Fatalf("absent-on-target needle was resurrected; expected no changes") + } +} + type testCommandVolumeCheckDisk struct { commandVolumeCheckDisk } diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go index e4af93d28..73497e530 100644 --- a/weed/shell/command_volume_fix_replication.go +++ b/weed/shell/command_volume_fix_replication.go @@ -219,7 +219,12 @@ func collectVolumeReplicaLocations(topologyInfo *master_pb.TopologyInfo) (map[ui type SelectOneVolumeFunc func(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica -func checkOneVolume(a *VolumeReplica, b *VolumeReplica, writer io.Writer, commandEnv *CommandEnv) (err error) { +// checkOneVolume compares the index of replica a against b. With +// applyChanges=false it is a read-only divergence check; the over-replication +// trim must use that mode so it does not push the soon-to-be-deleted replica's +// needles into the survivor (which would resurrect data and is the opposite of +// a safe trim). +func checkOneVolume(a *VolumeReplica, b *VolumeReplica, writer io.Writer, commandEnv *CommandEnv, applyChanges bool) (err error) { aDB, bDB := needle_map.NewMemDb(), needle_map.NewMemDb() defer func() { aDB.Close() @@ -232,7 +237,7 @@ func checkOneVolume(a *VolumeReplica, b *VolumeReplica, writer io.Writer, comman now: time.Now(), verbose: false, - applyChanges: true, + applyChanges: applyChanges, syncDeletions: false, nonRepairThreshold: float64(1), } @@ -261,6 +266,10 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replicas[0].info.ReplicaPlacement)) replica := selectOneVolumeFn(replicas, replicaPlacement) + if replica == nil { + fmt.Fprintf(writer, "skip trimming volume %d: no safe replica to delete (would leave only read-only survivors)\n", vid) + continue + } // check collection name pattern if *c.collectionPattern != "" { @@ -302,7 +311,9 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr if replicaB.location.dataNode == replica.location.dataNode { continue } - if checkErr = checkOneVolume(replica, replicaB, writer, commandEnv); checkErr != nil { + // Read-only divergence check only: never write the doomed + // replica's needles into a survivor while trimming. + if checkErr = checkOneVolume(replica, replicaB, writer, commandEnv, false); checkErr != nil { fmt.Fprintf(writer, "sync volume %d on %s and %s: %v\n", replica.info.Id, replica.location.dataNode.Id, replicaB.location.dataNode.Id, checkErr) break } @@ -625,8 +636,20 @@ func countReplicas(replicas []*VolumeReplica) (diffDc, diffRack, diffNode map[st return } -func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica { - slices.SortFunc(replicas, func(a, b *VolumeReplica) int { +// pickOneReplicaToDelete selects the replica to trim when over-replicated. +// It only ever removes the smallest of multiple healthy writable replicas: a +// ReadOnly/integrity-flagged replica is never chosen for deletion, and the +// trim is refused (returns nil) when removing a writable replica would leave +// only ReadOnly survivors. VolumeStatus file_count>0 alone cannot prove the +// survivors' .dat is readable, so we do not over-claim survivor health. +// pickSmallestReplica returns the smallest replica (ties broken by oldest then +// lowest compact revision), or nil for an empty set. +func pickSmallestReplica(replicas []*VolumeReplica) *VolumeReplica { + if len(replicas) == 0 { + return nil + } + sorted := slices.Clone(replicas) + slices.SortFunc(sorted, func(a, b *VolumeReplica) int { if a.info.Size != b.info.Size { return int(a.info.Size - b.info.Size) } @@ -638,9 +661,39 @@ func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_b } return 0 }) + return sorted[0] +} - return replicas[0] - +func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica { + // Over-replication trim: only ever remove a writable replica, and only + // when another writable one survives, so a healthy copy is never deleted + // down to a read-only (e.g. full or integrity-flagged) survivor. + var writable []*VolumeReplica + for _, r := range replicas { + if !r.info.ReadOnly { + writable = append(writable, r) + } + } + if len(writable) < 2 { + return nil + } + // Prefer a writable replica whose removal still satisfies placement, so the + // trim does not strip the only replica in a required failure domain. Fall + // back to the smallest writable if none keeps placement (a later misplaced + // cycle then re-balances). + var placementSafe []*VolumeReplica + for i, r := range replicas { + if r.info.ReadOnly { + continue + } + if !isMisplaced(otherThan(replicas, i), replicaPlacement) { + placementSafe = append(placementSafe, r) + } + } + if len(placementSafe) > 0 { + return pickSmallestReplica(placementSafe) + } + return pickSmallestReplica(writable) } // check and fix misplaced volumes @@ -669,6 +722,10 @@ func otherThan(replicas []*VolumeReplica, index int) (others []*VolumeReplica) { func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) (toDelete *VolumeReplica) { + // Relocation, not over-replication: pick the smallest replica to delete + // and recreate at a correct placement. Unlike the trim this must still act + // on read-only replicas (e.g. a full but misplaced volume), so it does not + // use pickOneReplicaToDelete's writable-survivor guard. var deletionCandidates []*VolumeReplica for i := 0; i < len(replicas); i++ { others := otherThan(replicas, i) @@ -676,10 +733,10 @@ func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_b deletionCandidates = append(deletionCandidates, replicas[i]) } } - if len(deletionCandidates) > 0 { - return pickOneReplicaToDelete(deletionCandidates, replicaPlacement) + if toDelete = pickSmallestReplica(deletionCandidates); toDelete != nil { + return toDelete } - return pickOneReplicaToDelete(replicas, replicaPlacement) + return pickSmallestReplica(replicas) } diff --git a/weed/shell/command_volume_fix_replication_test.go b/weed/shell/command_volume_fix_replication_test.go index 5f2318c32..46d3c008b 100644 --- a/weed/shell/command_volume_fix_replication_test.go +++ b/weed/shell/command_volume_fix_replication_test.go @@ -439,6 +439,127 @@ func TestPickingMisplacedVolumeToDelete(t *testing.T) { } +func TestPickOneReplicaToDeleteSkipsReadOnlySurvivor(t *testing.T) { + replicaPlacement, _ := super_block.NewReplicaPlacementFromString("001") + + // One writable replica and one read-only replica. The trim must never + // delete the only writable replica while the survivor is read-only: + // VolumeStatus alone cannot prove the read-only .dat is readable, so the + // trim is refused. + t.Run("refuses to delete the only writable replica while a read-only survivor remains", func(t *testing.T) { + replicas := []*VolumeReplica{ + { + location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn-writable"}}, + info: &master_pb.VolumeInformationMessage{Size: 90, ReadOnly: false}, + }, + { + location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn-readonly"}}, + info: &master_pb.VolumeInformationMessage{Size: 100, ReadOnly: true}, + }, + } + if got := pickOneReplicaToDelete(replicas, replicaPlacement); got != nil { + t.Fatalf("expected no replica to delete, got %s", got.location.dataNode.Id) + } + }) + + // All survivors read-only: never trim. + t.Run("refuses when all replicas are read-only", func(t *testing.T) { + replicas := []*VolumeReplica{ + { + location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}, + info: &master_pb.VolumeInformationMessage{Size: 100, ReadOnly: true}, + }, + { + location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn2"}}, + info: &master_pb.VolumeInformationMessage{Size: 100, ReadOnly: true}, + }, + } + if got := pickOneReplicaToDelete(replicas, replicaPlacement); got != nil { + t.Fatalf("expected no replica to delete, got %s", got.location.dataNode.Id) + } + }) + + // Two healthy writable replicas plus a read-only one: trim the smallest + // writable, never the read-only one. + t.Run("trims the smallest writable replica and never the read-only one", func(t *testing.T) { + replicas := []*VolumeReplica{ + { + location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn-big"}}, + info: &master_pb.VolumeInformationMessage{Size: 100, ReadOnly: false}, + }, + { + location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn-small"}}, + info: &master_pb.VolumeInformationMessage{Size: 90, ReadOnly: false}, + }, + { + location: &location{"dc1", "r3", &master_pb.DataNodeInfo{Id: "dn-readonly"}}, + info: &master_pb.VolumeInformationMessage{Size: 80, ReadOnly: true}, + }, + } + got := pickOneReplicaToDelete(replicas, replicaPlacement) + if got == nil { + t.Fatalf("expected a writable replica to delete, got nil") + } + if got.location.dataNode.Id != "dn-small" { + t.Fatalf("expected to delete smallest writable dn-small, got %s", got.location.dataNode.Id) + } + }) +} + +// The over-replication writable-survivor guard must NOT leak into the misplaced +// relocation path: a misplaced volume whose replicas are all read-only (e.g. a +// full volume) must still be relocated, picking the smallest replica to delete +// and recreate at a correct placement. +func TestPickOneMisplacedVolumeRelocatesReadOnlyReplicas(t *testing.T) { + replicaPlacement, _ := super_block.NewReplicaPlacementFromString("001") + replicas := []*VolumeReplica{ + { + location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}, + info: &master_pb.VolumeInformationMessage{Size: 100, ReadOnly: true}, + }, + { + location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn2"}}, + info: &master_pb.VolumeInformationMessage{Size: 99, ReadOnly: true}, + }, + } + got := pickOneMisplacedVolume(replicas, replicaPlacement) + if got == nil { + t.Fatal("misplaced read-only volume must still be relocated, got nil") + } + if got.location.dataNode.Id != "dn2" { + t.Fatalf("expected to relocate smallest replica dn2, got %s", got.location.dataNode.Id) + } +} + +// The over-replication trim must not strip the only replica in a required +// failure domain: with "100" and replicas in dc1 + two in dc2, deleting the +// smallest (dc1) leaves both survivors in dc2 and violates placement, so the +// trim must instead delete a dc2 writable replica. +func TestPickOneReplicaToDeletePreservesPlacement(t *testing.T) { + replicaPlacement, _ := super_block.NewReplicaPlacementFromString("100") + replicas := []*VolumeReplica{ + { + location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dc1-writable"}}, + info: &master_pb.VolumeInformationMessage{Size: 90, ReadOnly: false}, + }, + { + location: &location{"dc2", "r2", &master_pb.DataNodeInfo{Id: "dc2-readonly"}}, + info: &master_pb.VolumeInformationMessage{Size: 80, ReadOnly: true}, + }, + { + location: &location{"dc2", "r3", &master_pb.DataNodeInfo{Id: "dc2-writable"}}, + info: &master_pb.VolumeInformationMessage{Size: 100, ReadOnly: false}, + }, + } + got := pickOneReplicaToDelete(replicas, replicaPlacement) + if got == nil { + t.Fatal("expected a replica to delete") + } + if got.location.dataNode.Id != "dc2-writable" { + t.Fatalf("expected to delete dc2-writable to preserve placement, got %s", got.location.dataNode.Id) + } +} + func TestSatisfyReplicaCurrentLocation(t *testing.T) { var tests = []testcase{ diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 33312eb49..38d41d60a 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -250,8 +250,12 @@ func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind Ne if util.FileExists(noteFile) { note, _ := os.ReadFile(noteFile) glog.Warningf("volume %s was not completed: %s", volumeName, string(note)) - removeVolumeFiles(l.Directory+"/"+volumeName, false) - removeVolumeFiles(l.IdxDirectory+"/"+volumeName, false) + // Keep the .vif when an .ecx for this vid coexists on the disk: the + // regular and EC volumes share .vif, so removing the incomplete + // regular copy must not strip the EC volume's info file. + keepVif := l.hasEcxFile(volumeName) + removeVolumeFiles(l.Directory+"/"+volumeName, keepVif) + removeVolumeFiles(l.IdxDirectory+"/"+volumeName, keepVif) return false }