mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 04:27:49 +01:00
filer.remote.sync: no upload on an unanswered supersession lookup; bounded offset read
retriedWriteFile returns the supersession lookup's error instead of treating an unanswered lookup as "not superseded": in a replay an older snapshot could otherwise overwrite a newer remote object. The processor retries the event. isSuperseded keeps its boolean form for the other call sites. GetSyncOffset's KvGet carries a 30 s deadline so a startup that falls back to the saved offset cannot hang on it. TestGatewayUncachedRewriteIsStillUploaded asserted the upload on a failed lookup; it now asserts the withheld upload and the lookup error. (cherry picked from commit 8fc2e718f7cd8322615b4e40007be903ade2dbfa)
This commit is contained in:
1 parent
4f8f817461
commit
e4c0acfa61
4 files changed
+40
-19
No files matched your search
@@ -1,7 +1,7 @@
|
||||
package command
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
@@ -63,11 +63,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 +79,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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -973,15 +973,27 @@ 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 {
|
||||
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) {
|
||||
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
|
||||
}
|
||||
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
|
||||
@@ -1038,7 +1050,9 @@ 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 := supersessionLookup(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)
|
||||
}
|
||||
err = util.RetryOnError("writeFile", func(err error) bool {
|
||||
|
||||
@@ -580,8 +580,11 @@ 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 isSuperseded(&blockingFilerClient{}, dir, event) {
|
||||
t.Error("isSuperseded = true on a lookup that timed out; want false (the write failure is kept and retried)")
|
||||
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 _, 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)
|
||||
}
|
||||
if elapsed := time.Since(started); elapsed > 5*time.Second {
|
||||
t.Fatalf("two timed-out lookups took %v; the deadline did not bound them", elapsed)
|
||||
|
||||
Reference in new issue
Block a user