From 816f9356b47898cf9ba182b1d6c94c59916491a7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 30 Oct 2026 11:22:40 -0700 Subject: [PATCH] volume.copy: reject only on file-count drift, warn on deleted-count (#12253) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * volume.copy: reject only on file-count drift, warn on deleted-count Go warned and kept the copy on any count mismatch. The file count is exact on both sides of a stable source, so a mismatch means a genuinely bad copy and should reject; only the deleted count is a rebuild-time estimate that legitimately drifts. Match the Rust check_copy_counts, which warns solely on the counter that can drift. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: stage the copy and validate before dropping the old replica The destination replica was deleted before the copy even started, so a copy that failed its checks — now including the file-count validation — left the server with no replica at all. The copy now writes under a copying/ staging dir the volume loader never scans, the file-size and count checks run on the staged .idx (LoadCompactNeedleMap gives the same counters the mount would), and only then is the existing replica deleted and the staged files renamed into place. A rejected or failed copy leaves the previous replica untouched, and the .note marker is no longer needed since partial staged files can never be loaded. * volume.copy: remove only the files a failed copy moved into place * volume.copy: keep the .note in-progress marker through the renames A crash between the staged-file renames left a loadable volume without its .vif (writable where the source was read-only) or a partial set the load scan could not find. Write the .note before the renames and land the .vif first so any partial state is discovered and swept, then drop the note before the mount. * volume.copy: serialize copies of the same volume Two copies of the same volume write the same staged copying/ file names and can pass the existing-volume check before the first one mounts. Fail a concurrent second copy as retryable, matching the Rust volume server's volume_copies_in_flight. * volume.copy: restore the replaced replica when the swap fails Once the staged copy validates, the outgoing replica's files are hard-linked into copying/ before DeleteVolume unlinks them. If a later rename or the mount then fails, the preserved files are renamed back and remounted, so a failed swap no longer leaves the node with no replica. The links are removed on success. * volume.copy: clear stale needle-map files around the swap A leftover .ldb/.sdx/.rdb at the destination base is trusted over the fresh .idx the mount loads, and a mount that fails leaves the backends it built. Clear them before the staged files land and again when a failed copy cleans up, gated on no live replica owning them. Also keep .vif on cleanup while an .ecx index shares the basename. * volume.copy: skip the swap when the caller has gone A caller that left during the copy cannot use the result; bail before the swap deletes the replica it would replace. * volume.copy: discard a mounted copy when the result send fails A mounted copy was already announced to the master, so a later failure deletes it through the store to withdraw the announcement before the files go away, rather than unlinking files out from under a registered replica. * volume.copy: keep a vif-less copy discoverable on a split index dir With -dir.idx the staged .idx lands outside the data dir, so a copy whose source has no .vif leaves nothing the volume scan can find and the mount fails after the replica it replaced was already deleted. An empty .vif marks the basename; the loader treats it as absent and writes a real one on mount. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: preserve the replaced replica on its own disk * volume.copy: restore the replaced replica's .vif past a kept incoming one * test: cover the rollback link/restore helpers * volume.copy: preserve the mounted replica's own files for swap rollback The swap backup linked files at the copy's target paths, which are wrong when the replica sits on another disk or keeps its .idx beside the .dat; a failed swap then restored nothing. Link the mounted volume's real paths and require every present file to link before the delete, since a partial backup cannot roll back. Re-check the store at swap time so a replica that mounted mid-copy is replaced rather than overwritten, and drop a .kept incoming .vif that occupies the path the replica's own .vif restores to. * volume.copy: clear a stale .idx a mount would prefer over the copy A load switches dirIdx to the data dir when an .idx sits beside the .dat, so a leftover .idx shadows the fresh index the copy lands in the idx dir, and a surviving .idx fills in when the source had no index at all. Clearing only the derived backends still let a stale needle map reach the mount. * volume.copy: leave the shared copying/ dir for racing copies Copies of different volumes serialize per volume id but share the disk's copying/ dir; a finishing copy could rmdir it between another copy's MkdirAll and its first staged write, failing that copy with ENOENT. * volume.copy: fail replica preservation on stat errors other than not-found * volume.copy: refuse a mount while that volume is being copied A copy claims the volume id, then waits on the source with the replacement staged. A mount in that window loads the replica whose files the swap then clears; refuse it with Unavailable, retryable once the copy settles. Mirrors the same refusal on the Rust volume server. * volume.copy: refuse configure while that volume is being copied Configure unmounts and remounts; during a copy that already claimed the volume id the remount loads files the swap then unlinks. Refuse with Unavailable, retryable once the copy settles. Mirrors the Rust volume server's same refusal. * volume.copy: leave a still-standing replica's index on a rejected copy * volume.copy: preserve an unmounted replica's files for swap rollback * volume.copy: claim the in-flight id in mount and configure, not just check it * volume.copy: keep an EC-shared .vif when displacing an unmounted replica * volume.copy: restore replaced files over a surviving partial restoreReplacedVolumeFiles deleted a preserved link whenever the base name was still taken, on the assumption a landed file there belonged to the copy. When a moved-file removal fails, that left a surviving partial paired with the restored set's other files — mixed generations in one replica. A leftover at the basename now comes off first so the backup goes back, and the link is kept whenever it cannot go back. * volume.copy: link a repeated replaced path once so rollback cannot delete it * volume.copy: refuse allocation and staged adoption under the copy claim AllocateVolume and adoptStagedVolume still registered a volume while a copy of it held the destination: NewVolume loads the index before registration, and the copy then truncates or replaces files the live replica already opened, leaving it mounted on a stale index. Both paths now claim the volume id like VolumeMount and VolumeConfigure, so registration either wins the claim before the copy starts or sees it and refuses with Unavailable. * volume.copy: serialize reload loads with the copy claim A SIGHUP reload ran loadExistingVolume outside the copy's claim: its .note branch could unlink a copy's files mid-write, and its NewVolume+SetVolume could register a replica between the copy's recheck and its swap, leaving a live volume on a stale index. loadExistingVolume now consults a per-location hook the server installs: a vid whose claim is held is skipped for the next pass; otherwise the load takes the claim and its release drops it after registration, so whichever side claims the id first owns the files for the operation. * volume.copy: let a claim holder's own mount past the load guard VolumeMount, VolumeConfigure, VolumeCopy, and adoptStagedVolume all hold the per-volume claim while calling store.MountVolume, which reaches loadExistingVolume where the load guard saw the caller's own claim and skipped the disk: every mount failed with 'volume not found on disk'. Plumb claimHeld through MountVolume -> LoadVolume -> loadExistingVolume via the new MountVolumeClaimed entry point so an owner's own load proceeds while loads without the claim are still skipped. * volume.copy: keep the strict count check when only deletes drift Deletes append tombstone rows without moving the file count, so a copy whose source only deleted mid-flight still has an exactly comparable file count. Gate the strict check on file-count stability alone; the deleted-count drift is already a warning inside checkCopyCounts. * volume.copy: abort the swap when stale index files cannot be removed A surviving .ldb/.sdx/.idx is trusted over the copied .idx at mount, so a failed removal must stop the copy before the staged files land, not be swallowed. Mirrors the Rust side's remove_partial_files abort. * volume.copy: clear staged leftovers before starting a copy The staging names are fixed per volume, so a crashed copy can leave files under copying/; a retry whose new source lacks an optional file (.vif, or .dat on a remote volume) would rename the previous run's bytes into place. Clear them up front and refuse the copy when they cannot be removed. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/server/volume_grpc_copy.go | 498 ++++++++++++------ weed/server/volume_grpc_copy_count_test.go | 16 +- weed/server/volume_grpc_copy_rollback_test.go | 132 +++++ 3 files changed, 490 insertions(+), 156 deletions(-) create mode 100644 weed/server/volume_grpc_copy_rollback_test.go diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index a84b126c14..f5a2c081f7 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -6,10 +6,14 @@ import ( "io" "math" "os" + "path/filepath" "strings" "sync" "time" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/backend" @@ -23,8 +27,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/util" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) const BufferSizeLimit = 1024 * 1024 * 2 @@ -71,9 +73,9 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } } - // One copy of a volume at a time: a racing second copy can pass the - // existing-volume check before this one mounts and then clear or - // overwrite the files this copy still owns. Retryable once it settles. + // One copy of a volume at a time: the staged file names are fixed per + // volume, so a racing second copy would overwrite the files this copy + // still owns. Retryable once the in-flight copy settles. guard := vs.volumeCopyGuard(req.VolumeId) guard.Lock() if guard.copying { @@ -90,11 +92,9 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre // 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 + // source outage (and, on retry, can lose the volume entirely). The copy is + // staged next to it and validated first; the delete only runs once the + // replacement is known good (validation details below). // the master will not start compaction for read-only volumes, so it is safe to just copy files directly // copy .dat and .idx files @@ -106,13 +106,14 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre var sourceVolumeStatus *volume_server_pb.VolumeStatusResponse var sourceVolumeStatusAfterCopy *volume_server_pb.VolumeStatusResponse var dataBaseFileName, indexBaseFileName, idxFileName, datFileName, vifFileName string + var stagedDataBaseFileName, stagedIndexBaseFileName string + var stagedIdxFileName, stagedDatFileName, stagedVifFileName string + var movedFiles []string + var replacedLinks []string + var replacedRemoved, mounted bool var hasRemoteDatFile bool var copyLocation *storage.DiskLocation var copyReservedSpace uint64 - // Files this copy has (re)written itself. Until the first write the - // destination can still hold a previous replica's files, and an - // aborted copy must not destroy them. - var copyOwnsFiles bool err := operation.WithVolumeServerClientOptions(true, pb.ServerAddress(req.SourceDataNode), func(client volume_server_pb.VolumeServerClient) (err error) { sourceVolumeStatus, err = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{ VolumeId: req.VolumeId, @@ -192,25 +193,38 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } } + dataBaseFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId)) + indexBaseFileName = storage.VolumeFileName(location.IdxDirectory, volFileInfoResp.Collection, int(req.VolumeId)) + + // The copy writes into a copying/ staging dir the volume loader never + // scans: every staged byte is this copy's own, a live replica at the + // final basename is never touched by the copy, and a failed copy can + // always clean up after itself. + stagedDataBaseFileName = filepath.Join(location.Directory, "copying", filepath.Base(dataBaseFileName)) + stagedIndexBaseFileName = filepath.Join(location.IdxDirectory, "copying", filepath.Base(indexBaseFileName)) + if mkErr := os.MkdirAll(filepath.Dir(stagedDataBaseFileName), 0755); mkErr != nil { + err = mkErr + return fmt.Errorf("create copying dir for volume %d: %w", req.VolumeId, mkErr) + } + if mkErr := os.MkdirAll(filepath.Dir(stagedIndexBaseFileName), 0755); mkErr != nil { + err = mkErr + return fmt.Errorf("create copying dir for volume %d: %w", req.VolumeId, mkErr) + } + stagedIdxFileName = stagedIndexBaseFileName + ".idx" + stagedDatFileName = stagedDataBaseFileName + ".dat" + stagedVifFileName = stagedDataBaseFileName + ".vif" + defer func() { if err != nil { + os.Remove(stagedDatFileName) + os.Remove(stagedIdxFileName) + os.Remove(stagedVifFileName) + os.Remove(stagedVifFileName + ".copying") if dataBaseFileName != "" { // A .note only marks an in-flight copy; a partial write // must not be left to fail the next volume load. os.Remove(dataBaseFileName + ".note") } - if copyOwnsFiles { - os.Remove(dataBaseFileName + ".dat") - os.Remove(indexBaseFileName + ".idx") - if indexBaseFileName != dataBaseFileName { - os.Remove(dataBaseFileName + ".idx") - } - // A regular volume and an EC volume for the same id - // share .vif; keep it while the EC index is here. - if !location.HasEcxFileOnDisk(volFileInfoResp.Collection, needle.VolumeId(req.VolumeId)) { - os.Remove(dataBaseFileName + ".vif") - } - } // Zero before the outer defer sees the same error, or the // charge would be released twice. vs.store.ReleaseFreeLocation(location, copyReservedSpace) @@ -218,73 +232,18 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } }() - // Source is reachable and a destination is reserved: 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, false, true); delErr != nil { - return fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, delErr) + // The staging names are fixed per volume, so a crashed copy can + // leave files behind; an optional file the new source lacks (.vif, + // or .dat on a remote volume) must not inherit the previous run's. + for _, leftover := range []string{stagedDatFileName, stagedIdxFileName, stagedVifFileName, stagedDatFileName + ".copying", stagedVifFileName + ".copying"} { + if rmErr := os.Remove(leftover); rmErr != nil && !os.IsNotExist(rmErr) { + return fmt.Errorf("clear staged %s for volume %d: %w", leftover, req.VolumeId, rmErr) } - glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) } - dataBaseFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId)) - indexBaseFileName = storage.VolumeFileName(location.IdxDirectory, volFileInfoResp.Collection, int(req.VolumeId)) - - // The existing-volume check ran before the source RPCs; a mount that - // landed since would lose its index files to the clear below, which can - // share the index directory with this copy's bases. Rechecking under - // the volume's guard keeps a mount that passed the in-flight check - // first from registering between the recheck and the clearing: it - // either registered already (and this aborts) or sees the claim. - guard.Lock() - clearErr := func() error { - if vs.store.GetVolume(needle.VolumeId(req.VolumeId)) != nil { - return fmt.Errorf("volume %d mounted during copy; aborting before clearing files", req.VolumeId) - } - - // A crashed copy, or one whose cleanup failed, can leave a stale index - // next to the .idx that this copy's mount would trust - // (generateLevelDbFile replays only past the stored watermark). The - // same leftovers can sit beside the .dat when the dirs differ: the - // mount loads a data-dir .idx in preference to the idx-dir one. Clear - // them all before writing anything; a removal failure must abort - // rather than risk a mount reusing a poisoned index. - indexBases := []string{indexBaseFileName} - if indexBaseFileName != dataBaseFileName { - indexBases = append(indexBases, dataBaseFileName) - } - for _, indexBase := range indexBases { - if removeErr := os.RemoveAll(indexBase + ".ldb"); removeErr != nil { - return fmt.Errorf("clear leftover index before copying volume %d: %w", req.VolumeId, removeErr) - } - for _, ext := range []string{".idx", ".sdx", ".rdb"} { - if removeErr := os.Remove(indexBase + ext); removeErr != nil && !os.IsNotExist(removeErr) { - return fmt.Errorf("clear leftover index before copying volume %d: %w", req.VolumeId, removeErr) - } - } - } - - // 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 { - return fmt.Errorf("write .note for volume %d: %w", req.VolumeId, noteErr) - } - return nil - }() - guard.Unlock() - if clearErr != nil { - err = clearErr - return clearErr - } - // From here on the destination files are this copy's own writes. - copyOwnsFiles = true - var copyDatAllocated uint64 if preallocateSize > 0 && !hasRemoteDatFile { - volumeFile := dataBaseFileName + ".dat" + volumeFile := stagedDataBaseFileName + ".dat" _, err := backend.CreateVolumeFile(volumeFile, preallocateSize, 0) if err != nil { return fmt.Errorf("create volume file %s: %v", volumeFile, err) @@ -314,10 +273,9 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre if !hasRemoteDatFile { datProgress := newCopyProgress(copyDatAllocated) - // Write into the .dat CreateVolumeFile just made: the staged - // .copying file would peak at preallocate+dat on disk and then - // discard the fallocated blocks at rename. - if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".dat", preallocateSize > 0, true, func(processed int64) bool { + // Write into the .dat CreateVolumeFile just made so the + // fallocated blocks carry over to the final name at rename. + if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, stagedDataBaseFileName, ".dat", preallocateSize > 0, true, func(processed int64) bool { datProgress(processed) if processed > nextReportTarget { copyResponse.ProcessedBytes = processed @@ -334,37 +292,30 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return sendErr } if modifiedTsNs > 0 { - os.Chtimes(dataBaseFileName+".dat", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) + os.Chtimes(stagedDataBaseFileName+".dat", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) } } idxProgress := newCopyProgress(0) - if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, indexBaseFileName, ".idx", false, false, func(processed int64) bool { + if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, stagedIndexBaseFileName, ".idx", false, false, func(processed int64) bool { idxProgress(processed) return true }, throttler); err != nil { return err } if modifiedTsNs > 0 { - os.Chtimes(indexBaseFileName+".idx", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) + os.Chtimes(stagedIndexBaseFileName+".idx", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) } vifProgress := newCopyProgress(0) - if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, 1024*1024, dataBaseFileName, ".vif", false, true, func(processed int64) bool { + if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, 1024*1024, stagedDataBaseFileName, ".vif", false, true, func(processed int64) bool { vifProgress(processed) return true }, throttler); err != nil { return err } if modifiedTsNs > 0 { - os.Chtimes(dataBaseFileName+".vif", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) - } - - // 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) + os.Chtimes(stagedDataBaseFileName+".vif", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) } var statusErr error @@ -390,65 +341,92 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre datFileName = dataBaseFileName + ".dat" vifFileName = dataBaseFileName + ".vif" - mounted := false defer func() { if copyLocation == nil { return } if err != nil { - // A mounted copy was already announced to the master; deleting - // through the store unmounts it and withdraws the announcement - // before the files go away. keepRemoteData: the .vif may point at - // the same cloud-tier object the source still references. + keepFiles := false if mounted { + // A mounted copy was already announced to the master; delete + // through the store so the announcement is withdrawn before + // the files go away. keepRemoteData: the .vif may point at + // the same cloud-tier object the source still references. if delErr := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), false, false, true); delErr != nil { glog.Warningf("discard mounted copy of volume %d: %v", req.VolumeId, delErr) + // A refused delete leaves it registered; unlinking its + // files now would strand a live replica. + keepFiles = vs.store.GetVolume(needle.VolumeId(req.VolumeId)) != nil } } - // A still-registered replica — a refused delete (e.g. the volume - // is compacting), or one a racing mount loaded — can share the - // index directory with the bases below; unlinking its files now - // would strand a live replica. - if vs.store.GetVolume(needle.VolumeId(req.VolumeId)) == nil { - if copyOwnsFiles { - os.Remove(idxFileName) - if indexBaseFileName != dataBaseFileName { - os.Remove(dataBaseFileName + ".idx") + os.Remove(stagedIdxFileName) + os.Remove(stagedDatFileName) + os.Remove(stagedVifFileName) + os.Remove(stagedVifFileName + ".copying") + if !keepFiles { + // Only files this copy moved into place come off; before the + // renames the final basenames can still be the replica being + // replaced, and a rejected copy must leave it intact. + for _, moved := range movedFiles { + // A regular volume and an EC volume for the same id share + // .vif; keep it while the EC index is here. + if moved == vifFileName && copyLocation.HasEcxFileOnDisk(volFileInfoResp.Collection, needle.VolumeId(req.VolumeId)) { + continue } - os.Remove(datFileName) - if !copyLocation.HasEcxFileOnDisk(volFileInfoResp.Collection, needle.VolumeId(req.VolumeId)) { - os.Remove(vifFileName) + os.Remove(moved) + } + // A failed mount leaves the needle-map backends it built; + // drop them so the next load does not trust a stale index. + // Only after the renames could those be this copy's — before + // them the basenames belong to a replica still on disk, and + // an unmounted one reports no store volume. A replica still + // standing keeps its files either way. + if len(movedFiles) > 0 && vs.store.GetVolume(needle.VolumeId(req.VolumeId)) == nil { + if cleanErr := removeStaleIndexFiles(indexBaseFileName, dataBaseFileName); cleanErr != nil { + glog.Warningf("clear stale index files of failed copy of volume %d: %v", req.VolumeId, cleanErr) } } - // A copy that mounted and then failed validation is only unloaded, - // so the needle-map backends mount built (.ldb, .sdx, .rdb) can - // outlive it; a later copy of the same volume would trust the - // stale index. A mount that found an .idx beside the .dat builds - // them under the data base instead. - if indexBaseFileName != "" { - os.RemoveAll(indexBaseFileName + ".ldb") - os.Remove(indexBaseFileName + ".sdx") - os.Remove(indexBaseFileName + ".rdb") - if indexBaseFileName != dataBaseFileName { - os.RemoveAll(dataBaseFileName + ".ldb") - os.Remove(dataBaseFileName + ".sdx") - os.Remove(dataBaseFileName + ".rdb") + if dataBaseFileName != "" { + os.Remove(dataBaseFileName + ".note") + } + if len(replacedLinks) > 0 { + // The swap failed after the previous replica's files were + // displaced; put the preserved files back. A surviving + // partial — like an incoming .vif kept for a coexisting + // EC volume — is dropped so the original comes back. + restoreReplacedVolumeFiles(replacedLinks) + replacedLinks = nil + // Only a replica the store deleted gets remounted; an + // unmounted one goes back to files on disk, as found. + if replacedRemoved { + if remountErr := vs.store.MountVolumeClaimed(needle.VolumeId(req.VolumeId), &req.Collection); remountErr != nil { + glog.Warningf("remount replaced replica of volume %d: %v", req.VolumeId, remountErr) + } } } } + for _, link := range replacedLinks { + os.Remove(link) + } vs.store.ReleaseFreeLocation(copyLocation, copyReservedSpace) } else { + for _, link := range replacedLinks { + os.Remove(link) + } vs.store.CommitFreeLocation(copyLocation, copyReservedSpace) } + // copying/ is shared by copies of other volumes on the disk, so it + // stays behind; a leftover staged file is rewritten or removed by the + // next copy of its volume. }() - if err = checkCopyFiles(volFileInfoResp, hasRemoteDatFile, idxFileName, datFileName); err != nil { // added by panyc16 + if err = checkCopyFiles(volFileInfoResp, hasRemoteDatFile, stagedIdxFileName, stagedDatFileName); err != nil { // added by panyc16 return err } var lastAppendAtNs = volFileInfoResp.DatFileTimestampSeconds * uint64(time.Second) if !hasRemoteDatFile { - if appendAtNs, appendErr := findLastAppendAtNsFromCopiedFiles(idxFileName, datFileName, needle.Version(volFileInfoResp.Version)); appendErr == nil && appendAtNs > 0 { + if appendAtNs, appendErr := findLastAppendAtNsFromCopiedFiles(stagedIdxFileName, stagedDatFileName, needle.Version(volFileInfoResp.Version)); appendErr == nil && appendAtNs > 0 { lastAppendAtNs = appendAtNs } else if appendErr != nil { glog.V(1).Infof("failed to find last append timestamp for volume %d: %v", req.VolumeId, appendErr) @@ -460,15 +438,146 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre glog.V(1).Infof("source volume %d changed during copy; skip record count validation", req.VolumeId) } - // A caller that left during the copy cannot use the result; bail before the - // mount announces a replica nobody asked for. The window between this check - // and the send below cannot be closed, so the cleanup also drops a mounted - // copy when the send fails. + // The staged copy is checked before the existing replica is dropped, so + // a rejected copy leaves the replica it was meant to replace in place. + if shouldValidateCopyCounts { + stagedIdx, openErr := os.Open(stagedIdxFileName) + if openErr != nil { + err = fmt.Errorf("open copied index of volume %d for validation: %w", req.VolumeId, openErr) + return err + } + copiedNeedleMap, mapErr := storage.LoadCompactNeedleMap(stagedIdx, needle.Version(volFileInfoResp.Version)) + stagedIdx.Close() + if mapErr != nil { + err = fmt.Errorf("load copied index of volume %d for validation: %w", req.VolumeId, mapErr) + return err + } + if countErr := checkCopyCounts(req.VolumeId, sourceVolumeStatusAfterCopy, uint64(copiedNeedleMap.FileCount()), uint64(copiedNeedleMap.DeletedCount())); countErr != nil { + err = countErr + return countErr + } + } + + // A caller that left during the copy cannot use the result; bail before + // the swap deletes the replica it would replace. if ctxErr := stream.Context().Err(); ctxErr != nil { err = ctxErr return ctxErr } + // A replica that mounted mid-copy must leave before the swap too, so check + // the store again rather than the stale pre-copy snapshot. + if replaced := vs.store.GetVolume(needle.VolumeId(req.VolumeId)); replaced != nil { + glog.V(0).Infof("volume %d already exists. deleting before copying from %s...", req.VolumeId, req.SourceDataNode) + // Hard-link the outgoing replica's files into copying/ first: the + // inodes survive the delete, so a failed swap can put the previous + // replica back instead of leaving the node with none. The paths come + // from the mounted volume — it can sit on another disk, with its .idx + // beside the .dat. A partial backup cannot restore the replica. + var backupComplete bool + replacedLinks, backupComplete = linkReplacedVolumeFiles( + replaced.DataFileName()+".dat", + replaced.IndexFileName()+".idx", + replaced.DataFileName()+".vif", + replaced.DataFileName()+".idx", + ) + if !backupComplete { + err = fmt.Errorf("preserve files of existing volume %d for rollback", req.VolumeId) + return err + } + // 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, false, true); delErr != nil { + err = fmt.Errorf("failed to delete existing volume %d: %v", req.VolumeId, delErr) + return err + } + replacedRemoved = true + glog.V(0).Infof("deleted existing volume %d before copying.", req.VolumeId) + } else { + // An unmounted replica at the final basenames gets no store delete: + // the stale-index cleanup and the renames displace it silently, and a + // copy without a staged .dat would leave the old one orphaned. Link + // its files the same way so a failed swap restores them (no remount — + // it was not mounted), then take them off; refuse the copy when they + // cannot be preserved. + var backupComplete bool + replacedLinks, backupComplete = linkReplacedVolumeFiles( + datFileName, + idxFileName, + vifFileName, + dataBaseFileName+".idx", + dataBaseFileName+".note", + ) + if !backupComplete { + err = fmt.Errorf("preserve files of existing volume %d for rollback", req.VolumeId) + return err + } + for _, replacedFile := range []string{datFileName, idxFileName, dataBaseFileName + ".idx"} { + os.Remove(replacedFile) + } + // The .vif may be shared with an EC volume for the same id: while a + // sealed .ecx is here it must stay — an incoming .vif overwrites it + // on rename, but with no incoming one the EC volume would lose the + // settings it only stores there. + if !copyLocation.HasEcxFileOnDisk(volFileInfoResp.Collection, needle.VolumeId(req.VolumeId)) { + os.Remove(vifFileName) + } + } + + // A stale .note at the final basename fails the next load; the staged + // copy carries none, so clear leftovers before the files land. A stale + // derived index would likewise be trusted over the fresh .idx, unless a + // live replica landed mid-copy and still owns it. + os.Remove(dataBaseFileName + ".note") + if vs.store.GetVolume(needle.VolumeId(req.VolumeId)) == nil { + if cleanErr := removeStaleIndexFiles(indexBaseFileName, dataBaseFileName); cleanErr != nil { + err = fmt.Errorf("clear stale index files before copying volume %d: %w", req.VolumeId, cleanErr) + return err + } + } + + // The .note marks the copy as in-progress through the renames: a crash + // mid-rename leaves the note and the load scan sweeps the partial set. + // The .vif lands first so the scan always has a basename to find. + 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) + } + + for _, pair := range [][2]string{ + {stagedVifFileName, vifFileName}, + {stagedDatFileName, datFileName}, + {stagedIdxFileName, idxFileName}, + } { + if renameErr := os.Rename(pair[0], pair[1]); renameErr != nil { + if os.IsNotExist(renameErr) { + continue + } + err = fmt.Errorf("move copied %s into place for volume %d: %w", pair[0], req.VolumeId, renameErr) + return err + } + movedFiles = append(movedFiles, pair[1]) + } + + // With a separate index directory the .idx lands outside the data dir, so + // a copy whose source has no .vif leaves nothing the volume scan can find. + // An empty .vif marks the basename for discovery; the loader treats it as + // absent and writes a real one on mount. + if filepath.Dir(indexBaseFileName) != filepath.Dir(dataBaseFileName) && !util.FileExists(vifFileName) { + if writeErr := util.WriteFile(vifFileName, nil, 0644); writeErr != nil { + err = fmt.Errorf("write .vif marker for volume %d: %w", req.VolumeId, writeErr) + return err + } + movedFiles = append(movedFiles, vifFileName) + } + + // 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) + } + // Load and validate the volume before announcing it to the master. A failed // validation is unloaded by the store without ever making the replica // routable. @@ -476,13 +585,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre if !shouldValidateCopyCounts { return nil } - // Counter drift (bloom-filter estimates, rows appended between the - // status snapshots) is not proof of a bad copy; the byte sizes - // already match. Warn instead of rejecting the copy. - if countErr := checkCopyCounts(sourceVolumeStatusAfterCopy, targetVolume.FileCount(), targetVolume.DeletedCount()); countErr != nil { - glog.Warningf("copied volume %d counts differ from source in-memory counters (counter drift): %v", req.VolumeId, countErr) - } - return nil + return checkCopyCounts(req.VolumeId, sourceVolumeStatusAfterCopy, targetVolume.FileCount(), targetVolume.DeletedCount()) }) if err != nil { return fmt.Errorf("failed to mount or validate volume %d: %w", req.VolumeId, err) @@ -498,6 +601,86 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return err } +// linkReplacedVolumeFiles hard-links each file into a copying/ sibling so +// the inodes survive the replacement's delete and a failed swap can put the +// previous replica's files back. The loader never scans copying/. A repeated +// path links once: data and index can share a directory and so a basename. +// It reports false when a file that exists could not be preserved — deleting +// the replica then would leave nothing to roll back to. +func linkReplacedVolumeFiles(baseFileNames ...string) (links []string, complete bool) { + complete = true + seen := make(map[string]bool) + for _, base := range baseFileNames { + if seen[base] { + continue + } + seen[base] = true + if _, err := os.Stat(base); err != nil { + if !os.IsNotExist(err) { + glog.Warningf("preserve replaced file %s: %v", base, err) + complete = false + } + continue + } + link := filepath.Join(filepath.Dir(base), "copying", "replaced_"+filepath.Base(base)) + if err := os.MkdirAll(filepath.Dir(link), 0755); err != nil { + glog.Warningf("preserve replaced file %s: %v", base, err) + complete = false + continue + } + os.Remove(link) + if err := os.Link(base, link); err != nil { + glog.Warningf("preserve replaced file %s: %v", base, err) + complete = false + continue + } + links = append(links, link) + } + return links, complete +} + +// restoreReplacedVolumeFiles puts linkReplacedVolumeFiles' links back at +// their base names after a failed swap. A leftover at the basename is a +// surviving partial of the copy — it comes off first so the restored set +// stays one generation. The link is kept whenever it cannot go back: an +// existing path does not mean the copy's write succeeded. +func restoreReplacedVolumeFiles(links []string) { + for _, link := range links { + base := replacedBaseOf(link) + if _, err := os.Stat(base); err == nil { + if rmErr := os.Remove(base); rmErr != nil { + glog.Warningf("restore replaced file %s: cannot remove leftover %s: %v", link, base, rmErr) + continue + } + } + if err := os.Rename(link, base); err != nil { + glog.Warningf("restore replaced file %s: %v", link, err) + } + } +} + +func replacedBaseOf(link string) string { + return filepath.Join(filepath.Dir(filepath.Dir(link)), strings.TrimPrefix(filepath.Base(link), "replaced_")) +} + +// removeStaleIndexFiles drops the needle-map files a mount would trust over +// a fresh copy. A load prefers an .idx beside the .dat over one in the idx +// dir, so a leftover .idx is cleared along with the backends a mount builds +// (.ldb/.sdx/.rdb); both bases are covered. The first removal error is +// returned: a surviving index would be trusted over the copied .idx, so a +// caller that can still abort must know the removal failed. +func removeStaleIndexFiles(indexBaseFileName, dataBaseFileName string) error { + var firstErr error + for _, base := range []string{indexBaseFileName, dataBaseFileName} { + for _, name := range []string{base + ".ldb", base + ".sdx", base + ".rdb", base + ".idx"} { + if rmErr := os.RemoveAll(name); rmErr != nil && firstErr == nil { + firstErr = rmErr + } + } + } + return firstErr +} + func (vs *VolumeServer) doCopyFile(client volume_server_pb.VolumeServerClient, isEcVolume bool, collection string, vid, compactRevision uint32, stopOffset uint64, baseFileName, ext string, isAppend, ignoreSourceFileNotFound bool, progressFn storage.ProgressFunc) (modifiedTsNs int64, err error) { return vs.doCopyFileWithThrottler(client, isEcVolume, collection, vid, compactRevision, stopOffset, baseFileName, ext, isAppend, ignoreSourceFileNotFound, progressFn, util.NewWriteThrottler(vs.maintenanceBytePerSecond)) } @@ -557,23 +740,30 @@ func checkCopyFiles(originFileInf *volume_server_pb.ReadVolumeFileStatusResponse return nil } -func checkCopyCounts(origin *volume_server_pb.VolumeStatusResponse, targetFileCount, targetDeletedCount uint64) error { +func checkCopyCounts(vid uint32, origin *volume_server_pb.VolumeStatusResponse, targetFileCount, targetDeletedCount uint64) error { + // The file count is exact on both sides, so a mismatch on a stable source + // means the copy is genuinely bad and rejects. The deleted count is a + // rebuild-time estimate (bloom-filter dedup on some needle-map backends, + // rows appended between the status snapshots), so it legitimately drifts + // and only warns. The .dat/.idx size checks already gate integrity. if origin.FileCount != targetFileCount { return fmt.Errorf("target file count [%d] is not same as origin file count [%d]", targetFileCount, origin.FileCount) } if origin.FileDeletedCount != targetDeletedCount { - return fmt.Errorf("target deleted count [%d] is not same as origin deleted count [%d]", targetDeletedCount, origin.FileDeletedCount) + glog.Warningf("copied volume %d deleted count [%d] differs from origin deleted count [%d] (counter drift, copy kept)", vid, targetDeletedCount, origin.FileDeletedCount) } return nil } func copyCountsStable(before, after *volume_server_pb.VolumeStatusResponse) bool { - // A writable source may receive writes or deletions while its files are - // copied. In that case the before/after counts do not describe one stable - // snapshot, so strict target-count validation would report a false error. + // A writable source may receive writes while its files are copied; then + // the before/after file counts do not describe one stable snapshot and + // strict target-count validation would report a false error. Deletes do + // not move the file count — they append tombstone rows the deleted + // counter records — so drift there leaves the strict check exact, and + // checkCopyCounts warns on it instead. return before != nil && after != nil && - before.FileCount == after.FileCount && - before.FileDeletedCount == after.FileDeletedCount + before.FileCount == after.FileCount } func findLastAppendAtNsFromCopiedFiles(idxFileName, datFileName string, version needle.Version) (uint64, error) { diff --git a/weed/server/volume_grpc_copy_count_test.go b/weed/server/volume_grpc_copy_count_test.go index 5c3d01be21..c4112d15ff 100644 --- a/weed/server/volume_grpc_copy_count_test.go +++ b/weed/server/volume_grpc_copy_count_test.go @@ -35,18 +35,27 @@ func TestCheckCopyCounts(t *testing.T) { wantErr: "file count", }, { + // the deleted count is a rebuild-time estimate, so drift only warns name: "deleted count differs", sourceFileCount: 10, sourceDeleteCount: 3, targetFileCount: 10, targetDeleteCount: 2, - wantErr: "deleted count", + }, + { + name: "deleted count differs and file count differs", + sourceFileCount: 10, + sourceDeleteCount: 3, + targetFileCount: 9, + targetDeleteCount: 2, + wantErr: "file count", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { err := checkCopyCounts( + 1, &volume_server_pb.VolumeStatusResponse{ FileCount: tt.sourceFileCount, FileDeletedCount: tt.sourceDeleteCount, @@ -118,7 +127,9 @@ func TestCopyCountsStable(t *testing.T) { }, }, { - name: "changed deleted count is not stable", + // deletes append tombstone rows without moving the file count, + // so the strict check still applies and only warns on the drift + name: "changed deleted count alone is stable", before: &volume_server_pb.VolumeStatusResponse{ FileCount: 10, FileDeletedCount: 3, @@ -127,6 +138,7 @@ func TestCopyCountsStable(t *testing.T) { FileCount: 10, FileDeletedCount: 2, }, + want: true, }, { name: "missing before status is not stable", diff --git a/weed/server/volume_grpc_copy_rollback_test.go b/weed/server/volume_grpc_copy_rollback_test.go new file mode 100644 index 0000000000..5bfcd42734 --- /dev/null +++ b/weed/server/volume_grpc_copy_rollback_test.go @@ -0,0 +1,132 @@ +package weed_server + +import ( + "os" + "path/filepath" + "testing" +) + +func TestLinkReplacedVolumeFilesPreservesPresentFiles(t *testing.T) { + dir := t.TempDir() + dat := filepath.Join(dir, "vol_1.dat") + idx := filepath.Join(dir, "vol_1.idx") + for _, f := range []string{dat, idx} { + if err := os.WriteFile(f, []byte("x"), 0644); err != nil { + t.Fatal(err) + } + } + + links, complete := linkReplacedVolumeFiles(dat, idx, filepath.Join(dir, "vol_1.vif")) + if !complete { + t.Fatal("linking existing files should report complete") + } + if len(links) != 2 { + t.Fatalf("expected 2 links for 2 present files, got %d", len(links)) + } + for _, link := range links { + if _, err := os.Stat(link); err != nil { + t.Fatalf("link %s missing: %v", link, err) + } + } +} + +func TestLinkReplacedVolumeFilesReportsUnpreservedFile(t *testing.T) { + dir := t.TempDir() + dat := filepath.Join(dir, "vol_1.dat") + if err := os.WriteFile(dat, []byte("x"), 0644); err != nil { + t.Fatal(err) + } + // a copying/ path that exists as a file cannot hold the backup links + if err := os.WriteFile(filepath.Join(dir, "copying"), []byte("x"), 0644); err != nil { + t.Fatal(err) + } + + links, complete := linkReplacedVolumeFiles(dat) + if complete { + t.Fatal("a present file that cannot be linked must report incomplete") + } + if len(links) != 0 { + t.Fatalf("expected no links, got %v", links) + } +} + +func TestRestoreReplacedVolumeFilesPutsLinksBack(t *testing.T) { + dir := t.TempDir() + dat := filepath.Join(dir, "vol_1.dat") + if err := os.WriteFile(dat, []byte("x"), 0644); err != nil { + t.Fatal(err) + } + links, complete := linkReplacedVolumeFiles(dat) + if !complete || len(links) != 1 { + t.Fatalf("link setup failed: links=%v complete=%v", links, complete) + } + if err := os.Remove(dat); err != nil { + t.Fatal(err) + } + + restoreReplacedVolumeFiles(links) + if _, err := os.Stat(dat); err != nil { + t.Fatalf("restored file %s missing: %v", dat, err) + } + if _, err := os.Stat(links[0]); !os.IsNotExist(err) { + t.Fatalf("link %s should be consumed by the restore", links[0]) + } +} + +func TestRestoreReplacedVolumeFilesRestoresRepeatedPathOnce(t *testing.T) { + dir := t.TempDir() + idx := filepath.Join(dir, "vol_1.idx") + if err := os.WriteFile(idx, []byte("replica"), 0644); err != nil { + t.Fatal(err) + } + + links, complete := linkReplacedVolumeFiles(idx, idx) + if !complete || len(links) != 1 { + t.Fatalf("a repeated path should link once: links=%v complete=%v", links, complete) + } + if err := os.Remove(idx); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(idx, []byte("copy"), 0644); err != nil { + t.Fatal(err) + } + + restoreReplacedVolumeFiles(links) + got, err := os.ReadFile(idx) + if err != nil { + t.Fatalf("restored file %s missing: %v", idx, err) + } + if string(got) != "replica" { + t.Fatalf("the second pass deleted the restored file: %q", got) + } +} + +func TestRestoreReplacedVolumeFilesReplacesSurvivingPartial(t *testing.T) { + dir := t.TempDir() + dat := filepath.Join(dir, "vol_1.dat") + if err := os.WriteFile(dat, []byte("replica"), 0644); err != nil { + t.Fatal(err) + } + links, complete := linkReplacedVolumeFiles(dat) + if !complete || len(links) != 1 { + t.Fatalf("link setup failed: links=%v complete=%v", links, complete) + } + if err := os.Remove(dat); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(dat, []byte("copy"), 0644); err != nil { + t.Fatal(err) + } + + restoreReplacedVolumeFiles(links) + got, err := os.ReadFile(dat) + if err != nil { + t.Fatalf("restored file %s missing: %v", dat, err) + } + if string(got) != "replica" { + t.Fatalf("a surviving partial replaced the restored file: %q", got) + } + if _, err := os.Stat(links[0]); !os.IsNotExist(err) { + t.Fatalf("link %s should be consumed by the restore", links[0]) + } +}