diff --git a/weed/command/filer_sync_jobs.go b/weed/command/filer_sync_jobs.go
index a66604478..396b2af2c 100644
--- a/weed/command/filer_sync_jobs.go
+++ b/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()
 	}()
diff --git a/weed/command/filer_sync_jobs_test.go b/weed/command/filer_sync_jobs_test.go
index 8066cca08..1c6a1f3ce 100644
--- a/weed/command/filer_sync_jobs_test.go
+++ b/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)
+	}
+}
diff --git a/weed/util/retry.go b/weed/util/retry.go
index 4443cc27f..a8f7ffb41 100644
--- a/weed/util/retry.go
+++ b/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 {
diff --git a/weed/util/retry_test.go b/weed/util/retry_test.go
index 928eca30f..099293545 100644
--- a/weed/util/retry_test.go
+++ b/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) {