diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 8c9e89a5e8..6e10eb210b 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -53,12 +53,12 @@ import ( func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *source.FilerSource, mountedDir string) error { // read filer remote storage mount mappings - _, _, remoteStorageMountLocation, remoteStorage, detectErr := filer.DetectMountInfo(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir) + _, mountKey, remoteStorageMountLocation, remoteStorage, detectErr := filer.DetectMountInfo(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir) if detectErr != nil { return fmt.Errorf("read mount info: %w", detectErr) } - eachEntryFunc, err := option.makeEventProcessor(remoteStorage, mountedDir, remoteStorageMountLocation, filerSource) + eachEntryFunc, err := option.makeEventProcessor(remoteStorage, mountedDir, dirRemoteLocation(mountKey, mountedDir, remoteStorageMountLocation), filerSource) if err != nil { return err } @@ -67,13 +67,16 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour if err != nil { return err } + lastOffsetTs, err = startAfterMountOffset(lastOffsetTs, mountKey, mountedDir, *option.timeAgo, option.resumeTsNs, func() (int64, error) { + return remote_storage.GetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountKey) + }) + if err != nil { + return err + } processor := NewMetadataProcessor(eachEntryFunc, 128, lastOffsetTs.UnixNano()) var lastLogTsNs = time.Now().UnixNano() - processEventFnWithOffset := pb.AddOffsetFunc(func(resp *filer_pb.SubscribeMetadataResponse) error { - processor.AddSyncJob(resp) - return nil - }, 3*time.Second, func(counter int64, lastTsNs int64) error { + processEventFnWithOffset := pb.AddOffsetFunc(admitUntilMountEnds(processor, mountedDir, mountKey, remoteStorageMountLocation), 3*time.Second, func(counter int64, lastTsNs int64) error { offsetTsNs := processor.processedTsWatermark.Load() if offsetTsNs == 0 { return nil @@ -114,6 +117,37 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour return err } +// dirRemoteLocation is where dir, at or below the mount at mountKey, lives +// on the remote. +func dirRemoteLocation(mountKey, dir string, mountLocation *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation { + if strings.TrimSuffix(dir, "/") == strings.TrimSuffix(mountKey, "/") { + return mountLocation + } + return filer.MapFullPathToRemoteStorageLocation(util.FullPath(mountKey), mountLocation, util.FullPath(dir)) +} + +// startAfterMountOffset keeps a write-back of a directory below the mount at +// mountKey from starting before the offset saved for the mount itself. +// remote.unmount saves that one after its purge and leaves the offsets of +// directories below the mount alone, so resuming from the directory's own +// offset after a remount would replay the unmount and stop there on every +// start. Whatever it skips was purged, or was synced by a write-back of the +// whole mount. +func startAfterMountOffset(start time.Time, mountKey, dir string, timeAgo time.Duration, resumeTsNs int64, readMountOffset func() (int64, error)) (time.Time, error) { + if timeAgo != 0 || resumeTsNs > 0 || strings.TrimSuffix(dir, "/") == strings.TrimSuffix(mountKey, "/") { + return start, nil + } + mountOffsetTsNs, err := readMountOffset() + if err != nil { + return time.Time{}, fmt.Errorf("read the sync offset of %s: %w", mountKey, err) + } + if mountOffsetTs := time.Unix(0, mountOffsetTsNs); mountOffsetTsNs > 0 && mountOffsetTs.After(start) { + glog.V(0).Infof("resume %s from the offset of its mount %s: %v", dir, mountKey, mountOffsetTs) + return mountOffsetTs, nil + } + return start, nil +} + // saveResumeOffset records where the next subscription resumes once this one // ends. The periodic checkpoint only runs as events arrive, and the retry // recomputes -timeAgo from now, so a failure before the first checkpoint @@ -130,6 +164,101 @@ func saveResumeOffset(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddr return offsetTsNs } +// admitUntilMountEnds judges each mount-table version before the next event +// is admitted. remote.unmount purges the directory right after writing the +// table, and a purge delete admitted under a directory no longer mounted +// deletes the remote objects. At the end nothing more is admitted and no +// offset is saved, keeping the one remote.unmount writes after its purge; +// the admitted jobs drain before the exit. +func admitUntilMountEnds(processor *MetadataProcessor, mountedDir, mountKey string, mountLocation *remote_pb.RemoteStorageLocation) pb.ProcessMetadataFunc { + return func(resp *filer_pb.SubscribeMetadataResponse) error { + ended, moved, err := mountTableEnds(mountedDir, mountKey, mountLocation, resp) + if err != nil { + return err + } + if !ended { + processor.AddSyncJob(resp) + return nil + } + processor.Drain() + exitAtMountEnd(mountedDir, mountLocation, moved) + return nil + } +} + +var exitAtMountEnd = func(mountedDir string, from, to *remote_pb.RemoteStorageLocation) { + if to != nil { + glog.Fatalf("Unexpected mount changes %+v => %+v", from, to) + } + glog.V(0).Infof("unmounted %s exiting ...", mountedDir) + os.Exit(0) +} + +// mountTableEnds reports whether a mount-table event ends the write-back of +// mountedDir, which mountKey mounts at loc: a version in which another mount +// or none covers the directory. moved is the location of the covering mount, +// nil for an unmount. A version in which no mount covers it, replacing one +// in which none did, predates the mount: without a saved offset the follow +// starts at the directory's creation second and replays the table's earlier +// versions. Any earlier mount's unmount still ends it, since its purge +// follows. +func mountTableEnds(mountedDir, mountKey string, loc *remote_pb.RemoteStorageLocation, resp *filer_pb.SubscribeMetadataResponse) (ended bool, moved *remote_pb.RemoteStorageLocation, err error) { + message := resp.EventNotification + if isMountTable(filer_pb.MetadataEventTargetDirectory(resp), message.NewEntry) { + key, newLoc, readErr := mountOf(message.NewEntry, mountedDir) + if readErr != nil { + return false, nil, fmt.Errorf("unmarshal mappings: %w", readErr) + } + if key == mountKey && sameMountLocation(newLoc, loc) { + return false, nil, nil + } + moved = newLoc + } else if !isMountTable(resp.Directory, message.OldEntry) { + return false, nil, nil + } + if moved == nil && !replacedTableCovers(mountedDir, resp) { + return false, nil, nil + } + return true, moved, nil +} + +// replacedTableCovers reports whether a mount covered mountedDir in the +// mount table an event replaced. A created table replaced none; a replaced +// version the event does not carry, or that does not unmarshal, counts as +// covering it. +func replacedTableCovers(mountedDir string, resp *filer_pb.SubscribeMetadataResponse) bool { + oldEntry := resp.EventNotification.OldEntry + if oldEntry == nil { + return false + } + if !isMountTable(resp.Directory, oldEntry) { + return true + } + _, oldLoc, err := mountOf(oldEntry, mountedDir) + return err != nil || oldLoc != nil +} + +// mountOf resolves the mount covering dir in a mount-table version, as +// DetectMountInfo does in the live table: "" and nil when none does. +func mountOf(table *filer_pb.Entry, dir string) (string, *remote_pb.RemoteStorageLocation, error) { + mappings, err := filer.UnmarshalRemoteStorageMappings(table.Content) + if err != nil { + return "", nil, err + } + key, loc, _ := filer.FindMountedRemoteMapping(mappings, dir) + return key, loc, nil +} + +// isMountTable reports whether entry, in dir, is the mount table. A table +// renamed away, deleted, or written anywhere else no longer counts. +func isMountTable(dir string, entry *filer_pb.Entry) bool { + return entry != nil && dir == filer.DirectoryEtcRemote && entry.Name == filer.REMOTE_STORAGE_MOUNT_FILE +} + +func sameMountLocation(a, b *remote_pb.RemoteStorageLocation) bool { + return a != nil && b != nil && a.Name == b.Name && a.Bucket == b.Bucket && a.Path == b.Path +} + func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) { client, err := remote_storage.GetRemoteStorage(remoteStorage) if err != nil { @@ -139,20 +268,6 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem handleEtcRemoteChanges := func(resp *filer_pb.SubscribeMetadataResponse) error { message := resp.EventNotification if metadataEventUpdatesDirectory(resp, filer.DirectoryEtcRemote) { - if message.NewEntry.Name == filer.REMOTE_STORAGE_MOUNT_FILE { - mappings, readErr := filer.UnmarshalRemoteStorageMappings(message.NewEntry.Content) - if readErr != nil { - return fmt.Errorf("unmarshal mappings: %w", readErr) - } - if remoteLoc, found := mappings.Mappings[mountedDir]; found { - if remoteStorageMountLocation.Bucket != remoteLoc.Bucket || remoteStorageMountLocation.Path != remoteLoc.Path { - glog.Fatalf("Unexpected mount changes %+v => %+v", remoteStorageMountLocation, remoteLoc) - } - } else { - glog.V(0).Infof("unmounted %s exiting ...", mountedDir) - os.Exit(0) - } - } if message.NewEntry.Name == remoteStorage.Name+filer.REMOTE_STORAGE_CONF_SUFFIX { conf := &remote_pb.RemoteConf{} if err := proto.Unmarshal(message.NewEntry.Content, conf); err != nil { @@ -166,11 +281,6 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem } } } - if metadataEventRemovesFromDirectory(resp, filer.DirectoryEtcRemote) && - message.OldEntry.Name == filer.REMOTE_STORAGE_MOUNT_FILE { - glog.V(0).Infof("unmounted %s exiting ...", mountedDir) - os.Exit(0) - } return nil } diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index a8e0a3f7bb..9cfcd1c772 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -3158,3 +3158,268 @@ func TestKeptDeleteMarkerRetriesWhenTheLockIsReleased(t *testing.T) { }) } } + +func mountTableEvent(t *testing.T, tsNs int64, old, new map[string]*remote_pb.RemoteStorageLocation) *filer_pb.SubscribeMetadataResponse { + t.Helper() + tableEntry := func(mappings map[string]*remote_pb.RemoteStorageLocation) *filer_pb.Entry { + if mappings == nil { + return nil + } + content, err := proto.Marshal(&remote_pb.RemoteStorageMapping{Mappings: mappings}) + if err != nil { + t.Fatal(err) + } + return &filer_pb.Entry{Name: filer.REMOTE_STORAGE_MOUNT_FILE, Content: content} + } + message := &filer_pb.EventNotification{OldEntry: tableEntry(old), NewEntry: tableEntry(new)} + if message.NewEntry != nil { + message.NewParentPath = filer.DirectoryEtcRemote + } + return &filer_pb.SubscribeMetadataResponse{Directory: filer.DirectoryEtcRemote, TsNs: tsNs, EventNotification: message} +} + +func renamedTable(resp *filer_pb.SubscribeMetadataResponse, newParent, newName string) *filer_pb.SubscribeMetadataResponse { + resp.EventNotification.NewParentPath = newParent + resp.EventNotification.NewEntry.Name = newName + return resp +} + +func TestMountTableEnds(t *testing.T) { + mountedDir := "/buckets/b" + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "b", Path: "/"} + elsewhere := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "other", Path: "/"} + other := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "c", Path: "/"} + otherRemote := &remote_pb.RemoteStorageLocation{Name: "r2", Bucket: "b", Path: "/"} + mapped := map[string]*remote_pb.RemoteStorageLocation{mountedDir: loc} + mappedWithOther := map[string]*remote_pb.RemoteStorageLocation{mountedDir: loc, "/buckets/c": other} + + tests := []struct { + name string + resp *filer_pb.SubscribeMetadataResponse + wantEnded bool + wantMoved *remote_pb.RemoteStorageLocation + }{ + {"another mount added", mountTableEvent(t, 1, mapped, mappedWithOther), false, nil}, + {"unmounted", mountTableEvent(t, 1, mappedWithOther, map[string]*remote_pb.RemoteStorageLocation{"/buckets/c": other}), true, nil}, + {"mounted elsewhere", mountTableEvent(t, 1, mapped, map[string]*remote_pb.RemoteStorageLocation{mountedDir: elsewhere}), true, elsewhere}, + {"table deleted", mountTableEvent(t, 1, mapped, nil), true, nil}, + {"mounted on another remote", mountTableEvent(t, 1, mapped, map[string]*remote_pb.RemoteStorageLocation{mountedDir: otherRemote}), true, otherRemote}, + {"remounted here from another remote", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{mountedDir: otherRemote}, mapped), false, nil}, + {"replayed version from before the mount", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{"/buckets/c": other}, map[string]*remote_pb.RemoteStorageLocation{}), false, nil}, + {"replayed creation of the table", mountTableEvent(t, 1, nil, map[string]*remote_pb.RemoteStorageLocation{"/buckets/c": other}), false, nil}, + {"replayed deletion of the table", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{"/buckets/c": other}, nil), false, nil}, + {"the mount itself", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{"/buckets/c": other}, mappedWithOther), false, nil}, + {"replayed mount elsewhere", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{}, map[string]*remote_pb.RemoteStorageLocation{mountedDir: elsewhere}), true, elsewhere}, + {"replayed unmount from elsewhere: its purge follows", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{mountedDir: elsewhere}, map[string]*remote_pb.RemoteStorageLocation{}), true, nil}, + {"remounted here from elsewhere", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{mountedDir: elsewhere}, mapped), false, nil}, + {"table renamed away", renamedTable(mountTableEvent(t, 1, mapped, mapped), filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE+".bak"), true, nil}, + {"table moved out of the directory", renamedTable(mountTableEvent(t, 1, mapped, mapped), "/etc", filer.REMOTE_STORAGE_MOUNT_FILE), true, nil}, + {"a table-named file in a subdirectory", renamedTable(mountTableEvent(t, 1, nil, map[string]*remote_pb.RemoteStorageLocation{}), filer.DirectoryEtcRemote+"/old", filer.REMOTE_STORAGE_MOUNT_FILE), false, nil}, + {"remote conf updated", &filer_pb.SubscribeMetadataResponse{Directory: filer.DirectoryEtcRemote, EventNotification: &filer_pb.EventNotification{ + NewParentPath: filer.DirectoryEtcRemote, NewEntry: &filer_pb.Entry{Name: "r" + filer.REMOTE_STORAGE_CONF_SUFFIX}}}, false, nil}, + {"file under the mount", &filer_pb.SubscribeMetadataResponse{Directory: mountedDir, EventNotification: &filer_pb.EventNotification{ + NewParentPath: mountedDir, NewEntry: &filer_pb.Entry{Name: filer.REMOTE_STORAGE_MOUNT_FILE}}}, false, nil}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ended, moved, err := mountTableEnds(mountedDir, mountedDir, loc, tt.resp) + if err != nil { + t.Fatal(err) + } + if ended != tt.wantEnded || !proto.Equal(moved, tt.wantMoved) { + t.Errorf("mountTableEnds = %v, %v; want %v, %v", ended, moved, tt.wantEnded, tt.wantMoved) + } + }) + } + + t.Run("unreadable table", func(t *testing.T) { + resp := mountTableEvent(t, 1, mapped, mapped) + resp.EventNotification.NewEntry.Content = []byte{0xff, 0xff, 0xff} + if _, _, err := mountTableEnds(mountedDir, mountedDir, loc, resp); err == nil { + t.Error("want an error for a table that does not unmarshal") + } + }) + + t.Run("a replaced version it cannot read counts as covering", func(t *testing.T) { + unreadable := mountTableEvent(t, 1, mapped, map[string]*remote_pb.RemoteStorageLocation{}) + unreadable.EventNotification.OldEntry.Content = []byte{0xff, 0xff, 0xff} + renamedOver := mountTableEvent(t, 1, mapped, map[string]*remote_pb.RemoteStorageLocation{}) + renamedOver.EventNotification.OldEntry.Name = filer.REMOTE_STORAGE_MOUNT_FILE + ".new" + for _, resp := range []*filer_pb.SubscribeMetadataResponse{unreadable, renamedOver} { + if ended, _, err := mountTableEnds(mountedDir, mountedDir, loc, resp); err != nil || !ended { + t.Errorf("mountTableEnds(%s) = %v, %v; want the mount ended", resp.EventNotification.OldEntry.Name, ended, err) + } + } + }) +} + +func TestMountTableEndsBelowAMount(t *testing.T) { + dir, mountKey := "/buckets/b/sub", "/buckets/b" + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "b", Path: "/"} + nested := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "nested", Path: "/"} + other := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "c", Path: "/"} + mapped := map[string]*remote_pb.RemoteStorageLocation{mountKey: loc} + + tests := []struct { + name string + resp *filer_pb.SubscribeMetadataResponse + wantEnded bool + wantMoved *remote_pb.RemoteStorageLocation + }{ + {"another mount added", mountTableEvent(t, 1, mapped, map[string]*remote_pb.RemoteStorageLocation{mountKey: loc, "/buckets/c": other}), false, nil}, + {"the parent unmounted", mountTableEvent(t, 1, mapped, map[string]*remote_pb.RemoteStorageLocation{}), true, nil}, + {"the directory mounted itself", mountTableEvent(t, 1, mapped, map[string]*remote_pb.RemoteStorageLocation{mountKey: loc, dir: nested}), true, nested}, + {"replayed nested unmount from before", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{mountKey: loc, dir: nested}, mapped), false, nil}, + {"replayed version from before the mount", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{}, map[string]*remote_pb.RemoteStorageLocation{"/buckets/c": other}), false, nil}, + {"replayed unmount of an earlier parent mount", mountTableEvent(t, 1, map[string]*remote_pb.RemoteStorageLocation{mountKey: other}, map[string]*remote_pb.RemoteStorageLocation{}), true, nil}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ended, moved, err := mountTableEnds(dir, mountKey, loc, tt.resp) + if err != nil { + t.Fatal(err) + } + if ended != tt.wantEnded || !proto.Equal(moved, tt.wantMoved) { + t.Errorf("mountTableEnds = %v, %v; want %v, %v", ended, moved, tt.wantEnded, tt.wantMoved) + } + }) + } +} + +func TestDirRemoteLocationBelowAMount(t *testing.T) { + tests := []struct { + name, mountKey, dir, mountPath, want string + }{ + {"the mount itself", "/buckets/b", "/buckets/b", "/", "/x.txt"}, + {"the mount with a trailing slash", "/buckets/b", "/buckets/b/", "/", "/x.txt"}, + {"a directory below the mount", "/buckets/b", "/buckets/b/sub", "/", "/sub/x.txt"}, + {"a directory below a mount at a remote prefix", "/buckets/b", "/buckets/b/sub/deeper", "/prefix", "/prefix/sub/deeper/x.txt"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mountLoc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "rb", Path: tt.mountPath} + dirLoc := dirRemoteLocation(tt.mountKey, tt.dir, mountLoc) + got := toRemoteStorageLocation(util.FullPath(tt.dir), util.FullPath(tt.dir).Child("x.txt"), dirLoc) + if got.Name != "r" || got.Bucket != "rb" || got.Path != tt.want { + t.Errorf("%s/x.txt maps to %s/%s%s; want r/rb%s", tt.dir, got.Name, got.Bucket, got.Path, tt.want) + } + }) + } +} + +func TestStartAfterMountOffset(t *testing.T) { + start := time.Unix(100, 5) + later, earlier := time.Unix(200, 7), time.Unix(50, 3) + tests := []struct { + name string + dir string + timeAgo time.Duration + resumeTsNs int64 + mountOffset time.Time + want time.Time + wantRead bool + }{ + {"below the mount, unmounted since", "/buckets/b/sub", 0, 0, later, later, true}, + {"below the mount, mount offset older", "/buckets/b/sub", 0, 0, earlier, start, true}, + {"below the mount, no mount offset", "/buckets/b/sub", 0, 0, time.Time{}, start, true}, + {"the mount itself", "/buckets/b", 0, 0, later, start, false}, + {"the mount with a trailing slash", "/buckets/b/", 0, 0, later, start, false}, + {"a -timeAgo replay", "/buckets/b/sub", time.Hour, 0, later, start, false}, + {"a retry in this process", "/buckets/b/sub", 0, start.UnixNano(), later, start, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + read := false + got, err := startAfterMountOffset(start, "/buckets/b", tt.dir, tt.timeAgo, tt.resumeTsNs, func() (int64, error) { + read = true + if tt.mountOffset.IsZero() { + return 0, nil + } + return tt.mountOffset.UnixNano(), nil + }) + if err != nil { + t.Fatal(err) + } + if !got.Equal(tt.want) || read != tt.wantRead { + t.Errorf("startAfterMountOffset = %v (read %v); want %v (read %v)", got, read, tt.want, tt.wantRead) + } + }) + } + + if _, err := startAfterMountOffset(start, "/buckets/b", "/buckets/b/sub", 0, 0, func() (int64, error) { + return 0, errors.New("filer unavailable") + }); err == nil || !strings.Contains(err.Error(), "filer unavailable") { + t.Errorf("an unread mount offset must fail the start, got %v", err) + } +} + +func TestAdmitUntilMountEndsDrainsBeforeTheExit(t *testing.T) { + mountedDir := "/buckets/b" + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "b", Path: "/"} + elsewhere := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "other", Path: "/"} + mapped := map[string]*remote_pb.RemoteStorageLocation{mountedDir: loc} + + for _, tt := range []struct { + name string + end *filer_pb.SubscribeMetadataResponse + wantMoved *remote_pb.RemoteStorageLocation + }{ + {"unmount", mountTableEvent(t, 2, mapped, map[string]*remote_pb.RemoteStorageLocation{}), nil}, + {"move", mountTableEvent(t, 2, mapped, map[string]*remote_pb.RemoteStorageLocation{mountedDir: elsewhere}), elsewhere}, + } { + t.Run(tt.name, func(t *testing.T) { + release := make(chan struct{}) + ran := make(chan string, 4) + processor := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error { + <-release + ran <- resp.EventNotification.NewEntry.Name + return nil + }, 128, 0) + type exitCall struct { + to *remote_pb.RemoteStorageLocation + jobsDone int + } + exits := make(chan exitCall, 1) + defer func(exit func(string, *remote_pb.RemoteStorageLocation, *remote_pb.RemoteStorageLocation)) { + exitAtMountEnd = exit + }(exitAtMountEnd) + exitAtMountEnd = func(_ string, _, to *remote_pb.RemoteStorageLocation) { exits <- exitCall{to, len(ran)} } + + admit := admitUntilMountEnds(processor, mountedDir, mountedDir, loc) + if err := admit(&filer_pb.SubscribeMetadataResponse{Directory: mountedDir, TsNs: 1, EventNotification: &filer_pb.EventNotification{ + NewParentPath: mountedDir, NewEntry: &filer_pb.Entry{Name: "a.txt"}}}); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- admit(tt.end) }() + select { + case <-exits: + t.Fatal("exited while an admitted job was still running") + case <-time.After(100 * time.Millisecond): + } + close(release) + select { + case exit := <-exits: + if exit.jobsDone != 1 { + t.Errorf("exited with %d of 1 admitted jobs finished", exit.jobsDone) + } + if !proto.Equal(exit.to, tt.wantMoved) { + t.Errorf("exit with %v, want %v", exit.to, tt.wantMoved) + } + case <-time.After(5 * time.Second): + t.Fatal("no exit after the admitted job finished") + } + if err := <-done; err != nil { + t.Fatal(err) + } + close(ran) + var names []string + for name := range ran { + names = append(names, name) + } + if !slices.Equal(names, []string{"a.txt"}) { + t.Errorf("jobs run = %v, want only the write admitted before the mount table", names) + } + }) + } +}