diff --git a/weed/command/filer_remote_gateway.go b/weed/command/filer_remote_gateway.go index 4f03b7927f..040e5b0555 100644 --- a/weed/command/filer_remote_gateway.go +++ b/weed/command/filer_remote_gateway.go @@ -31,6 +31,7 @@ type RemoteGatewayOptions struct { bucketsDir string clientId int32 clientEpoch int32 + resumeTsNs int64 } var _ = filer_pb.FilerClient(&RemoteGatewayOptions{}) diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index db4b276a98..7a9453acfb 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -33,7 +33,10 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo return err } - lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo) + lastOffsetTs, err := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo, option.resumeTsNs) + if err != nil { + return err + } processor := NewMetadataProcessor(eachEntryFunc, 128, lastOffsetTs.UnixNano()) var lastLogTsNs = time.Now().UnixNano() @@ -70,7 +73,9 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo Resubscribe: processor.ResubscribeCh(), } - return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) + err = pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) + option.resumeTsNs = saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, processor) + return err } // makeBucketedEventProcessor is the gateway half of the sync protocol diff --git a/weed/command/filer_remote_gateway_buckets_test.go b/weed/command/filer_remote_gateway_buckets_test.go index 253b765189..32d0b8aed2 100644 --- a/weed/command/filer_remote_gateway_buckets_test.go +++ b/weed/command/filer_remote_gateway_buckets_test.go @@ -2,6 +2,7 @@ package command import ( "slices" + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -63,11 +64,13 @@ func TestGatewayDroppedLocalCopyIsNotUploaded(t *testing.T) { } } -// TestGatewayUncachedRewriteIsStillUploaded: a rewrite of an uncached entry -// holds local data, so the dropped-copy skip does not apply and the rewrite -// uploads. The stamp cannot reach a filer in the test, so the call errors; -// the upload is what is asserted. -func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) { +// TestGatewayUncachedRewriteWaitsForTheSupersessionLookup: a rewrite of an +// uncached entry holds local data, so the dropped-copy skip does not apply +// and the rewrite is an upload. The lookup that guards the upload cannot +// reach a filer in the test, so the event fails with that error and nothing +// is uploaded: an unanswered lookup never lets a snapshot overwrite the +// remote object, and the processor retries the event. +func TestGatewayUncachedRewriteWaitsForTheSupersessionLookup(t *testing.T) { option, remote := gatewayTestOption("gatewaytest-rewrite") t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rewrite") }) process, err := option.makeBucketedEventProcessor(&source.FilerSource{}) @@ -77,14 +80,12 @@ func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) { rewrite := entryWith(droppedCopyName, droppedCopyStamp(), chunk("4,02", "e2")) resp := droppedCopyUpdate(droppedCopyEntry(droppedCopyStamp()), rewrite) - if err := process(resp); err != nil { - t.Logf("stamp against the test filer fails as expected: %v", err) + err = process(resp) + if err == nil || !strings.Contains(err.Error(), "supersession lookup") { + t.Fatalf("process = %v, want the supersession lookup error", err) } - if len(remote.writes) != 1 { - t.Fatalf("writes = %+v, want the rewrite uploaded once", remote.writes) - } - if !slices.Equal(remote.writtenChunks[0], []string{"4,02"}) { - t.Errorf("uploaded chunks %v, want the rewrite's [4,02]", remote.writtenChunks[0]) + if len(remote.writes) != 0 { + t.Fatalf("writes = %+v, want none until the lookup answers", remote.writes) } } diff --git a/weed/command/filer_remote_sync.go b/weed/command/filer_remote_sync.go index 9e96f6d069..60e683454d 100644 --- a/weed/command/filer_remote_sync.go +++ b/weed/command/filer_remote_sync.go @@ -22,6 +22,7 @@ type RemoteSyncOptions struct { dir *string clientId int32 clientEpoch int32 + resumeTsNs int64 } var _ = filer_pb.FilerClient(&RemoteSyncOptions{}) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 4a9c950d56..e7629157b2 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -62,7 +62,10 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour return err } - lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo) + lastOffsetTs, err := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo, option.resumeTsNs) + if err != nil { + return err + } processor := NewMetadataProcessor(eachEntryFunc, 128, lastOffsetTs.UnixNano()) var lastLogTsNs = time.Now().UnixNano() @@ -105,7 +108,25 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour Resubscribe: processor.ResubscribeCh(), } - return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) + err = pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) + option.resumeTsNs = saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, processor) + return err +} + +// 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 +// would otherwise start the retry after the event it failed on. The offset is +// returned for the retry in this process, which must not depend on the write. +func saveResumeOffset(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, dir string, processor *MetadataProcessor) int64 { + offsetTsNs := resumeTsNsFor(processor) + if offsetTsNs <= 0 { + return 0 + } + if err := remote_storage.SetSyncOffset(grpcDialOption, filerAddress, dir, offsetTsNs); err != nil { + glog.Errorf("save the sync offset of %s: %v", dir, err) + } + return offsetTsNs } func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) { @@ -612,10 +633,21 @@ func releaseRenamedSource(filerSource filer_pb.FilerClient, client remote_storag return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry) } +// 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) +} + // currentEntry returns what the filer holds at dir/name now, nil when the // entry is gone. 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 +775,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 } @@ -958,28 +992,35 @@ func readRemoteStorageConf(filerClient filer_pb.FilerClient, storageName string) // after it in the log is never persisted. Skipping is safe: the event that // superseded this one follows in the log, and the delete removes the remote // 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)) +// for any other reason is returned: no upload starts on a lookup that did not +// answer, or an older snapshot in a replay overwrites a newer remote object. +func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) (bool, error) { + 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 + return true, nil } if err != nil { - return false + return false, err } if !filer.HasData(entry) && filer.HasData(current) { // The event described an entry without data (remote-only, or empty); // the filer has written to it since. Uploading the snapshot would put // an empty object where the write belongs. - return true + return true, nil } if len(entry.Content) > 0 || len(current.Content) > 0 { - return !bytes.Equal(entry.Content, current.Content) + return !bytes.Equal(entry.Content, current.Content), nil } return len(filer.DoMinusChunks(entry.GetChunks(), current.GetChunks())) > 0 || - len(filer.DoMinusChunks(current.GetChunks(), entry.GetChunks())) > 0 + len(filer.DoMinusChunks(current.GetChunks(), entry.GetChunks())) > 0, nil } +// 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) + // remoteUploadReader streams the entry's full declared size through // ChunkReadAt, which zero-fills any range no chunk view covers: a sparse // file, a mid-write snapshot flushed while its chunk uploads were still @@ -999,7 +1040,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 } @@ -1010,6 +1051,10 @@ func remoteUploadReader(filerClient filer_pb.FilerClient, entry *filer_pb.Entry) // follows in the log and uploads them. var errSuperseded = errors.New("superseded or incomplete since the event was logged") +// errLookupUnanswered marks an upload stopped because the filer did not +// answer the supersession lookup; the processor retries the event. +var errLookupUnanswered = errors.New("supersession lookup unanswered") + // retriedWriteFile uploads the entry, retrying transient failures. The filer is // asked whether the entry is superseded before the first attempt and after // every failed one, and the upload stops at once when it is. Before: a backlog @@ -1019,18 +1064,30 @@ var errSuperseded = errors.New("superseded or incomplete since the event was log // still holds is worth sending. After: a dead chunk reads as a transient // "RequestError" from the SDK, and waiting out the backoff on it buys nothing. func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource filer_pb.FilerClient, dir string, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) { - if isSuperseded(filerSource, dir, newEntry) { + if superseded, lookupErr := isSuperseded(filerSource, dir, newEntry); lookupErr != nil { + return nil, fmt.Errorf("%s: supersession lookup: %w", util.NewFullPath(dir, newEntry.Name), lookupErr) + } else if superseded { return nil, fmt.Errorf("%s %w", util.NewFullPath(dir, newEntry.Name), errSuperseded) } + // A failed attempt is retried only once the filer has answered that the + // entry is still the one the event described; an unanswered lookup + // ends the loop so no further attempt uploads an unverified snapshot. + afterFailure := func(attemptErr error) error { + superseded, lookupErr := isSuperseded(filerSource, dir, newEntry) + if lookupErr != nil { + return fmt.Errorf("%s %w: %w (after: %v)", util.NewFullPath(dir, newEntry.Name), errLookupUnanswered, lookupErr, attemptErr) + } + if superseded { + return fmt.Errorf("%s %w: %w", util.NewFullPath(dir, newEntry.Name), errSuperseded, attemptErr) + } + return attemptErr + } err = util.RetryOnError("writeFile", func(err error) bool { - return !errors.Is(err, errSuperseded) && util.IsTransientError(err) + return !errors.Is(err, errSuperseded) && !errors.Is(err, errLookupUnanswered) && util.IsTransientError(err) }, func() error { reader, closeFn, readerErr := remoteUploadReader(filerSource, newEntry) if readerErr != nil { - if isSuperseded(filerSource, dir, newEntry) { - return fmt.Errorf("%s %w: %w", util.NewFullPath(dir, newEntry.Name), errSuperseded, readerErr) - } - return readerErr + return afterFailure(readerErr) } if closeFn != nil { defer closeFn() @@ -1038,10 +1095,10 @@ func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource fil glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) var writeErr error remoteEntry, writeErr = client.WriteFile(dest, newEntry, reader) - if writeErr != nil && isSuperseded(filerSource, dir, newEntry) { - return fmt.Errorf("%s %w: %w", util.NewFullPath(dir, newEntry.Name), errSuperseded, writeErr) + if writeErr != nil { + return afterFailure(writeErr) } - return writeErr + return nil }) if err != nil && !errors.Is(err, errSuperseded) { glog.Errorf("write to %s: %v", dest, err) @@ -1049,38 +1106,79 @@ func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource fil return } -func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration) time.Time { - // 1. specified by timeAgo - // 2. last offset timestamp for this directory - // 3. directory creation time +func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration, resumeTsNs int64) (time.Time, error) { + if resumeTsNs > 0 { + // the previous attempt pinned where it stopped, which can sit behind + // both the saved offset and a recomputed -timeAgo window + glog.V(0).Infof("resuming previous attempt at %v", time.Unix(0, resumeTsNs)) + return time.Unix(0, resumeTsNs), nil + } + return resolveLastSyncOffset(mountedDir, timeAgo, + func() (*filer_pb.Entry, error) { + ctx, cancel := lookupContext() + defer cancel() + entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(mountedDir)) + return entry, err + }, + func() (int64, error) { return remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir) }) +} + +// resolveLastSyncOffset picks the position to subscribe from: +// 1. now minus timeAgo, or the saved offset when it is older +// 2. the saved offset for this directory +// 3. the directory's creation time +// A mount lookup that fails never resolves to now: the saved offset still +// orders the replay, and without one the error goes back to the caller's +// retry loop instead of skipping every event since the failure. Only a +// directory that does not exist yet, with no saved offset, starts now. +func resolveLastSyncOffset(mountedDir string, timeAgo time.Duration, lookupMount func() (*filer_pb.Entry, error), readOffset func() (int64, error)) (time.Time, error) { + startedAt := time.Now() var lastOffsetTs time.Time if timeAgo == 0 { - mountedDirEntry, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.FullPath(mountedDir)) + mountedDirEntry, err := lookupMount() if err != nil { - glog.V(0).Infof("get mounted directory %s: %v", mountedDir, err) - return time.Now() + lastOffsetTsNs, offsetErr := readOffset() + if offsetErr == nil && lastOffsetTsNs > 0 { + glog.V(0).Infof("get mounted directory %s: %v; resume from the saved offset %v", mountedDir, err, time.Unix(0, lastOffsetTsNs)) + return time.Unix(0, lastOffsetTsNs), nil + } + if offsetErr == nil && errors.Is(err, filer_pb.ErrNotFound) { + return startedAt, nil + } + return time.Time{}, fmt.Errorf("get mounted directory %s: %w; no saved sync offset (%v)", mountedDir, err, offsetErr) } - lastOffsetTsNs, err := remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir) + lastOffsetTsNs, err := readOffset() + if err != nil { + return time.Time{}, fmt.Errorf("read the sync offset of %s: %w", mountedDir, err) + } if mountedDirEntry != nil { - if err == nil && mountedDirEntry.Attributes.Crtime < lastOffsetTsNs/1000000 { + // a saved offset wins even over a newer creation time: remote.unmount + // saves one after recreating the directory, often in the same second + if lastOffsetTsNs > 0 { lastOffsetTs = time.Unix(0, lastOffsetTsNs) glog.V(0).Infof("resume from %v", lastOffsetTs) } else { lastOffsetTs = time.Unix(mountedDirEntry.Attributes.Crtime, 0) } } else { - lastOffsetTs = time.Now() + lastOffsetTs = startedAt } } else { - lastOffsetTs = time.Now().Add(-timeAgo) - if lastOffsetTsNs, err := remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir); err == nil && lastOffsetTsNs > 0 { + lastOffsetTs = startedAt.Add(-timeAgo) + lastOffsetTsNs, err := readOffset() + if err != nil { + // an unread offset must not resolve to now minus timeAgo: the + // saved one may be older, and everything between would be skipped + return time.Time{}, fmt.Errorf("read the sync offset of %s: %w", mountedDir, err) + } + if lastOffsetTsNs > 0 { if savedOffsetTs := time.Unix(0, lastOffsetTsNs); savedOffsetTs.Before(lastOffsetTs) { lastOffsetTs = savedOffsetTs } } } - return lastOffsetTs + return lastOffsetTs, nil } func toRemoteStorageLocation(mountDir, sourcePath util.FullPath, remoteMountLocation *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation { @@ -1126,7 +1224,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 } @@ -1174,10 +1274,13 @@ func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer remoteEntry = proto.Clone(remoteEntry).(*filer_pb.RemoteEntry) remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano() expected := proto.Clone(entry).(*filer_pb.Entry) + entry = proto.Clone(entry).(*filer_pb.Entry) entry.RemoteEntry = remoteEntry delete(entry.Extended, filer.ExtRemoteRenameSourceKey) err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ + ctx, cancel := lookupContext() + defer cancel() + _, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{ Directory: dir, Entry: entry, Condition: ifEntryEqual(expected), diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 69371c25b4..c4f17b6a73 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" @@ -455,6 +456,11 @@ func (c *stubFilerClient) UpdateEntry(_ context.Context, req *filer_pb.UpdateEnt if c.updateErr != nil { return &filer_pb.UpdateEntryResponse{}, c.updateErr } + if key := req.Directory + "/" + req.Entry.GetName(); c.entries[key] != nil { + c.entries[key] = req.Entry + } else if c.entry != nil && c.entry.Name == req.Entry.GetName() { + c.entry = req.Entry + } return &filer_pb.UpdateEntryResponse{}, nil } @@ -537,6 +543,230 @@ func entryWith(name string, remote *filer_pb.RemoteEntry, chunks ...*filer_pb.Fi return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: remote, Chunks: chunks} } +// 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) UpdateEntry(ctx context.Context, _ *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + c.updates++ + <-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 superseded, lookupErr := isSuperseded(&blockingFilerClient{}, dir, event); superseded || !errors.Is(lookupErr, context.DeadlineExceeded) { + t.Errorf("isSuperseded = %v, %v on a lookup that timed out; want false, deadline exceeded (no upload starts on an unanswered lookup)", superseded, lookupErr) + } + if _, writeErr := retriedWriteFile(nil, &blockingFilerClient{}, dir, event, &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "b", Path: "/lease.json"}); !errors.Is(writeErr, context.DeadlineExceeded) { + t.Errorf("retriedWriteFile = %v on an unanswered supersession lookup; want the lookup error, not an upload", writeErr) + } + stamp := &blockingFilerClient{} + if stampErr := updateLocalEntry(stamp, dir, event, &filer_pb.RemoteEntry{StorageName: "b2"}); stamp.updates != 1 || !errors.Is(stampErr, context.DeadlineExceeded) { + t.Errorf("updateLocalEntry = %v after %d updates on an unanswered stamp; want deadline exceeded after 1", stampErr, stamp.updates) + } + if elapsed := time.Since(started); elapsed > 5*time.Second { + t.Fatalf("four timed-out filer calls took %v; the deadline did not bound them", elapsed) + } +} + +// answerThenBlockFilerClient answers the first lookup and blocks every later +// one until its context ends. +type answerThenBlockFilerClient struct { + stubFilerClient + entry *filer_pb.Entry +} + +func (c *answerThenBlockFilerClient) LookupDirectoryEntry(ctx context.Context, _ *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + c.lookups++ + if c.lookups == 1 { + return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entry}, nil + } + <-ctx.Done() + return nil, ctx.Err() +} + +func (c *answerThenBlockFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { + return fn(c) +} + +func TestRetriedWriteFileStopsWhenTheLookupAfterAFailureIsUnanswered(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")) + dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "tier", Path: "/" + event.Name} + transient := errors.New("RequestError: send request failed\ncaused by: Put \"https://s3.example.com/tier/x\": 503 Service Unavailable") + remote := &failingRemote{err: transient} + filerClient := &answerThenBlockFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"))} + + started := time.Now() + _, err := retriedWriteFile(remote, filerClient, dir, event, dest) + if !errors.Is(err, errLookupUnanswered) || !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want the unanswered-lookup error wrapping the deadline", err) + } + if remote.writes != 1 { + t.Errorf("wrote %d times, want 1: no further attempt until the filer answers", remote.writes) + } + if filerClient.lookups != 2 { + t.Errorf("looked up the filer %d times, want one before the attempt and one after it failed", filerClient.lookups) + } + if elapsed := time.Since(started); elapsed > 5*time.Second { + t.Fatalf("took %v; the retry loop did not stop on the unanswered lookup", elapsed) + } +} + +func TestResolveLastSyncOffsetKeepsTheSavedOffsetWhenTheLookupFails(t *testing.T) { + saved := time.Date(2026, 11, 4, 17, 57, 10, 123_000_000, time.UTC) + lookupFails := func() (*filer_pb.Entry, error) { return nil, context.DeadlineExceeded } + savedOffset := func() (int64, error) { return saved.UnixNano(), nil } + + got, err := resolveLastSyncOffset("/buckets/x", 0, lookupFails, savedOffset) + if err != nil { + t.Fatal(err) + } + if !got.Equal(saved) { + t.Fatalf("resolved %v, want the saved offset %v: a failed mount lookup must not start the replay at now", got, saved) + } + + // a mount that answers resumes from the saved offset when it postdates the directory + entry := &filer_pb.Entry{Name: "x", Attributes: &filer_pb.FuseAttributes{Crtime: saved.Add(-time.Hour).Unix()}} + got, err = resolveLastSyncOffset("/buckets/x", 0, func() (*filer_pb.Entry, error) { return entry, nil }, savedOffset) + if err != nil { + t.Fatal(err) + } + if !got.Equal(saved) { + t.Fatalf("resolved %v, want %v", got, saved) + } + + // remote.unmount recreates the directory, then saves the offset in the same + // second: starting at the creation second would replay the unmount's purge + unmounted := &filer_pb.Entry{Name: "x", Attributes: &filer_pb.FuseAttributes{Crtime: saved.Unix()}} + got, err = resolveLastSyncOffset("/buckets/x", 0, func() (*filer_pb.Entry, error) { return unmounted, nil }, savedOffset) + if err != nil { + t.Fatal(err) + } + if !got.Equal(saved) { + t.Fatalf("resolved %v, want the saved offset %v", got, saved) + } +} + +func TestResolveLastSyncOffsetReturnsTheErrorToTheRetryLoop(t *testing.T) { + lookupFails := func() (*filer_pb.Entry, error) { return nil, context.DeadlineExceeded } + noOffset := func() (int64, error) { return 0, nil } + readFails := func() (int64, error) { return 0, errors.New("filer unavailable") } + + // no saved offset to fall back on: the error goes back to the caller + if _, err := resolveLastSyncOffset("/buckets/x", 0, lookupFails, noOffset); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("got %v, want the lookup error: without a saved offset the caller's retry loop decides, not an exit", err) + } + + // no directory and no saved offset yet: a first run starts no later than + // the lookup, so a bucket created while the offset is read is replayed + started := time.Now() + var lookedUp time.Time + notFound := func() (*filer_pb.Entry, error) { + lookedUp = time.Now() + return nil, fmt.Errorf("lookup /buckets: %w", filer_pb.ErrNotFound) + } + slowNoOffset := func() (int64, error) { time.Sleep(10 * time.Millisecond); return 0, nil } + if got, err := resolveLastSyncOffset("/buckets", 0, notFound, slowNoOffset); err != nil || got.Before(started) || got.After(lookedUp) { + t.Fatalf("got %v, %v; want a start between %v and the lookup at %v", got, err, started, lookedUp) + } + + // the mount answers but the saved offset is unreadable: never its creation time + mount := &filer_pb.Entry{Name: "x", Attributes: &filer_pb.FuseAttributes{Crtime: 1}} + if _, err := resolveLastSyncOffset("/buckets/x", 0, func() (*filer_pb.Entry, error) { return mount, nil }, readFails); err == nil || !strings.Contains(err.Error(), "filer unavailable") { + t.Fatalf("got %v, want the offset read error: the creation time would replay everything already synced", err) + } + + // -timeAgo set, the saved offset unreadable: never now minus timeAgo + if _, err := resolveLastSyncOffset("/buckets/x", time.Hour, func() (*filer_pb.Entry, error) { return nil, nil }, readFails); err == nil || !strings.Contains(err.Error(), "filer unavailable") { + t.Fatalf("got %v, want the offset read error", err) + } +} + +// TestSaveResumeOffsetKeepsTheStartForTheRetry: a subscription that ends +// before its first checkpoint still records where it started, so the retry's +// -timeAgo, recomputed from a later now, cannot start after it. +func TestSaveResumeOffsetKeepsTheStartForTheRetry(t *testing.T) { + filerAddress, dialOption := startKvFiler(t) + const dir = "/buckets/x" + start := time.Now().Add(-time.Hour) + processor := NewMetadataProcessor(func(*filer_pb.SubscribeMetadataResponse) error { return nil }, 1, start.UnixNano()) + saveResumeOffset(dialOption, filerAddress, dir, processor) + + got, err := resolveLastSyncOffset(dir, time.Hour, func() (*filer_pb.Entry, error) { return nil, nil }, func() (int64, error) { + return remote_storage.GetSyncOffset(dialOption, filerAddress, dir) + }) + if err != nil { + t.Fatal(err) + } + if got.After(start) { + t.Fatalf("the retry starts at %v, after the first subscription's start %v", got, start) + } + + // the write fails and the retry reads no saved offset: the position + // returned for this process still holds the start + failingAddress, failingDialOption := startFiler(t, &failingKvFilerServer{}) + resumeTsNs := saveResumeOffset(failingDialOption, failingAddress, dir, processor) + emptyAddress, emptyDialOption := startKvFiler(t) + got, err = collectLastSyncOffset(&stubFilerClient{}, emptyDialOption, emptyAddress, dir, time.Hour, resumeTsNs) + if err != nil { + t.Fatal(err) + } + if got.After(start) { + t.Fatalf("after a failed write the retry starts at %v, after the first subscription's start %v", got, start) + } +} + +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) + } +} + // TestIsSuperseded decides what a failed upload means for its event (#11148). // A replay from an earlier offset re-emits creates for entries the filer has // since deleted or rewritten; their chunks are gone, so the upload can never @@ -571,12 +801,12 @@ func TestIsSuperseded(t *testing.T) { {name: "still as described", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"))}, want: false}, {name: "still as described, since replicated", filer: &stubFilerClient{entry: entryWith(event.Name, stamped, chunk("3,01", "e1"), chunk("3,02", "e2"))}, want: false}, {name: "appended since", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"), chunk("3,03", "e3"))}, want: true}, - {name: "filer lookup failed", filer: &stubFilerClient{err: errors.New("rpc error: code = Unavailable")}, want: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := isSuperseded(tt.filer, dir, event); got != tt.want { - t.Errorf("isSuperseded = %v, want %v", got, tt.want) + got, err := isSuperseded(tt.filer, dir, event) + if err != nil || got != tt.want { + t.Errorf("isSuperseded = %v, %v; want %v, nil", got, err, tt.want) } if tt.filer.lookups != 1 { t.Errorf("looked up the filer %d times, want 1", tt.filer.lookups) @@ -584,19 +814,35 @@ func TestIsSuperseded(t *testing.T) { }) } + t.Run("filer lookup failed", func(t *testing.T) { + lookupErr := errors.New("rpc error: code = Unavailable") + if got, err := isSuperseded(&stubFilerClient{err: lookupErr}, dir, event); got || !errors.Is(err, lookupErr) { + t.Errorf("isSuperseded = %v, %v; want false and the lookup error", got, err) + } + }) + + superseded := func(t *testing.T, filer *stubFilerClient, entry *filer_pb.Entry) bool { + t.Helper() + got, err := isSuperseded(filer, dir, entry) + if err != nil { + t.Fatalf("isSuperseded: %v", err) + } + return got + } + t.Run("written since: the event described an entry without data", func(t *testing.T) { remoteOnly := entryWith("video.mp4", &filer_pb.RemoteEntry{StorageName: "b2", RemoteSize: 20971520, RemoteMtime: 1786096669}) - if !isSuperseded(&stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry, chunk("3,09", "e9"))}, dir, remoteOnly) { + if !superseded(t, &stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry, chunk("3,09", "e9"))}, remoteOnly) { t.Error("isSuperseded = false, want true: the filer wrote local data to a remote-only entry") } - if isSuperseded(&stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry)}, dir, remoteOnly) { + if superseded(t, &stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry)}, remoteOnly) { t.Error("isSuperseded = true, want false: the entry is still remote-only") } empty := &filer_pb.Entry{Name: "touch.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}} - if !isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt", Content: []byte("now")}}, dir, empty) { + if !superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt", Content: []byte("now")}}, empty) { t.Error("isSuperseded = false, want true: the empty file was written to") } - if isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt"}}, dir, empty) { + if superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt"}}, empty) { t.Error("isSuperseded = true, want false: the file is still empty") } }) @@ -604,20 +850,20 @@ func TestIsSuperseded(t *testing.T) { t.Run("mid-write snapshot with a hole, filled in since", func(t *testing.T) { snapshot := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("220,03", "e3", 2048)) filled := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("216,02", "e2", 1024), chunkAt("220,03", "e3", 2048)) - if !isSuperseded(&stubFilerClient{entry: filled}, dir, snapshot) { + if !superseded(t, &stubFilerClient{entry: filled}, snapshot) { t.Error("isSuperseded = false, want true: the filer holds the finished write, the snapshot can never be read past its hole") } - if isSuperseded(&stubFilerClient{entry: filled}, dir, filled) { + if superseded(t, &stubFilerClient{entry: filled}, filled) { t.Error("isSuperseded = true, want false for the event that describes the finished write") } }) t.Run("inline content rewritten since", func(t *testing.T) { event := &filer_pb.Entry{Name: "note.txt", Content: []byte("v1")} - if !isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v2")}}, dir, event) { + if !superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v2")}}, event) { t.Error("isSuperseded = false, want true for an entry whose inline content has changed") } - if isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v1")}}, dir, event) { + if superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v1")}}, event) { t.Error("isSuperseded = true, want false for an entry whose inline content is unchanged") } }) @@ -1366,6 +1612,28 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { }) } +// TestUpdateLocalEntryLeavesTheEventEntryAlone: a transient stamp failure +// retries the same event in place, so the event must still read as unsynced +// and keep its rename source, or the retry skips the upload and the stamp. +func TestUpdateLocalEntryLeavesTheEventEntryAlone(t *testing.T) { + const dir = "/buckets/b/vol" + event := entryWith("state.db", nil, chunk("3,01", "e1")) + event.Extended = map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("src")} + filerClient := &stubFilerClient{updateErr: status.Error(codes.Unavailable, "filer is shutting down")} + if err := updateLocalEntry(filerClient, dir, event, &filer_pb.RemoteEntry{StorageName: "gcs", RemoteSize: 7}); err == nil { + t.Fatal("err = nil, want the transient failure") + } + if event.RemoteEntry != nil || len(event.Extended[filer.ExtRemoteRenameSourceKey]) == 0 { + t.Errorf("event entry = %v, want it untouched by the failed stamp", event) + } + if !shouldSendToRemote(event) { + t.Error("shouldSendToRemote = false after a failed stamp: the in-place retry would skip the upload") + } + if sent := filerClient.updated.GetEntry(); sent.GetRemoteEntry() == nil || len(sent.GetExtended()[filer.ExtRemoteRenameSourceKey]) > 0 { + t.Errorf("stamp request entry = %v, want the remote entry set and the rename source dropped", sent) + } +} + // TestUpdateLocalEntrySkipsDeletedEntry: the stamp that follows an upload lands // on an entry a later event already deleted. The filer answers "not found"; the // delete's own event follows in the log, so the stamp is skipped rather than @@ -1651,7 +1919,7 @@ func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) { t.Run("a marker still in place keeps the source", func(t *testing.T) { // the stamp did not land: the filer still has the marker - filerClient := &stubFilerClient{entry: pending()} + filerClient := &stubFilerClient{entry: pending(), updateErr: status.Error(codes.FailedPrecondition, "entry changed")} remote := &recordingRemote{objects: objects(), statETag: `"copy"`} if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", pending(), renameSource{path: "/buckets/b/src/a.jpg", loc: oldDest}, dest, ""); err == nil { t.Fatal("err = nil, want the event retried") diff --git a/weed/remote_storage/track_sync_offset.go b/weed/remote_storage/track_sync_offset.go index 7fa77a468a..a762a815e5 100644 --- a/weed/remote_storage/track_sync_offset.go +++ b/weed/remote_storage/track_sync_offset.go @@ -3,6 +3,7 @@ package remote_storage import ( "context" "errors" + "time" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -23,10 +24,16 @@ func SyncOffsetKey(dir string) []byte { return syncKey } +// syncOffsetTimeout bounds an offset read or write: one that never answers +// would hold the daemon's start, or its event stream behind the checkpoint. +const syncOffsetTimeout = 30 * time.Second + func GetSyncOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, dir string) (lastOffsetTsNs int64, readErr error) { readErr = pb.WithFilerClient(false, 0, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: SyncOffsetKey(dir)}) + ctx, cancel := context.WithTimeout(context.Background(), syncOffsetTimeout) + defer cancel() + resp, err := client.KvGet(ctx, &filer_pb.KvGetRequest{Key: SyncOffsetKey(dir)}) if err != nil { return err } @@ -54,7 +61,9 @@ func SetSyncOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, dir s valueBuf := make([]byte, 8) util.Uint64toBytes(valueBuf, uint64(offsetTsNs)) - resp, err := client.KvPut(context.Background(), &filer_pb.KvPutRequest{ + ctx, cancel := context.WithTimeout(context.Background(), syncOffsetTimeout) + defer cancel() + resp, err := client.KvPut(ctx, &filer_pb.KvPutRequest{ Key: SyncOffsetKey(dir), Value: valueBuf, })