From 4fb3e22a01d03c350b07cf21a13393f52b2477d1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 13 Jun 2026 20:09:00 -0700 Subject: [PATCH] fix(tiering): never delete a shared remote object while replicas still reference it (#9942) * tiering: stop a shared remote object being deleted while replicas still point at it A remote-tiered volume's .dat content lives only in one cloud object that all N replica .vif files point at. Deleting that object while destroying any one replica, or before a downloaded replica is durable, bricks the survivors. - volume.tier.move cleanup now deletes old replicas with keepRemoteData=true so surviving replicas keep the shared object. Document why the alreadyPlaced anchor needs no replica sync (same-object replicas are byte-identical). - VolumeTierMoveDatFromRemote now fsyncs the downloaded .dat, fsyncs the containing directory, trims the .vif (fsynced) and swaps to the local DiskFile BEFORE deleting the remote object, on both the keep-remote and delete paths. Only the final DeleteFile is gated by keep_remote_dat_file, so a keep-remote download leaves the replica served from local disk rather than the shared object, and a crash before delete merely leaks the object. - volume.tier.download keeps the shared object for every replica except the last, which deletes it. - s3 and rclone download paths fsync the .dat before close. * storage: swap the volume data backend under the data lock The tier-download swap closed v.DataBackend and assigned the new local DiskFile without holding dataFileAccessLock, racing concurrent reads/writes (use of a closed file / nil deref). Add an exported Volume.SwapDataBackend that performs the close-and-replace under the lock, and call it from the tier download. * server: skip directory fsync on Windows in the tier download path os.Open(dir).Sync() is unsupported on Windows and returns an error, which would fail VolumeTierMoveDatFromRemote entirely there. Skip the directory fsync on Windows, matching how the storage-side helper tolerates the unsupported case. * shell: make multi-replica tier.download resilient to already-local replicas If a multi-replica download is interrupted and retried, a replica made local in the prior attempt returns "already on local disk", which aborted the whole command and left the remaining remote replicas dangling. Treat that case as a skip-and-continue so a retry completes the rest. * server: assert downloaded .dat content, not just length, in the tier test A length-only check passes even if the bytes are corrupted; compare the full content of the local .dat against the original. --- weed/server/volume_grpc_tier_download.go | 77 ++++- weed/server/volume_grpc_tier_download_test.go | 313 ++++++++++++++++++ weed/shell/command_volume_tier_compact.go | 4 +- weed/shell/command_volume_tier_download.go | 21 +- weed/shell/command_volume_tier_move.go | 10 +- .../backend/rclone_backend/rclone_backend.go | 7 + .../storage/backend/s3_backend/s3_download.go | 6 + weed/storage/volume.go | 13 + 8 files changed, 429 insertions(+), 22 deletions(-) create mode 100644 weed/server/volume_grpc_tier_download_test.go diff --git a/weed/server/volume_grpc_tier_download.go b/weed/server/volume_grpc_tier_download.go index afde9a44f..af07f80d6 100644 --- a/weed/server/volume_grpc_tier_download.go +++ b/weed/server/volume_grpc_tier_download.go @@ -3,10 +3,13 @@ package weed_server import ( "fmt" "os" + "path/filepath" + "runtime" "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/needle" ) @@ -60,10 +63,11 @@ func (vs *VolumeServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.Volume ProcessedPercentage: percentage, }) } - // copy the data file - _, err := backendStorage.DownloadFile(v.FileName(".dat"), storageKey, fn) + // copy the data file (DownloadFile opens, fsyncs, and closes the .dat internally) + datFileName := v.FileName(".dat") + _, err := backendStorage.DownloadFile(datFileName, storageKey, fn) if err != nil { - return fmt.Errorf("backend %s copy file %s: %v", storageName, v.FileName(".dat"), err) + return fmt.Errorf("backend %s copy file %s: %v", storageName, datFileName, err) } if remoteFileModifiedTime > 0 { modifiedTime := time.Unix(int64(remoteFileModifiedTime), 0) @@ -74,23 +78,70 @@ func (vs *VolumeServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.Volume } } - if req.KeepRemoteDatFile { - return nil + // fsync the containing directory so the new .dat and the about-to-be-rewritten + // .vif are durably linked before we touch the shared remote object. + if err := fsyncDir(filepath.Dir(datFileName)); err != nil { + return fmt.Errorf("volume %d fsync dir for %s: %v", v.Id, datFileName, err) } - // remove remote file - if err := backendStorage.DeleteFile(storageKey); err != nil { - return fmt.Errorf("volume %d failed to delete remote file %s: %v", v.Id, storageKey, err) - } - - // forget remote file + // Trim the remote file reference and persist the .vif (util.WriteFile fsyncs it) + // BEFORE deleting the remote object. After this point hasRemoteFile is false, so a + // crash before DeleteFile merely leaks the remote object while the volume reloads + // its local .dat. The volume must NEVER be left with a .vif referencing the remote + // object while that object is deleted. v.GetVolumeInfo().Files = v.GetVolumeInfo().Files[1:] if err := v.SaveVolumeInfo(); err != nil { return fmt.Errorf("volume %d failed to save remote file info: %v", v.Id, err) } - v.DataBackend.Close() - v.DataBackend = nil + // fsync the directory again so the rewritten .vif is durable. + if err := fsyncDir(filepath.Dir(datFileName)); err != nil { + return fmt.Errorf("volume %d fsync dir after saving volume info: %v", v.Id, err) + } + + // Swap the data backend from the remote storage to the now-local .dat on BOTH + // paths, so a KeepRemoteDatFile=true download still leaves the replica serving + // from local disk (hasRemoteFile=false) rather than the shared remote object. + if err := swapToLocalDatBackend(v, datFileName); err != nil { + return fmt.Errorf("volume %d failed to open local dat file %s: %v", v.Id, datFileName, err) + } + + if req.KeepRemoteDatFile { + // Surviving replicas still reference this object; keep it intact. + return nil + } + + // remove remote file: only the last replica to download deletes the shared object. + if err := backendStorage.DeleteFile(storageKey); err != nil { + return fmt.Errorf("volume %d failed to delete remote file %s: %v", v.Id, storageKey, err) + } return nil } + +// swapToLocalDatBackend closes the remote data backend and opens the downloaded +// local .dat as a DiskFile so reads are served from local disk. +func swapToLocalDatBackend(v *storage.Volume, datFileName string) error { + dataFile, err := os.OpenFile(datFileName, os.O_RDWR, 0644) + if err != nil { + return err + } + // Swap under the volume's data lock so concurrent reads never see a closed + // or half-swapped backend. + v.SwapDataBackend(backend.NewDiskFile(dataFile)) + return nil +} + +// fsyncDir flushes a directory entry so renamed/created files within it survive +// a crash. Directory fsync is unsupported on Windows, so it is skipped there. +func fsyncDir(dir string) error { + if runtime.GOOS == "windows" { + return nil + } + d, err := os.Open(dir) + if err != nil { + return err + } + defer d.Close() + return d.Sync() +} diff --git a/weed/server/volume_grpc_tier_download_test.go b/weed/server/volume_grpc_tier_download_test.go new file mode 100644 index 000000000..43401ca26 --- /dev/null +++ b/weed/server/volume_grpc_tier_download_test.go @@ -0,0 +1,313 @@ +package weed_server + +import ( + "bytes" + "fmt" + "io" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" + + "google.golang.org/grpc" +) + +const tierTestBackendName = "tier_test_local_dir.default" + +// tierTestBackend is a fake BackendStorage backed by files in a temp dir. It +// records DeleteFile calls so a test can assert that a shared remote object is +// (or is not) removed. +type tierTestBackend struct { + root string + + mu sync.Mutex + deletes []string +} + +func (b *tierTestBackend) ToProperties() map[string]string { return map[string]string{"root": b.root} } + +func (b *tierTestBackend) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile { + return &tierTestBackendFile{backend: b, key: key, tierInfo: tierInfo} +} + +func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) { + key = fmt.Sprintf("obj-%d", time.Now().UnixNano()) + dst := filepath.Join(b.root, key) + out, err := os.Create(dst) + if err != nil { + return "", 0, err + } + defer out.Close() + if _, err = f.Seek(0, io.SeekStart); err != nil { + return "", 0, err + } + written, err := io.Copy(out, f) + if err != nil { + return "", 0, err + } + return key, written, nil +} + +func (b *tierTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) { + in, err := os.Open(filepath.Join(b.root, key)) + if err != nil { + return 0, err + } + defer in.Close() + out, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return 0, err + } + written, err := io.Copy(out, in) + if err != nil { + out.Close() + return 0, err + } + // mirror the real backends: fsync the .dat before close so the caller can + // trim the .vif and delete the remote object without risking a torn write. + if syncErr := out.Sync(); syncErr != nil { + out.Close() + return 0, syncErr + } + if closeErr := out.Close(); closeErr != nil { + return 0, closeErr + } + return written, nil +} + +func (b *tierTestBackend) DeleteFile(key string) error { + b.mu.Lock() + b.deletes = append(b.deletes, key) + b.mu.Unlock() + return os.Remove(filepath.Join(b.root, key)) +} + +func (b *tierTestBackend) deleteHistory() []string { + b.mu.Lock() + defer b.mu.Unlock() + out := make([]string, len(b.deletes)) + copy(out, b.deletes) + return out +} + +func (b *tierTestBackend) objectExists(key string) bool { + _, err := os.Stat(filepath.Join(b.root, key)) + return err == nil +} + +type tierTestBackendFile struct { + backend *tierTestBackend + key string + tierInfo *volume_server_pb.VolumeInfo +} + +func (f *tierTestBackendFile) ReadAt(p []byte, off int64) (int, error) { + in, err := os.Open(filepath.Join(f.backend.root, f.key)) + if err != nil { + return 0, err + } + defer in.Close() + return in.ReadAt(p, off) +} +func (f *tierTestBackendFile) WriteAt(p []byte, off int64) (int, error) { panic("not implemented") } +func (f *tierTestBackendFile) Truncate(off int64) error { panic("not implemented") } +func (f *tierTestBackendFile) Close() error { return nil } +func (f *tierTestBackendFile) Name() string { return f.key } +func (f *tierTestBackendFile) Sync() error { return nil } +func (f *tierTestBackendFile) GetStat() (int64, time.Time, error) { + files := f.tierInfo.GetFiles() + if len(files) == 0 { + return 0, time.Time{}, fmt.Errorf("remote file info not found") + } + return int64(files[0].FileSize), time.Unix(int64(files[0].ModifiedTime), 0), nil +} + +// fakeTierStream is a no-op server stream for the tier-download RPC. +type fakeTierStream struct { + grpc.ServerStream +} + +func (s *fakeTierStream) Send(*volume_server_pb.VolumeTierMoveDatFromRemoteResponse) error { return nil } + +func newTierTestStore(t *testing.T, dir string) *storage.Store { + t.Helper() + diskIOProbeConfig := stats.DefaultDiskIOProbeConfig() + store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "", + []string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "", + storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, diskIOProbeConfig) + + done := make(chan bool) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.DeletedVolumesChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + store.Close() + close(done) + }) + return store +} + +// tierUpVolumeOnDisk creates a real local volume in dir, writes a few needles, +// uploads the .dat to the fake backend, rewrites the .vif to remote mode, and +// removes the local .dat — mirroring volume_grpc_tier_upload.go. It returns the +// remote object key and the original local .dat bytes for later comparison. +func tierUpVolumeOnDisk(t *testing.T, dir string, vid needle.VolumeId, b *tierTestBackend) (key string, localDat []byte) { + t.Helper() + store := newTierTestStore(t, dir) + if err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add volume: %v", err) + } + for i := 1; i <= 5; i++ { + n := new(needle.Needle) + n.Id = types.Uint64ToNeedleId(uint64(i)) + n.Data = []byte(fmt.Sprintf("payload-%d-localdisk", i)) + n.Checksum = needle.NewCRC(n.Data) + if _, err := store.WriteVolumeNeedle(vid, n, true, false); err != nil { + t.Fatalf("write needle %d: %v", i, err) + } + } + + v := store.GetVolume(vid) + if v == nil { + t.Fatal("volume not found after add") + } + diskFile, ok := v.DataBackend.(*backend.DiskFile) + if !ok { + t.Fatalf("expected on-disk backend before tier-up, got %T", v.DataBackend) + } + datPath := v.FileName(".dat") + key, size, err := b.CopyFile(diskFile.File, nil) + if err != nil { + t.Fatalf("upload to fake backend: %v", err) + } + + bType, bId := backend.BackendNameToTypeId(tierTestBackendName) + v.GetVolumeInfo().Files = append(v.GetVolumeInfo().GetFiles(), &volume_server_pb.RemoteFile{ + BackendType: bType, + BackendId: bId, + Key: key, + FileSize: uint64(size), + ModifiedTime: uint64(time.Now().Unix()), + Extension: ".dat", + }) + if err := v.SaveVolumeInfo(); err != nil { + t.Fatalf("save volume info: %v", err) + } + + localDat, err = os.ReadFile(datPath) + if err != nil { + t.Fatalf("read local dat: %v", err) + } + if err := store.UnmountVolume(vid); err != nil { + t.Fatalf("unmount: %v", err) + } + if err := os.Remove(datPath); err != nil { + t.Fatalf("remove local dat: %v", err) + } + return key, localDat +} + +// TestTierMoveDatFromRemote_KeepRemote_LeavesReplicaLocal reproduces the +// multi-replica tier-download data loss: when KeepRemoteDatFile=true (every +// replica except the last), the replica must end up served from its local .dat +// with the shared remote object intact. Before the fix the keep-remote path +// returned right after the download without trimming the .vif or swapping the +// data backend, so the replica stayed remote-backed and a subsequent delete of +// the shared object on the final replica bricked it. +func TestTierMoveDatFromRemote_KeepRemote_LeavesReplicaLocal(t *testing.T) { + b := &tierTestBackend{root: t.TempDir()} + backend.BackendStorages[tierTestBackendName] = b + t.Cleanup(func() { delete(backend.BackendStorages, tierTestBackendName) }) + + dir := t.TempDir() + const vid = needle.VolumeId(71) + key, localDat := tierUpVolumeOnDisk(t, dir, vid, b) + + if !b.objectExists(key) { + t.Fatal("remote object missing after tier-up") + } + + // Mount the tiered volume the way a volume server does at startup. + store := newTierTestStore(t, dir) + v := store.GetVolume(vid) + if v == nil { + t.Fatal("tiered volume not loaded by store") + } + if !v.HasRemoteFile() { + t.Fatal("volume should load in remote mode before download") + } + + vs := &VolumeServer{store: store} + req := &volume_server_pb.VolumeTierMoveDatFromRemoteRequest{ + VolumeId: uint32(vid), + Collection: "", + KeepRemoteDatFile: true, + } + if err := vs.VolumeTierMoveDatFromRemote(req, &fakeTierStream{}); err != nil { + t.Fatalf("VolumeTierMoveDatFromRemote: %v", err) + } + + // keepRemote: the shared object must survive untouched. + if len(b.deleteHistory()) != 0 { + t.Fatalf("KeepRemoteDatFile=true must not delete the shared remote object, deletes: %v", b.deleteHistory()) + } + if !b.objectExists(key) { + t.Fatal("remote object must still exist after a keep-remote download") + } + + // The in-memory volume must now be served from a local DiskFile, not the + // remote backend — this is the assertion that fails on the pre-fix code. + if _, ok := v.DataBackend.(*backend.DiskFile); !ok { + t.Fatalf("after keep-remote download the data backend must be local DiskFile, got %T", v.DataBackend) + } + + // Remount and confirm the volume is no longer remote-backed and reads come + // from the local .dat (matching the bytes uploaded before tiering). + if err := store.UnmountVolume(vid); err != nil { + t.Fatalf("unmount after download: %v", err) + } + if err := store.MountVolume(vid); err != nil { + t.Fatalf("remount after download: %v", err) + } + v2 := store.GetVolume(vid) + if v2 == nil { + t.Fatal("volume missing after remount") + } + if v2.HasRemoteFile() { + t.Fatal("after a keep-remote download the remounted replica must not be remote-backed") + } + if _, ok := v2.DataBackend.(*backend.DiskFile); !ok { + t.Fatalf("remounted replica must read from local DiskFile, got %T", v2.DataBackend) + } + + // The .vif must no longer reference the remote object. + storageName, storageKey := v2.RemoteStorageNameKey() + if storageName != "" || storageKey != "" { + t.Fatalf("trimmed .vif must not reference remote object, got %q/%q", storageName, storageKey) + } + + // The local .dat must hold the downloaded content (the original bytes). + gotDat, err := os.ReadFile(v2.FileName(".dat")) + if err != nil { + t.Fatalf("read local dat after download: %v", err) + } + if !bytes.Equal(gotDat, localDat) { + t.Fatalf("local .dat content mismatch: got %d bytes, want %d", len(gotDat), len(localDat)) + } +} diff --git a/weed/shell/command_volume_tier_compact.go b/weed/shell/command_volume_tier_compact.go index 1dd501833..f26af60a6 100644 --- a/weed/shell/command_volume_tier_compact.go +++ b/weed/shell/command_volume_tier_compact.go @@ -213,9 +213,9 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu fmt.Fprintf(writer, "volume %d garbage ratio %.4f, starting compaction...\n", rv.vid, garbageRatio) // step 2: download .dat from remote to local - // this deletes the remote file and reloads the volume as local + // this deletes the remote file and reloads the volume as local, then re-uploads below fmt.Fprintf(writer, " downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName) - err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress) + err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false) if err != nil { return fmt.Errorf("download volume %d from remote: %v", rv.vid, err) } diff --git a/weed/shell/command_volume_tier_download.go b/weed/shell/command_volume_tier_download.go index 4626bd383..d261747c3 100644 --- a/weed/shell/command_volume_tier_download.go +++ b/weed/shell/command_volume_tier_download.go @@ -5,6 +5,7 @@ import ( "flag" "fmt" "io" + "strings" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -124,11 +125,20 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s return fmt.Errorf("volume %d not found", vid) } + // All replicas point at the same remote object; only the final download may delete + // it. Every earlier replica keeps it so the survivors are not left dangling. // TODO parallelize this - for _, loc := range locations { + for i, loc := range locations { + keepRemote := i < len(locations)-1 // copy the .dat file from remote tier to local - err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress()) + err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote) if err != nil { + // A replica already made local by a prior interrupted run is not a + // failure; skip it so the remaining remote replicas still download. + if strings.Contains(err.Error(), "already on local disk") { + fmt.Fprintf(writer, "volume %d on %s is already on local disk, skipping\n", vid, loc.Url) + continue + } return fmt.Errorf("download dat file for volume %d to %s: %v", vid, loc.Url, err) } } @@ -136,12 +146,13 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s return nil } -func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress) error { +func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool) error { err := operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { stream, downloadErr := volumeServerClient.VolumeTierMoveDatFromRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatFromRemoteRequest{ - VolumeId: uint32(volumeId), - Collection: collection, + VolumeId: uint32(volumeId), + Collection: collection, + KeepRemoteDatFile: keepRemote, }) var lastProcessed int64 diff --git a/weed/shell/command_volume_tier_move.go b/weed/shell/command_volume_tier_move.go index f12fcb222..0d098756f 100644 --- a/weed/shell/command_volume_tier_move.go +++ b/weed/shell/command_volume_tier_move.go @@ -251,7 +251,11 @@ func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer } // a replica already on the target tier (e.g. left by an interrupted earlier run) - // anchors the move: skip the copy, only fulfill replication and clean up old replicas + // anchors the move: skip the copy, only fulfill replication and clean up old replicas. + // The anchor must match both the target disk type AND (when set) the target data + // center, so a bare presence elsewhere never short-circuits a cross-DC move. + // No replica sync is needed here: remote-tiered replicas share one cloud object, + // so their content is byte-identical and there is no local divergence to reconcile. for _, r := range replicas { if types.ToDiskType(r.info.DiskType) != toDiskType || (toDataCenter != "" && r.location.dc != toDataCenter) { continue @@ -372,7 +376,9 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i if preserveServers[loc.Url] { continue } - if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false, false); err != nil { + // keepRemoteData=true: remote-tiered replicas share one cloud object, so + // deleting a replica must not delete the object the survivors still point at. + if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false, true); err != nil { fmt.Fprintf(writer, "failed to delete volume %d on %s: %v\n", vid, loc.Url, err) } } diff --git a/weed/storage/backend/rclone_backend/rclone_backend.go b/weed/storage/backend/rclone_backend/rclone_backend.go index 24e8620e5..1fe7aca76 100644 --- a/weed/storage/backend/rclone_backend/rclone_backend.go +++ b/weed/storage/backend/rclone_backend/rclone_backend.go @@ -190,6 +190,13 @@ func downloadViaRclone(fs fs.Fs, filename string, key string, fn func(progressed tr := accounting.NewStats(ctx).NewTransfer(obj, fs) defer func() { + // fsync the .dat before closing so its content is durable before the caller + // trims the remote reference and deletes the shared remote object. + if syncer, ok := file.(interface{ Sync() error }); ok { + if syncErr := syncer.Sync(); err == nil && syncErr != nil { + err = syncErr + } + } if closeErr := file.Close(); err == nil && closeErr != nil { err = closeErr } diff --git a/weed/storage/backend/s3_backend/s3_download.go b/weed/storage/backend/s3_backend/s3_download.go index af6f7b4f3..82d73d71b 100644 --- a/weed/storage/backend/s3_backend/s3_download.go +++ b/weed/storage/backend/s3_backend/s3_download.go @@ -50,6 +50,12 @@ func downloadFromS3(sess s3iface.S3API, destFileName string, sourceBucket string return fileSize, fmt.Errorf("failed to download /buckets/%s/%s to %s: %v", sourceBucket, sourceKey, destFileName, err) } + // fsync the downloaded .dat so its content is durable before the caller trims the + // remote reference and deletes the shared remote object. + if syncErr := f.Sync(); syncErr != nil { + return fileSize, fmt.Errorf("failed to fsync %s: %v", destFileName, syncErr) + } + glog.V(1).Infof("downloaded file %s\n", destFileName) return diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 10249482f..b326f9ab3 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -310,6 +310,19 @@ func (v *Volume) Close() { v.doClose() } +// SwapDataBackend atomically replaces the data backend (e.g. swapping a +// remote-tier backend for a freshly downloaded local .dat), closing the old +// one. Held under dataFileAccessLock so a concurrent read/write never observes +// a half-swapped or closed backend. +func (v *Volume) SwapDataBackend(newBackend backend.BackendStorageFile) { + v.dataFileAccessLock.Lock() + defer v.dataFileAccessLock.Unlock() + if v.DataBackend != nil { + v.DataBackend.Close() + } + v.DataBackend = newBackend +} + func (v *Volume) doClose() { if v.nm != nil { if err := v.nm.Sync(); err != nil {