diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index 5d138e494..2bacf14b8 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -2,6 +2,7 @@ package command import ( "context" + "errors" "fmt" "math" "math/rand" @@ -263,7 +264,11 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour return client.WriteDirectory(dest, message.NewEntry) } glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) + return nil + } if writeErr != nil { return writeErr } @@ -349,7 +354,11 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour return client.UpdateFileMetadata(oldDest, message.OldEntry, message.NewEntry) } else { newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation) - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, newDest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(newDest), writeErr) + return nil + } if writeErr != nil { return writeErr } @@ -386,7 +395,11 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour if message.NewEntry.IsDirectory { return client.WriteDirectory(newDest, message.NewEntry) } - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, newDest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(newDest), writeErr) + return nil + } if writeErr != nil { return writeErr } diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index bab0770ee..d682a7e2e 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -1,6 +1,7 @@ package command import ( + "bytes" "context" "errors" "fmt" @@ -177,7 +178,11 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return client.WriteDirectory(dest, message.NewEntry) } glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) + return nil + } if writeErr != nil { return writeErr } @@ -247,7 +252,11 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return nil } } - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) + return nil + } if writeErr != nil { return writeErr } @@ -259,18 +268,55 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return eachEntryFunc, nil } -func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource *source.FilerSource, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) { - var writeErr error - err = util.Retry("writeFile", func() error { +// isSuperseded reports whether the filer has moved past the entry an event +// described: it is deleted, or it no longer references every chunk the event +// named. Those are the chunks the filer deletes when an entry is updated, so +// they are gone from the volume servers, or about to be, and no retry of the +// upload can succeed. Chunks are compared by file id, not content: a rewrite +// stores even identical bytes under new ids and drops the old ones. A replay +// from an earlier offset (-timeAgo) re-emits such events. Failing one holds the +// sync offset before it, so every restart of the subscription replays it into +// the same dead chunks, and progress on everything 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 rewrite 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)) + if errors.Is(err, filer_pb.ErrNotFound) { + return true + } + if err != nil { + return false + } + if len(entry.Content) > 0 || len(current.Content) > 0 { + return !bytes.Equal(entry.Content, current.Content) + } + return len(filer.DoMinusChunks(entry.GetChunks(), current.GetChunks())) > 0 +} + +// errSuperseded marks an upload that failed for an entry the filer has since +// moved past (isSuperseded). The caller skips the event instead of failing it. +var errSuperseded = errors.New("deleted or rewritten since the event was logged") + +// retriedWriteFile uploads the entry, retrying transient failures. Every failed +// attempt first asks the filer whether the entry is superseded, and stops at +// once when it is: 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) { + err = util.RetryOnError("writeFile", func(err error) bool { + return !errors.Is(err, errSuperseded) && util.IsTransientError(err) + }, func() error { reader := filer.NewFileReader(filerSource, newEntry) glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) + var writeErr error remoteEntry, writeErr = client.WriteFile(dest, newEntry, reader) - if writeErr != nil { - return writeErr + if writeErr != nil && isSuperseded(filerSource, dir, newEntry) { + return fmt.Errorf("%s %w: %w", util.NewFullPath(dir, newEntry.Name), errSuperseded, writeErr) } - return nil + return writeErr }) - if err != nil { + if err != nil && !errors.Is(err, errSuperseded) { glog.Errorf("write to %s: %v", dest, err) } return diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 2cdd5bf77..de08100d9 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -4,12 +4,15 @@ import ( "context" "errors" "fmt" + "io" "strings" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" @@ -405,15 +408,20 @@ func TestIsMetadataOnlyUpdate(t *testing.T) { } } -// stubFilerClient serves one entry; nil means not found. +// stubFilerClient serves one entry; nil means not found. A non-nil err fails +// every lookup instead. type stubFilerClient struct { filer_pb.SeaweedFilerClient entry *filer_pb.Entry + err error lookups int } func (c *stubFilerClient) LookupDirectoryEntry(context.Context, *filer_pb.LookupDirectoryEntryRequest, ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { c.lookups++ + if c.err != nil { + return nil, c.err + } return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entry}, nil } @@ -483,3 +491,141 @@ func TestLiveRemoteEntry(t *testing.T) { } }) } + +func chunk(fileId, etag string) *filer_pb.FileChunk { + return &filer_pb.FileChunk{FileId: fileId, Size: 1024, ETag: etag} +} + +func entryWith(name string, remote *filer_pb.RemoteEntry, chunks ...*filer_pb.FileChunk) *filer_pb.Entry { + return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: remote, Chunks: chunks} +} + +// 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 +// succeed, and failing the event pins the offset before it forever. Those are +// skipped. A failure to upload an entry the filer still holds as described is +// a real failure and must keep failing the event so the subscription retries. +// +// Every write stores its chunks under new file ids, so the entry is compared +// by chunk identity, the way the filer itself decides which chunks an update +// leaves for deletion. Content fingerprints would miss a delete-and-recreate +// of identical bytes and fail the event forever on its dead chunks. +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")) + stamped := &filer_pb.RemoteEntry{StorageName: "b2", RemoteMtime: 1786096669} + + tests := []struct { + name string + filer *stubFilerClient + want bool + }{ + {name: "deleted since", filer: &stubFilerClient{}, want: true}, + {name: "rewritten since", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("4,07", "e3"))}, want: true}, + {name: "recreated with identical content", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("5,11", "e1"), chunk("5,12", "e2"))}, want: true}, + {name: "recreated with other content", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("6,01", "e9"))}, want: true}, + {name: "truncated since", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"))}, want: true}, + {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, chunks still live", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"), chunk("3,03", "e3"))}, want: false}, + {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) + } + if tt.filer.lookups != 1 { + t.Errorf("looked up the filer %d times, want 1", tt.filer.lookups) + } + }) + } + + 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) { + 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) { + t.Error("isSuperseded = true, want false for an entry whose inline content is unchanged") + } + }) +} + +// failingRemote fails every WriteFile with err and counts the attempts. +type failingRemote struct { + remote_storage.RemoteStorageClient + err error + writes int +} + +func (r *failingRemote) WriteFile(*remote_pb.RemoteStorageLocation, *filer_pb.Entry, io.Reader) (*filer_pb.RemoteEntry, error) { + r.writes++ + return nil, r.err +} + +// TestRetriedWriteFileStopsWhenSuperseded checks that a failed upload asks the +// filer at once, and stops retrying when the entry is gone: the ~13s of backoff +// util.Retry would spend on a dead chunk buys nothing. An upload of an entry the +// filer still holds keeps the retry policy it had. +func TestRetriedWriteFileStopsWhenSuperseded(t *testing.T) { + const dir = "/buckets/ingest" + event := entryWith("tmpjt9req69__Alan_Doe_Resume-1.pdf", nil, chunk("3,01", "e1")) + dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "tier", Path: "/" + event.Name} + // What the S3 SDK reports when the volume server no longer has the chunk: + // "requesterror" makes IsTransientError retry it to exhaustion. + deadChunk := errors.New("RequestError: send request failed\ncaused by: Put \"https://s3.example.com/tier/x\": http://volume:8444/3,01?readDeleted=true: 404 Not Found: not found") + + t.Run("superseded, one attempt", func(t *testing.T) { + remote := &failingRemote{err: deadChunk} + filerClient := &stubFilerClient{} + start := time.Now() + _, err := retriedWriteFile(remote, filerClient, dir, event, dest) + if !errors.Is(err, errSuperseded) { + t.Errorf("err = %v, want errSuperseded", err) + } + if !strings.Contains(err.Error(), "404 Not Found") { + t.Errorf("err = %v, want it to keep the write failure", err) + } + if remote.writes != 1 { + t.Errorf("wrote %d times, want 1: retrying a dead chunk cannot succeed", remote.writes) + } + if filerClient.lookups != 1 { + t.Errorf("looked up the filer %d times, want 1", filerClient.lookups) + } + if elapsed := time.Since(start); elapsed > 500*time.Millisecond { + t.Errorf("took %v, want no backoff", elapsed) + } + }) + + t.Run("still as described, transient failure keeps retrying", func(t *testing.T) { + prev := util.RetryWaitTime + util.RetryWaitTime = 1600 * time.Millisecond // two attempts: 1s, then 1.5s + t.Cleanup(func() { util.RetryWaitTime = prev }) + remote := &failingRemote{err: deadChunk} + filerClient := &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"))} + _, err := retriedWriteFile(remote, filerClient, dir, event, dest) + if errors.Is(err, errSuperseded) { + t.Errorf("err = %v, want the write failure itself", err) + } + if remote.writes != 2 { + t.Errorf("wrote %d times, want 2: a transient failure on a live entry is still retried", remote.writes) + } + if filerClient.lookups != 2 { + t.Errorf("looked up the filer %d times, want one per failed attempt", filerClient.lookups) + } + }) + + t.Run("still as described, permanent failure not retried", func(t *testing.T) { + remote := &failingRemote{err: errors.New("AccessDenied: Access Denied")} + filerClient := &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"))} + _, err := retriedWriteFile(remote, filerClient, dir, event, dest) + if err == nil || errors.Is(err, errSuperseded) { + t.Errorf("err = %v, want the write failure itself", err) + } + if remote.writes != 1 { + t.Errorf("wrote %d times, want 1", remote.writes) + } + }) +}