filer.remote.sync: bounded filer lookups, one chunk cache per process

The filer lookups on the sync path (currentEntry, isSuperseded,
uploadCurrentEntry, liveRemoteEntry, the mount directory read) ran with no
deadline. A lookup that never answers keeps its job alive, the job keeps
its path in the MetadataProcessor conflict index, AddSyncJob waits on the
condition and the daemon stops consuming events with no log line. Each
lookup now carries a 30 s deadline, so the job fails and the processor's
stop-and-resubscribe path runs.

remoteUploadReader built a chunk cache per upload; the cache's worker
goroutine is never stopped, so each upload leaked one goroutine and its
cache (21k goroutines, ~150 MB/h under lock-file churn, OOM at the memory
limit). One chunk cache per process.

(cherry picked from commit ff77fd222772d4f3b68f34d5ee39a66684f90e78)

weed/command/filer_remote_sync_dir.go

@@ -614,8 +614,19 @@ func releaseRenamedSource(filerSource filer_pb.FilerClient, client remote_storag
 
 // currentEntry returns what the filer holds at dir/name now, nil when the
 // entry is gone.
+// remoteSyncLookupTimeout bounds a filer lookup on the sync path. A lookup
+// that never answers keeps its job alive, the job keeps its path in the
+// processor's conflict index, and the stream stops behind it.
+var remoteSyncLookupTimeout = 30 * time.Second
+
+func lookupContext() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), remoteSyncLookupTimeout)
+}
+
 func currentEntry(filerSource filer_pb.FilerClient, dir, name string) (*filer_pb.Entry, error) {
-	current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name))
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerSource, util.NewFullPath(dir, name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return nil, nil
 	}
@@ -743,7 +754,9 @@ func openRemoteObject(client remote_storage.RemoteStorageClient, loc *remote_pb.
 // 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))
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerSource, util.NewFullPath(dir, name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return nil
 	}
@@ -960,7 +973,9 @@ func readRemoteStorageConf(filerClient filer_pb.FilerClient, storageName string)
 // object or the later write uploads the current content. A lookup that fails
 // for any other reason keeps the write failure, so the event is retried.
 func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) bool {
-	current, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.NewFullPath(dir, entry.Name))
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.NewFullPath(dir, entry.Name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return true
 	}
@@ -988,6 +1003,10 @@ func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.
 // the backend a short body on every retry — a hole that never fills wedges
 // the event, and a backend that does not verify the length publishes a
 // truncated object.
+// remoteUploadChunkCache is one per process: a chunk cache runs a worker
+// goroutine that nothing stops, so a cache per upload leaks one per file.
+var remoteUploadChunkCache = chunk_cache.NewChunkCacheInMemory(8)
+
 func remoteUploadReader(filerClient filer_pb.FilerClient, entry *filer_pb.Entry) (reader io.Reader, closeFn func() error, err error) {
 	if len(entry.Content) > 0 {
 		return bytes.NewReader(entry.Content), nil, nil
@@ -999,7 +1018,7 @@ func remoteUploadReader(filerClient filer_pb.FilerClient, entry *filer_pb.Entry)
 		return nil, nil, err
 	}
 	readerAt := filer.NewChunkReaderAtFromClient(context.Background(),
-		filer.NewReaderCache(8, chunk_cache.NewChunkCacheInMemory(8), lookupFileIdFn, nil),
+		filer.NewReaderCache(8, remoteUploadChunkCache, lookupFileIdFn, nil),
 		chunkViews, size, filer.DefaultPrefetchCount)
 	return io.NewSectionReader(readerAt, 0, size), readerAt.Close, nil
 }
@@ -1055,7 +1074,9 @@ func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc
 	// 3. directory creation time
 	var lastOffsetTs time.Time
 	if timeAgo == 0 {
-		mountedDirEntry, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.FullPath(mountedDir))
+		ctx, cancel := lookupContext()
+		mountedDirEntry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(mountedDir))
+		cancel()
 		if err != nil {
 			glog.V(0).Infof("get mounted directory %s: %v", mountedDir, err)
 			return time.Now()
@@ -1126,7 +1147,9 @@ func liveRemoteEntry(filerClient filer_pb.FilerClient, dir string, entry *filer_
 	if entry.RemoteEntry != nil {
 		return entry.RemoteEntry, nil
 	}
-	current, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.NewFullPath(dir, entry.Name))
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.NewFullPath(dir, entry.Name))
 	if err != nil {
 		return nil, err
 	}

weed/command/filer_remote_sync_dir_test.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"slices"
 	"strings"
 	"testing"
@@ -553,6 +554,67 @@ func entryWith(name string, remote *filer_pb.RemoteEntry, chunks ...*filer_pb.Fi
 // of a write in progress: a mount flushes metadata while chunk uploads are
 // still running, so the snapshot can have holes, and the stream reader stops
 // at the first one on every retry.
+// blockingFilerClient answers a lookup only when its context ends.
+type blockingFilerClient struct {
+	stubFilerClient
+}
+
+func (c *blockingFilerClient) LookupDirectoryEntry(ctx context.Context, _ *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	c.lookups++
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (c *blockingFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fn(c)
+}
+
+func TestFilerLookupsOnTheSyncPathCarryADeadline(t *testing.T) {
+	remoteSyncLookupTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { remoteSyncLookupTimeout = 30 * time.Second })
+	const dir = "/buckets/ingest"
+	event := entryWith("lease.json", nil, chunk("3,01", "e1"))
+
+	started := time.Now()
+	current, err := currentEntry(&blockingFilerClient{}, dir, event.Name)
+	if current != nil || !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("currentEntry = %v, %v; want nil, deadline exceeded", current, err)
+	}
+	if isSuperseded(&blockingFilerClient{}, dir, event) {
+		t.Error("isSuperseded = true on a lookup that timed out; want false (the write failure is kept and retried)")
+	}
+	if elapsed := time.Since(started); elapsed > 5*time.Second {
+		t.Fatalf("two timed-out lookups took %v; the deadline did not bound them", elapsed)
+	}
+}
+
+func TestRemoteUploadReaderSharesOneChunkCache(t *testing.T) {
+	entry := entryWith("blob.bin", nil, chunk("3,01", "e1"))
+	for i := 0; i < 3; i++ {
+		if _, closeFn, err := remoteUploadReader(&stubFilerClient{}, entry); err != nil {
+			t.Fatal(err)
+		} else if closeFn != nil {
+			_ = closeFn()
+		}
+	}
+	runtime.GC()
+	before := runtime.NumGoroutine()
+	for i := 0; i < 200; i++ {
+		_, closeFn, err := remoteUploadReader(&stubFilerClient{}, entry)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if closeFn != nil {
+			_ = closeFn()
+		}
+	}
+	runtime.GC()
+	time.Sleep(100 * time.Millisecond)
+	if after := runtime.NumGoroutine(); after-before > 20 {
+		t.Fatalf("200 upload readers left %d goroutines behind (%d -> %d); one chunk cache per reader leaks its worker", after-before, before, after)
+	}
+}
+
 func TestIsSuperseded(t *testing.T) {
 	const dir = "/buckets/ingest"
 	event := entryWith("tmpjt9req69__Alan_Doe_Resume-1.pdf", nil, chunk("3,01", "e1"), chunk("3,02", "e2"))