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") +}