filer.remote.sync: an unanswered lookup after a failed attempt ends the retry loop; an unread offset never resolves to now minus timeAgo

After a failed upload attempt the supersession lookup's error ended in
isSuperseded's boolean and the loop could upload the snapshot again
unverified. The lookup error now ends the loop (errLookupUnanswered, not
retried) and the processor retries the event. With -timeAgo set, a failed
read of the saved offset resolved to now minus timeAgo, skipping everything
older; the daemon exits instead so a restart retries.

(cherry picked from commit 55ca2a88e01d7b556d494a404b9cc263c27b2566)

weed/command/filer_remote_sync_dir.go

@@ -1041,6 +1041,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
@@ -1055,15 +1059,25 @@ func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource fil
 	} 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 := supersessionLookup(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()
@@ -1071,10 +1085,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)
@@ -1126,7 +1140,13 @@ func resolveLastSyncOffset(mountedDir string, timeAgo time.Duration, lookupMount
 		}
 	} else {
 		lastOffsetTs = time.Now().Add(-timeAgo)
-		if lastOffsetTsNs, err := readOffset(); err == nil && lastOffsetTsNs > 0 {
+		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
+			glog.Fatalf("read the sync offset of %s: %v", mountedDir, err)
+		}
+		if lastOffsetTsNs > 0 {
 			if savedOffsetTs := time.Unix(0, lastOffsetTsNs); savedOffsetTs.Before(lastOffsetTs) {
 				lastOffsetTs = savedOffsetTs
 			}

weed/command/filer_remote_sync_dir_test.go

@@ -591,6 +591,52 @@ func TestFilerLookupsOnTheSyncPathCarryADeadline(t *testing.T) {
 	}
 }
 
+// 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, 0, time.UTC)
 	lookupFails := func() (*filer_pb.Entry, error) { return nil, context.DeadlineExceeded }