filer sync: do not advance the sync offset past a failed event (#10424)

* util: retry transient errors, not just the ones containing "transport"

util.Retry only retried when the error string contained "transport", so a
plain "read: connection reset by peer" from S3 got zero retries. Classify
the error instead: net timeouts, connection resets, and the throttling and
overload replies S3 and gRPC return are all worth another attempt, while a
cancelled or expired context is not.

* filer sync: hold the sync offset behind a failed event

A sync job that returned an error was logged and forgotten, and the
watermark advanced past it anyway. The offset is the durable resume point,
so the event was never replayed: for filer.remote.sync that left the file
present locally, absent on the remote, with no RemoteEntry and nothing to
retry it.

Pin the watermark at the oldest failed event. Later events keep flowing,
but the persisted offset stays behind the failure, so a restart replays it.

weed/command/filer_sync_jobs.go

@@ -5,6 +5,7 @@ import (
 	"path"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -86,6 +87,12 @@ type MetadataProcessor struct {
 	// tsHeap is a min-heap of active job timestamps with lazy deletion,
 	// used for O(log n) amortized watermark tracking.
 	tsHeap tsMinHeap
+
+	// oldestFailedTsNs is the timestamp of the oldest event whose job returned
+	// an error, or 0 when none has. The watermark is never advanced to it or
+	// past it, so the persisted sync offset stays behind the failure and a
+	// restart replays the event instead of skipping it forever.
+	oldestFailedTsNs int64
 }
 
 func NewMetadataProcessor(fn pb.ProcessMetadataFunc, concurrency int, offsetTsNs int64) *MetadataProcessor {
@@ -254,15 +261,22 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
 
 	go func() {
 
-		if err := util.Retry("metadata processor", func() error {
+		jobErr := util.Retry("metadata processor", func() error {
 			return t.fn(resp)
-		}); err != nil {
-			glog.Errorf("process %v: %v", resp, err)
-		}
+		})
 
 		t.activeJobsLock.Lock()
 		defer t.activeJobsLock.Unlock()
 
+		if jobErr != nil {
+			if t.oldestFailedTsNs == 0 || resp.TsNs < t.oldestFailedTsNs {
+				t.oldestFailedTsNs = resp.TsNs
+				glog.Errorf("process %v: %v; holding sync offset at %v so this event is replayed on restart", resp, jobErr, time.Unix(0, resp.TsNs))
+			} else {
+				glog.Errorf("process %v: %v", resp, jobErr)
+			}
+		}
+
 		delete(t.activeJobs, resp.TsNs)
 		t.removePathFromIndex(jobPaths.path, jobPaths.kind)
 		if jobPaths.newPath != "" {
@@ -277,9 +291,13 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
 			}
 			heap.Pop(&t.tsHeap)
 		}
-		// If this was the oldest job, advance the watermark.
+		// If this was the oldest job, advance the watermark, but never to or
+		// past an event that failed: the offset is the durable resume point,
+		// and moving it over a failure drops that event for good.
 		if t.tsHeap.Len() == 0 || resp.TsNs < t.tsHeap[0] {
-			t.processedTsWatermark.Store(resp.TsNs)
+			if t.oldestFailedTsNs == 0 || resp.TsNs < t.oldestFailedTsNs {
+				t.processedTsWatermark.Store(resp.TsNs)
+			}
 		}
 		t.activeJobsCond.Signal()
 	}()

weed/command/filer_sync_jobs_test.go

@@ -2,6 +2,7 @@ package command
 
 import (
 	"container/heap"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -610,3 +611,76 @@ func TestMetadataProcessorEmptyMarkerKeepsWatermarkStale(t *testing.T) {
 	}
 	t.Logf("marker carried fresh ts %d but watermark stayed stale at %d", freshTs, staleOffset)
 }
+
+// waitForJobsToDrain blocks until every job goroutine has finished bookkeeping.
+func waitForJobsToDrain(t *testing.T, p *MetadataProcessor) {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		p.activeJobsLock.Lock()
+		remaining := len(p.activeJobs)
+		p.activeJobsLock.Unlock()
+		if remaining == 0 {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatal("timed out waiting for sync jobs to drain")
+}
+
+// TestFailedJobHoldsWatermark verifies that a job that returns an error keeps
+// the watermark — and therefore the persisted sync offset — behind the failed
+// event, so a restart replays it. Advancing past it drops the event for good:
+// the file stays local-only and nothing ever retries the upload.
+func TestFailedJobHoldsWatermark(t *testing.T) {
+	const failedTsNs = int64(200)
+	// a permanent error, so util.Retry gives up on the first attempt
+	fn := func(resp *filer_pb.SubscribeMetadataResponse) error {
+		if resp.TsNs == failedTsNs {
+			return errors.New("AccessDenied: Access Denied")
+		}
+		return nil
+	}
+	// concurrency 1 runs the jobs serially in timestamp order
+	p := NewMetadataProcessor(fn, 1, 0)
+
+	p.AddSyncJob(makeResp("/dir", "a.txt", false, 100, true))
+	waitForJobsToDrain(t, p)
+	if got := p.processedTsWatermark.Load(); got != 100 {
+		t.Fatalf("watermark = %d after a successful job, want 100", got)
+	}
+
+	p.AddSyncJob(makeResp("/dir", "b.txt", false, failedTsNs, true))
+	waitForJobsToDrain(t, p)
+	if got := p.processedTsWatermark.Load(); got != 100 {
+		t.Fatalf("watermark = %d after a failed job, want it held at 100", got)
+	}
+
+	// later events keep flowing, but the offset stays behind the failure
+	p.AddSyncJob(makeResp("/dir", "c.txt", false, 300, true))
+	waitForJobsToDrain(t, p)
+	if got := p.processedTsWatermark.Load(); got != 100 {
+		t.Fatalf("watermark = %d after a later success, want it held at 100", got)
+	}
+}
+
+// TestFailedJobHoldsWatermarkAtOldestFailure verifies that the watermark is
+// pinned by the oldest failure, not the most recent one.
+func TestFailedJobHoldsWatermarkAtOldestFailure(t *testing.T) {
+	fn := func(resp *filer_pb.SubscribeMetadataResponse) error {
+		if resp.TsNs == 200 || resp.TsNs == 400 {
+			return errors.New("AccessDenied: Access Denied")
+		}
+		return nil
+	}
+	p := NewMetadataProcessor(fn, 1, 0)
+
+	for _, ts := range []int64{100, 200, 300, 400, 500} {
+		p.AddSyncJob(makeResp("/dir", fmt.Sprintf("f%d.txt", ts), false, ts, true))
+		waitForJobsToDrain(t, p)
+	}
+
+	if got := p.processedTsWatermark.Load(); got != 100 {
+		t.Fatalf("watermark = %d, want it held at 100 by the failure at 200", got)
+	}
+}

weed/util/retry.go

@@ -2,7 +2,11 @@ package util
 
 import (
 	"context"
+	"errors"
+	"io"
+	"net"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -10,6 +14,53 @@ import (
 
 var RetryWaitTime = 6 * time.Second
 
+// transientErrorMessages are substrings of failures that a later attempt is
+// likely to get past: connection resets, timeouts, and the throttling or
+// overload replies S3 and gRPC hand back. Cloud SDKs bury the underlying net
+// error in an opaque wrapper with no Unwrap, so the message is often all
+// that is left to match on.
+var transientErrorMessages = []string{
+	"transport",
+	"connection reset",
+	"connection refused",
+	"broken pipe",
+	"unexpected EOF",
+	"i/o timeout",
+	"TLS handshake timeout",
+	"no such host",
+	"Client.Timeout",
+	"RequestError",
+	"RequestTimeout",
+	"SlowDown",
+	"Throttling",
+	"InternalError",
+	"ServiceUnavailable",
+	"ResourceExhausted",
+	"Unavailable",
+}
+
+// IsTransientError reports whether err is a network or service condition worth
+// retrying. A cancelled or expired context never is: the caller is already gone.
+func IsTransientError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return false
+	}
+	if errors.Is(err, io.ErrUnexpectedEOF) ||
+		errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.ECONNABORTED) ||
+		errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.EPIPE) ||
+		errors.Is(err, syscall.ETIMEDOUT) {
+		return true
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) && netErr.Timeout() {
+		return true
+	}
+	return containErr(err.Error(), transientErrorMessages)
+}
+
 func Retry(name string, job func() error) (err error) {
 	waitTime := time.Second
 	hasErr := false
@@ -22,7 +73,7 @@ func Retry(name string, job func() error) (err error) {
 			waitTime = time.Second
 			break
 		}
-		if strings.Contains(err.Error(), "transport") {
+		if IsTransientError(err) {
 			hasErr = true
 			glog.V(0).Infof("retry %s: err: %v", name, err)
 		} else {

weed/util/retry_test.go

@@ -3,10 +3,75 @@ package util
 import (
 	"context"
 	"errors"
+	"fmt"
+	"io"
+	"net"
+	"syscall"
 	"testing"
 	"time"
 )
 
+func TestIsTransientError(t *testing.T) {
+	transient := []error{
+		// the S3 failure that motivated widening the gate: aws-sdk-go wraps the
+		// net error in an opaque type, so only the message survives
+		errors.New(`RequestError: send request failed caused by: Post "https://s3.eu-west-2.amazonaws.com/b/k?uploads=": read tcp 10.0.0.1:53868->1.2.3.4:443: read: connection reset by peer`),
+		errors.New("rpc error: code = Unavailable desc = transport is closing"),
+		errors.New("SlowDown: Please reduce your request rate."),
+		errors.New("InternalError: We encountered an internal error. Please try again."),
+		fmt.Errorf("send: %w", syscall.ETIMEDOUT),
+		&net.DNSError{Err: "operation timed out", IsTimeout: true},
+		io.ErrUnexpectedEOF,
+	}
+	for _, err := range transient {
+		if !IsTransientError(err) {
+			t.Errorf("expected transient: %v", err)
+		}
+	}
+
+	permanent := []error{
+		nil,
+		errors.New("AccessDenied: Access Denied"),
+		errors.New("NoSuchBucket: The specified bucket does not exist"),
+		context.Canceled,
+		fmt.Errorf("write: %w", context.DeadlineExceeded),
+	}
+	for _, err := range permanent {
+		if IsTransientError(err) {
+			t.Errorf("expected permanent: %v", err)
+		}
+	}
+}
+
+func TestRetryTransientError(t *testing.T) {
+	callCount := 0
+	err := Retry("test", func() error {
+		callCount++
+		if callCount < 2 {
+			return errors.New("read: connection reset by peer")
+		}
+		return nil
+	})
+	if err != nil {
+		t.Errorf("expected success, got %v", err)
+	}
+	if callCount != 2 {
+		t.Errorf("expected 2 calls, got %d", callCount)
+	}
+
+	callCount = 0
+	err = Retry("test", func() error {
+		callCount++
+		return errors.New("AccessDenied: Access Denied")
+	})
+	if err == nil {
+		t.Error("expected error")
+	}
+	if callCount != 1 {
+		t.Errorf("expected 1 call for a permanent error, got %d", callCount)
+	}
+}
+
 func TestRetryUntil(t *testing.T) {
 	// Test case 1: Function succeeds immediately
 	t.Run("SucceedsImmediately", func(t *testing.T) {