diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index c08f97a380..90e40ed8da 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -5052,13 +5052,18 @@ impl VolumeServer for VolumeGrpcService { // for and referenced by nothing. Go does not gate here either: // its progress callback only runs during the transfer. // Update volume info with remote file reference, unless the - // volume was compacted, replaced or removed mid-upload: the - // object would not match what now holds this id. + // volume was compacted, replaced, removed or written to + // mid-upload: the object would not match what now holds this + // id. One write lease spans the checks, the .vif save and the + // backend swap, so no write or commit lands in between. let abort = { let store = state.store.read().unwrap(); - let abort = match store.find_volume(vid) { + let vol = store.find_volume(vid).map(|(_, v)| v); + let _sync_guard = vol.as_ref().map(|v| v.data_file_sync_lock().lock().unwrap()); + let _lease = vol.as_ref().map(|v| v.instance().write_lock()); + let abort = match &vol { None => Some(Status::not_found(format!("volume {} not found", vid))), - Some((_, v)) + Some(v) if !v.is_instance(&instance) || v.super_block().compaction_revision != compaction_revision => { @@ -5067,12 +5072,22 @@ impl VolumeServer for VolumeGrpcService { vid ))) } - Some(_) => None, + Some(v) => match v.dat_file_size() { + Ok(dat_size) if dat_size == size => None, + Ok(dat_size) => Some(Status::failed_precondition(format!( + "volume {} data file is {} bytes, the copy holds {}: written during the copy", + vid, dat_size, size + ))), + Err(e) => Some(Status::internal(format!( + "stat volume {} data file: {}", + vid, e + ))), + }, }; if abort.is_none() - && let Some((_, vol)) = store.find_volume(vid) + && let Some(vol) = &vol { - vol.update_remote_files(|files| { + vol.update_remote_files_held(|files| { files.push(volume_server_pb::RemoteFile { backend_type: backend_type.clone(), backend_id: backend_id.clone(), @@ -5100,7 +5115,7 @@ impl VolumeServer for VolumeGrpcService { // Switch the data backend from the local .dat to the remote // object so reads keep working after upload (matches Go's // LoadRemoteFile). - if let Err(e) = vol.load_remote_dat_file() { + if let Err(e) = vol.load_remote_dat_file_held() { return Err(Status::internal(format!( "volume {} failed to load remote file: {}", vid, e @@ -9691,6 +9706,68 @@ mod tests { let _ = s3.shutdown.send(()); } + // A write acknowledged while the upload runs is not in the object; the + // volume must keep serving the local .dat that holds it. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_tier_move_to_remote_aborts_when_written_mid_upload() { + let (service, tmp) = make_local_service_with_volume("", None); + seed_compactable_volume(&service); + let mut s3 = spawn_multipart_s3_server(); + register_tier_backend("s3.tier_up_written", s3.endpoint.clone()); + + let stream = start_parked_tier_move(&service, &mut s3, "s3.tier_up_written", false).await; + let data = b"needle-3".to_vec(); + let mut n = Needle { + id: NeedleId(3), + cookie: Cookie(3), + data_size: data.len() as u32, + data: data.clone(), + ..Needle::default() + }; + service + .state + .store + .read() + .unwrap() + .write_volume_needle(VolumeId(1), &mut n, true) + .expect("the write lands while the upload is parked"); + s3.release.add_permits(1); + + let terminal = tier_move_error(stream).await; + global_s3_tier_registry() + .write() + .unwrap() + .remove("s3.tier_up_written"); + let err = terminal.expect("the tier move must fail once the volume was written"); + assert_eq!(err.code(), tonic::Code::FailedPrecondition, "{err:?}"); + assert_eq!( + s3.delete_count.load(std::sync::atomic::Ordering::SeqCst), + 1, + "the stale object must be deleted" + ); + { + let store = service.state.store.read().unwrap(); + let (_, vol) = store.find_volume(VolumeId(1)).unwrap(); + assert!(!vol.has_remote_file()); + assert!(vol.volume_info().files.is_empty()); + } + assert!(tmp.path().join("1.dat").exists()); + let mut n = Needle { + id: NeedleId(3), + cookie: Cookie(3), + ..Needle::default() + }; + service + .state + .store + .read() + .unwrap() + .read_volume_needle(VolumeId(1), &mut n) + .expect("the acknowledged write must stay readable"); + assert_eq!(n.data, data); + let _ = s3.shutdown.send(()); + } + // Go fails here too: unmounting closes the descriptor its copy reads. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_tier_move_to_remote_fails_when_volume_is_unmounted_mid_upload() { diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 774d651baf..df565c3cfc 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -2498,9 +2498,15 @@ impl Volume { Ok(()) } + #[cfg(test)] pub(crate) fn load_remote_dat_file(&self) -> Result<(), VolumeError> { let _sync_guard = self.data_file_sync_lock.lock().unwrap(); let _guard = self.data_file_access_control.write_lock(); + self.load_remote_dat_file_held() + } + + /// `load_remote_dat_file` for callers already holding the write lease. + pub(crate) fn load_remote_dat_file_held(&self) -> Result<(), VolumeError> { let mut io = self.io.write().unwrap(); self.load_remote_dat_file_io(&mut io) } diff --git a/weed/server/volume_grpc_tier_upload_append_test.go b/weed/server/volume_grpc_tier_upload_append_test.go new file mode 100644 index 0000000000..ff8da25e34 --- /dev/null +++ b/weed/server/volume_grpc_tier_upload_append_test.go @@ -0,0 +1,85 @@ +package weed_server + +import ( + "bytes" + "os" + "testing" + + "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" +) + +// A write acknowledged after the tier upload copied the .dat is not in the +// copy. The upload must not switch the volume to that copy or remove the local +// .dat that holds the write. +func TestTierUploadRefusesADatWrittenDuringTheCopy(t *testing.T) { + root := t.TempDir() + b := &afterCopyBackend{tierTimestampTestBackend: tierTimestampTestBackend{root: root}} + backend.BackendStorages[tierTimestampTestBackendName] = b + t.Cleanup(func() { delete(backend.BackendStorages, tierTimestampTestBackendName) }) + + store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{t.TempDir()}, []int32{10}, []util.MinFreeSpace{{}}, "", storage.NeedleMapInMemory, + []types.DiskType{types.HardDriveType}, nil, 0, stats.DefaultDiskIOProbeConfig()) + t.Cleanup(store.Close) + if err := store.AddVolume(1, "", storage.NeedleMapInMemory, "000", "", 0, needle.Version3, 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add volume: %v", err) + } + written := map[uint64][]byte{} + write := func(i uint64) error { + n := &needle.Needle{Id: types.Uint64ToNeedleId(i), Cookie: 7, Data: bytes.Repeat([]byte{byte(i)}, 1024)} + n.Checksum = needle.NewCRC(n.Data) + if _, err := store.WriteVolumeNeedle(1, n, false, false); err != nil { + return err + } + written[i] = n.Data + return nil + } + for i := uint64(1); i <= 5; i++ { + if err := write(i); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + v := store.GetVolume(1) + + appended := false + b.afterCopy = func() { + if err := write(6); err != nil { + t.Errorf("append after the copy: %v", err) + return + } + appended = true + } + vs := &VolumeServer{store: store} + err := vs.VolumeTierMoveDatToRemote(&volume_server_pb.VolumeTierMoveDatToRemoteRequest{ + VolumeId: 1, DestinationBackendName: tierTimestampTestBackendName, + }, &discardServerStream[volume_server_pb.VolumeTierMoveDatToRemoteResponse]{}) + if !appended { + t.Fatal("no write was acknowledged between the copy and the attach") // positive control + } + if err == nil { + t.Errorf("the upload attached a remote copy that misses an acknowledged write") + } + if v.HasRemoteFile() { + t.Errorf("the volume is tiered to the stale remote copy") + } + if entries, _ := os.ReadDir(root); len(entries) != 0 { + t.Errorf("the abandoned remote copy was left behind: %v", entries) + } + if !util.FileExists(v.FileName(".dat")) { + t.Fatalf("the local .dat is gone") + } + for i := uint64(1); i <= 6; i++ { + n := &needle.Needle{Id: types.Uint64ToNeedleId(i), Cookie: 7} + if _, err := store.ReadVolumeNeedle(1, n, nil, nil); err != nil { + t.Errorf("needle %d does not read after the upload: %v", i, err) + } else if !bytes.Equal(n.Data, written[i]) { + t.Errorf("needle %d reads different data", i) + } + } +} diff --git a/weed/storage/volume_tier.go b/weed/storage/volume_tier.go index cd046fdd01..fac67bf0f0 100644 --- a/weed/storage/volume_tier.go +++ b/weed/storage/volume_tier.go @@ -146,9 +146,11 @@ func (v *Volume) loadRemoteFileLocked() error { // AttachRemoteDatFile records remoteFile, a copy of the .dat read from source, // in the volume's .vif and switches the data backend to it. It refuses when // source is no longer the volume's data backend: a compaction commit since the -// copy replaced the .dat and the index, and the copy has the old layout. The -// check, the .vif write and the switch happen under the data-file locks, so a -// commit either lands before (and is seen here) or after (and refuses a tiered +// copy replaced the .dat and the index, and the copy has the old layout. It +// also refuses when source is no longer the size of the copy: a write landed +// after the copy read the file, and the copy does not hold it. The checks, the +// .vif write and the switch happen under the data-file locks, so a commit or a +// write either lands before (and is seen here) or after (and refuses a tiered // volume). func (v *Volume) AttachRemoteDatFile(source backend.BackendStorageFile, remoteFile *volume_server_pb.RemoteFile) error { v.dataFileSyncLock.Lock() @@ -158,6 +160,18 @@ func (v *Volume) AttachRemoteDatFile(source backend.BackendStorageFile, remoteFi if v.DataBackend != source { return fmt.Errorf("volume %d data file changed during the copy (compacted or reloaded)", v.Id) } + diskFile, ok := source.(*backend.DiskFile) + if !ok { + return fmt.Errorf("volume %d data file is not on local disk", v.Id) + } + // The physical size, as the copy read it: GetStat pads an unaligned tail. + stat, err := diskFile.File.Stat() + if err != nil { + return fmt.Errorf("stat volume %d data file: %w", v.Id, err) + } + if uint64(stat.Size()) != remoteFile.FileSize { + return fmt.Errorf("volume %d data file is %d bytes, the copy holds %d: written during the copy", v.Id, stat.Size(), remoteFile.FileSize) + } files := v.volumeInfo.GetFiles() v.volumeInfo.Files = append(files, remoteFile) if err := v.SaveVolumeInfo(); err != nil { diff --git a/weed/storage/volume_tier_attach_unaligned_test.go b/weed/storage/volume_tier_attach_unaligned_test.go new file mode 100644 index 0000000000..0457a25fd1 --- /dev/null +++ b/weed/storage/volume_tier_attach_unaligned_test.go @@ -0,0 +1,58 @@ +package storage + +import ( + "os" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// A read-only load keeps an unaligned .dat tail, and the data backend pads its +// size past it. The copy holds the physical bytes, so AttachRemoteDatFile must +// not read the padding as a write that landed during the copy. +func TestAttachRemoteDatFileAcceptsAnUnalignedDatTail(t *testing.T) { + b := newLocalDirBackend(t) + registerTestBackend(t, b) + + dir := t.TempDir() + open := func() *Volume { + v, err := NewVolume(dir, dir, "", 88, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + return v + } + v := open() + for i := 1; i <= 3; i++ { + _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false) + require.NoError(t, err) + } + require.NoError(t, v.PersistReadOnly(true, false)) + datName := v.FileName(".dat") + v.Close() + f, err := os.OpenFile(datName, os.O_WRONLY|os.O_APPEND, 0) + require.NoError(t, err) + _, err = f.Write([]byte{1, 2, 3}) + require.NoError(t, err) + require.NoError(t, f.Close()) + + v = open() + defer v.Close() + diskFile, ok := v.DataBackend.(*backend.DiskFile) + require.True(t, ok) + key, size, err := b.CopyFile(diskFile.File, nil, 0) + require.NoError(t, err) + require.NotZero(t, size%types.NeedlePaddingSize, "the .dat must end unaligned") // positive control + + bType, bId := backend.BackendNameToTypeId(testBackendName) + require.NoError(t, v.AttachRemoteDatFile(diskFile, &volume_server_pb.RemoteFile{ + BackendType: bType, BackendId: bId, Key: key, FileSize: uint64(size), + ModifiedTime: uint64(time.Now().Unix()), Extension: ".dat", + })) + require.True(t, v.HasRemoteFile()) +}