From cc04c805a4f11e9512c0cd2dd1e360d80933e1f3 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 30 Oct 2026 09:20:55 -0700 Subject: [PATCH] volume.copy: drop the needle-map index a failed copy leaves behind (#12252) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * volume.copy: drop the needle-map index a failed copy leaves behind A VolumeCopy that mounts and then fails validation is only unloaded, not deleted, so a leveldb (.ldb) or sorted-file (.sdx) index built by the mount outlives the copy. The next copy of the same volume to this server then trusts the stale index (generateLevelDbFile replays only past its watermark), which is wrong once the source has compacted in between. Mirrors the Rust server's remove_partial_files. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: clear a stale index before a copy writes A crashed copy, or one whose cleanup failed, leaves a stale .ldb/.sdx/.rdb next to the .idx that this copy's mount would trust (generateLevelDbFile replays only past the stored watermark, and a same-size .idx keeps every un-replayed offset). Abort on a removal error rather than risk a mount reusing a poisoned index. Mirrors the Rust server's pre-write remove_partial_files call. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: keep a mounted replica when the final response send fails Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: discard a mounted replica the caller abandoned A send failure after MountVolume means the caller is gone, but the replica was already announced to the master. Deleting its files under the live mount breaks the loaded index; leaving it mounted keeps an orphan the master routes to. Mirror the Rust copy cleanup: track the mount, and on error delete through the store (unmount + withdraw the announcement + remove files, keeping remote data), and check the stream context before mounting. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: serialize copies of a volume on one server A second VolumeCopy of the same volume could pass the existing-volume check before the first copy mounts, then clear or overwrite the files the first copy still owns. Mirror ecDecodesInFlight: the claim covers the whole copy, and a duplicate fails fast so the caller can retry. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: remove only files the copy owns, and every stale index Three cleanup gaps around which files a copy may remove: - The failure cleanup ran on any error, so an abort before the first write (a failed stale-index clear or a failed .note write) could delete a previous replica's files the copy had not touched. File removal is now gated on the copy having started writing. - A mount that finds an .idx beside the .dat builds the needle-map index under the data base, so leftovers there poison the next copy the same way the idx-dir ones do. Clear and clean up .idx/.ldb/ .sdx/.rdb under both bases when the dirs differ. - A regular volume and an EC volume for the same id share .vif. Keep it while the EC index is on the disk, as the destroy path does. - When the discard of a mounted copy is refused (e.g. it is compacting) the volume stays registered; unlinking its files would strand a live replica. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: do not unlink a previous replica on an early abort An abort before the first write now leaves the destination files alone, but three removal paths still leaked past that gate: a partially-written .note, and the post-copy defer's .idx/.vif at the data base. Remove the .note unconditionally (it only ever marks an in-flight copy) and gate the data-file removals in the outer defer on copyOwnsFiles too. * volume.copy: refuse a mount or configure while a copy is in flight A copy claims the volume id, then waits on the source before it clears or overwrites the destination files. A mount or configure landing in that window loads the replica whose index files the copy is about to remove. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: abort when a replica registers during the source RPCs The existing-volume snapshot runs before the copy calls the source, so a mount that lands in between escapes it. Recheck registration just before the destructive clear, which can share the index directory with the mounted replica's bases. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: keep files whenever a replica of the volume stays registered The cleanup already kept files after a refused delete; a replica a racing mount registered during the copy needs the same protection, since its index directory can coincide with the bases being cleaned. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: skip the index cleanup when the copy never picked bases An early failure leaves the base names empty, so the removals would otherwise run against ".ldb"/".sdx"/".rdb" in the process directory. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: serialize mount and configure registration under a per-volume guard The in-flight flag check in VolumeMount and VolumeConfigure was not atomic with volume registration, and the copy's registration recheck was not atomic with its stale-index clearing. A mount that passed the check just before the copy claimed the volume could register after the recheck and then lose its index files to the clearing. Replace the bare flag with a per-volume guard: the copy claims it for the copy's life, mounts and configures hold it across check plus registration, and the copy holds it across the recheck plus clearing. Whichever side takes the guard first wins, so a mount either registers before the recheck (and aborts the copy) or sees the claim and refuses. Guards persist in the map so a mutex handed to a waiter is never replaced mid-wait. * volume.copy: refuse allocation and staged adoption under the copy guard 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 take the per-volume guard through their registration, so they either register before the copy's recheck (and make it abort) or see the claim and refuse with Unavailable. * volume.copy: serialize reload loads with the per-volume guard A SIGHUP reload ran loadExistingVolume outside the copy's guard: 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 file clearing, leaving a live volume on a stale index. loadExistingVolume now consults a per-location hook the server installs: a vid with a copy in flight (or whose guard the copy currently holds) is skipped for the next pass; otherwise the returned release keeps the guard held through registration, so the copy's recheck either sees the volume and aborts or clears first. * volume.copy: skip compaction-state reconcile while a copy owns the volume * volume.copy: let a guarded operation's own mount past the load guard VolumeMount, VolumeConfigure, adoptStagedVolume hold the volume guard locked, and VolumeCopy holds copying=true, while calling store.MountVolume. The mount reaches loadExistingVolume where the new load guard saw the caller's own hold (TryLock fails, or copying set) and skipped the disk: every guarded mount failed 'volume not found on disk' — a completed copy deleted its copied files, configure left the volume unmounted, staged adoption failed after renaming. Plumb ownLoad through MountVolume -> LoadVolume -> loadExistingVolume via the new MountVolumeClaimed entry point so the owner's own load proceeds while unrelated loads are still skipped. * volume.copy: fix own-load test call, assert unguarded mounts still skip Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume.copy: refuse unmount and delete while a copy owns the volume --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/server/volume_grpc_admin.go | 56 +++++- weed/server/volume_grpc_copy.go | 178 ++++++++++++++++-- weed/server/volume_grpc_copy_inflight_test.go | 104 ++++++++++ weed/server/volume_grpc_erasure_coding.go | 13 +- weed/server/volume_server.go | 18 +- weed/storage/disk_location.go | 47 ++++- weed/storage/disk_location_ec_test.go | 2 +- weed/storage/disk_location_scan_order_test.go | 61 +++++- weed/storage/store.go | 21 ++- weed/storage/volume_vacuum_crash_safe_test.go | 37 ++++ 10 files changed, 505 insertions(+), 32 deletions(-) create mode 100644 weed/server/volume_grpc_copy_inflight_test.go diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index d4d96c4124..634d56cf3c 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -101,6 +101,15 @@ func (vs *VolumeServer) AllocateVolume(ctx context.Context, req *volume_server_p return resp, err } + // Same window as VolumeMount: allocation registers the volume, and a + // copy in flight clears or rewrites the files it would open. + guard := vs.volumeCopyGuard(req.VolumeId) + guard.Lock() + defer guard.Unlock() + if guard.copying { + return resp, status.Errorf(codes.Unavailable, "volume %d is being copied", req.VolumeId) + } + err := vs.store.AddVolume( needle.VolumeId(req.VolumeId), req.Collection, @@ -132,7 +141,19 @@ func (vs *VolumeServer) VolumeMount(ctx context.Context, req *volume_server_pb.V return resp, err } - err := vs.store.MountVolume(needle.VolumeId(req.VolumeId), req.Collection) + // A copy claims the volume, then waits on the source. A mount in that + // window loads files the copy is about to clear or overwrite, so the + // check and the registration share the copy's guard: a mount either + // registers before the copy's recheck (and makes it abort) or sees the + // claim and refuses. + guard := vs.volumeCopyGuard(req.VolumeId) + guard.Lock() + if guard.copying { + guard.Unlock() + return resp, status.Errorf(codes.Unavailable, "volume %d is being copied", req.VolumeId) + } + err := vs.store.MountVolumeClaimed(needle.VolumeId(req.VolumeId), req.Collection) + guard.Unlock() if err != nil { glog.Errorf("volume mount %v: %v", req, err) @@ -152,6 +173,16 @@ func (vs *VolumeServer) VolumeUnmount(ctx context.Context, req *volume_server_pb return resp, err } + // Same window as VolumeMount: the copy announces its replica before the + // claim drops, and an unmount inside that window unregisters a volume the + // master was just told is here. + guard := vs.volumeCopyGuard(req.VolumeId) + guard.Lock() + defer guard.Unlock() + if guard.copying { + return resp, status.Errorf(codes.Unavailable, "volume %d is being copied", req.VolumeId) + } + err := vs.store.UnmountVolume(needle.VolumeId(req.VolumeId)) if err != nil { @@ -198,6 +229,15 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb. return resp, err } + // Same window as VolumeMount: a delete inside the copy's post-mount + // window removes a replica the master was just told is here. + guard := vs.volumeCopyGuard(req.VolumeId) + guard.Lock() + defer guard.Unlock() + if guard.copying { + return resp, status.Errorf(codes.Unavailable, "volume %d is being copied", req.VolumeId) + } + err := vs.store.DeleteVolume(needle.VolumeId(req.VolumeId), req.OnlyEmpty, req.OnlyGarbage, req.KeepRemoteData) if err != nil { @@ -245,6 +285,16 @@ func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_ return resp, nil } + // Same window as VolumeMount, and wider: configure unmounts, rewrites + // the .vif, and remounts, so the whole sequence shares the copy's guard + // rather than only the entry check. + guard := vs.volumeCopyGuard(req.VolumeId) + guard.Lock() + defer guard.Unlock() + if guard.copying { + return resp, status.Errorf(codes.Unavailable, "volume %d is being copied", req.VolumeId) + } + // unmount if err := vs.store.UnmountVolume(needle.VolumeId(req.VolumeId)); err != nil { glog.Errorf("volume configure unmount %v: %v", req, err) @@ -257,7 +307,7 @@ func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_ glog.Errorf("volume configure %v: %v", req, err) resp.Error = fmt.Sprintf("volume configure %v: %v", req, err) // Try to re-mount to restore the volume state - if mountErr := vs.store.MountVolume(needle.VolumeId(req.VolumeId), nil); mountErr != nil { + if mountErr := vs.store.MountVolumeClaimed(needle.VolumeId(req.VolumeId), nil); mountErr != nil { glog.Errorf("volume configure failed to restore mount %v: %v", req, mountErr) resp.Error += fmt.Sprintf(". Also failed to restore mount: %v", mountErr) } @@ -265,7 +315,7 @@ func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_ } // mount - if err := vs.store.MountVolume(needle.VolumeId(req.VolumeId), nil); err != nil { + if err := vs.store.MountVolumeClaimed(needle.VolumeId(req.VolumeId), nil); err != nil { glog.Errorf("volume configure mount %v: %v", req, err) resp.Error = fmt.Sprintf("volume configure mount %v: %v", req, err) return resp, nil diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 55bd0d5bbf..a84b126c14 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -7,6 +7,7 @@ import ( "math" "os" "strings" + "sync" "time" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -22,10 +23,40 @@ 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 +// volumeCopyGuard serializes a copy's registration recheck and file +// clearing against mounts and configures of the same volume. +type volumeCopyGuard struct { + sync.Mutex + copying bool +} + +func (vs *VolumeServer) volumeCopyGuard(volumeId uint32) *volumeCopyGuard { + guard, _ := vs.volumeCopyGuards.LoadOrStore(volumeId, &volumeCopyGuard{}) + return guard.(*volumeCopyGuard) +} + +// volumeLoadGuard is the storage-layer hook for volume loads: a vid with a +// copy in flight is skipped, otherwise the returned release keeps the load's +// registration ordered against the copy's own recheck. A copy holding the +// guard means the same skip. +func (vs *VolumeServer) volumeLoadGuard(vid needle.VolumeId) (func(), bool) { + guard := vs.volumeCopyGuard(uint32(vid)) + if !guard.TryLock() { + return func() {}, true + } + if guard.copying { + guard.Unlock() + return func() {}, true + } + return guard.Unlock, false +} + // VolumeCopy copy the .idx .dat .vif files, and mount the volume func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stream volume_server_pb.VolumeServer_VolumeCopyServer) error { if err := vs.checkGrpcAdminAuth(stream.Context()); err != nil { @@ -40,6 +71,23 @@ 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. + guard := vs.volumeCopyGuard(req.VolumeId) + guard.Lock() + if guard.copying { + guard.Unlock() + return status.Errorf(codes.Unavailable, "volume %d is already being copied", req.VolumeId) + } + guard.copying = true + guard.Unlock() + defer func() { + guard.Lock() + guard.copying = false + guard.Unlock() + }() + // 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 @@ -57,10 +105,14 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre var volFileInfoResp *volume_server_pb.ReadVolumeFileStatusResponse var sourceVolumeStatus *volume_server_pb.VolumeStatusResponse var sourceVolumeStatusAfterCopy *volume_server_pb.VolumeStatusResponse - var dataBaseFileName, indexBaseFileName, idxFileName, datFileName string + var dataBaseFileName, indexBaseFileName, idxFileName, datFileName, vifFileName string 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, @@ -143,10 +195,21 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre defer func() { if err != nil { 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") - os.Remove(dataBaseFileName + ".vif") - os.Remove(dataBaseFileName + ".note") + 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. @@ -170,12 +233,54 @@ 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 .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) + // 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 { @@ -283,15 +388,54 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre idxFileName = indexBaseFileName + ".idx" datFileName = dataBaseFileName + ".dat" + vifFileName = dataBaseFileName + ".vif" + mounted := false defer func() { if copyLocation == nil { return } if err != nil { - os.Remove(idxFileName) - os.Remove(datFileName) - os.Remove(dataBaseFileName + ".vif") + // 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. + if mounted { + 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 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(datFileName) + if !copyLocation.HasEcxFileOnDisk(volFileInfoResp.Collection, needle.VolumeId(req.VolumeId)) { + os.Remove(vifFileName) + } + } + // 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") + } + } + } vs.store.ReleaseFreeLocation(copyLocation, copyReservedSpace) } else { vs.store.CommitFreeLocation(copyLocation, copyReservedSpace) @@ -316,10 +460,19 @@ 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. + if ctxErr := stream.Context().Err(); ctxErr != nil { + err = ctxErr + return ctxErr + } + // 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. - err = vs.store.MountVolume(needle.VolumeId(req.VolumeId), &req.Collection, func(targetVolume *storage.Volume) error { + err = vs.store.MountVolumeClaimed(needle.VolumeId(req.VolumeId), &req.Collection, func(targetVolume *storage.Volume) error { if !shouldValidateCopyCounts { return nil } @@ -334,6 +487,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre if err != nil { return fmt.Errorf("failed to mount or validate volume %d: %w", req.VolumeId, err) } + mounted = true if err = stream.Send(&volume_server_pb.VolumeCopyResponse{ LastAppendAtNs: lastAppendAtNs, diff --git a/weed/server/volume_grpc_copy_inflight_test.go b/weed/server/volume_grpc_copy_inflight_test.go new file mode 100644 index 0000000000..843456fcfb --- /dev/null +++ b/weed/server/volume_grpc_copy_inflight_test.go @@ -0,0 +1,104 @@ +package weed_server + +import ( + "context" + "testing" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestVolumeMountRefusesAVolumeBeingCopied(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + guard := vs.volumeCopyGuard(17) + guard.copying = true + + _, err := vs.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{VolumeId: 17}) + assert.Equal(t, codes.Unavailable, status.Code(err), err) + + guard.copying = false + _, err = vs.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{VolumeId: 17}) + assert.NotEqual(t, codes.Unavailable, status.Code(err), err) +} + +func TestVolumeConfigureRefusesAVolumeBeingCopied(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + vs.volumeCopyGuard(17).copying = true + + _, err := vs.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{VolumeId: 17, Replication: "000"}) + assert.Equal(t, codes.Unavailable, status.Code(err), err) +} + +func TestAllocateVolumeRefusesAVolumeBeingCopied(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + vs.volumeCopyGuard(17).copying = true + + _, err := vs.AllocateVolume(context.Background(), &volume_server_pb.AllocateVolumeRequest{VolumeId: 17}) + assert.Equal(t, codes.Unavailable, status.Code(err), err) +} + +func TestVolumeUnmountRefusesAVolumeBeingCopied(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + vs.volumeCopyGuard(17).copying = true + + _, err := vs.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{VolumeId: 17}) + assert.Equal(t, codes.Unavailable, status.Code(err), err) +} + +func TestVolumeDeleteRefusesAVolumeBeingCopied(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + vs.volumeCopyGuard(17).copying = true + + _, err := vs.VolumeDelete(context.Background(), &volume_server_pb.VolumeDeleteRequest{VolumeId: 17}) + assert.Equal(t, codes.Unavailable, status.Code(err), err) +} + +func TestAdoptStagedVolumeRefusesAVolumeBeingCopied(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + vs.volumeCopyGuard(17).copying = true + + _, err := vs.adoptStagedVolume(&volume_server_pb.VolumeEcShardsToVolumeRequest{VolumeId: 17}) + assert.Equal(t, codes.Unavailable, status.Code(err), err) +} + +func TestVolumeMountSucceedsWhileHoldingTheGuard(t *testing.T) { + dir := t.TempDir() + volume, err := storage.NewVolume(dir, dir, "", 17, storage.NeedleMapInMemory, + &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + volume.Close() + + vs := &VolumeServer{store: newTraversalTestStore(dir)} + vs.store.SetVolumeLoadGuard(vs.volumeLoadGuard) + + _, err = vs.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{VolumeId: 17}) + assert.NoError(t, err) +} + +func TestVolumeLoadGuardSkipsACopyInFlight(t *testing.T) { + vs := &VolumeServer{store: newTraversalTestStore(t.TempDir())} + guard := vs.volumeCopyGuard(17) + guard.copying = true + + release, skip := vs.volumeLoadGuard(17) + release() + assert.True(t, skip) + + guard.copying = false + release, skip = vs.volumeLoadGuard(17) + assert.False(t, skip) + release() + + // A copy's own recheck holds the guard; the load must not slip inside. + guard.Lock() + _, skip = vs.volumeLoadGuard(17) + guard.Unlock() + assert.True(t, skip) +} diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 0fb0e32e96..86842309e3 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -1389,6 +1389,17 @@ func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_ // as an EC/normal twin in one directory. func (vs *VolumeServer) adoptStagedVolume(req *volume_server_pb.VolumeEcShardsToVolumeRequest) (*volume_server_pb.VolumeEcShardsToVolumeResponse, error) { vid := needle.VolumeId(req.VolumeId) + + // Same window as VolumeMount: the renames and the mount below register + // the volume, and an in-flight copy of it rewrites the destination + // files. The check shares the copy's guard through the mount. + guard := vs.volumeCopyGuard(req.VolumeId) + guard.Lock() + defer guard.Unlock() + if guard.copying { + return nil, status.Errorf(codes.Unavailable, "volume %d is being copied", req.VolumeId) + } + if vs.store.GetVolume(vid) != nil { return nil, fmt.Errorf("staged volume %d already exists on this server", req.VolumeId) } @@ -1433,7 +1444,7 @@ func (vs *VolumeServer) adoptStagedVolume(req *volume_server_pb.VolumeEcShardsTo } os.Remove(noteFile) - if err := vs.store.MountVolume(vid, &req.Collection); err != nil { + if err := vs.store.MountVolumeClaimed(vid, &req.Collection); err != nil { return nil, fmt.Errorf("mount staged volume %d: %w", req.VolumeId, err) } glog.V(0).Infof("VolumeEcShardsToVolume: adopted decoded volume %d from staging (%s)", req.VolumeId, base) diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index e4814f6752..bb5ab57016 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -46,12 +46,12 @@ type VolumeServer struct { // by a leader; before first registration the retry loop reconnects // promptly instead of pacing a full pulse. confirmedLeaderOnce atomic.Bool - pulsePeriod time.Duration - dataCenter string - rack string - store *storage.Store - guard *security.Guard - grpcDialOption grpc.DialOption + pulsePeriod time.Duration + dataCenter string + rack string + store *storage.Store + guard *security.Guard + grpcDialOption grpc.DialOption needleMapKind storage.NeedleMapKind ldbTimout int64 @@ -72,6 +72,11 @@ type VolumeServer struct { // stop the handler; this keeps a retry from racing it on the same // volume files. ecDecodesInFlight sync.Map + // A copy claims its volume's guard, so mounts and configures either + // register before the copy's registration recheck (and abort it) or see + // the claim and refuse. Entries are never removed: a guard handed out + // while one side waits must stay the same mutex. + volumeCopyGuards sync.Map } func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, @@ -153,6 +158,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, vs.checkWithMaster() vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout, diskProbeConfig) + vs.store.SetVolumeLoadGuard(vs.volumeLoadGuard) vs.store.MasterFn = vs.getCurrentMaster vs.store.SetSyncBatchOption(syncBatch) vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 72d8783785..967632f51d 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -60,6 +60,14 @@ type DiskLocation struct { ecShardNotifyHandler func(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume) + // loadGuard serializes a volume's load-and-register against an in-flight + // VolumeCopy of the same id on the server. When set, the hook runs before + // any file cleanup in loadExistingVolume and its release is held through + // SetVolume; a skipped vid is left for the next pass. An explicit mount + // is the claim holder's own load and passes ownLoad instead: its hold on + // the guard already serializes it, and consulting the hook would skip it. + loadGuard func(vid needle.VolumeId) (release func(), skip bool) + isDiskSpaceLow atomic.Bool isDiskUnavailable atomic.Bool // readOnlyMount: dir or idxDir is on a read-only filesystem. Volumes load @@ -239,7 +247,7 @@ func vifIsEcVolume(vifPath string) bool { return err == nil && vi.GetEcShardConfig() != nil } -func (l *DiskLocation) loadExistingVolume(basename string, needleMapKind NeedleMapKind, skipIfEcVolumesExists bool, ldbTimeout int64, diskId uint32) bool { +func (l *DiskLocation) loadExistingVolume(basename string, needleMapKind NeedleMapKind, skipIfEcVolumesExists bool, ldbTimeout int64, diskId uint32, ownLoad bool) bool { volumeName := getValidVolumeName(basename) if volumeName == "" { return false @@ -252,6 +260,20 @@ func (l *DiskLocation) loadExistingVolume(basename string, needleMapKind NeedleM return false } + // A copy of this volume may be mid-flight: the .note cleanup below would + // unlink files it is still writing, and a register here could slip past + // its own recheck. Skip the volume while a copy owns it and hold the + // guard through registration so the copy's recheck either sees the + // volume (and aborts) or runs first. The claim holder's own load skips + // the consult; its hold on the guard is the serialization. + if l.loadGuard != nil && !ownLoad { + release, skip := l.loadGuard(vid) + if skip { + return false + } + defer release() + } + // Sweep a leftover empty .dat stub before any EC presence checks below. // It must go first: next to an .ecx it would otherwise make // validateEcVolume mistake a healthy distributed EC volume for an @@ -380,7 +402,7 @@ func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapKind, con go func() { defer wg.Done() for basename := range task_queue { - _ = l.loadExistingVolume(basename, needleMapKind, true, ldbTimeout, diskId) + _ = l.loadExistingVolume(basename, needleMapKind, true, ldbTimeout, diskId, false) } }() } @@ -470,10 +492,25 @@ func (l *DiskLocation) reconcileCompactStates() { if loaded { continue } + // A VolumeCopy of this id can be mid-flight at the same basenames: + // rolling a leftover commit forward would rename compaction temps + // over the files it is writing. Same skip the load path takes; a + // later pass reconciles the leftover. + var release func() + if l.loadGuard != nil { + var skip bool + release, skip = l.loadGuard(k.vid) + if skip { + continue + } + } v := &Volume{dir: l.Directory, dirIdx: l.IdxDirectory, Collection: k.collection, Id: k.vid} if err := v.reconcileCompactState(); err != nil { glog.Errorf("volume %d: reconcile interrupted compaction failed: %v", k.vid, err) } + if release != nil { + release() + } } } @@ -549,17 +586,17 @@ func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId, onlyEmpty bool, onl return } -func (l *DiskLocation) LoadVolume(diskId uint32, vid needle.VolumeId, needleMapKind NeedleMapKind, collection *string) bool { +func (l *DiskLocation) LoadVolume(diskId uint32, vid needle.VolumeId, needleMapKind NeedleMapKind, collection *string, ownLoad bool) bool { if collection != nil { for _, ext := range []string{".vif", ".idx"} { filename := VolumeFileName(l.Directory, *collection, int(vid)) + ext if fi, err := os.Stat(filename); err == nil && !fi.IsDir() { - return l.loadExistingVolume(fi.Name(), needleMapKind, false, 0, diskId) + return l.loadExistingVolume(fi.Name(), needleMapKind, false, 0, diskId, ownLoad) } } } if fileInfo, found := l.LocateVolume(vid); found { - return l.loadExistingVolume(fileInfo.Name(), needleMapKind, false, 0, diskId) + return l.loadExistingVolume(fileInfo.Name(), needleMapKind, false, 0, diskId, ownLoad) } return false } diff --git a/weed/storage/disk_location_ec_test.go b/weed/storage/disk_location_ec_test.go index 4c3e77a758..335a09e190 100644 --- a/weed/storage/disk_location_ec_test.go +++ b/weed/storage/disk_location_ec_test.go @@ -723,7 +723,7 @@ func TestLoadExistingVolumeSkipsVifWhenEcxPresent(t *testing.T) { t.Fatalf("write .ecx: %v", err) } - loaded := diskLocation.loadExistingVolume(filepath.Base(vifPath), NeedleMapInMemory, false, 0, 0) + loaded := diskLocation.loadExistingVolume(filepath.Base(vifPath), NeedleMapInMemory, false, 0, 0, false) if loaded { t.Fatalf("loadExistingVolume should refuse to load a .vif-only entry when .ecx is present (volume %d)", vid) } diff --git a/weed/storage/disk_location_scan_order_test.go b/weed/storage/disk_location_scan_order_test.go index 10724fa1ec..fc2d8ca20c 100644 --- a/weed/storage/disk_location_scan_order_test.go +++ b/weed/storage/disk_location_scan_order_test.go @@ -28,7 +28,7 @@ func TestScanOrderDoesNotDecideWhetherAVolumeLoads(t *testing.T) { t.Fatal(err) } - location.loadExistingVolume("9"+ext, NeedleMapInMemory, true, 0, 0) + location.loadExistingVolume("9"+ext, NeedleMapInMemory, true, 0, 0, false) _, found := location.FindVolume(needle.VolumeId(9)) loaded[ext] = found } @@ -40,3 +40,62 @@ func TestScanOrderDoesNotDecideWhetherAVolumeLoads(t *testing.T) { t.Error("a volume with an .idx beside its .vif did not load") } } + +// A volume mid-copy must not be picked up by a reload: the copy clears and +// rewrites its files, so registering now pairs a mounted volume with data +// that is still changing, and the .note cleanup would unlink the copy's +// own output. +func TestLoadExistingVolumeSkipsACopyInFlight(t *testing.T) { + store := newTestStore(t, 1) + location := store.Locations[0] + mountTestVolume(t, location, 9, "") + location.UnloadVolume(needle.VolumeId(9)) + + location.loadGuard = func(vid needle.VolumeId) (func(), bool) { + return func() {}, true + } + location.loadExistingVolume("9.idx", NeedleMapInMemory, true, 0, 0, false) + if _, found := location.FindVolume(needle.VolumeId(9)); found { + t.Fatal("a volume the guard skips was loaded") + } + + // A release the guard hands out is held through the whole load and only + // dropped on return. + releaseRan := false + location.loadGuard = func(vid needle.VolumeId) (func(), bool) { + return func() { releaseRan = true }, false + } + location.loadExistingVolume("9.idx", NeedleMapInMemory, true, 0, 0, false) + if _, found := location.FindVolume(needle.VolumeId(9)); !found { + t.Fatal("a clear guard pass did not load the volume") + } + if !releaseRan { + t.Fatal("the guard release was not called when the load returned") + } +} + +// The claim holder's own mount must not consult the guard: VolumeCopy mounts +// while its claim still stands, and VolumeMount/VolumeConfigure hold the +// guard's lock through the mount. The skip is for scan-driven loads; an +// explicit LoadVolume is serialized by the caller's own hold. +func TestLoadVolumeIgnoresTheGuardForTheHoldersOwnMount(t *testing.T) { + store := newTestStore(t, 1) + location := store.Locations[0] + mountTestVolume(t, location, 9, "") + location.UnloadVolume(needle.VolumeId(9)) + + location.loadGuard = func(vid needle.VolumeId) (func(), bool) { + return func() {}, true + } + if !location.LoadVolume(0, needle.VolumeId(9), NeedleMapInMemory, nil, true) { + t.Fatal("the claim holder's own mount was skipped by its own guard") + } + if _, found := location.FindVolume(needle.VolumeId(9)); !found { + t.Fatal("explicit mount did not register the volume") + } + + location.UnloadVolume(needle.VolumeId(9)) + if location.LoadVolume(0, needle.VolumeId(9), NeedleMapInMemory, nil, false) { + t.Fatal("an unguarded mount slipped past the claim's skip") + } +} diff --git a/weed/storage/store.go b/weed/storage/store.go index 6756983e0b..1c395c69e6 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -945,6 +945,15 @@ func (s *Store) LoadNewVolumes() { } } +// SetVolumeLoadGuard installs the server-side hook every location consults +// before loading a volume from disk, so a reload cannot register or clean +// files out from under an in-flight VolumeCopy. +func (s *Store) SetVolumeLoadGuard(guard func(vid needle.VolumeId) (release func(), skip bool)) { + for _, location := range s.Locations { + location.loadGuard = guard + } +} + func (s *Store) Close() { for _, location := range s.Locations { location.Close() @@ -1109,12 +1118,18 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { // succeed. A validator failure unloads the volume before returning the error, // so an invalid newly copied replica is never announced to the master. func (s *Store) MountVolume(i needle.VolumeId, collection *string, validators ...func(*Volume) error) error { - return s.mountVolume(i, collection, validators...) + return s.mountVolume(i, collection, false, validators...) } -func (s *Store) mountVolume(i needle.VolumeId, collection *string, validators ...func(*Volume) error) error { +// MountVolumeClaimed mounts a volume whose copy guard the caller already +// holds, so the load guard does not skip the owner's own load. +func (s *Store) MountVolumeClaimed(i needle.VolumeId, collection *string, validators ...func(*Volume) error) error { + return s.mountVolume(i, collection, true, validators...) +} + +func (s *Store) mountVolume(i needle.VolumeId, collection *string, ownLoad bool, validators ...func(*Volume) error) error { for diskId, location := range s.Locations { - if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind, collection); found == true { + if found := location.LoadVolume(uint32(diskId), i, s.NeedleMapKind, collection, ownLoad); found == true { glog.V(0).Infof("mount volume %d", i) v := s.findVolume(i) v.diskId = uint32(diskId) // Set disk ID when mounting diff --git a/weed/storage/volume_vacuum_crash_safe_test.go b/weed/storage/volume_vacuum_crash_safe_test.go index 74279017a5..deca5c7c60 100644 --- a/weed/storage/volume_vacuum_crash_safe_test.go +++ b/weed/storage/volume_vacuum_crash_safe_test.go @@ -340,3 +340,40 @@ func TestDestroyRemovesCommitMarker(t *testing.T) { location.reconcileCompactStates() mustNotExist(t, cpc) } + +// A VolumeCopy mid-flight writes the same basenames an interrupted commit +// would swap: reconcileCompactStates must skip the vid while the load guard +// reports a copy, and roll the leftover forward once it clears. +func TestReconcileSkipsVidWithCopyInFlight(t *testing.T) { + dir := t.TempDir() + + location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig()) + defer location.Close() + + base := filepath.Join(dir, "7") + for _, ext := range []string{".cpc", ".cpd", ".cpx"} { + if err := os.WriteFile(base+ext, []byte("x"), 0644); err != nil { + t.Fatal(err) + } + } + if err := os.WriteFile(base+".dat", []byte("copy"), 0644); err != nil { + t.Fatal(err) + } + + vid := needle.VolumeId(7) + inFlight := true + location.loadGuard = func(v needle.VolumeId) (func(), bool) { + return func() {}, v == vid && inFlight + } + + location.reconcileCompactStates() + mustExist(t, base+".cpd") + if got, err := os.ReadFile(base + ".dat"); err != nil || string(got) != "copy" { + t.Fatalf("reconcile rolled a swap over a copy's .dat: %q %v", got, err) + } + + inFlight = false + location.reconcileCompactStates() + mustNotExist(t, base+".cpd") + mustExist(t, base+".dat") +}