filer.remote.sync: skip an upload whose source entry was deleted or rewritten (#11149)

* filer.remote.sync: skip an upload whose source entry was deleted or rewritten A replay from an earlier offset (-timeAgo) re-emits create and update events for entries the filer has since deleted or rewritten. Their chunks are gone from the volume servers, so the upload can never succeed, and failing the event holds the sync offset before it: every restart of the subscription replays it into the same dead chunks, and progress on everything after it in the log is never persisted. One such entry stops replication for the whole mount. When the upload fails, look the entry up on the filer. Gone, or holding other content than the event described, the event is superseded and is skipped with an error log; the event that superseded it follows in the log and brings the remote to the current state. Otherwise the failure stands and the event is retried as before. Fixes #11148 * filer.remote.sync: compare chunks by file id when deciding an event is superseded filer.IsSameData compares chunk ETags, so a delete-and-recreate of identical bytes, which stores the same content under new file ids and drops the old ones, looked still as described and kept failing the event on its dead chunks. Compare by file id with DoMinusChunks, the way the filer itself decides which chunks an update leaves for deletion: the event is superseded when the current entry no longer references every chunk it named, and still as described when it does, including when more chunks were appended after it. * filer.remote.sync: ask the filer on the first failed upload attempt, not after the backoff The superseded check ran after util.Retry had given up, so every dead entry still cost the full retry cycle, about 13s, before it was skipped: the SDK reports a missing chunk as "RequestError", which IsTransientError takes as worth retrying. Move the check into the retry loop with util.RetryOnError. Any failed attempt asks the filer, and the loop stops at once when the entry is gone, surfacing errSuperseded for the caller to skip. An entry the filer still holds keeps the retry policy it had. filer.remote.gateway shares retriedWriteFile and the same offset-pinning processor, so its three call sites skip a superseded event the same way.
2026-09-08 15:41:15 +02:00 · 2026-09-04 00:18:37 -07:00
parent 06838e28b2
commit ed9d58873e
3 changed files with 218 additions and 13 deletions
@@ -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
 				}
@@ -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
@@ -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)
+		}
+	})
+}