diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 4a9c950d56..8841b4fee3 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/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 } diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 69371c25b4..dcab97f07e 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/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"))