filer.remote.sync: skip pre-mount mount-table versions, decide the unmount before admitting later events (#12472)

* filer.remote.sync: judge the mount table before admitting later events

The unmount was decided inside a processor job, so events behind the
mapping write, such as remote.unmount's purge deletes, were admitted and
could delete remote objects before the exit landed. The stream callback
now reads each mount-table version before the next event, admits nothing
after the one that ends the mount, saves no offset, and drains the
admitted jobs before exiting.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: ignore mount-table versions from before the mount

Without a saved offset the follow starts at the mounted directory's
creation second, so it replays mount-table versions written before
remote.mount added the directory. Each read as an unmount and the
process exited on every start. A version now ends the write-back only
when the version it replaced mapped the directory to its location.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: judge the mount table by the mount's own key

-dir may lie below a mount, and DetectMountInfo resolves it to the
enclosing mount. Looking the table up by -dir found no mapping in any
version, so a replaced table never counted as mapped and the parent's
unmount was missed, admitting its purge deletes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: a table renamed away ends the mount

A rename of /etc/remote/mount.mapping to another name inside /etc/remote
was ignored, although the filer no longer has a mount table. The table is
now matched by its exact path on both sides of the event, so renaming it
away counts as removing it, and a file of that name elsewhere under
/etc/remote no longer counts as the table.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: a mount on another remote is not the same mount

The mount-table check compared only bucket and path, so remounting the
directory on another remote storage at the same bucket and path kept
writing through the old client, and a replayed unmount from the other
remote stopped the write-back. The remote storage name now counts too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: check the drain at the exit itself

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: resolve the mount covering -dir in each table version

Judging the table by the startup mount key alone missed a mount added at
-dir itself below that key: the key stayed mapped, so the nested mount and
its later removal were both admitted, and the removal's purge deletes ran
against the startup remote. Each version is now resolved the way
DetectMountInfo resolves the live table, and a different covering mount
ends the follow as a move.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: an earlier mount's unmount still ends the follow

Skipping every version whose replaced table did not map the directory to
this mount also skipped the unmount of an earlier mount of the directory,
which a -timeAgo replay reaches, and admitted its purge deletes against
this mount's remote. Only a version in which no mount covers the
directory, replacing one in which none did, is now skipped as history
from before the mount.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: map paths below -dir under its place in the mount

With -dir below a mount, every remote key was taken relative to -dir and
appended to the mount's location, so /mount/sub/x.txt uploaded to x.txt
instead of sub/x.txt and could overwrite another object. The processor
now gets the remote location of -dir itself; the mount's own location
still judges mount-table changes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.remote.sync: resume below a mount no earlier than the mount's offset

remote.unmount saves the offset of the directory it unmounts, after its
purge, and leaves the offsets of directories below it alone. A
write-back of -dir=/mount/sub resumed from its own offset after a
remount, replayed the unmount and exited, on every start. Without
-timeAgo it now starts no earlier than the mount's offset. Whatever that
skips was purged, or was synced by a write-back of the whole mount.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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
 	}

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)
+			}
+		})
+	}
+}