diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 90d7514222..38261cac68 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -140,63 +140,24 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem if filer_pb.IsEmpty(resp) { return nil } + mustSend := false + var renameSrc util.FullPath + if message.OldEntry != nil && message.NewEntry != nil { + reshaped, inMount := reshapeCrossMountRename(mountedDir, resp) + if !inMount { + return nil + } + // a create reshaped from a rename into the mount can carry a + // RemoteEntry earned at the old path: it stamps the old key, so + // it cannot vouch for this destination + mustSend = filer_pb.IsCreate(reshaped) + if mustSend { + renameSrc = util.NewFullPath(resp.Directory, message.OldEntry.Name) + } + resp, message = reshaped, reshaped.EventNotification + } if filer_pb.IsCreate(resp) { - if isMultipartUploadFile(message.NewParentPath, message.NewEntry.Name) { - return nil - } - if isSystemPath(message.NewParentPath, message.NewEntry.Name) { - glog.V(2).Infof("skipping create of system path: %s/%s", message.NewParentPath, message.NewEntry.Name) - return nil - } - // Propagate delete markers as deletions on the remote. - // Delete markers are zero-content version entries, so they - // would be filtered out by the HasData check below. - if isDeleteMarker(message.NewEntry) { - if newParent, newName, ok := rewriteVersionedSourcePath(message.NewParentPath, message.NewEntry.Name); ok { - dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(newParent, newName), remoteStorageMountLocation) - return syncDeleteMarker(client, option, message, dest) - } - return nil - } - if !filer.HasData(message.NewEntry) { - return nil - } - glog.V(2).Infof("create: %+v", resp) - if !shouldSendToRemote(message.NewEntry) { - glog.V(2).Infof("skipping creating: %+v", resp) - return nil - } - // Rewrite internal versioning paths to the original S3 key - // to prevent double-versioning when central also has versioning enabled - parentPath, entryName := message.NewParentPath, message.NewEntry.Name - isRewrittenVersion := false - if newParent, newName, ok := rewriteVersionedSourcePath(parentPath, entryName); ok { - glog.V(0).Infof("rewrite versioned path %s/%s -> %s/%s", parentPath, entryName, newParent, newName) - parentPath, entryName = newParent, newName - isRewrittenVersion = true - } - dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(parentPath, entryName), remoteStorageMountLocation) - if message.NewEntry.IsDirectory { - glog.V(0).Infof("mkdir %s", remote_storage.FormatLocation(dest)) - return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, *option.storageClass)) - } - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, *option.storageClass), dest) - if errors.Is(writeErr, errSuperseded) { - glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) - return nil - } - if writeErr != nil { - return writeErr - } - // Skip updateLocalEntry for versioned rewrites: the logical - // object (e.g. file.xml) has no filer entry in versioned - // buckets, and stamping the internal v_* entry with a - // RemoteEntry for the logical key is semantically wrong. - // Replay is safe because S3 PutObject is idempotent. - if isRewrittenVersion { - return nil - } - return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) + return processCreateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp, mustSend, renameSrc) } if filer_pb.IsDelete(resp) { return processDeleteEvent(filerSource, client, mountedDir, remoteStorageMountLocation, resp) @@ -210,6 +171,136 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return eachEntryFunc, nil } +// reshapeCrossMountRename maps an update event onto the side of it the mount +// covers. The subscription delivers a rename when either of its paths lies +// under the mounted directory, and only a path under it maps to a remote +// key: toRemoteStorageLocation slices the mount prefix off, which panics on +// a shorter path and names a key in the wrong tree on a longer one. A rename +// into the mount is the creation of the new entry, a rename out of it the +// deletion of the old entry's object, unless that entry is remote-only and +// the object holds its only copy. An update with both paths inside is +// returned as is; one with neither is not the mount's (false). +func reshapeCrossMountRename(mountedDir string, resp *filer_pb.SubscribeMetadataResponse) (*filer_pb.SubscribeMetadataResponse, bool) { + message := resp.EventNotification + oldInMount := isUnderMount(mountedDir, resp.Directory, message.OldEntry.Name) + newInMount := isUnderMount(mountedDir, message.NewParentPath, message.NewEntry.Name) + switch { + case oldInMount && newInMount: + return resp, true + case newInMount: + glog.V(0).Infof("rename into mount %s: %s/%s -> %s/%s, creating", mountedDir, resp.Directory, message.OldEntry.Name, message.NewParentPath, message.NewEntry.Name) + return &filer_pb.SubscribeMetadataResponse{ + Directory: message.NewParentPath, + EventNotification: &filer_pb.EventNotification{NewEntry: message.NewEntry, NewParentPath: message.NewParentPath, Signatures: message.Signatures}, + TsNs: resp.TsNs, + }, true + case oldInMount: + if isRemoteOnly(message.OldEntry) { + glog.V(0).Infof("rename out of mount %s: %s/%s -> %s/%s, keeping the remote-only object", mountedDir, resp.Directory, message.OldEntry.Name, message.NewParentPath, message.NewEntry.Name) + return nil, false + } + glog.V(0).Infof("rename out of mount %s: %s/%s -> %s/%s, deleting", mountedDir, resp.Directory, message.OldEntry.Name, message.NewParentPath, message.NewEntry.Name) + return &filer_pb.SubscribeMetadataResponse{ + Directory: resp.Directory, + EventNotification: &filer_pb.EventNotification{OldEntry: message.OldEntry, Signatures: message.Signatures}, + TsNs: resp.TsNs, + }, true + default: + return nil, false + } +} + +// isUnderMount reports whether dir/name is the mounted directory or lies +// below it; a trailing slash on the mount flag does not change the answer. +func isUnderMount(mountedDir string, dir string, name string) bool { + mount := strings.TrimSuffix(mountedDir, "/") + path := string(util.NewFullPath(dir, name)) + return path == mount || strings.HasPrefix(path, mount+"/") +} + +// processCreateEvent uploads a created entry to the remote key its path maps +// to. mustSend skips the sync-stamp check: a create reshaped out of a rename +// into the mount inherits a RemoteEntry stamped at the old path's key, which +// says nothing about whether this destination holds the content. renameSrc is +// the path that rename came from, and is empty for a genuine create. +func processCreateEvent( + filerClient filer_pb.FilerClient, + filerSource filer_pb.FilerClient, + storageClass string, + client remote_storage.RemoteStorageClient, + mountedDir string, + remoteStorageMountLocation *remote_pb.RemoteStorageLocation, + resp *filer_pb.SubscribeMetadataResponse, + mustSend bool, + renameSrc util.FullPath, +) error { + message := resp.EventNotification + if isMultipartUploadFile(message.NewParentPath, message.NewEntry.Name) { + return nil + } + if isSystemPath(message.NewParentPath, message.NewEntry.Name) { + glog.V(2).Infof("skipping create of system path: %s/%s", message.NewParentPath, message.NewEntry.Name) + return nil + } + // Propagate delete markers as deletions on the remote. + // Delete markers are zero-content version entries, so they + // would be filtered out by the HasData check below. + if isDeleteMarker(message.NewEntry) { + if newParent, newName, ok := rewriteVersionedSourcePath(message.NewParentPath, message.NewEntry.Name); ok { + dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(newParent, newName), remoteStorageMountLocation) + return syncDeleteMarker(client, filerClient, message, dest) + } + return nil + } + // A create reshaped from a rename into the mount (mustSend) writes an empty + // entry as a zero-byte object; a dropped local copy still skips since the + // rename already copied its object to the new key. A stamped entry whose + // size diverged since (a truncate) is not a dropped copy and sends its + // new content. + if !filer.HasData(message.NewEntry) && !(mustSend && !isDroppedLocalCopy(message.NewEntry)) { + return nil + } + glog.V(2).Infof("create: %+v", resp) + if !mustSend && !shouldSendToRemote(message.NewEntry) { + glog.V(2).Infof("skipping creating: %+v", resp) + return nil + } + // Rewrite internal versioning paths to the original S3 key + // to prevent double-versioning when central also has versioning enabled + parentPath, entryName := message.NewParentPath, message.NewEntry.Name + isRewrittenVersion := false + if newParent, newName, ok := rewriteVersionedSourcePath(parentPath, entryName); ok { + glog.V(0).Infof("rewrite versioned path %s/%s -> %s/%s", parentPath, entryName, newParent, newName) + parentPath, entryName = newParent, newName + isRewrittenVersion = true + } + dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(parentPath, entryName), remoteStorageMountLocation) + if message.NewEntry.IsDirectory { + glog.V(0).Infof("mkdir %s", remote_storage.FormatLocation(dest)) + return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, storageClass)) + } + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, storageClass), dest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) + if mustSend { + return recoverSupersededCreate(filerClient, filerSource, client, mountedDir, remoteStorageMountLocation, message.NewParentPath, message.NewEntry.Name, renameSrc, dest, storageClass) + } + return nil + } + if writeErr != nil { + return writeErr + } + // Skip updateLocalEntry for versioned rewrites: the logical + // object (e.g. file.xml) has no filer entry in versioned + // buckets, and stamping the internal v_* entry with a + // RemoteEntry for the logical key is semantically wrong. + // Replay is safe because S3 PutObject is idempotent. + if isRewrittenVersion { + return nil + } + return updateLocalEntry(filerClient, message.NewParentPath, message.NewEntry, remoteEntry) +} + // processDeleteEvent removes the remote object an entry mapped to. A remote // object that is already absent counts as deleted: the entry was never // uploaded (created and deleted faster than the sync ran, or a replay of an @@ -651,6 +742,176 @@ func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.F return updateLocalEntry(filerClient, dir, current, remoteEntry) } +// recoverSupersededCreate finishes a create reshaped from a rename into the +// mount when the filer has moved past the entry the event described. A plain +// superseded create skips because the superseding event follows in the log; a +// reshaped one has no later event for this destination. remote.uncache accepts +// the rename's inherited stamp and drops the chunks, and the update it logs +// skips the upload through isDroppedLocalCopy, so the content would exist +// nowhere. What the filer holds now decides: local content uploads here only +// when the superseding event will not (the stamp-edge case uploadCurrentEntry +// covers), and remote-only content copies from the object its stamp +// describes, the way completeRemoteOnlyRename finishes a rename. +func recoverSupersededCreate( + filerClient filer_pb.FilerClient, + filerSource filer_pb.FilerClient, + client remote_storage.RemoteStorageClient, + mountedDir string, + remoteStorageMountLocation *remote_pb.RemoteStorageLocation, + dir, name string, + renameSrc util.FullPath, + dest *remote_pb.RemoteStorageLocation, + storageClass string, +) error { + current, err := currentEntry(filerSource, dir, name) + if err != nil { + return err + } + if current == nil || current.IsDirectory { + return nil + } + if isRemoteOnly(current) { + return copyRenameInSource(filerClient, client, mountedDir, remoteStorageMountLocation, dir, current, renameSrc, dest, storageClass) + } + if shouldSendToRemote(current) { + glog.V(0).Infof("leaving %s to the event that superseded the rename", remote_storage.FormatLocation(dest)) + return nil + } + glog.V(0).Infof("uploading the current %s in place of the superseded rename", remote_storage.FormatLocation(dest)) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, dir, remoteWriteEntry(current, storageClass), dest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) + return nil + } + if writeErr != nil { + return writeErr + } + return updateLocalEntry(filerClient, dir, current, remoteEntry) +} + +// copyRenameInSource gives the destination key the content of a remote-only +// entry a rename carried into the mount. The object sits at the key the +// renamed-from path maps to on its own mount's remote, or where a +// pending-rename marker under this mount points. The source object is left +// alone: the source side removes it through its own delete event. When the +// destination already holds what the stamp describes (an earlier run copied +// it), only the stamp is refreshed. With no object on either side the event +// fails and holds the offset. +func copyRenameInSource( + filerClient filer_pb.FilerClient, + client remote_storage.RemoteStorageClient, + mountedDir string, + remoteStorageMountLocation *remote_pb.RemoteStorageLocation, + dir string, + current *filer_pb.Entry, + renameSrc util.FullPath, + dest *remote_pb.RemoteStorageLocation, + storageClass string, +) error { + srcClient, srcLoc, err := renameInSource(filerClient, client, mountedDir, remoteStorageMountLocation, current, renameSrc) + if err != nil { + return err + } + if existing, err := client.StatFile(dest); err == nil { + if remote_storage.Describes(current.RemoteEntry, existing) { + glog.V(0).Infof("%s already holds the renamed content", remote_storage.FormatLocation(dest)) + return updateLocalEntry(filerClient, dir, current, existing) + } + 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(srcLoc)) + } else if !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return err + } + stat, err := srcClient.StatFile(srcLoc) + if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return fmt.Errorf("%s: content is on neither %s nor %s", util.NewFullPath(dir, current.Name), remote_storage.FormatLocation(srcLoc), remote_storage.FormatLocation(dest)) + } + if err != nil { + return err + } + reader, err := openRemoteObject(srcClient, srcLoc, stat.RemoteSize) + if err != nil { + return err + } + defer reader.Close() + glog.V(0).Infof("copy %s -> %s", remote_storage.FormatLocation(srcLoc), remote_storage.FormatLocation(dest)) + remoteEntry, err := client.WriteFile(dest, remoteWriteEntry(current, storageClass), reader) + if err != nil { + return err + } + return updateLocalEntry(filerClient, dir, current, remoteEntry) +} + +// renameInSource locates the object holding a remote-only rename-in entry's +// content and the client that serves it. A pending-rename marker under this +// mount names a key on this remote; otherwise the renamed-from path is +// resolved through the filer's mount mappings to its own mount's remote +// location, served by this client when the source shares this storage. +func renameInSource( + filerClient filer_pb.FilerClient, + client remote_storage.RemoteStorageClient, + mountedDir string, + remoteStorageMountLocation *remote_pb.RemoteStorageLocation, + current *filer_pb.Entry, + renameSrc util.FullPath, +) (remote_storage.RemoteStorageClient, *remote_pb.RemoteStorageLocation, error) { + if src := util.FullPath(current.Extended[filer.ExtRemoteRenameSourceKey]); strings.HasPrefix(string(src), strings.TrimSuffix(mountedDir, "/")+"/") { + return client, toRemoteStorageLocation(util.FullPath(mountedDir), src, remoteStorageMountLocation), nil + } + mappings, err := readMountMappings(filerClient) + if err != nil { + return nil, nil, err + } + srcMount, srcRemoteLoc, err := filer.FindMountedRemoteMapping(mappings, string(renameSrc)) + if err != nil { + return nil, nil, fmt.Errorf("rename source %s: %w", renameSrc, err) + } + srcLoc := filer.MapFullPathToRemoteStorageLocation(util.FullPath(srcMount), srcRemoteLoc, renameSrc) + if srcRemoteLoc.Name == remoteStorageMountLocation.Name { + return client, srcLoc, nil + } + conf, err := readRemoteStorageConf(filerClient, srcRemoteLoc.Name) + if err != nil { + return nil, nil, fmt.Errorf("rename source %s: %w", renameSrc, err) + } + srcClient, err := remote_storage.GetRemoteStorage(conf) + if err != nil { + return nil, nil, fmt.Errorf("rename source %s: %w", renameSrc, err) + } + return srcClient, srcLoc, nil +} + +// readMountMappings reads the filer's remote mount table, the file +// DetectMountInfo reads at startup, through the event's filer client. +func readMountMappings(filerClient filer_pb.FilerClient) (mappings *remote_pb.RemoteStorageMapping, err error) { + var content []byte + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + var readErr error + content, readErr = filer.ReadInsideFiler(context.Background(), client, filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE) + return readErr + }) + if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + return nil, err + } + return filer.UnmarshalRemoteStorageMappings(content) +} + +// readRemoteStorageConf reads a remote storage's configuration entry. +func readRemoteStorageConf(filerClient filer_pb.FilerClient, storageName string) (*remote_pb.RemoteConf, error) { + var content []byte + if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + var readErr error + content, readErr = filer.ReadInsideFiler(context.Background(), client, filer.DirectoryEtcRemote, storageName+filer.REMOTE_STORAGE_CONF_SUFFIX) + return readErr + }); err != nil { + return nil, err + } + conf := &remote_pb.RemoteConf{} + if err := proto.Unmarshal(content, conf); err != nil { + return nil, fmt.Errorf("unmarshal %s/%s: %w", filer.DirectoryEtcRemote, storageName+filer.REMOTE_STORAGE_CONF_SUFFIX, err) + } + return conf, nil +} + // isSuperseded reports whether the filer has moved past the entry an event // described: it is deleted, or it no longer references every chunk the event // named. Those are the chunks the filer deletes when an entry is updated, so diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 93c8340b47..5b004bd068 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -413,10 +413,12 @@ func TestIsMetadataOnlyUpdate(t *testing.T) { } // stubFilerClient serves one entry; nil means not found. A non-nil err fails -// every lookup instead. +// every lookup instead. entries, when set, answers lookups by "dir/name" +// before entry is consulted. type stubFilerClient struct { filer_pb.SeaweedFilerClient entry *filer_pb.Entry + entries map[string]*filer_pb.Entry err error updateErr error lookups int @@ -424,11 +426,14 @@ type stubFilerClient struct { updated *filer_pb.UpdateEntryRequest } -func (c *stubFilerClient) LookupDirectoryEntry(context.Context, *filer_pb.LookupDirectoryEntryRequest, ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { +func (c *stubFilerClient) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { c.lookups++ if c.err != nil { return nil, c.err } + if c.entries != nil { + return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entries[req.Directory+"/"+req.Name]}, nil + } return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entry}, nil } @@ -2017,6 +2022,354 @@ func TestMetadataUpdateKeepsTheStoredStampOnNilResult(t *testing.T) { } } +// TestReshapeCrossMountRename covers an update event with a path outside +// the mount. The subscription delivers a rename when either path is under +// the mount, and an update of the /etc/remote directory entry itself through +// the daemon's /etc/remote prefix; processUpdateEvent then sliced the mount +// prefix off both paths, which panics on a path shorter than the mount (seen +// on a replay: "slice bounds out of range [20:11]" on the 11-byte +// /etc/remote, every restart dying on the same event) and names a key in the +// wrong tree on a longer one. +func TestReshapeCrossMountRename(t *testing.T) { + const mount = "/buckets/mount" + chunked := func(name string) *filer_pb.Entry { + return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669, FileSize: 7}, Chunks: []*filer_pb.FileChunk{{FileId: "3,01", Size: 7}}} + } + event := func(oldDir, oldName, newDir, newName string, oldEntry *filer_pb.Entry) *filer_pb.SubscribeMetadataResponse { + if oldEntry == nil { + oldEntry = chunked(oldName) + } + return &filer_pb.SubscribeMetadataResponse{ + Directory: oldDir, + TsNs: 42, + EventNotification: &filer_pb.EventNotification{OldEntry: oldEntry, NewParentPath: newDir, NewEntry: chunked(newName), Signatures: []int32{7}}, + } + } + + t.Run("into the mount from a shorter path is a create of the new entry", func(t *testing.T) { + resp := event("/plain", "in.txt", mount+"/lock", "in.txt", nil) + got, inMount := reshapeCrossMountRename(mount, resp) + if !inMount || !filer_pb.IsCreate(got) { + t.Fatalf("got %+v inMount=%v, want a create", got, inMount) + } + if got.Directory != mount+"/lock" || got.EventNotification.NewParentPath != mount+"/lock" || got.EventNotification.NewEntry.Name != "in.txt" { + t.Errorf("create at %s/%s (dir %s), want %s/in.txt", got.EventNotification.NewParentPath, got.EventNotification.NewEntry.Name, got.Directory, mount+"/lock") + } + if got.TsNs != 42 || len(got.EventNotification.Signatures) != 1 { + t.Errorf("timestamp and signatures must carry over: %+v", got) + } + }) + + t.Run("into the mount from a longer path outside it is a create, not a key in the wrong tree", func(t *testing.T) { + resp := event("/buckets/other/deeper/tree", "x.txt", mount, "x.txt", nil) + got, inMount := reshapeCrossMountRename(mount, resp) + if !inMount || !filer_pb.IsCreate(got) || got.EventNotification.NewParentPath != mount { + t.Fatalf("got %+v inMount=%v, want a create under %s", got, inMount, mount) + } + }) + + t.Run("out of the mount is a delete of the old entry", func(t *testing.T) { + resp := event(mount+"/lock", "out.txt", "/plain", "out.txt", nil) + got, inMount := reshapeCrossMountRename(mount, resp) + if !inMount || !filer_pb.IsDelete(got) { + t.Fatalf("got %+v inMount=%v, want a delete", got, inMount) + } + if got.Directory != mount+"/lock" || got.EventNotification.OldEntry.Name != "out.txt" { + t.Errorf("delete of %s/%s, want %s/out.txt", got.Directory, got.EventNotification.OldEntry.Name, mount+"/lock") + } + }) + + t.Run("out of the mount keeps the object of a remote-only entry", func(t *testing.T) { + remoteOnly := &filer_pb.Entry{Name: "out.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669, FileSize: 7}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteSize: 7, RemoteMtime: 1786096669}} + resp := event(mount+"/lock", "out.txt", "/plain", "out.txt", remoteOnly) + if got, inMount := reshapeCrossMountRename(mount, resp); inMount || got != nil { + t.Fatalf("got %+v inMount=%v, want the event dropped", got, inMount) + } + }) + + t.Run("both paths inside is the update itself", func(t *testing.T) { + resp := event(mount+"/a", "x.txt", mount+"/b", "y.txt", nil) + if got, inMount := reshapeCrossMountRename(mount, resp); !inMount || got != resp { + t.Fatalf("got %+v inMount=%v, want the same event", got, inMount) + } + }) + + t.Run("an update of the /etc/remote directory entry itself is not the mount's event", func(t *testing.T) { + dirEntry := &filer_pb.Entry{Name: "remote", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}} + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/etc", + EventNotification: &filer_pb.EventNotification{OldEntry: dirEntry, NewParentPath: "/etc", NewEntry: dirEntry}, + } + if got, inMount := reshapeCrossMountRename("/buckets/c4x-storage", resp); inMount || got != nil { + t.Fatalf("got %+v inMount=%v, want the event dropped", got, inMount) + } + }) + + t.Run("neither path inside is not the mount's event", func(t *testing.T) { + resp := event("/plain", "x.txt", "/buckets/mounted-not", "x.txt", nil) + if got, inMount := reshapeCrossMountRename(mount, resp); inMount || got != nil { + t.Fatalf("got %+v inMount=%v, want the event dropped", got, inMount) + } + }) + + t.Run("a trailing slash on the mount flag changes nothing", func(t *testing.T) { + resp := event("/plain", "in.txt", mount, "in.txt", nil) + if got, inMount := reshapeCrossMountRename(mount+"/", resp); !inMount || !filer_pb.IsCreate(got) { + t.Fatalf("got %+v inMount=%v, want a create", got, inMount) + } + }) +} + +// TestRenameIntoMountUploadsDespiteTheOldStamp covers a file synced at a path +// under the mount, renamed out (which deleted its remote object), and renamed +// back in: the entry still carries the RemoteEntry the first sync stamped, so +// shouldSendToRemote would skip the create the reshape made, leaving the new +// key without an object. A reshaped create must send unconditionally. +func TestRenameIntoMountUploadsDespiteTheOldStamp(t *testing.T) { + const mountedDir = "/buckets/mount" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"} + + stamp := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: "abc", RemoteSize: 7, RemoteMtime: 1786096669, LastLocalSyncTsNs: 1786096669*1e9 + 1} + newEntry := &filer_pb.Entry{ + Name: "back.txt", + Content: []byte("payload"), + Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, + RemoteEntry: stamp, + } + resp, inMount := reshapeCrossMountRename(mountedDir, &filer_pb.SubscribeMetadataResponse{ + Directory: "/plain", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "back.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: stamp}, + NewParentPath: mountedDir + "/lock", + NewEntry: newEntry, + }, + }) + if !inMount || !filer_pb.IsCreate(resp) { + t.Fatalf("reshape = %+v inMount=%v, want a create", resp, inMount) + } + if shouldSendToRemote(resp.EventNotification.NewEntry) { + t.Fatal("precondition: the stamp the rename carried should fail shouldSendToRemote") + } + + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: newEntry} + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp, true, "/plain/back.txt"); err != nil { + t.Fatal(err) + } + wantWrite := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/lock/back.txt"} + if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], wantWrite) { + t.Fatalf("writes = %+v, want the new key %s written", remote.writes, remote_storage.FormatLocation(wantWrite)) + } + + // a plain create keeps trusting the stamp + remote = &recordingRemote{} + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp, false, ""); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 0 { + t.Errorf("writes = %+v, want the stamped create skipped", remote.writes) + } +} + +// TestRenameIntoMountWritesAnEmptyFile: a rename into the mount is reshaped +// to a create with mustSend, and an empty entry has no data for the HasData +// gate; the new key still takes a zero-byte object. A remote-only entry holds +// nothing local to send and is skipped as before. +func TestRenameIntoMountWritesAnEmptyFile(t *testing.T) { + const mountedDir = "/buckets/mount" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"} + renameIn := func(entry *filer_pb.Entry) *filer_pb.SubscribeMetadataResponse { + resp, inMount := reshapeCrossMountRename(mountedDir, &filer_pb.SubscribeMetadataResponse{ + Directory: "/plain", + EventNotification: &filer_pb.EventNotification{OldEntry: entry, NewParentPath: mountedDir + "/lock", NewEntry: entry}, + }) + if !inMount || !filer_pb.IsCreate(resp) { + t.Fatalf("reshape = %+v inMount=%v, want a create", resp, inMount) + } + return resp + } + wantWrite := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/lock/empty.txt"} + + t.Run("an empty entry writes a zero-byte object", func(t *testing.T) { + entry := &filer_pb.Entry{Name: "empty.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}} + if filer.HasData(entry) { + t.Fatal("precondition: an empty entry carries no data") + } + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: entry} + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn(entry), true, "/plain/empty.txt"); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], wantWrite) { + t.Errorf("writes = %+v, want the new key %s written", remote.writes, remote_storage.FormatLocation(wantWrite)) + } + }) + + t.Run("a remote-only entry still skips", func(t *testing.T) { + stamp := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: "abc", RemoteSize: 7, RemoteMtime: 1786096669, LastLocalSyncTsNs: 1786096669*1e9 + 1} + entry := &filer_pb.Entry{Name: "empty.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669, FileSize: 7}, RemoteEntry: stamp} + if !isRemoteOnly(entry) { + t.Fatal("precondition: the entry should be remote-only") + } + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: entry} + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn(entry), true, "/plain/empty.txt"); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 0 { + t.Errorf("writes = %+v, want none for a remote-only entry", remote.writes) + } + }) + + // a stamped entry truncated to zero is not remote-only content: its stamp + // describes bytes the file no longer holds, so the new key takes a + // zero-byte object + t.Run("a truncated stamped entry writes a zero-byte object", func(t *testing.T) { + stamp := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: "abc", RemoteSize: 7, RemoteMtime: 1786096669, LastLocalSyncTsNs: 1786096669*1e9 + 1} + entry := &filer_pb.Entry{Name: "empty.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: stamp} + if !isRemoteOnly(entry) || isDroppedLocalCopy(entry) { + t.Fatal("precondition: a truncated stamped entry reports remote-only but is not a dropped copy") + } + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: entry} + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn(entry), true, "/plain/empty.txt"); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], wantWrite) { + t.Errorf("writes = %+v, want the new key %s written", remote.writes, remote_storage.FormatLocation(wantWrite)) + } + }) + + // an ordinary create of an empty entry is unchanged: no object + t.Run("a plain empty create still writes nothing", func(t *testing.T) { + entry := &filer_pb.Entry{Name: "empty.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}} + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: entry} + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn(entry), false, ""); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 0 { + t.Errorf("writes = %+v, want the empty create skipped", remote.writes) + } + }) +} + +// TestSupersededRenameInRecoversTheCurrentEntry covers a rename into the +// mount whose create event the sync reaches only after remote.uncache dropped +// the moved file's chunks: the upload is superseded, and skipping it leaves +// the destination object missing once the source side's delete runs. +func TestSupersededRenameInRecoversTheCurrentEntry(t *testing.T) { + const mountedDir = "/buckets/mount" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/mnt"} + srcMountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/plain-remote"} + + stamp := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: "abc", RemoteSize: 4, RemoteMtime: 1786096669, LastLocalSyncTsNs: 1786096669*1e9 + 1} + // the entry as the rename logged it: local chunks plus the stamp earned + // under the source mount + eventEntry := &filer_pb.Entry{ + Name: "a.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669, FileSize: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,01111111", Size: 4}}, + RemoteEntry: stamp, + } + // the entry as remote.uncache left it: the chunks are gone and the stamp + // still describes the source mount's object + uncached := &filer_pb.Entry{ + Name: "a.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669, FileSize: 4}, + RemoteEntry: stamp, + } + mappingContent, err := proto.Marshal(&remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{ + mountedDir: mountLoc, + "/plain": srcMountLoc, + }}) + if err != nil { + t.Fatal(err) + } + mappingEntry := &filer_pb.Entry{Name: filer.REMOTE_STORAGE_MOUNT_FILE, Content: mappingContent} + renameIn := &filer_pb.SubscribeMetadataResponse{ + Directory: mountedDir + "/lock", + EventNotification: &filer_pb.EventNotification{NewParentPath: mountedDir + "/lock", NewEntry: eventEntry}, + } + dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/mnt/lock/a.txt"} + + newFilerClient := func(entry *filer_pb.Entry) *stubFilerClient { + return &stubFilerClient{entries: map[string]*filer_pb.Entry{ + mountedDir + "/lock/a.txt": proto.Clone(entry).(*filer_pb.Entry), + filer.DirectoryEtcRemote + "/" + filer.REMOTE_STORAGE_MOUNT_FILE: mappingEntry, + }} + } + + t.Run("uncached meanwhile: the source object is copied and the entry stamped", func(t *testing.T) { + remote := &recordingRemote{objects: map[string][]byte{"/plain-remote/a.txt": []byte("data")}} + filerClient := newFilerClient(uncached) + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn, true, "/plain/a.txt"); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], dest) { + t.Fatalf("writes = %+v, want the source copied to %s", remote.writes, remote_storage.FormatLocation(dest)) + } + if string(remote.written[0]) != "data" { + t.Errorf("written = %q, want the source object's bytes", remote.written[0]) + } + if filerClient.updates != 1 { + t.Errorf("updates = %d, want the entry stamped with the copy", filerClient.updates) + } + }) + + t.Run("the destination already holds it: only a stamp", func(t *testing.T) { + remote := &recordingRemote{objects: map[string][]byte{"/mnt/lock/a.txt": []byte("data")}, statETags: map[string]string{"/mnt/lock/a.txt": "abc"}} + filerClient := newFilerClient(uncached) + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn, true, "/plain/a.txt"); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 0 { + t.Errorf("writes = %+v, want none once the destination holds the content", remote.writes) + } + if filerClient.updates != 1 { + t.Errorf("updates = %d, want the entry stamped", filerClient.updates) + } + }) + + t.Run("no object on either side: the event fails", func(t *testing.T) { + remote := &recordingRemote{objects: map[string][]byte{}} + filerClient := newFilerClient(uncached) + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn, true, "/plain/a.txt"); err == nil { + t.Fatal("want the event to fail and hold the offset") + } + }) + + t.Run("a pending marker under the mount names the source", func(t *testing.T) { + marked := proto.Clone(uncached).(*filer_pb.Entry) + marked.Extended = map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(mountedDir + "/lock/old.txt")} + remote := &recordingRemote{objects: map[string][]byte{"/mnt/lock/old.txt": []byte("data")}} + filerClient := newFilerClient(marked) + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn, true, "/plain/a.txt"); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], dest) { + t.Fatalf("writes = %+v, want the marked source copied to %s", remote.writes, remote_storage.FormatLocation(dest)) + } + }) + + t.Run("rewritten since: the superseding event uploads", func(t *testing.T) { + rewritten := &filer_pb.Entry{ + Name: "a.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 1786096700, FileSize: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "2,02222222", Size: 4}}, + RemoteEntry: stamp, + } + remote := &recordingRemote{} + filerClient := newFilerClient(rewritten) + if err := processCreateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, renameIn, true, "/plain/a.txt"); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 0 { + t.Errorf("writes = %+v, want the rewrite's own event to upload", remote.writes) + } + }) +} + func TestIsSystemPath(t *testing.T) { tests := []struct { dir string @@ -2086,4 +2439,3 @@ func TestRemoteSyncSkipsSystemFolder(t *testing.T) { t.Fatalf("expected 0 remote deletes for system path, got %d deletes", len(remote.deletes)) } } -