Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-11-07 04:27:49 +01:00
Code Issues Packages Projects Releases Wiki Activity

filer.remote.sync: isSuperseded returns the lookup error itself

Browse Source 
Its only production callers needed the error, so the boolean wrapper was
left to the tests. The tests now assert the error on a failed lookup.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chris LuandClaude Opus 5.5 committed 2026-11-06 00:51:07 -08:00
1 parent 952d80c4d3
commit 51f603b5ec
2 files changed
+37 -32
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

weed/command/filer_remote_sync_dir.go
+8 -19
View File
@@ -974,16 +974,9 @@ 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 {
superseded, _ := supersessionLookup(filerClient, dir, entry)
return superseded
}
// supersessionLookup is isSuperseded with the lookup failure surfaced: an
// upload must not start on a lookup that did not answer, or an older
// snapshot in a replay overwrites a newer remote object.
func supersessionLookup(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) (bool, error) {
// 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))
@@ -993,21 +986,17 @@ func supersessionLookup(filerClient filer_pb.FilerClient, dir string, entry *fil
if err != nil {
return false, err
}
return entryDiffers(entry, current), nil
}
func entryDiffers(entry, current *filer_pb.Entry) bool {
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
@@ -1057,7 +1046,7 @@ var errLookupUnanswered = errors.New("supersession lookup unanswered")
// 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 superseded, lookupErr := supersessionLookup(filerSource, dir, newEntry); lookupErr != nil {
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)
@@ -1066,7 +1055,7 @@ func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource fil
// 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 := supersessionLookup(filerSource, dir, newEntry)
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)
}
weed/command/filer_remote_sync_dir_test.go
+29 -13
View File
@@ -564,8 +564,8 @@ func TestFilerLookupsOnTheSyncPathCarryADeadline(t *testing.T) {
if current != nil || !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("currentEntry = %v, %v; want nil, deadline exceeded", current, err)
}
if superseded, lookupErr := supersessionLookup(&blockingFilerClient{}, dir, event); superseded || !errors.Is(lookupErr, context.DeadlineExceeded) {
t.Errorf("supersessionLookup = %v, %v on a lookup that timed out; want false, deadline exceeded (no upload starts on an unanswered lookup)", superseded, lookupErr)
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)
@@ -722,12 +722,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)
@@ -735,19 +735,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")
}
})
@@ -755,20 +771,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")
}
})