diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index f730fb7b7..81b5f916c 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -195,7 +195,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) } if filer_pb.IsDelete(resp) { - return processDeleteEvent(client, mountedDir, remoteStorageMountLocation, resp) + return processDeleteEvent(filerSource, client, mountedDir, remoteStorageMountLocation, resp) } if message.OldEntry != nil && message.NewEntry != nil { return processUpdateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp) @@ -212,8 +212,11 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem // event the inline delete already handled), and GCS reports that case as // ErrRemoteObjectNotFound where S3 and Azure answer an idempotent success. // Returning the error would pin the sync offset on an event that has nothing -// left to do. +// left to do. A file whose path a remote-only entry has taken since keeps its +// object: a rename copied that entry's content to this key, and it exists +// nowhere else. func processDeleteEvent( + filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, @@ -232,6 +235,36 @@ func processDeleteEvent( } glog.V(2).Infof("delete: %+v", resp) dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation) + if !message.OldEntry.IsDirectory { + // a deleted entry whose rename copy was still pending read its + // content from the source its marker names, which nothing else + // releases + if err := releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry); err != nil { + return err + } + // The object at the key may have been replaced since the delete: a + // rename copies another entry's content onto it, a rewrite uploads + // over it, and a copy can land before its entry is visible. Delete + // only an object the event's stamp still describes. + if live, statErr := client.StatFile(dest); statErr == nil { + if oldRemote := message.OldEntry.GetRemoteEntry(); oldRemote == nil || + !describes(oldRemote, live) || + oldRemote.RemoteETag != "" && live.RemoteETag == "" { + glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest)) + return nil + } + } else if !errors.Is(statErr, remote_storage.ErrRemoteObjectNotFound) { + return statErr + } + current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) + if err != nil { + return err + } + if isRemoteOnly(current) { + glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest)) + return nil + } + } if message.OldEntry.IsDirectory { glog.V(0).Infof("rmdir %s", remote_storage.FormatLocation(dest)) return client.RemoveDirectory(dest) @@ -240,6 +273,41 @@ func processDeleteEvent( return deleteRemoteFile(client, dest) } +// releasePendingSource deletes the source a pending-rename marker on entry +// names, once nothing needs it. The filer never deletes a marked source: a +// concurrent rename can copy onto that key before its entry is visible. So +// the object goes only while it is still the one entry's stamp describes, and +// no remote-only entry holds its path. +func releasePendingSource(filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) error { + src := util.FullPath(entry.GetExtended()[filer.ExtRemoteRenameSourceKey]) + if src == "" || !strings.HasPrefix(string(src), strings.TrimSuffix(mountedDir, "/")+"/") { + return nil + } + srcDest := toRemoteStorageLocation(util.FullPath(mountedDir), src, remoteStorageMountLocation) + srcDir, srcName := src.DirAndName() + current, err := currentEntry(filerSource, srcDir, srcName) + if err != nil { + return err + } + if isRemoteOnly(current) { + glog.V(0).Infof("keep %s: a remote-only entry holds the pending rename's source", remote_storage.FormatLocation(srcDest)) + return nil + } + live, err := client.StatFile(srcDest) + if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return nil + } + if err != nil { + return err + } + if stamp := entry.GetRemoteEntry(); !describes(stamp, live) || stamp.RemoteETag != "" && live.RemoteETag == "" { + glog.V(0).Infof("keep %s: the pending rename's source changed since", remote_storage.FormatLocation(srcDest)) + return nil + } + glog.V(0).Infof("delete %s, the source of a pending rename", remote_storage.FormatLocation(srcDest)) + return deleteRemoteFile(client, srcDest) +} + // deleteRemoteFile deletes the object and treats an already-absent object as // deleted. func deleteRemoteFile(client remote_storage.RemoteStorageClient, dest *remote_pb.RemoteStorageLocation) error { @@ -268,6 +336,17 @@ func processUpdateEvent( } oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation) dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation) + eventSource := renameSource{path: util.NewFullPath(resp.Directory, message.OldEntry.Name), loc: oldDest, stamp: message.OldEntry.GetRemoteEntry()} + // sourceOf is where a remote-only entry's content is: the source its + // pending-rename marker names, which may be an earlier rename's old key, + // or else the event's old key + sourceOf := func(entry *filer_pb.Entry, fallback renameSource) renameSource { + if src := string(entry.Extended[filer.ExtRemoteRenameSourceKey]); strings.HasPrefix(src, strings.TrimSuffix(mountedDir, "/")+"/") { + // a pending entry carries the stamp of the source it reads + return renameSource{path: util.FullPath(src), loc: toRemoteStorageLocation(util.FullPath(mountedDir), util.FullPath(src), remoteStorageMountLocation), stamp: entry.GetRemoteEntry()} + } + return fallback + } if proto.Equal(oldDest, dest) && !shouldSendToRemote(message.NewEntry) { glog.V(2).Infof("skipping updating: %+v", resp) return nil @@ -307,7 +386,23 @@ func processUpdateEvent( return err } if isRemoteOnly(current) { - return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, oldDest, dest, storageClass) + source := sourceOf(current, eventSource) + if err := completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, source, dest, storageClass); err != nil { + return err + } + // a pending entry the filer copied on this rename leaves the + // source its marker named + if pending := sourceOf(message.OldEntry, renameSource{}); pending.loc != nil && !proto.Equal(pending.loc, source.loc) { + return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry) + } + return nil + } + if current == nil && len(message.NewEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + // the rename's copy was pending and the entry has moved on: a + // later rename carries the source in its own marker and finishes + // or releases it, so this one leaves it + glog.V(0).Infof("keep %s: the pending rename moved on", remote_storage.FormatLocation(oldDest)) + return nil } glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest)) if err := client.DeleteFile(oldDest); err != nil { @@ -323,14 +418,43 @@ func processUpdateEvent( if errors.Is(writeErr, errSuperseded) { glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) if !proto.Equal(oldDest, dest) { - return uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, oldDest, dest, storageClass) + if err := uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, eventSource, dest, storageClass, sourceOf); err != nil { + return err + } + return releaseRenamedSource(filerSource, client, mountedDir, remoteStorageMountLocation, message) } return nil } if writeErr != nil { return writeErr } - return updateLocalEntry(filerClient, message.NewParentPath, message.NewEntry, remoteEntry) + if err := updateLocalEntry(filerClient, message.NewParentPath, message.NewEntry, remoteEntry); err != nil { + return err + } + if !proto.Equal(oldDest, dest) { + return releaseRenamedSource(filerSource, client, mountedDir, remoteStorageMountLocation, message) + } + return nil +} + +// releaseRenamedSource releases a pending source the renamed entry's old +// version read, once the destination holds the content on the remote under +// its own key: only after the upload and its stamp, which clears the marker, +// and only if the entry does not read through a marker any more. Released +// earlier, an upload that fails followed by remote.uncache would leave the +// entry reading a deleted source. +func releaseRenamedSource(filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, message *filer_pb.EventNotification) error { + if len(message.OldEntry.GetExtended()[filer.ExtRemoteRenameSourceKey]) == 0 { + return nil + } + current, err := currentEntry(filerSource, message.NewParentPath, message.NewEntry.Name) + if err != nil { + return err + } + if current != nil && len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + return nil + } + return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry) } // currentEntry returns what the filer holds at dir/name now, nil when the @@ -359,11 +483,24 @@ func isRemoteOnly(entry *filer_pb.Entry) bool { // copied over the destination, the entry is stamped with the copy, and the // old key goes. With neither object present the content is lost: the event // fails and holds the offset for recovery. -func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir string, current *filer_pb.Entry, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string) error { +func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir string, current *filer_pb.Entry, source renameSource, dest *remote_pb.RemoteStorageLocation, storageClass string) error { + oldDest := source.loc + // the source's recorded stamp; a retry can find the entry already + // stamped with the copy, whose ETag a multipart source does not share + sourceStamp := source.stamp + if sourceStamp == nil { + sourceStamp = proto.Clone(current.RemoteEntry).(*filer_pb.RemoteEntry) + } if existing, err := client.StatFile(dest); err == nil { if describes(current.RemoteEntry, existing) { glog.V(0).Infof("%s already holds the renamed content", remote_storage.FormatLocation(dest)) - return deleteRemoteFile(client, oldDest) + if len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + // reads still go to the source until the stamp clears the marker + if err := updateLocalEntry(filerClient, dir, current, existing); err != nil { + return err + } + } + return deleteRenameSource(filerClient, client, dir, current.Name, source, sourceStamp) } glog.V(0).Infof("%s holds an object the entry does not describe (size %d, want %d); replacing it from %s", remote_storage.FormatLocation(dest), existing.RemoteSize, current.RemoteEntry.RemoteSize, remote_storage.FormatLocation(oldDest)) } else if !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { @@ -389,7 +526,42 @@ func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_st if err := updateLocalEntry(filerClient, dir, current, remoteEntry); err != nil { return err } - return deleteRemoteFile(client, oldDest) + return deleteRenameSource(filerClient, client, dir, current.Name, source, sourceStamp) +} + +// renameSource is where a renamed remote-only entry's content was read +// from: the path, the object it maps to, and the stamp describing it. +type renameSource struct { + path util.FullPath + loc *remote_pb.RemoteStorageLocation + stamp *filer_pb.RemoteEntry +} + +// deleteRenameSource deletes a rename's old key once the renamed entry no +// longer reads it: an entry still carrying the pending-source marker (its +// stamp lost a race) keeps the source, and the event is retried. The key is +// left alone when the object there is no longer the one stamp describes, as +// when another rename copied onto it, visible yet or not. +func deleteRenameSource(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, source renameSource, stamp *filer_pb.RemoteEntry) error { + current, err := currentEntry(filerClient, dir, name) + if err != nil { + return err + } + if current != nil && len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + return fmt.Errorf("%s still reads %s", util.NewFullPath(dir, name), remote_storage.FormatLocation(source.loc)) + } + live, err := client.StatFile(source.loc) + if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return nil + } + if err != nil { + return err + } + if !describes(stamp, live) || stamp.GetRemoteETag() != "" && live.RemoteETag == "" { + glog.V(0).Infof("keep %s: it is no longer the renamed content", remote_storage.FormatLocation(source.loc)) + return nil + } + return deleteRemoteFile(client, source.loc) } // describes reports whether the object a stat returned is the one the entry's @@ -417,15 +589,15 @@ func openRemoteObject(client remote_storage.RemoteStorageClient, loc *remote_pb. } // uploadCurrentEntry uploads what the filer holds at dir/name now, when the -// event that superseded this rename will not. A superseded rename has -// already deleted the old key; the rewrite behind it in the log uploads the -// destination itself whenever shouldSendToRemote sees it as unsynced, so -// only a current entry that reads synced is uploaded here and the content -// goes up once. A remote-only entry is finished the way -// completeRemoteOnlyRename finishes a rename: its stamp can already describe -// the destination (a sync plus remote.uncache in the meantime), and only -// then is it complete. -func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string) error { +// event that superseded this rename will not. A rename whose snapshot is +// superseded has already deleted the old key; the rewrite behind it in the +// log uploads the destination itself unless shouldSendToRemote skips it on +// the inherited RemoteEntry, whose RemoteMtime can equal the rewrite's mtime +// within the same second. Only that case uploads here, so the content goes +// up once. A remote-only entry is finished the way completeRemoteOnlyRename +// finishes a rename: its stamp can already describe the destination (a sync +// plus remote.uncache in the meantime), and only then is it complete. +func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldSource renameSource, dest *remote_pb.RemoteStorageLocation, storageClass string, sourceOf func(*filer_pb.Entry, renameSource) renameSource) error { current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name)) if errors.Is(err, filer_pb.ErrNotFound) { return nil @@ -437,7 +609,7 @@ func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.F return nil } if isRemoteOnly(current) { - return completeRemoteOnlyRename(filerClient, client, dir, current, oldDest, dest, storageClass) + return completeRemoteOnlyRename(filerClient, client, dir, current, sourceOf(current, oldSource), dest, storageClass) } if shouldSendToRemote(current) { glog.V(0).Infof("leaving %s to the rewrite that superseded the rename", remote_storage.FormatLocation(dest)) @@ -634,6 +806,7 @@ func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano() expected := proto.Clone(entry).(*filer_pb.Entry) entry.RemoteEntry = remoteEntry + delete(entry.Extended, filer.ExtRemoteRenameSourceKey) err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ Directory: dir, diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index a2486d251..0a33aeb30 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "slices" "strings" "testing" "time" @@ -686,12 +687,20 @@ type recordingRemote struct { // chunk ids of each written entry: which version of the file went up writtenChunks [][]string deleteErr error + writeErr error // objects present on the remote, by path: StatFile and ReadFile answer // from it, everything else is ErrRemoteObjectNotFound. objects map[string][]byte + // statETag is the ETag StatFile reports for a present object, unless + // statETags names one for its path. + statETag string + statETags map[string]string } func (r *recordingRemote) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (*filer_pb.RemoteEntry, error) { + if r.writeErr != nil { + return nil, r.writeErr + } r.writes = append(r.writes, loc) var ids []string for _, c := range entry.GetChunks() { @@ -714,7 +723,11 @@ func (r *recordingRemote) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer if !ok { return nil, remote_storage.ErrRemoteObjectNotFound } - return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body))}, nil + etag := r.statETag + if e, ok := r.statETags[loc.Path]; ok { + etag = e + } + return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body)), RemoteETag: etag}, nil } func (r *recordingRemote) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) ([]byte, error) { @@ -1008,7 +1021,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { t.Run("absent object", func(t *testing.T) { remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound} - if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err != nil { + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err != nil { t.Fatalf("err = %v, want nil: deleting an absent object is complete", err) } if len(remote.deletes) != 1 || !proto.Equal(remote.deletes[0], wantDelete) { @@ -1020,7 +1033,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { marked := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) marked.EventNotification.OldEntry.Extended = map[string][]byte{filer.ExtKeepRemoteObjectKey: []byte("true")} remote := &recordingRemote{} - if err := processDeleteEvent(remote, mountedDir, mountLoc, marked); err != nil { + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, marked); err != nil { t.Fatalf("err = %v", err) } if len(remote.deletes) != 0 { @@ -1028,9 +1041,108 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { } }) + t.Run("a remote-only entry that took the path keeps the object", func(t *testing.T) { + // deleted, then a remote-only file renamed onto the path: the rename + // copied its content to this key before the sync got to the delete + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: &filer_pb.Entry{ + Name: "state.db.tmp", + Attributes: &filer_pb.FuseAttributes{FileSize: 10}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10}, + }} + if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + + t.Run("a replaced object is kept", func(t *testing.T) { + // the delete recorded the old object's etag; a rename copied another + // entry's content onto the key since, and it must not go with it + deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`} + filerClient := &stubFilerClient{entry: &filer_pb.Entry{Name: "state.db.tmp", Content: []byte("copied")}} + if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, deleted); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + + t.Run("a changed object with no entry at the path yet is kept", func(t *testing.T) { + // a directory rename copies a child onto the key before its new entry + // becomes visible; the copy may be the content's only remaining home + deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + + t.Run("an unchanged object still goes", func(t *testing.T) { + deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 3} + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("old")}, statETag: `"orig"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 1 { + t.Errorf("deletes = %+v, want one", remote.deletes) + } + }) + + t.Run("a copy that landed before its entry is kept", func(t *testing.T) { + // the deleted entry was never replicated, so its event records no + // stamp to match the fresh rename copy waiting for its filer entry + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + + t.Run("an unverifiable object is kept", func(t *testing.T) { + // the delete recorded an etag the live stat does not report; the + // object cannot be confirmed as the one the delete meant + deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 3} + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("old")}} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + + t.Run("an entry with local data at the path does not stop the delete", func(t *testing.T) { + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: &filer_pb.Entry{ + Name: "state.db.tmp", + Content: []byte("local"), + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10}, + }} + if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 1 { + t.Errorf("deletes = %+v, want one", remote.deletes) + } + }) + t.Run("other failure still fails", func(t *testing.T) { remote := &recordingRemote{deleteErr: errors.New("AccessDenied: Access Denied")} - if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err == nil { + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err == nil { t.Fatal("err = nil, want the delete failure") } }) @@ -1281,7 +1393,7 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) { current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 1024}) remote := &recordingRemote{} filerClient := &stubFilerClient{entry: current} - err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "") + err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", renameSource{loc: wantDelete}, wantWrite, "", eventSource) if err == nil || !strings.Contains(err.Error(), "neither") { t.Errorf("err = %v, want the lost-content failure", err) } @@ -1292,9 +1404,9 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) { t.Run("uncached with the destination already synced: complete without a write", func(t *testing.T) { current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 7}) - remote := &recordingRemote{objects: map[string][]byte{"/b/dir/b.txt": []byte("payload")}} + remote := &recordingRemote{objects: map[string][]byte{"/b/dir/a.txt": []byte("payload"), "/b/dir/b.txt": []byte("payload")}} filerClient := &stubFilerClient{entry: current} - if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, ""); err != nil { + if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", renameSource{loc: wantDelete}, wantWrite, "", eventSource); err != nil { t.Fatalf("err = %v, want nil: the destination object matches the entry's stamp", err) } if len(remote.writes) != 0 { @@ -1306,6 +1418,228 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) { }) } +func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) { + oldDest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/src/a.jpg"} + dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/dst/a.jpg"} + pending := func() *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "a.jpg", + Attributes: &filer_pb.FuseAttributes{FileSize: 4}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteETag: `"copy"`, RemoteSize: 4}, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/src/a.jpg")}, + } + } + objects := func() map[string][]byte { + return map[string][]byte{"/b/src/a.jpg": []byte("data"), "/b/dst/a.jpg": []byte("data")} + } + + t.Run("stamped, then the source goes", func(t *testing.T) { + // the destination already holds the copy; the entry still reads the + // source until its stamp clears the marker + current := pending() + filerClient := &stubFilerClient{entry: current} + remote := &recordingRemote{objects: objects(), statETag: `"copy"`} + if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", current, renameSource{path: "/buckets/b/src/a.jpg", loc: oldDest}, dest, ""); err != nil { + t.Fatalf("err = %v", err) + } + if filerClient.updates != 1 { + t.Errorf("updates = %d, want the stamp that clears the marker", filerClient.updates) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/src/a.jpg" { + t.Errorf("deletes = %+v, want the source", remote.deletes) + } + }) + + t.Run("a marker still in place keeps the source", func(t *testing.T) { + // the stamp did not land: the filer still has the marker + filerClient := &stubFilerClient{entry: pending()} + remote := &recordingRemote{objects: objects(), statETag: `"copy"`} + if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", pending(), renameSource{path: "/buckets/b/src/a.jpg", loc: oldDest}, dest, ""); err == nil { + t.Fatal("err = nil, want the event retried") + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none while the entry reads the source", remote.deletes) + } + }) +} + +// eventSource resolves a rename's source to the event's old key. +func eventSource(_ *filer_pb.Entry, source renameSource) renameSource { + return source +} + +func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) { + // A was renamed over B and B over C, both copies failing: C reads A's + // object through the marker, and B's key holds the old target's bytes + const mountedDir = "/buckets" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"} + remoteOnly := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6} + pendingC := func() *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "c.jpg", + Attributes: &filer_pb.FuseAttributes{FileSize: 6}, + RemoteEntry: remoteOnly, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")}, + } + } + event := func(oldName, newName string) *filer_pb.SubscribeMetadataResponse { + newEntry := pendingC() + newEntry.Name = newName + return &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/b", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: oldName, RemoteEntry: remoteOnly}, + NewParentPath: "/buckets/b", + NewEntry: newEntry, + }, + } + } + objects := func() map[string][]byte { + return map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("old-c")} + } + + t.Run("the first rename leaves the source to the entry that moved on", func(t *testing.T) { + remote := &recordingRemote{objects: objects()} + filerClient := &stubFilerClient{} // nothing at b.jpg any more + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("a.jpg", "b.jpg")); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 0 || len(remote.writes) != 0 { + t.Errorf("deletes = %+v, writes = %+v, want neither: c.jpg still reads a.jpg", remote.deletes, remote.writes) + } + }) + + t.Run("the second rename copies from the marker's source", func(t *testing.T) { + current := pendingC() + remote := &recordingRemote{objects: objects(), statETag: `"a"`} + filerClient := &stubFilerClient{entry: current} + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("b.jpg", "c.jpg")); err != nil { + t.Fatal(err) + } + if len(remote.written) != 1 || string(remote.written[0]) != "data-a" { + t.Errorf("written = %q, want a.jpg's content", remote.written) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/a.jpg" { + t.Errorf("deletes = %+v, want the source once the stamp cleared the marker", remote.deletes) + } + }) +} + +func TestPendingRenameSourceIsReleasedOnlyWhileUnchanged(t *testing.T) { + const mountedDir = "/buckets" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"} + // b.jpg's copy failed after it replaced a target: it read a.jpg's object + pendingB := func() *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "b.jpg", + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")}, + } + } + deleted := func() *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{OldEntry: pendingB()}} + } + deletedPaths := func(r *recordingRemote) (paths []string) { + for _, d := range r.deletes { + paths = append(paths, d.Path) + } + return + } + + t.Run("deleting the pending entry releases its source", func(t *testing.T) { + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a")}, statETag: `"a"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil { + t.Fatal(err) + } + if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the source", got) + } + }) + + t.Run("a source replaced since is kept", func(t *testing.T) { + // another rename copied onto a.jpg before its entry became visible + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-d")}, statETag: `"d"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil { + t.Fatal(err) + } + if got := deletedPaths(remote); slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the replaced source kept", got) + } + }) + + t.Run("a destination cached since releases the source too", func(t *testing.T) { + // c.jpg was cached before the sync reached the rename: it is no + // longer remote-only, and a.jpg is still left over + cached := &filer_pb.Entry{Name: "c.jpg", Content: []byte("data-a"), Attributes: &filer_pb.FuseAttributes{FileSize: 6}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}} + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("data-a")}, statETag: `"a"`} + event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{ + OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: cached, + }} + filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": cached}} + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event); err != nil { + t.Fatal(err) + } + if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the pending source released", got) + } + }) + + t.Run("a cached entry still reading the source keeps it until its upload lands", func(t *testing.T) { + // c.jpg was cached through its marker; until the upload is stamped + // and the marker cleared, a.jpg is what remote.uncache would fall + // back to + marked := func() *filer_pb.Entry { + return &filer_pb.Entry{Name: "c.jpg", Content: []byte("data-a"), Attributes: &filer_pb.FuseAttributes{FileSize: 6}, + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")}} + } + for name, writeErr := range map[string]error{"upload fails": errors.New("upload failed"), "stamp does not land": nil} { + t.Run(name, func(t *testing.T) { + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b")}, statETag: `"a"`, writeErr: writeErr} + event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{ + OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: marked(), + }} + // the filer keeps the marker: the stub does not store updates + filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": marked()}} + _ = processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event) + if got := deletedPaths(remote); slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the source kept while the entry reads it", got) + } + }) + } + }) + + t.Run("a later rename the filer copied releases the source", func(t *testing.T) { + // b.jpg was renamed to c.jpg and the filer copied a.jpg there at once + copied := &filer_pb.Entry{Name: "c.jpg", Attributes: &filer_pb.FuseAttributes{FileSize: 6}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}} + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("data-a")}, statETag: `"a"`} + event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{ + OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: copied, + }} + filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": copied}} + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event); err != nil { + t.Fatal(err) + } + if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the pending source released", got) + } + }) +} + +// pathFilerClient serves entries by path; a path it lacks is not found. +type pathFilerClient struct { + stubFilerClient + entries map[string]*filer_pb.Entry +} + +func (c *pathFilerClient) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entries[req.Directory+"/"+req.Name]}, nil +} + +func (c *pathFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { + return fn(c) +} + func TestShouldSendToRemote(t *testing.T) { const mtime = 1786096669 @@ -1383,3 +1717,37 @@ func TestMetadataOnlyUpdateRestampsSyncTs(t *testing.T) { t.Error("the restamped entry should read synced") } } + +func TestCompleteRemoteOnlyRenameKeepsAReusedOldKey(t *testing.T) { + // A was renamed to B, then D to A before the sync got to the first + // rename: the object at A is now D's, and the only copy of it + oldDest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/a.jpg"} + dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/b.jpg"} + current := &filer_pb.Entry{Name: "b.jpg", RemoteEntry: &filer_pb.RemoteEntry{RemoteETag: `"a"`, RemoteSize: 6}} + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-d"), "/b/b.jpg": []byte("data-a")}, statETag: `"a"`} + remote.statETags = map[string]string{"/b/a.jpg": `"d"`} + filerClient := &stubFilerClient{entry: current} + if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b", current, renameSource{path: "/buckets/b/a.jpg", loc: oldDest}, dest, ""); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want the reused key kept", remote.deletes) + } +} + +func TestCompleteRemoteOnlyRenameRetryUsesTheSourceStamp(t *testing.T) { + // an earlier run copied a multipart source and stamped the entry with + // the copy's ETag, then failed to delete the source + oldDest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/a.jpg"} + dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/b.jpg"} + current := &filer_pb.Entry{Name: "b.jpg", RemoteEntry: &filer_pb.RemoteEntry{RemoteETag: `"copy"`, RemoteSize: 6}} + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("data-a")}, + statETags: map[string]string{"/b/a.jpg": `"a-2"`, "/b/b.jpg": `"copy"`}} + source := renameSource{path: "/buckets/b/a.jpg", loc: oldDest, stamp: &filer_pb.RemoteEntry{RemoteETag: `"a-2"`, RemoteSize: 6}} + if err := completeRemoteOnlyRename(&stubFilerClient{entry: current}, remote, "/buckets/b", current, source, dest, ""); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/a.jpg" { + t.Errorf("deletes = %+v, want the unchanged source deleted on the retry", remote.deletes) + } +} diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 7c642cd6a..09110adfa 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -225,3 +225,97 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, glog.V(3).InfofCtx(ctx, "maybeDeleteFromRemote: deleted %s from remote", entry.FullPath) return true, nil } + +// DeleteFromRemote deletes the object, or for a directory every object, the +// entry maps to on its mounted remote storage. +func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error { + _, err := f.maybeDeleteFromRemote(ctx, entry) + return err +} + +// ExtRemoteRenameSourceKey marks a remote-only entry whose rename committed +// while its copy to the new key was still pending: it holds the path whose +// object remote-only reads must serve until the copy stamps the entry. +const ExtRemoteRenameSourceKey = "Seaweed-X-Remote-Rename-Source" + +// RemoteSourcePath returns the path whose remote object holds the content of +// the remote-only entry at p: the renamed-from path a pending deferred copy +// still reads, or p itself. A marker leading out of p's mount is ignored, so +// whoever can set extended attributes cannot point reads at another mount. +func (f *Filer) RemoteSourcePath(entry *Entry, p util.FullPath) util.FullPath { + src := util.FullPath(entry.Extended[ExtRemoteRenameSourceKey]) + if src == "" || f.RemoteStorage == nil { + return p + } + mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(p) + if remoteLoc == nil { + return p + } + if srcMountDir, _ := f.RemoteStorage.FindMountDirectory(src); srcMountDir != mountDir { + return p + } + return src +} + +// RemoteReadPath returns the dir and name whose object a remote-only read of +// the entry at dir/name serves. +func (f *Filer) RemoteReadPath(entry *Entry, dir, name string) (string, string) { + return f.RemoteSourcePath(entry, util.NewFullPath(dir, name)).DirAndName() +} + +// CopyRemoteOnlyEntry copies the remote object of a renamed entry whose +// content exists only on the remote to the key of newPath, and returns the +// RemoteEntry of the copy. The rename deletes the old key along with the old +// entry, and with no local data nothing would write the new key. It returns +// nil when no copy is needed: the entry has local data, which the remote sync +// uploads to the new key, it is empty, which the sync writes as an empty +// object, or it is not under a remote mount. A rename that +// moves such an entry out of its mount, or onto a remote that cannot copy, is +// refused, since it would delete the only copy of the content. +func (f *Filer) CopyRemoteOnlyEntry(ctx context.Context, entry *Entry, oldPath, newPath util.FullPath) (*filer_pb.RemoteEntry, error) { + src, dst, copier, err := f.RemoteCopyLocations(ctx, entry, oldPath, newPath) + if err != nil || copier == nil { + return nil, err + } + remoteEntry, err := copier.CopyFile(src, dst) + if err != nil { + return nil, fmt.Errorf("copy remote %s to %s: %w", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst), err) + } + glog.V(3).InfofCtx(ctx, "CopyRemoteOnlyEntry: copied %s to %s", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst)) + return remoteEntry, nil +} + +// RemoteCopyLocations resolves the object locations a remote-only rename copy +// of entry between oldPath and newPath would use, and the client able to copy +// it. A nil copier means the entry needs no copy; an error means it needs one +// that cannot be served, which a rename must refuse before committing. +func (f *Filer) RemoteCopyLocations(ctx context.Context, entry *Entry, oldPath, newPath util.FullPath) (src, dst *remote_pb.RemoteStorageLocation, copier remote_storage.RemoteStorageObjectCopier, err error) { + if f.RemoteStorage == nil || entry.IsDirectory() || !entry.IsInRemoteOnly() || len(entry.Content) > 0 { + return nil, nil, nil, nil + } + + mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(oldPath) + if remoteLoc == nil { + return nil, nil, nil, nil + } + if newMountDir, _ := f.RemoteStorage.FindMountDirectory(newPath); newMountDir != mountDir { + return nil, nil, nil, fmt.Errorf("%s exists only on the remote of mount %s and cannot be renamed out of it; cache it first", oldPath, mountDir) + } + + remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name) + if !found { + return nil, nil, nil, fmt.Errorf("resolve remote storage client for %s: not found", oldPath) + } + client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf) + if clientErr != nil { + return nil, nil, nil, fmt.Errorf("resolve remote storage client for %s: %w", oldPath, clientErr) + } + copier, ok := client.(remote_storage.RemoteStorageObjectCopier) + if !ok { + return nil, nil, nil, fmt.Errorf("%s exists only on the remote and remote storage type %s cannot copy it to the new name; cache it first", oldPath, remoteConf.Type) + } + + src = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, f.RemoteSourcePath(entry, oldPath)) + dst = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, newPath) + return src, dst, copier, nil +} diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index f1845e1b0..0cefe362a 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -110,6 +110,14 @@ type RemoteStorageStreamReader interface { ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) } +// RemoteStorageObjectCopier is an optional interface for remote storage clients +// that copy an object to another key within the storage, without the content +// passing through the client. A rename of an entry whose content exists only +// on the remote needs it. +type RemoteStorageObjectCopier interface { + CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) +} + // CacheWaitTimeout is how long a read of an uncached remote-only object waits // for the local cache before serving another way: small files wait longer since // their cache completes quickly, large files fail fast for better TTFB. A mount diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index a57a063cf..4166df8a4 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -610,7 +610,7 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL copyInput := &s3.CopyObjectInput{ Bucket: aws.String(loc.Bucket), Key: aws.String(key), - CopySource: aws.String(url.PathEscape(loc.Bucket + "/" + key)), + CopySource: aws.String(s3CopySource(loc.Bucket, key)), MetadataDirective: aws.String(s3.MetadataDirectiveReplace), Metadata: headOut.Metadata, ContentType: headOut.ContentType, @@ -663,6 +663,168 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL } return } + +// s3CopyPartSize is the smallest part a multipart copy uses; parts grow so +// the largest object stays within the 10000-part limit +const s3CopyPartSize = 512 * 1024 * 1024 + +// CopyFile copies the object at src to dst on the server side, with a single +// CopyObject up to the 5 GB limit and a multipart copy above it. The copy +// keeps the source's metadata. +func (s *s3RemoteStorageClient) CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) { + srcKey, dstKey := src.Path[1:], dst.Path[1:] + headOut, err := s.conn.HeadObject(&s3.HeadObjectInput{ + Bucket: aws.String(src.Bucket), + Key: aws.String(srcKey), + }) + if err != nil { + if reqErr, ok := err.(awserr.RequestFailure); ok && reqErr.StatusCode() == http.StatusNotFound { + return nil, remote_storage.ErrRemoteObjectNotFound + } + return nil, fmt.Errorf("stat s3 %s/%s before copy: %w", src.Bucket, srcKey, err) + } + copySource := aws.String(s3CopySource(src.Bucket, srcKey)) + var storageClass *string + if s.conf.S3StorageClass != "" { + storageClass = aws.String(s.conf.S3StorageClass) + } + + if size := aws.Int64Value(headOut.ContentLength); size <= s3CopyObjectSizeLimit { + // a copy takes the bucket's default encryption unless told otherwise + if _, err = s.conn.CopyObject(&s3.CopyObjectInput{ + Bucket: aws.String(dst.Bucket), + Key: aws.String(dstKey), + CopySource: copySource, + StorageClass: storageClass, + ServerSideEncryption: headOut.ServerSideEncryption, + SSEKMSKeyId: headOut.SSEKMSKeyId, + }); err != nil { + return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err) + } + } else if err = s.multipartCopy(headOut, src, copySource, size, dst.Bucket, dstKey, storageClass); err != nil { + return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err) + } + + return s.readFileRemoteEntry(dst) +} + +// s3CopySource escapes each segment of the copy source and keeps the slashes, +// which every S3 implementation reads back +func s3CopySource(bucket, key string) string { + segments := strings.Split(bucket+"/"+key, "/") + for i, segment := range segments { + segments[i] = url.PathEscape(segment) + } + return strings.Join(segments, "/") +} + +// s3TaggingHeader encodes tags as the query string x-amz-tagging takes, with +// spaces as %20: a + is read back as a space by some implementations and +// literally by others +func s3TaggingHeader(tagSet []*s3.Tag) string { + escape := func(s string) string { + return strings.ReplaceAll(url.QueryEscape(s), "+", "%20") + } + pairs := make([]string, 0, len(tagSet)) + for _, tag := range tagSet { + pairs = append(pairs, escape(aws.StringValue(tag.Key))+"="+escape(aws.StringValue(tag.Value))) + } + return strings.Join(pairs, "&") +} + +// multipartCopy copies an object above the CopyObject limit part by part. A +// multipart upload does not inherit the source's metadata or tags, so they are +// carried over like CopyObject would: metadata from the source's HeadObject, +// tags when the remote supports tagging. +func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src *remote_pb.RemoteStorageLocation, copySource *string, size int64, bucket, key string, storageClass *string) error { + createInput := &s3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + Metadata: headOut.Metadata, + ContentType: headOut.ContentType, + ContentEncoding: headOut.ContentEncoding, + CacheControl: headOut.CacheControl, + ContentDisposition: headOut.ContentDisposition, + ContentLanguage: headOut.ContentLanguage, + WebsiteRedirectLocation: headOut.WebsiteRedirectLocation, + ServerSideEncryption: headOut.ServerSideEncryption, + SSEKMSKeyId: headOut.SSEKMSKeyId, + StorageClass: storageClass, + } + if headOut.Expires != nil { + if expires, parseErr := http.ParseTime(*headOut.Expires); parseErr == nil { + createInput.Expires = aws.Time(expires) + } + } + if s.conf.S3SupportTagging { + tagOut, tagErr := s.conn.GetObjectTagging(&s3.GetObjectTaggingInput{ + Bucket: aws.String(src.Bucket), + Key: aws.String(src.Path[1:]), + }) + if tagErr != nil { + return fmt.Errorf("read tags: %w", tagErr) + } + if len(tagOut.TagSet) > 0 { + createInput.Tagging = aws.String(s3TaggingHeader(tagOut.TagSet)) + } + } + createOut, err := s.conn.CreateMultipartUpload(createInput) + if err != nil { + return err + } + + partSize := int64(s3CopyPartSize) + for partSize*10000 < size { + partSize *= 2 + } + var parts []*s3.CompletedPart + for offset, partNumber := int64(0), int64(1); offset < size; offset, partNumber = offset+partSize, partNumber+1 { + partOut, partErr := s.conn.UploadPartCopy(&s3.UploadPartCopyInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: createOut.UploadId, + PartNumber: aws.Int64(partNumber), + CopySource: copySource, + CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", offset, min(offset+partSize, size)-1)), + // every part from the object the HeadObject saw, not a mix of + // versions if it is overwritten mid-copy + CopySourceIfMatch: headOut.ETag, + }) + if partErr == nil && partOut.CopyPartResult == nil { + partErr = errors.New("no part result") + } + if partErr != nil { + s.abortMultipartUpload(bucket, key, createOut.UploadId) + return fmt.Errorf("copy part %d: %w", partNumber, partErr) + } + parts = append(parts, &s3.CompletedPart{ + ETag: partOut.CopyPartResult.ETag, + PartNumber: aws.Int64(partNumber), + }) + } + + if _, err = s.conn.CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: createOut.UploadId, + MultipartUpload: &s3.CompletedMultipartUpload{Parts: parts}, + }); err != nil { + s.abortMultipartUpload(bucket, key, createOut.UploadId) + return err + } + return nil +} + +func (s *s3RemoteStorageClient) abortMultipartUpload(bucket, key string, uploadId *string) { + if _, err := s.conn.AbortMultipartUpload(&s3.AbortMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: uploadId, + }); err != nil { + glog.Warningf("abort multipart copy to s3 %s/%s: %v", bucket, key, err) + } +} + func (s *s3RemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) { _, err = s.conn.DeleteObject(&s3.DeleteObjectInput{ Bucket: aws.String(loc.Bucket), diff --git a/weed/remote_storage/s3/s3_storage_client_copy_test.go b/weed/remote_storage/s3/s3_storage_client_copy_test.go new file mode 100644 index 000000000..a20fbf57a --- /dev/null +++ b/weed/remote_storage/s3/s3_storage_client_copy_test.go @@ -0,0 +1,169 @@ +package s3 + +import ( + "errors" + "net/http" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/awserr" + awss3 "github.com/aws/aws-sdk-go/service/s3" + "github.com/aws/aws-sdk-go/service/s3/s3iface" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/stretchr/testify/require" +) + +// copyObjectMock serves HeadObject for a source of size bytes and records the +// copy calls. +type copyObjectMock struct { + s3iface.S3API + size int64 + missing bool + partErr error + copyInputs []*awss3.CopyObjectInput + partInputs []*awss3.UploadPartCopyInput + createInput *awss3.CreateMultipartUploadInput + tagInputs []*awss3.GetObjectTaggingInput + completed *awss3.CompleteMultipartUploadInput + aborted bool +} + +func (m *copyObjectMock) HeadObject(input *awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) { + if m.missing { + return nil, awserr.NewRequestFailure(awserr.New("NotFound", "not found", nil), http.StatusNotFound, "") + } + return &awss3.HeadObjectOutput{ + ContentLength: aws.Int64(m.size), + ContentType: aws.String("image/jpeg"), + Expires: aws.String("Wed, 21 Oct 2026 07:28:00 GMT"), + WebsiteRedirectLocation: aws.String("/elsewhere"), + ServerSideEncryption: aws.String("aws:kms"), + SSEKMSKeyId: aws.String("key-1"), + ETag: aws.String(`"etag-` + aws.StringValue(input.Key) + `"`), + }, nil +} + +func (m *copyObjectMock) GetObjectTagging(input *awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { + m.tagInputs = append(m.tagInputs, input) + return &awss3.GetObjectTaggingOutput{TagSet: []*awss3.Tag{ + {Key: aws.String("album"), Value: aws.String("2026 & co")}, + {Key: aws.String("a+b"), Value: aws.String("x=y")}, + }}, nil +} + +func (m *copyObjectMock) CopyObject(input *awss3.CopyObjectInput) (*awss3.CopyObjectOutput, error) { + m.copyInputs = append(m.copyInputs, input) + return &awss3.CopyObjectOutput{}, nil +} + +func (m *copyObjectMock) CreateMultipartUpload(input *awss3.CreateMultipartUploadInput) (*awss3.CreateMultipartUploadOutput, error) { + m.createInput = input + return &awss3.CreateMultipartUploadOutput{UploadId: aws.String("upload-1")}, nil +} + +func (m *copyObjectMock) UploadPartCopy(input *awss3.UploadPartCopyInput) (*awss3.UploadPartCopyOutput, error) { + m.partInputs = append(m.partInputs, input) + if m.partErr != nil && len(m.partInputs) == 2 { + return nil, m.partErr + } + return &awss3.UploadPartCopyOutput{CopyPartResult: &awss3.CopyPartResult{ETag: aws.String(`"part"`)}}, nil +} + +func (m *copyObjectMock) CompleteMultipartUpload(input *awss3.CompleteMultipartUploadInput) (*awss3.CompleteMultipartUploadOutput, error) { + m.completed = input + return &awss3.CompleteMultipartUploadOutput{}, nil +} + +func (m *copyObjectMock) AbortMultipartUpload(*awss3.AbortMultipartUploadInput) (*awss3.AbortMultipartUploadOutput, error) { + m.aborted = true + return &awss3.AbortMultipartUploadOutput{}, nil +} + +func newCopyTestClient(mock *copyObjectMock) *s3RemoteStorageClient { + return &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} +} + +var ( + copySrc = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/src/a b.jpg"} + copyDst = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dst/a b.jpg"} +) + +func TestS3ClientImplementsObjectCopier(t *testing.T) { + var _ remote_storage.RemoteStorageObjectCopier = &s3RemoteStorageClient{} +} + +func TestS3CopyFileUsesCopyObjectUpToTheLimit(t *testing.T) { + mock := ©ObjectMock{size: s3CopyObjectSizeLimit} + + remoteEntry, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Len(t, mock.copyInputs, 1) + require.Equal(t, "bucket/src/a%20b.jpg", aws.StringValue(mock.copyInputs[0].CopySource)) + require.Equal(t, "dst/a b.jpg", aws.StringValue(mock.copyInputs[0].Key)) + require.Equal(t, "aws:kms", aws.StringValue(mock.copyInputs[0].ServerSideEncryption)) + require.Equal(t, "key-1", aws.StringValue(mock.copyInputs[0].SSEKMSKeyId)) + require.Nil(t, mock.createInput) + // the RemoteEntry describes the destination + require.Equal(t, `"etag-dst/a b.jpg"`, remoteEntry.RemoteETag) + require.Equal(t, int64(s3CopyObjectSizeLimit), remoteEntry.RemoteSize) +} + +func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) { + size := int64(s3CopyObjectSizeLimit) + 1 + mock := ©ObjectMock{size: size} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Empty(t, mock.copyInputs) + require.NotNil(t, mock.createInput) + require.Equal(t, "image/jpeg", aws.StringValue(mock.createInput.ContentType)) + require.Equal(t, "/elsewhere", aws.StringValue(mock.createInput.WebsiteRedirectLocation)) + require.Equal(t, "aws:kms", aws.StringValue(mock.createInput.ServerSideEncryption)) + require.Equal(t, "key-1", aws.StringValue(mock.createInput.SSEKMSKeyId)) + require.Equal(t, int64(1792567680), aws.TimeValue(mock.createInput.Expires).Unix()) + // tags only where the remote supports tagging + require.Empty(t, mock.tagInputs) + require.Nil(t, mock.createInput.Tagging) + require.Len(t, mock.partInputs, 11) + require.Equal(t, "bytes=0-536870911", aws.StringValue(mock.partInputs[0].CopySourceRange)) + require.Equal(t, "bytes=5368709120-5368709120", aws.StringValue(mock.partInputs[10].CopySourceRange)) + for _, part := range mock.partInputs { + require.Equal(t, `"etag-src/a b.jpg"`, aws.StringValue(part.CopySourceIfMatch)) + } + require.Len(t, mock.completed.MultipartUpload.Parts, 11) + require.Equal(t, int64(11), aws.Int64Value(mock.completed.MultipartUpload.Parts[10].PartNumber)) + require.False(t, mock.aborted) +} + +func TestS3CopyFileCarriesTagsIntoAMultipartCopy(t *testing.T) { + mock := ©ObjectMock{size: s3CopyObjectSizeLimit + 1} + client := newCopyTestClient(mock) + client.conf.S3SupportTagging = true + + _, err := client.CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Len(t, mock.tagInputs, 1) + require.Equal(t, "src/a b.jpg", aws.StringValue(mock.tagInputs[0].Key)) + require.Equal(t, "album=2026%20%26%20co&a%2Bb=x%3Dy", aws.StringValue(mock.createInput.Tagging)) +} + +func TestS3CopyFileAbortsAFailedMultipartCopy(t *testing.T) { + mock := ©ObjectMock{size: 2 * s3CopyObjectSizeLimit, partErr: errors.New("part failed")} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.ErrorContains(t, err, "part failed") + require.True(t, mock.aborted) + require.Nil(t, mock.completed) +} + +func TestS3CopyFileMissingSource(t *testing.T) { + mock := ©ObjectMock{missing: true} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.ErrorIs(t, err, remote_storage.ErrRemoteObjectNotFound) + require.Empty(t, mock.copyInputs) +} diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go index 8a56d466a..892789fb7 100644 --- a/weed/server/filer_grpc_server_remote.go +++ b/weed/server/filer_grpc_server_remote.go @@ -96,7 +96,8 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re glog.V(1).Infof("CacheRemoteObjectToLocalCluster: caching %s/%s (remote size: %d)", req.Directory, req.Name, entry.Remote.RemoteSize) - storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, req.Directory, req.Name) + readDir, readName := fs.filer.RemoteReadPath(entry, req.Directory, req.Name) + storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, readDir, readName) if err != nil { return nil, err } diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 020a59c71..c55ddd46b 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -2,7 +2,9 @@ package weed_server import ( "context" + "errors" "fmt" + "maps" "path/filepath" "time" @@ -45,6 +47,9 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom defer fs.acquireRenamePathLocks("AtomicRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))() + // a SQL store keeps the transaction in the context and refuses calls + // through it once committed; the work after the commit uses this one + afterCommitCtx := ctx ctx, err := fs.filer.BeginTransaction(ctx) if err != nil { return nil, err @@ -58,7 +63,8 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom var metadataEvents []metadataEvent var pendingChunkDeletes []*filer_pb.FileChunk - moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes) + var pendingRemoteCopies []*pendingRemoteCopy + moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies) if moveErr != nil { fs.filer.RollbackTransaction(ctx) return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr) @@ -75,8 +81,10 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom if len(pendingChunkDeletes) > 0 { fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } + fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) + applyRemoteCopies(metadataEvents, pendingRemoteCopies) for _, event := range metadataEvents { - event.notify(fs.filer, ctx, req.Signatures) + event.notify(fs.filer, afterCommitCtx, req.Signatures) } return &filer_pb.AtomicRenameEntryResponse{}, nil @@ -97,6 +105,7 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, ctx := context.Background() + afterCommitCtx := ctx ctx, err = fs.filer.BeginTransaction(ctx) if err != nil { return err @@ -125,7 +134,8 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, var metadataEvents []metadataEvent var pendingChunkDeletes []*filer_pb.FileChunk - moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes) + var pendingRemoteCopies []*pendingRemoteCopy + moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies) if moveErr != nil { fs.filer.RollbackTransaction(ctx) return fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr) @@ -138,8 +148,10 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, if len(pendingChunkDeletes) > 0 { fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } + fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) + applyRemoteCopies(metadataEvents, pendingRemoteCopies) for _, event := range metadataEvents { - event.notify(fs.filer, ctx, req.Signatures) + event.notify(fs.filer, afterCommitCtx, req.Signatures) } return nil @@ -149,29 +161,119 @@ type metadataEvent struct { oldEntry *filer.Entry newEntry *filer.Entry deleteChunks bool + // the old entry's remote object is deleted once the rename has committed: + // it was copied to the new key, and a rename that fails must not lose it + deleteOldRemote bool +} + +// pendingRemoteCopy is a remote-only rename over a remote-only target whose +// copy runs after the commit: the store transaction cannot roll a remote +// write back, so the new entry keeps the old key's stamp until the copy lands. +type pendingRemoteCopy struct { + source *filer.Entry + oldPath util.FullPath + newPath util.FullPath + failed bool + copied *filer_pb.RemoteEntry +} + +// finishRemoteCopies runs the deferred remote copies after the commit. A +// failure leaves the new entry pointing at the old key, whose object the +// caller then keeps: remote.sync completes the copy the same way it does a +// rename whose event it finds unfinished. +func (fs *FilerServer) finishRemoteCopies(ctx context.Context, pendings []*pendingRemoteCopy) { + for _, pending := range pendings { + remoteEntry, err := fs.filer.CopyRemoteOnlyEntry(ctx, pending.source, pending.oldPath, pending.newPath) + if err != nil { + glog.WarningfCtx(ctx, "remote rename copy of %s: %v", pending.newPath, err) + pending.failed = true + continue + } + entry, findErr := fs.filer.FindEntry(ctx, pending.newPath) + if findErr != nil { + glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, findErr) + pending.failed = true + continue + } + entry.Remote = remoteEntry + delete(entry.Extended, filer.ExtRemoteRenameSourceKey) + if updateErr := fs.filer.UpdateEntry(ctx, entry, entry, false); updateErr != nil { + glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, updateErr) + pending.failed = true + continue + } + pending.copied = remoteEntry + } +} + +// applyRemoteCopies brings the queued rename events in line with the deferred +// copies. The stamp is written with a plain UpdateEntry, which publishes no +// event, so the rename event carries it, without the pending-source marker, +// to peer filers and subscribers. A failed copy keeps the old object, which +// the entry still reads through the marker. +func applyRemoteCopies(events []metadataEvent, pendings []*pendingRemoteCopy) { + for i := range events { + event := &events[i] + for _, pending := range pendings { + if event.oldEntry.FullPath != pending.oldPath { + continue + } + if pending.failed { + event.deleteOldRemote = false + } else if pending.copied != nil && event.newEntry != nil { + newEntry := event.newEntry.ShallowClone() + newEntry.Remote = pending.copied + newEntry.Extended = maps.Clone(newEntry.Extended) + delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey) + event.newEntry = newEntry + } + } + } } func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signatures []int32) { + if event.deleteOldRemote { + deleteRenamedRemoteObject(f, ctx, event.oldEntry) + } f.NotifyUpdateEvent(ctx, event.oldEntry, event.newEntry, event.deleteChunks, false, signatures) } -func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error { +// deleteRenamedRemoteObject deletes the remote object a committed rename no +// longer needs. A child path is not locked by its directory's rename, so +// another rename may have put a remote-only entry at the key meanwhile; the +// store is asked directly, a lazy fetch would revive the key. When in doubt +// the object stays: an extra object loses nothing. +func deleteRenamedRemoteObject(f *filer.Filer, ctx context.Context, entry *filer.Entry) { + current, findErr := f.Store.FindEntry(ctx, entry.FullPath) + switch { + case findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound): + glog.WarningfCtx(ctx, "keep remote object of renamed %s: look up its old path: %v", entry.FullPath, findErr) + case findErr == nil && current.IsInRemoteOnly(): + glog.V(0).InfofCtx(ctx, "keep remote object of renamed %s: a remote-only entry has taken its path", entry.FullPath) + default: + if err := f.DeleteFromRemote(ctx, entry); err != nil { + glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", entry.FullPath, err) + } + } +} + +func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error { if err := fs.moveSelfEntry(ctx, stream, oldParent, entry, newParent, newName, func() error { if entry.IsDirectory() { - if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures, metadataEvents, pendingChunkDeletes); err != nil { + if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures, metadataEvents, pendingChunkDeletes, pendingRemoteCopies); err != nil { return err } } return nil - }, signatures, skipTargetLookup, metadataEvents, pendingChunkDeletes); err != nil { + }, signatures, skipTargetLookup, metadataEvents, pendingChunkDeletes, pendingRemoteCopies); err != nil { return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(entry.Name()), newParent.Child(newName), err) } return nil } -func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error { +func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error { currentDirPath := oldParent.Child(entry.Name()) newDirPath := newParent.Child(newName) @@ -194,7 +296,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb // println("processing", lastFileName) newChildPath := newDirPath.Child(item.Name()) skipTarget := fs.filer.Store.SameActualStore(newDirPath, newChildPath) - err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures, skipTarget, metadataEvents, pendingChunkDeletes) + err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures, skipTarget, metadataEvents, pendingChunkDeletes, pendingRemoteCopies) if err != nil { return err } @@ -206,7 +308,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb return nil } -func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error { +func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error { oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName) @@ -235,8 +337,48 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee case !existingTarget.IsDirectory() && entry.IsDirectory(): return fmt.Errorf("%s: %w", existingTarget.FullPath, filer_pb.ErrExistingIsFile) } + } + + // a remote-only entry is copied on the remote before anything changes: + // deleting the old entry deletes the old object. Over a remote-only + // target the copy runs after the commit instead — it overwrites the + // target's only content, which a rolled-back commit cannot restore — + // and the new entry reads the old key until the copy stamps it. + deferredCopy := false + var copiedRemote *filer_pb.RemoteEntry + if existingTarget != nil && existingTarget.IsInRemoteOnly() && entry.IsInRemoteOnly() { + if _, _, copier, copyErr := fs.filer.RemoteCopyLocations(ctx, entry, oldPath, newPath); copyErr != nil { + return copyErr + } else if copier != nil { + deferredCopy = true + *pendingRemoteCopies = append(*pendingRemoteCopies, &pendingRemoteCopy{source: sourceEntry, oldPath: oldPath, newPath: newPath}) + } + } + if !deferredCopy { + var copyErr error + copiedRemote, copyErr = fs.filer.CopyRemoteOnlyEntry(ctx, entry, oldPath, newPath) + if copyErr != nil { + return copyErr + } + } + remoteEntry := entry.Remote + if copiedRemote != nil { + remoteEntry = copiedRemote + } + + if existingTarget != nil { + targetCtx := filer.WithSuppressedMetadataEvents(ctx) + if copiedRemote != nil || deferredCopy { + // the copy has replaced or still needs the target's object; keep it + targetCtx = filer.WithKeepRemoteObject(targetCtx) + existingTarget.Extended = maps.Clone(existingTarget.Extended) + if existingTarget.Extended == nil { + existingTarget.Extended = map[string][]byte{} + } + existingTarget.Extended[filer.ExtKeepRemoteObjectKey] = []byte("true") + } if deleteErr := fs.filer.DeleteEntryMetaAndData( - filer.WithSuppressedMetadataEvents(ctx), + targetCtx, newPath, false, false, @@ -258,9 +400,23 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee Content: entry.Content, HardLinkCounter: entry.HardLinkCounter, HardLinkId: entry.HardLinkId, - Remote: entry.Remote, + Remote: remoteEntry, Quota: entry.Quota, } + if deferredCopy { + // a pending entry renamed again still reads its original source + newEntry.Extended = maps.Clone(newEntry.Extended) + if newEntry.Extended == nil { + newEntry.Extended = map[string][]byte{} + } + newEntry.Extended[filer.ExtRemoteRenameSourceKey] = []byte(fs.filer.RemoteSourcePath(entry, oldPath)) + } else if copiedRemote != nil && len(newEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + // the copy came from the pending source; the new key holds it now. + // The source itself is left to remote.sync, which deletes it only + // while it is still the object the rename event recorded + newEntry.Extended = maps.Clone(newEntry.Extended) + delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey) + } if skipTargetLookup { if newEntry.FullPath.IsLongerFileName(fs.filer.MaxFilenameLength) { return filer_pb.ErrEntryNameTooLong @@ -311,19 +467,36 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee }) } *metadataEvents = append(*metadataEvents, metadataEvent{ - oldEntry: sourceEntry, - newEntry: newEntry, + oldEntry: sourceEntry, + newEntry: newEntry, + deleteOldRemote: copiedRemote != nil || deferredCopy, }) + selfEvent := len(*metadataEvents) - 1 + keepOldRemote := copiedRemote != nil || deferredCopy if moveFolderSubEntries != nil { if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil { return moveChildrenErr } + // removing the old directory on the remote removes every object under + // its prefix, the copied children's old keys included, which must wait + // for the commit. Those are deleted one by one after it instead, and + // nothing else under the prefix is touched. + for _, event := range (*metadataEvents)[selfEvent+1:] { + if event.deleteOldRemote { + keepOldRemote = true + break + } + } } // delete old entry ctx = context.WithValue(ctx, "OP", "MV") - deleteErr := fs.filer.DeleteEntryMetaAndData(filer.WithSuppressedMetadataEvents(ctx), oldPath, false, false, false, false, signatures, 0) + deleteCtx := filer.WithSuppressedMetadataEvents(ctx) + if keepOldRemote { + deleteCtx = filer.WithKeepRemoteObject(deleteCtx) + } + deleteErr := fs.filer.DeleteEntryMetaAndData(deleteCtx, oldPath, false, false, false, false, signatures, 0) if deleteErr != nil { return deleteErr } diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go new file mode 100644 index 000000000..445b1b9ac --- /dev/null +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -0,0 +1,430 @@ +package weed_server + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +// renameRemoteClient records the remote calls a rename makes. It cannot copy; +// copyingRenameRemoteClient adds CopyFile. +type renameRemoteClient struct { + remote_storage.RemoteStorageClient + mu sync.Mutex + calls []string + copyErr error +} + +func (c *renameRemoteClient) record(call string) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, call) +} + +func (c *renameRemoteClient) recorded() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.calls...) +} + +func (c *renameRemoteClient) StatFile(*remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) { + return nil, remote_storage.ErrRemoteObjectNotFound +} + +func (c *renameRemoteClient) ListDirectory(context.Context, *remote_pb.RemoteStorageLocation, remote_storage.VisitFunc) error { + return nil +} + +func (c *renameRemoteClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) error { + c.record("delete " + loc.Bucket + loc.Path) + return nil +} + +func (c *renameRemoteClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) error { + c.record("rmdir " + loc.Bucket + loc.Path) + return nil +} + +type copyingRenameRemoteClient struct { + *renameRemoteClient +} + +func (c copyingRenameRemoteClient) CopyFile(src, dst *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) { + if c.copyErr != nil { + return nil, c.copyErr + } + c.record("copy " + src.Bucket + src.Path + " " + dst.Bucket + dst.Path) + return &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1800000000, RemoteETag: `"copy"`}, nil +} + +// newRenameRemoteTestServer mounts /buckets/b, /buckets/c and /data/m on r1, +// through client. +func newRenameRemoteTestServer(t *testing.T, client remote_storage.RemoteStorageClient) (*FilerServer, *renameTestStore) { + t.Helper() + store := newRenameTestStore() + f := newRenameTestFiler(t, store) + f.BuildGuardedRemoteClient = func(context.Context, *remote_pb.RemoteConf, bool) (remote_storage.RemoteStorageClient, error) { + return client, nil + } + mapping, err := proto.Marshal(&remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/b": {Name: "r1", Bucket: "origin", Path: "/"}, + "/buckets/c": {Name: "r1", Bucket: "other", Path: "/"}, + "/data/m": {Name: "r1", Bucket: "origin", Path: "/m"}, + }}) + require.NoError(t, err) + conf, err := proto.Marshal(&remote_pb.RemoteConf{Name: "r1", Type: "s3"}) + require.NoError(t, err) + for name, content := range map[string][]byte{filer.REMOTE_STORAGE_MOUNT_FILE: mapping, "r1" + filer.REMOTE_STORAGE_CONF_SUFFIX: conf} { + entry := newFileEntry(filer.DirectoryEtcRemote+"/"+name, 1) + entry.Content = content + store.entries[string(entry.FullPath)] = entry + } + require.NoError(t, f.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(f)) + for _, dir := range []string{"/buckets", "/buckets/b", "/buckets/b/src", "/buckets/b/dst", "/buckets/c", "/data", "/data/m"} { + store.entries[dir] = newDirectoryEntry(dir, 2) + } + return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}, store +} + +func remoteOnlyRenameEntry(path string, inode uint64) *filer.Entry { + entry := newFileEntry(path, inode) + entry.FileSize = 10 + entry.Remote = &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1700000000, RemoteETag: `"orig"`} + return entry +} + +func renameFile(server *FilerServer, oldPath, newPath util.FullPath) error { + oldDir, oldName := oldPath.DirAndName() + newDir, newName := newPath.DirAndName() + _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ + OldDirectory: oldDir, + OldName: oldName, + NewDirectory: newDir, + NewName: newName, + }) + return err +} + +func TestRenameRemoteOnlyEntryCopiesBeforeDeletingTheOldObject(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + _, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + assert.ErrorIs(t, err, filer_pb.ErrNotFound) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) + assert.Equal(t, uint64(101), dst.Attr.Inode) +} + +func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + queue := &captureQueue{} + swapNotificationQueue(t, queue) + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + // the target's delete must not take the copy that replaced its object + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + events := queue.snapshot() + require.Len(t, events, 2) + assert.True(t, filer.IsMetadataOnlyDelete(events[0].notification.OldEntry), "the target's delete event must leave the remote object alone") + assert.False(t, filer.IsMetadataOnlyDelete(events[1].notification.OldEntry)) + // the copy lands after the commit; the rename event still carries its stamp + assert.Equal(t, `"copy"`, events[1].notification.NewEntry.RemoteEntry.GetRemoteETag()) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, uint64(101), dst.Attr.Inode) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) + assert.Empty(t, dst.Extended[filer.ExtRemoteRenameSourceKey]) + // peers replay the post-copy state, not the source's stamp the entry + // committed with + newEvent := events[1].notification.NewEntry + assert.Equal(t, `"copy"`, newEvent.RemoteEntry.RemoteETag) + assert.Empty(t, newEvent.Extended[filer.ExtRemoteRenameSourceKey]) +} + +func TestRenameRemoteOnlyEntryCopyFailureKeepsTheOldObject(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{copyErr: errors.New("copy refused")}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + // the copy runs after the commit over a remote-only target, so its + // failure cannot roll anything back: the entry still stamps the old + // object, which is kept for remote.sync to finish the copy + assert.Empty(t, client.recorded()) + _, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + assert.ErrorIs(t, err, filer_pb.ErrNotFound) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, dst.Remote.RemoteETag) + // reads keep resolving the source's key until the copy stamps the entry: + // its own key still holds the target's old object + assert.Equal(t, "/buckets/b/src/a.jpg", string(dst.Extended[filer.ExtRemoteRenameSourceKey])) + readDir, readName := server.filer.RemoteReadPath(dst, "/buckets/b/dst", "a.jpg") + assert.Equal(t, "/buckets/b/src", readDir) + assert.Equal(t, "a.jpg", readName) +} + +func TestRenameRemoteOnlyEntryWithoutCopierIsRefused(t *testing.T) { + client := &renameRemoteClient{} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + + err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") + require.ErrorContains(t, err, "cannot copy") + + assert.Empty(t, client.recorded()) + _, err = store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + require.NoError(t, err) + _, err = store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + assert.ErrorIs(t, err, filer_pb.ErrNotFound) +} + +func TestRenameRemoteOnlyEntryOutOfItsMountIsRefused(t *testing.T) { + for _, paths := range [][2]util.FullPath{ + {"/buckets/b/src/a.jpg", "/buckets/c/a.jpg"}, + {"/data/m/a.jpg", "/data/a.jpg"}, + } { + t.Run(string(paths[1]), func(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries[string(paths[0])] = remoteOnlyRenameEntry(string(paths[0]), 101) + + err := renameFile(server, paths[0], paths[1]) + require.ErrorContains(t, err, "cannot be renamed out of it") + + assert.Empty(t, client.recorded()) + _, err = store.FindEntry(context.Background(), paths[0]) + require.NoError(t, err) + _, err = store.FindEntry(context.Background(), paths[1]) + assert.ErrorIs(t, err, filer_pb.ErrNotFound) + }) + } +} + +func TestRenameEmptyRemoteEntryDoesNotCopy(t *testing.T) { + // nothing to lose: no copy, and no refusal even out of the mount or + // without a copier; the sync writes an empty object at a new key + client := &renameRemoteClient{} + server, store := newRenameRemoteTestServer(t, client) + for _, path := range []string{"/buckets/b/src/empty", "/data/m/empty"} { + entry := remoteOnlyRenameEntry(path, 101) + entry.FileSize = 0 + entry.Remote.RemoteSize = 0 + store.entries[path] = entry + } + + require.NoError(t, renameFile(server, "/buckets/b/src/empty", "/buckets/b/dst/empty")) + require.NoError(t, renameFile(server, "/data/m/empty", "/data/empty")) + + assert.Equal(t, []string{"delete origin/src/empty", "delete origin/m/empty"}, client.recorded()) +} + +func TestRenameRemoteOnlyEntryFailingAfterTheCopyKeepsTheSource(t *testing.T) { + for _, src := range []util.FullPath{"/buckets/b/src/a.jpg", "/buckets/b/src"} { + t.Run(string(src), func(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.commitErr = errors.New("commit failed") + + err := renameFile(server, src, "/buckets/b/moved") + require.ErrorContains(t, err, "commit failed") + + // the old object, and the old directory holding it, are deleted + // only once the rename has committed + dst := "origin/moved" + if src == "/buckets/b/src" { + dst = "origin/moved/a.jpg" + } + assert.Equal(t, []string{"copy origin/src/a.jpg " + dst}, client.recorded()) + }) + } +} + +func TestRenameEntryWithLocalDataDoesNotCopy(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + entry := remoteOnlyRenameEntry("/data/m/a.jpg", 101) + entry.Chunks = []*filer_pb.FileChunk{{FileId: "1,0123456789", Size: 10}} + store.entries["/data/m/a.jpg"] = entry + + // even out of the mount: the old object goes, the content stays local + require.NoError(t, renameFile(server, "/data/m/a.jpg", "/data/a.jpg")) + + assert.Equal(t, []string{"delete origin/m/a.jpg"}, client.recorded()) + dst, err := store.FindEntry(context.Background(), "/data/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, dst.Remote.RemoteETag) + assert.Len(t, dst.GetChunks(), 1) +} + +func TestRenameDirectoryCopiesRemoteOnlyChildren(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + local := remoteOnlyRenameEntry("/buckets/b/src/b.jpg", 102) + local.Content = []byte("0123456789") + store.entries["/buckets/b/src/b.jpg"] = local + + require.NoError(t, renameFile(server, "/buckets/b/src", "/buckets/b/moved")) + + // b.jpg's old object goes with its entry, a.jpg's waits for the commit, + // and the old prefix is not swept: it could hold a newer object by then + assert.Equal(t, []string{ + "copy origin/src/a.jpg origin/moved/a.jpg", + "delete origin/src/b.jpg", + "delete origin/src/a.jpg", + }, client.recorded()) + moved, err := store.FindEntry(context.Background(), "/buckets/b/moved/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, moved.Remote.RemoteETag) + moved, err = store.FindEntry(context.Background(), "/buckets/b/moved/b.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, moved.Remote.RemoteETag) +} + +func TestDeferredOldKeyDeleteKeepsAKeyRetakenSince(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + event := metadataEvent{ + oldEntry: remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101), + newEntry: remoteOnlyRenameEntry("/buckets/b/moved/a.jpg", 101), + deleteOldRemote: true, + } + + // another rename put a remote-only entry at the old path after the commit + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 303) + event.notify(server.filer, context.Background(), nil) + assert.Empty(t, client.recorded()) + + // a lookup that fails says nothing about the path + delete(store.entries, "/buckets/b/src/a.jpg") + store.findErr = errors.New("store unavailable") + event.notify(server.filer, context.Background(), nil) + assert.Empty(t, client.recorded()) + + store.findErr = nil + event.notify(server.filer, context.Background(), nil) + assert.Equal(t, []string{"delete origin/src/a.jpg"}, client.recorded()) +} + +func TestRenameRemoteOnlyEntryFinishesAfterTheCommitOnASQLStore(t *testing.T) { + // a SQL store keeps the transaction in the context and refuses calls + // through it once committed, so the work after the commit must not use it + for _, overTarget := range []bool{false, true} { + t.Run(map[bool]string{false: "new path", true: "over a remote-only target"}[overTarget], func(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.strictTx = true + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + if overTarget { + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + } + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + // the copy is stamped and the old key deleted by the filer itself + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) + }) + } +} + +func TestRenamePendingEntryAgainCopiesFromItsSource(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{copyErr: errors.New("copy refused")}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + store.entries["/buckets/b/dst/d.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/d.jpg", 303) + + // the deferred copy fails: dst/a.jpg reads src/a.jpg's object + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + // renamed again over another remote-only file, failing again: the marker + // keeps naming the original source, not dst/a.jpg's old object + require.NoError(t, renameFile(server, "/buckets/b/dst/a.jpg", "/buckets/b/dst/d.jpg")) + d, err := store.FindEntry(context.Background(), "/buckets/b/dst/d.jpg") + require.NoError(t, err) + assert.Equal(t, "/buckets/b/src/a.jpg", string(d.Extended[filer.ExtRemoteRenameSourceKey])) + + // renamed to a new path, the copy comes from the source and clears the marker + client.copyErr = nil + require.NoError(t, renameFile(server, "/buckets/b/dst/d.jpg", "/buckets/b/dst/c.jpg")) + assert.Contains(t, client.recorded(), "copy origin/src/a.jpg origin/dst/c.jpg") + // the source is left to remote.sync: a concurrent rename may have copied + // onto that key before its entry is visible + assert.NotContains(t, client.recorded(), "delete origin/src/a.jpg") + c, err := store.FindEntry(context.Background(), "/buckets/b/dst/c.jpg") + require.NoError(t, err) + assert.Empty(t, c.Extended[filer.ExtRemoteRenameSourceKey]) + assert.Equal(t, `"copy"`, c.Remote.RemoteETag) +} + +func TestRemoteSourcePathStaysInTheMount(t *testing.T) { + server, _ := newRenameRemoteTestServer(t, &renameRemoteClient{}) + entry := remoteOnlyRenameEntry("/buckets/b/x.jpg", 101) + for marker, want := range map[string]util.FullPath{ + "/buckets/b/src/y.jpg": "/buckets/b/src/y.jpg", // same mount: the pending source + "/buckets/c/y.jpg": "/buckets/b/x.jpg", // another mount: ignored + "/tmp/y.jpg": "/buckets/b/x.jpg", // no mount: ignored + } { + entry.Extended = map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(marker)} + assert.Equal(t, want, server.filer.RemoteSourcePath(entry, "/buckets/b/x.jpg"), marker) + } +} + +func TestTaggingCannotSetThePendingSourceMarker(t *testing.T) { + server, store := newRenameRemoteTestServer(t, &renameRemoteClient{}) + store.entries["/buckets/b/x.jpg"] = remoteOnlyRenameEntry("/buckets/b/x.jpg", 101) + + r := httptest.NewRequest(http.MethodPut, "/buckets/b/x.jpg?tagging", nil) + r.Header.Set("Seaweed-X-Remote-Rename-Source", "/buckets/b/other.jpg") + r.Header.Set("Seaweed-Album", "2026") + server.PutTaggingHandler(httptest.NewRecorder(), r) + + x, err := store.FindEntry(context.Background(), "/buckets/b/x.jpg") + require.NoError(t, err) + assert.Equal(t, "2026", string(x.Extended["Seaweed-Album"])) + assert.Empty(t, x.Extended[filer.ExtRemoteRenameSourceKey], "only the filer sets the pending source") +} + +func TestDeletingTagsKeepsThePendingSourceMarker(t *testing.T) { + server, store := newRenameRemoteTestServer(t, &renameRemoteClient{}) + entry := remoteOnlyRenameEntry("/buckets/b/x.jpg", 101) + entry.Extended = map[string][]byte{"Seaweed-Album": []byte("2026"), filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/src/x.jpg")} + store.entries["/buckets/b/x.jpg"] = entry + + // no tag names: delete every tag + server.DeleteTaggingHandler(httptest.NewRecorder(), httptest.NewRequest(http.MethodDelete, "/buckets/b/x.jpg?tagging", nil)) + + x, err := store.FindEntry(context.Background(), "/buckets/b/x.jpg") + require.NoError(t, err) + assert.Empty(t, x.Extended["Seaweed-Album"]) + assert.Equal(t, "/buckets/b/src/x.jpg", string(x.Extended[filer.ExtRemoteRenameSourceKey]), "the marker is not a tag") +} diff --git a/weed/server/filer_grpc_server_rename_test.go b/weed/server/filer_grpc_server_rename_test.go index 2c36c9785..1a36bc45d 100644 --- a/weed/server/filer_grpc_server_rename_test.go +++ b/weed/server/filer_grpc_server_rename_test.go @@ -31,6 +31,21 @@ type renameTestStore struct { deleteErr error listErr error // simulates a transient store/RPC failure from a directory listing findDelay time.Duration // optional: widen check-then-act windows in tests + findErr error // simulates a failing entry lookup + strictTx bool // like a SQL store: calls through a committed transaction fail +} + +type renameTestTxKey struct{} + +type renameTestTx struct{ done bool } + +// txDone reports a call through a transaction that has already committed or +// rolled back, which a SQL store refuses. +func (s *renameTestStore) txDone(ctx context.Context) error { + if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok && tx.done { + return errors.New("sql: transaction has already been committed or rolled back") + } + return nil } func newRenameTestStore() *renameTestStore { @@ -44,10 +59,23 @@ func (s *renameTestStore) GetName() string { return func (s *renameTestStore) Initialize(util.Configuration, string) error { return nil } func (s *renameTestStore) Shutdown() {} func (s *renameTestStore) BeginTransaction(ctx context.Context) (context.Context, error) { + if s.strictTx { + return context.WithValue(ctx, renameTestTxKey{}, &renameTestTx{}), nil + } return ctx, nil } -func (s *renameTestStore) CommitTransaction(context.Context) error { return s.commitErr } -func (s *renameTestStore) RollbackTransaction(context.Context) error { return nil } +func (s *renameTestStore) CommitTransaction(ctx context.Context) error { + if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok { + tx.done = true + } + return s.commitErr +} +func (s *renameTestStore) RollbackTransaction(ctx context.Context) error { + if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok { + tx.done = true + } + return nil +} func (s *renameTestStore) KvPut(context.Context, []byte, []byte) error { return nil } @@ -56,27 +84,39 @@ func (s *renameTestStore) KvGet(context.Context, []byte) ([]byte, error) { } func (s *renameTestStore) KvDelete(context.Context, []byte) error { return nil } -func (s *renameTestStore) InsertEntry(_ context.Context, entry *filer.Entry) error { +func (s *renameTestStore) InsertEntry(ctx context.Context, entry *filer.Entry) error { + if err := s.txDone(ctx); err != nil { + return err + } s.mu.Lock() defer s.mu.Unlock() s.entries[string(entry.FullPath)] = entry.ShallowClone() return nil } -func (s *renameTestStore) UpdateEntry(_ context.Context, entry *filer.Entry) error { +func (s *renameTestStore) UpdateEntry(ctx context.Context, entry *filer.Entry) error { + if err := s.txDone(ctx); err != nil { + return err + } s.mu.Lock() defer s.mu.Unlock() s.entries[string(entry.FullPath)] = entry.ShallowClone() return nil } -func (s *renameTestStore) FindEntry(_ context.Context, p util.FullPath) (*filer.Entry, error) { +func (s *renameTestStore) FindEntry(ctx context.Context, p util.FullPath) (*filer.Entry, error) { + if err := s.txDone(ctx); err != nil { + return nil, err + } if s.findDelay > 0 { time.Sleep(s.findDelay) } s.mu.Lock() defer s.mu.Unlock() s.findCalls[string(p)]++ + if s.findErr != nil { + return nil, s.findErr + } entry, found := s.entries[string(p)] if !found { return nil, filer_pb.ErrNotFound @@ -84,7 +124,10 @@ func (s *renameTestStore) FindEntry(_ context.Context, p util.FullPath) (*filer. return entry.ShallowClone(), nil } -func (s *renameTestStore) DeleteEntry(_ context.Context, p util.FullPath) error { +func (s *renameTestStore) DeleteEntry(ctx context.Context, p util.FullPath) error { + if err := s.txDone(ctx); err != nil { + return err + } if s.deleteErr != nil { return s.deleteErr } diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go index ff7e1d09b..2fa32fa2f 100644 --- a/weed/server/filer_server_handlers_read.go +++ b/weed/server/filer_server_handlers_read.go @@ -206,15 +206,18 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) // Every part of a multipart Range is prepared on its own, so the origin // preflight below is memoized to one stat per request. - statOrigin := sync.OnceValue(func() error { - dir, name := entry.FullPath.DirAndName() - client, remoteLocation, err := fs.mountedRemoteClient(ctx, dir, name) - if err != nil { + statOriginAt := func(readDir, readName string) func() error { + return sync.OnceValue(func() error { + client, remoteLocation, err := fs.mountedRemoteClient(ctx, readDir, readName) + if err != nil { + return err + } + _, err = client.StatFile(remoteLocation) return err - } - _, err = client.StatFile(remoteLocation) - return err - }) + }) + } + entryDir, entryName := entry.FullPath.DirAndName() + statOrigin := statOriginAt(fs.filer.RemoteReadPath(entry, entryDir, entryName)) ProcessRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) { if offset+size <= int64(len(entry.Content)) { @@ -230,13 +233,23 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) chunks := entry.GetChunks() if entry.IsInRemoteOnly() { dir, name := entry.FullPath.DirAndName() + readDir, readName := fs.filer.RemoteReadPath(entry, dir, name) var mountedLocation *remote_pb.RemoteStorageLocation if fs.filer.RemoteStorage != nil { _, mountedLocation = fs.filer.RemoteStorage.FindMountDirectory(entry.FullPath) } cacheWait := remote_storage.CacheWaitTimeout(entry.Remote.RemoteSize, mountedLocation) if cacheWait <= 0 { - return fs.streamFromRemoteOnly(ctx, r, dir, name, offset, size, statOrigin) + streamFn, err := fs.streamFromRemoteOnly(ctx, r, readDir, readName, offset, size, statOrigin) + if errors.Is(err, filer_pb.ErrNotFound) && (readDir != dir || readName != name) { + // the pending copy finished and its source went away since + // the entry was read; serve what the entry names now + if fresh, findErr := fs.filer.FindEntry(ctx, entry.FullPath); findErr == nil { + freshDir, freshName := fs.filer.RemoteReadPath(fresh, dir, name) + return fs.streamFromRemoteOnly(ctx, r, freshDir, freshName, offset, size, statOriginAt(freshDir, freshName)) + } + } + return streamFn, err } // Bounded wait: a large download outlasts any client timeout, so // serve straight from the origin once the wait expires while the @@ -264,7 +277,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) // when a later prepare fails; keep those on the 503 retry path. if !strings.Contains(r.Header.Get("Range"), ",") { glog.V(1).InfofCtx(ctx, "stream %s from remote while caching: %v", entry.FullPath, err) - if streamFn, remoteErr := fs.streamFromRemote(ctx, dir, name, offset, size); remoteErr == nil { + if streamFn, remoteErr := fs.streamFromRemote(ctx, readDir, readName, offset, size); remoteErr == nil { return streamFn, nil } else { glog.WarningfCtx(ctx, "stream %s from remote: %v", entry.FullPath, remoteErr) diff --git a/weed/server/filer_server_handlers_tagging.go b/weed/server/filer_server_handlers_tagging.go index 06cd1295e..d2ef239f9 100644 --- a/weed/server/filer_server_handlers_tagging.go +++ b/weed/server/filer_server_handlers_tagging.go @@ -4,6 +4,7 @@ import ( "net/http" "strings" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" @@ -35,7 +36,7 @@ func (fs *FilerServer) PutTaggingHandler(w http.ResponseWriter, r *http.Request) } for header, values := range r.Header { - if strings.HasPrefix(header, needle.PairNamePrefix) { + if strings.HasPrefix(header, needle.PairNamePrefix) && header != filer.ExtRemoteRenameSourceKey { for _, value := range values { existingEntry.Extended[header] = []byte(value) } @@ -89,7 +90,7 @@ func (fs *FilerServer) DeleteTaggingHandler(w http.ResponseWriter, r *http.Reque // delete all tags or specific tags hasDeletion := false for header, _ := range existingEntry.Extended { - if strings.HasPrefix(header, needle.PairNamePrefix) { + if strings.HasPrefix(header, needle.PairNamePrefix) && header != filer.ExtRemoteRenameSourceKey { if len(deletions) == 0 { delete(existingEntry.Extended, header) hasDeletion = true diff --git a/weed/server/filer_server_handlers_write_autochunk.go b/weed/server/filer_server_handlers_write_autochunk.go index 2c86affe2..d0abfd7b0 100644 --- a/weed/server/filer_server_handlers_write_autochunk.go +++ b/weed/server/filer_server_handlers_write_autochunk.go @@ -468,6 +468,9 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa } for k, v := range r.Header { if len(v) > 0 && len(v[0]) > 0 { + if k == filer.ExtRemoteRenameSourceKey { + continue // set by the filer only: it picks the remote key reads serve + } if strings.HasPrefix(k, needle.PairNamePrefix) || k == "Cache-Control" || k == "Expires" || k == "Content-Disposition" { entry.Extended[k] = []byte(v[0]) }