s3 lifecycle: bound the daily-replay pass so a quiet cluster stops wedging the job (#10578)

* s3 lifecycle: bound the daily-replay subscription at the pass boundary

A pass opens one meta-log subscription and 16 shard drains, then waits
on all of them. Nothing told the subscription where the pass ends, so
the only exit was the fan-out spotting an event past runNow — i.e. some
unrelated write landing under /buckets after the pass started. On a
cluster that goes quiet the reader parks in Recv, every shard drain
starves on an empty channel, and Run never returns. The job sits at
stage "starting" with the executor slot held and no log line, so
expiry stops cluster-wide until someone restarts the worker.

The pass covers (globalStartTsNs, runNow], so say that: UntilNs on the
subscribe request makes the filer end the stream once it has shipped
that range. The reader then closes the event channel on the way out,
which is what unblocks the fan-out and the drains when the stream
finishes on its own rather than by cancellation.

Same fix retires the other silent hang: a reader that failed early
(subscribe error, stream error) also left every drain waiting forever.

* s3 lifecycle: keep a halted shard from starving the shared fan-out

A drain that halts mid-stream (BLOCKED / RETRY_LATER / an RPC error on
dispatch) returns while the fan-out is still routing that shard's
events. After 256 of them the per-shard buffer is full and the fan-out
blocks on the send, so no other shard sees another event. Run's
WaitGroup never drains, and the teardown that would cancel the reader
sits behind that wait — the pass wedges exactly like an idle
subscription did, with one S3 hiccup as the trigger.

Keep discarding the channel after runShard returns. The events are
past this shard's saved cursor and get re-scanned next pass anyway.

* s3 lifecycle: assert the starved shard actually made progress

The fan-out test only checked that Run returned, which a version that
quietly dropped the second shard's events would also satisfy. Assert
the dispatch landed and the cursor moved.

recordingClient gains a per-object outcome map: the two shards dispatch
from separate goroutines, so pinning BLOCKED by call index was a race
waiting to pick the wrong shard.

* s3 lifecycle: fail the pass when the shared subscription dies

Closing the event channel on reader exit is what unblocks the shard
drains, but it also means a subscribe that never opened, or a stream
that broke mid-pass, now ends every drain cleanly. Run logged that at
V(2) and returned the shard result — so a filer failure produced a
green lifecycle job that had processed nothing.

Surface it as the pass error. Cursors still hold what was processed and
tomorrow resumes there; what changes is that the job stops claiming
success.

Cancellation has to stay a non-error — the shell driver's -runtime cap
is a truncated pass, not a failed one — and a canceled gRPC stream
arrives as a status code, not a wrapped context.Canceled, so isCanceled
checks both forms the way the rest of the tree does.

* s3 lifecycle: decide reader cancellation by intent, not status code

A stream we cancel and a stream the filer cancels both arrive as
codes.Canceled, so classifying the reader's exit by its error let a
truncated pass report success whenever the failure happened to carry a
cancellation status.

Intent is knowable exactly, so read that instead: the pass stops on
purpose only when the caller's context ended (the shell driver's
-runtime cap) or the fan-out hit the pass boundary itself. Everything
else is a broken subscription and fails the pass.

TestRun_ServerSideCancelFailsThePass and TestRun_CappedPassIsNotAFailure
are the same codes.Canceled from the reader with opposite verdicts —
the pair only passes because the decision no longer looks at the error.

* s3 lifecycle: time out a subscription that stops delivering

UntilNs ends a healthy stream and gRPC keepalive catches a dead
connection, but neither reaches a filer that keeps answering pings while
its handler has stopped producing. The pass would wait on that forever,
since s3_lifecycle is the one job type with no execution timeout.

Bound the wait for each response at 20 minutes, and opt into the filer's
idle heartbeats so a caught-up stream proves liveness instead of looking
stalled. The default sits above the filer's 15-minute metadata-gap
recovery budget, so a subscriber legitimately parked on a gap is never
mistaken for a stalled one.

Recv is only interruptible by killing the RPC, so it moves to its own
goroutine behind a per-response deadline. The timer covers only the wait
on the filer — dispatch to Events happens outside it, so a slow consumer
can't trip the watchdog.

Approach and the 20-minute figure are from #10577 by way of comparing
the two fixes; the wiring differs because the reader here ends the pass
by closing its event channel rather than cancelling the fan-out.

* s3 lifecycle: trim the comments added by this branch

Keep the non-obvious why, drop the prose restating what the code says.

* s3 lifecycle: snapshot reader intent where the reader stops

Sampling ctx.Err() during teardown reads it after the drains and cursor
saves have run. A reader that failed while the deadline was still live,
on a pass whose teardown then outlives that deadline, was classified as
an intentional stop and reported success.

Sampling earlier in Run is not the fix either: before the shard wait, a
legitimately capped pass has not reached its deadline yet and would be
misclassified the other way. Intent belongs where the reader actually
stops, so the reader goroutine records it next to the error it returns.

Reported by greptile on #10578.

* s3 lifecycle: cover the worker-dispatched pass with nothing due

The e2e suite drives the shell command in 14 of 15 files; the one test
on the real admin->worker path backdates an object, so its own delete
pushes a meta-log event past the pass boundary and ends the pass. The
branch where a pass has nothing to dispatch was never exercised through
the worker.

Cover it, asserting the pass returns on its own: no admin cancellation,
and the executor slot free for the next one.

This is not a regression test for the wedge. A pass used to end when any
write landed past its boundary, and on a shared test cluster something
usually does — the whole suite passes on the unfixed build, verified.
The deterministic guards stay the dailyrun unit tests; this one would
catch a pass that hangs unconditionally.

test/s3/lifecycle/s3_lifecycle_admin_dispatch_test.go

@@ -36,12 +36,65 @@ func TestLifecycleAdminDispatchSucceedsWithCustomFilerGrpcPort(t *testing.T) {
 
 	waitForLifecycleWorkerReady(t, adminEndpoint)
 
-	// Lifecycle is a long-running batch; the run endpoint cancels it at
-	// this timeout, which converts a healthy run into canceled_count=1.
-	const runTimeoutSeconds = 30
-	body, err := json.Marshal(map[string]any{
-		"timeout_seconds": runTimeoutSeconds,
-	})
+	payload := runLifecycleJob(t, adminEndpoint, 30)
+
+	require.GreaterOrEqual(t, jsonNumber(t, payload, "detected_count"), 1)
+	require.Equal(t, 0, jsonNumber(t, payload, "error_count"),
+		"dispatched job errored — likely filer_grpc_address was raw host:httpPort.grpcPort")
+
+	require.Eventuallyf(t, func() bool {
+		_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucket), Key: aws.String(oldKey),
+		})
+		return err != nil
+	}, 30*time.Second, 500*time.Millisecond,
+		"expected %s/%s to be deleted after admin-dispatched lifecycle run", bucket, oldKey)
+}
+
+// A worker-dispatched pass with rules in place but nothing due: the
+// invariant is that a pass returns on its own, so the admin never has to
+// cancel it and the executor slot is free for the next one.
+//
+// Not a regression test for the wedge this scenario comes from. Before
+// the subscription was bounded a pass ended only when some write landed
+// past its boundary, and on a shared test cluster something usually
+// does — verified: the whole suite passes on the unfixed build. The
+// deterministic guards are the dailyrun unit tests. This covers the
+// worker path end-to-end and would catch a pass that hangs
+// unconditionally.
+func TestLifecycleAdminDispatchCompletesWithNothingDue(t *testing.T) {
+	adminEndpoint := envOr("ADMIN_ENDPOINT", defaultAdminEndpoint)
+
+	c := s3Client(t)
+	bucket := uniqueBucket("admin-idle")
+	mustCreateBucket(t, c, bucket)
+	putExpirationLifecycle(t, c, bucket, "expire/", 1)
+	// Deliberately not backdated: the rule matches the prefix but nothing
+	// is due, so the pass dispatches nothing.
+	putObject(t, c, bucket, "expire/fresh.txt", "fresh")
+
+	waitForLifecycleWorkerReady(t, adminEndpoint)
+
+	// Let the writes above settle below the pass boundary, or one of them
+	// ends the pass and the wedge is masked.
+	time.Sleep(2 * time.Second)
+
+	payload := runLifecycleJob(t, adminEndpoint, 30)
+
+	require.Equal(t, 0, jsonNumber(t, payload, "canceled_count"),
+		"pass did not return on its own; the admin cancelled it at the run timeout")
+	require.GreaterOrEqual(t, jsonNumber(t, payload, "success_count"), 1,
+		"expected the dispatched pass to complete")
+	require.Equal(t, 0, jsonNumber(t, payload, "skipped_active_count"),
+		"a previous pass still held the executor slot")
+}
+
+// runLifecycleJob drives the admin's run endpoint and returns its decoded
+// response. timeoutSeconds bounds the run admin-side: a pass that never
+// returns comes back as canceled_count=1 after that long.
+func runLifecycleJob(t *testing.T, adminEndpoint string, timeoutSeconds int) map[string]any {
+	t.Helper()
+	body, err := json.Marshal(map[string]any{"timeout_seconds": timeoutSeconds})
 	require.NoError(t, err)
 	req, err := http.NewRequestWithContext(
 		context.Background(), http.MethodPost,
@@ -59,18 +112,7 @@ func TestLifecycleAdminDispatchSucceedsWithCustomFilerGrpcPort(t *testing.T) {
 	require.NoError(t, json.NewDecoder(resp.Body).Decode(&payload))
 	t.Logf("admin /api/plugin/job-types/s3_lifecycle/run response: %v", payload)
 	require.Equal(t, http.StatusOK, resp.StatusCode, "admin run endpoint failed: %v", payload)
-
-	require.GreaterOrEqual(t, jsonNumber(t, payload, "detected_count"), 1)
-	require.Equal(t, 0, jsonNumber(t, payload, "error_count"),
-		"dispatched job errored — likely filer_grpc_address was raw host:httpPort.grpcPort")
-
-	require.Eventuallyf(t, func() bool {
-		_, err := c.HeadObject(context.Background(), &s3.HeadObjectInput{
-			Bucket: aws.String(bucket), Key: aws.String(oldKey),
-		})
-		return err != nil
-	}, 30*time.Second, 500*time.Millisecond,
-		"expected %s/%s to be deleted after admin-dispatched lifecycle run", bucket, oldKey)
+	return payload
 }
 
 func jsonNumber(t *testing.T, payload map[string]any, key string) int {

weed/s3api/s3lifecycle/DESIGN.md

@@ -108,7 +108,7 @@ One filer `SubscribeMetadata` stream per `dailyrun.Run()` call, covering every s
 
 `globalStartTsNs = min(per-shard cursor, runNow - maxTTL)`. Pre-loaded once at pass start so the subscription's `StartTsNs` covers every shard's needed range; per-shard drains then filter `ev.TsNs <= shard.startTsNs` locally.
 
-Fan-out cancels the reader on the first `ev.TsNs > runNow` (meta-log events arrive in TsNs order; everything after is past the pass boundary). Per-shard channels are buffered to 256 events — large enough to absorb bursts without back-pressuring the fan-out.
+The subscription is bounded: `UntilNs = runNow`, so the filer ends the stream once it has delivered everything up to the pass boundary and the pass ends on its own. Fan-out also cancels the reader on the first `ev.TsNs > runNow` as a backstop (meta-log events arrive in TsNs order; everything after is past the boundary) — that used to be the *only* way a pass ended, which wedged the job for as long as the cluster stayed quiet. Per-shard channels are buffered to 256 events — large enough to absorb bursts without back-pressuring the fan-out.
 
 ## Action kinds and dispatch paths
 
@@ -140,7 +140,7 @@ type Cursor struct {
 
 `LastWalkedNs` is JSON-omitempty, so cursor files written before that field existed decode cleanly as zero (treated as "never walked steady-state" → next pass seeds the anchor).
 
-Cursor save uses a fresh `context.Background()` with a 5s timeout because the steady-state drain exits via passCtx cancellation (the only way an idle subscription ends). Saving with the canceled passCtx would silently drop the cursor and the next pass would re-replay from the same floor.
+Cursor save uses a fresh `context.Background()` with a 5s timeout because a caller-imposed wall-clock cap on the pass (the shell driver's `-runtime`) cancels the drain's context. Saving with the canceled context would silently drop the cursor and the next pass would re-replay from the same floor.
 
 In steady state the start position honors the cursor verbatim — the floor `runNow - maxTTL` is applied only on cold start (`!found`). The drain freezes the cursor at the last pre-skip event so pending matches with `DueTime == TsNs + maxTTL` stay in scope across passes; bumping forward in steady state would orphan exactly those events.
 

weed/s3api/s3lifecycle/dailyrun/process_matches_test.go

@@ -19,19 +19,25 @@ import (
 
 // recordingClient captures every LifecycleDelete request the
 // daily-run path emits. Default outcome is DONE; tests that need
-// other outcomes set responses by index.
+// other outcomes set responses by index, or outcomeByObject when
+// shards dispatch concurrently and call order isn't fixed.
 type recordingClient struct {
-	mu        sync.Mutex
-	requests  []*s3_lifecycle_pb.LifecycleDeleteRequest
-	responses []s3_lifecycle_pb.LifecycleDeleteOutcome
-	calls     atomic.Int32
+	mu              sync.Mutex
+	requests        []*s3_lifecycle_pb.LifecycleDeleteRequest
+	responses       []s3_lifecycle_pb.LifecycleDeleteOutcome
+	outcomeByObject map[string]s3_lifecycle_pb.LifecycleDeleteOutcome
+	calls           atomic.Int32
 }
 
 func (c *recordingClient) LifecycleDelete(_ context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
 	c.mu.Lock()
 	c.requests = append(c.requests, req)
 	idx := int(c.calls.Add(1)) - 1
+	byObject, pinned := c.outcomeByObject[req.ObjectPath]
 	c.mu.Unlock()
+	if pinned {
+		return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: byObject}, nil
+	}
 	out := s3_lifecycle_pb.LifecycleDeleteOutcome_DONE
 	if idx < len(c.responses) {
 		out = c.responses[idx]

weed/s3api/s3lifecycle/dailyrun/run.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -88,8 +89,15 @@ type Config struct {
 
 	// 0 -> unbounded.
 	EventBudget int
+
+	// 0 -> defaultSubscriptionReceiveTimeout.
+	SubscriptionReceiveTimeout time.Duration
 }
 
+// Above the filer's 15m maxGapStall, so a subscriber parked on a gap is
+// not mistaken for a stalled one.
+const defaultSubscriptionReceiveTimeout = 20 * time.Minute
+
 // Run executes the daily replay for every shard in cfg.Shards
 // concurrently. Returns the first shard error; the rest log and run to
 // completion so one shard's transient failure doesn't lose other shards'
@@ -132,6 +140,9 @@ func Run(ctx context.Context, cfg Config) error {
 		fanoutDone      chan struct{}
 		cancelRead      context.CancelFunc
 		globalStartTsNs int64
+		// Set by the reader goroutine when it stops for a reason we asked
+		// for, rather than a broken stream.
+		stoppedOnPurpose atomic.Bool
 	)
 	if rsh != [32]byte{} {
 		// Pre-load all per-shard cursors so the shared subscription
@@ -141,7 +152,7 @@ func Run(ctx context.Context, cfg Config) error {
 		// exactly the case where a rule's TTL equals maxTTL and an
 		// older event still has DueTime <= runNow.
 		globalStartTsNs = computeGlobalStartTsNs(ctx, cfg, runNow, maxTTL)
-		shardEvents, readerDone, fanoutDone, cancelRead = startSharedSubscription(ctx, cfg, runNow, globalStartTsNs)
+		shardEvents, readerDone, fanoutDone, cancelRead = startSharedSubscription(ctx, cfg, runNow, globalStartTsNs, &stoppedOnPurpose)
 		defer cancelRead()
 	}
 
@@ -159,6 +170,13 @@ func Run(ctx context.Context, cfg Config) error {
 			if err := runShard(ctx, cfg, snap, runNow, sh, ch); err != nil {
 				errCh <- err
 			}
+			// A halted dispatch returns runShard mid-stream. Keep
+			// discarding, or the fan-out blocks on this shard's full
+			// buffer and starves every other drain.
+			if ch != nil {
+				for range ch {
+				}
+			}
 		}()
 	}
 	wg.Wait()
@@ -167,11 +185,16 @@ func Run(ctx context.Context, cfg Config) error {
 	// Tear down the shared subscription. cancelRead unblocks both the
 	// reader's gRPC stream and the fan-out's send loop; we wait on both
 	// so their goroutines don't outlive Run.
+	var readerErr error
 	if cancelRead != nil {
 		cancelRead()
 		<-fanoutDone
-		if rerr := <-readerDone; rerr != nil && !errors.Is(rerr, context.Canceled) && !errors.Is(rerr, context.DeadlineExceeded) {
-			glog.V(2).Infof("daily_run: shared reader returned: %v", rerr)
+		if rerr := <-readerDone; rerr != nil && !stoppedOnPurpose.Load() {
+			// The drains ended on a closed channel, so the pass is
+			// truncated. Cursors hold what was processed; what matters is
+			// that the job doesn't go green on a dead subscription.
+			readerErr = fmt.Errorf("shared meta-log subscription: %w", rerr)
+			glog.Warningf("daily_run: %v", readerErr)
 		}
 	}
 
@@ -185,6 +208,12 @@ func Run(ctx context.Context, cfg Config) error {
 			glog.V(1).Infof("daily_run: additional shard error: %v", err)
 		}
 	}
+	if readerErr != nil {
+		errCount++
+		if first == nil {
+			first = readerErr
+		}
+	}
 	status := "ok"
 	if first != nil {
 		status = "error"
@@ -280,7 +309,9 @@ func computeGlobalStartTsNs(ctx context.Context, cfg Config, runNow time.Time, m
 // Events arriving with TsNs > runUpTo (the pass boundary) cause the
 // fan-out to cancel the reader and close all per-shard channels,
 // ending the pass.
-func startSharedSubscription(ctx context.Context, cfg Config, runNow time.Time, globalStartTsNs int64) (map[int]chan *reader.Event, chan error, chan struct{}, context.CancelFunc) {
+func startSharedSubscription(ctx context.Context, cfg Config, runNow time.Time, globalStartTsNs int64, stoppedOnPurpose *atomic.Bool) (map[int]chan *reader.Event, chan error, chan struct{}, context.CancelFunc) {
+	// Set by the fan-out when it ends the pass on an event past runUpTo.
+	var pastBoundary atomic.Bool
 	shardSet := make(map[int]bool, len(cfg.Shards))
 	shardEvents := make(map[int]chan *reader.Event, len(cfg.Shards))
 	for _, sh := range cfg.Shards {
@@ -301,22 +332,40 @@ func startSharedSubscription(ctx context.Context, cfg Config, runNow time.Time,
 		clientID = int32(util.RandomInt32())
 	}
 
+	runUpTo := runNow.UnixNano()
+	receiveTimeout := cfg.SubscriptionReceiveTimeout
+	if receiveTimeout == 0 {
+		receiveTimeout = defaultSubscriptionReceiveTimeout
+	}
 	events := make(chan *reader.Event, 4*len(cfg.Shards))
 	rd := &reader.Reader{
 		ShardPredicate: func(id int) bool { return shardSet[id] },
 		BucketsPath:    cfg.BucketsPath,
 		StartTsNs:      globalStartTsNs,
+		// The pass covers (globalStartTsNs, runUpTo]. Unbounded, it ends
+		// only when an unrelated write pushes an event past runUpTo — so
+		// a quiet cluster wedges it.
+		UntilTsNs:      runUpTo,
 		Events:         events,
 		EventBudget:    cfg.EventBudget,
+		ReceiveTimeout: receiveTimeout,
 	}
 
 	readerCtx, cancelReader := context.WithCancel(ctx)
 	readerDone := make(chan error, 1)
 	go func() {
-		readerDone <- rd.Run(readerCtx, cfg.FilerClient, clientName, clientID)
+		rerr := rd.Run(readerCtx, cfg.FilerClient, clientName, clientID)
+		// Not decidable from the error: a stream we cancel and one the
+		// filer cancels both arrive as codes.Canceled. Snapshot intent
+		// here rather than after teardown, or a deadline expiring during
+		// the drains and cursor saves masks a real failure.
+		stoppedOnPurpose.Store(ctx.Err() != nil || pastBoundary.Load())
+		readerDone <- rerr
+		// Only sender. Closing is what ends the fan-out, and through it
+		// every drain, when the stream finishes on its own.
+		close(events)
 	}()
 
-	runUpTo := runNow.UnixNano()
 	fanoutDone := make(chan struct{})
 	go func() {
 		defer close(fanoutDone)
@@ -341,6 +390,7 @@ func startSharedSubscription(ctx context.Context, cfg Config, runNow time.Time,
 				// too. Cancel the reader so subsequent passes don't
 				// pay for stream tail we'd drop anyway.
 				if ev.TsNs > runUpTo {
+					pastBoundary.Store(true)
 					cancelReader()
 					return
 				}
@@ -389,6 +439,10 @@ func validate(cfg Config) error {
 	if cfg.WalkerInterval < 0 {
 		return fmt.Errorf("daily_run: negative WalkerInterval %v (0 = unthrottled, positive values throttle)", cfg.WalkerInterval)
 	}
+	// Negative would reach the reader as "disabled" by accident.
+	if cfg.SubscriptionReceiveTimeout < 0 {
+		return fmt.Errorf("daily_run: negative SubscriptionReceiveTimeout %v (0 = default %v)", cfg.SubscriptionReceiveTimeout, defaultSubscriptionReceiveTimeout)
+	}
 	for _, sh := range cfg.Shards {
 		if sh < 0 || sh >= s3lifecycle.ShardCount {
 			return fmt.Errorf("daily_run: shard %d out of [0, %d)", sh, s3lifecycle.ShardCount)
@@ -554,18 +608,15 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim
 	}
 
 	lastOK, _, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, snap, startTsNs, events)
-	// Cursor save uses a fresh ctx because the steady-state drain exits
-	// via passCtx cancellation (the only signal the filer subscription
-	// gets when no new events arrive). Saving with the canceled passCtx
-	// would silently drop the cursor and the next pass would re-replay
-	// from the same floor — defeating advancement entirely.
+	// Fresh ctx: a wall-clock cap on the pass (the shell driver's
+	// -runtime) cancels the drain's, and saving with it would drop the
+	// cursor and re-replay from the same floor next pass.
 	saveCtx, saveCancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer saveCancel()
 	if drainErr != nil {
 		_ = saveCursorAndPublish(saveCtx, cfg.Persister, shardID, Cursor{TsNs: lastOK, RuleSetHash: rsh, PromotedHash: promoted, LastWalkedNs: lastWalkedNs})
-		// passCtx timeout is the expected end-of-pass for an idle
-		// subscription; not a real error. Other drain errors still
-		// propagate.
+		// Cut short by its wall-clock cap: truncated, not failed — the
+		// cursor above carries progress forward.
 		if errors.Is(drainErr, context.DeadlineExceeded) || errors.Is(drainErr, context.Canceled) {
 			return nil
 		}

weed/s3api/s3lifecycle/dailyrun/run_bounded_test.go

@@ -0,0 +1,297 @@
+package dailyrun
+
+import (
+	"context"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// idleSubscribeStream: nothing to deliver past the pass boundary. A
+// bounded subscription gets EOF; an unbounded one blocks forever.
+type idleSubscribeStream struct {
+	grpc.ClientStream
+	ctx     context.Context
+	untilNs int64
+}
+
+func (s *idleSubscribeStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	if s.untilNs != 0 {
+		return nil, io.EOF
+	}
+	<-s.ctx.Done()
+	return nil, s.ctx.Err()
+}
+
+type idleSubscribeClient struct {
+	filer_pb.SeaweedFilerClient
+	gotReq chan *filer_pb.SubscribeMetadataRequest
+}
+
+func (c *idleSubscribeClient) SubscribeMetadata(ctx context.Context, req *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_SubscribeMetadataClient, error) {
+	select {
+	case c.gotReq <- req:
+	default:
+	}
+	return &idleSubscribeStream{ctx: ctx, untilNs: req.UntilNs}, nil
+}
+
+// The reported wedge: the pass ended only when an unrelated write pushed
+// an event past its boundary, so a quiet cluster starved every drain and
+// Run's WaitGroup never reached zero.
+func TestRun_EndsOnIdleSubscription(t *testing.T) {
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 30},
+		}},
+	}, engine.CompileOptions{})
+	require.NotEqual(t, [32]byte{}, engine.ReplayContentHash(e.Snapshot()),
+		"rule must be replay-eligible or Run skips the subscription entirely")
+
+	client := &idleSubscribeClient{gotReq: make(chan *filer_pb.SubscribeMetadataRequest, 1)}
+	runNow := time.Now().UTC()
+	cfg := Config{
+		Shards:      []int{0, 1, 2},
+		BucketsPath: "/buckets",
+		Engine:      e,
+		FilerClient: client,
+		Client:      stubLifecycleClient{},
+		Persister:   newMemPersister(),
+		Lister:      stubSiblingLister{},
+		Now:         func() time.Time { return runNow },
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- Run(context.Background(), cfg) }()
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run did not return on an idle subscription")
+	}
+
+	req := <-client.gotReq
+	assert.Equal(t, runNow.UnixNano(), req.UntilNs, "subscription must stop at the pass boundary")
+}
+
+// failingSubscribeClient fails the subscribe call outright.
+type failingSubscribeClient struct {
+	filer_pb.SeaweedFilerClient
+	err error
+}
+
+func (c *failingSubscribeClient) SubscribeMetadata(_ context.Context, _ *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_SubscribeMetadataClient, error) {
+	return nil, c.err
+}
+
+// Flip side of closing the event channel: the drains now end cleanly on
+// a dead subscription, so the pass would otherwise go green having
+// processed nothing.
+func TestRun_ReaderFailureFailsThePass(t *testing.T) {
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 30},
+		}},
+	}, engine.CompileOptions{})
+
+	cfg := Config{
+		Shards:      []int{0, 1},
+		BucketsPath: "/buckets",
+		Engine:      e,
+		FilerClient: &failingSubscribeClient{err: status.Error(codes.Unavailable, "filer down")},
+		Client:      stubLifecycleClient{},
+		Persister:   newMemPersister(),
+		Lister:      stubSiblingLister{},
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- Run(context.Background(), cfg) }()
+	select {
+	case err := <-done:
+		require.Error(t, err, "a dead subscription must not report a successful pass")
+		assert.Contains(t, err.Error(), "subscription")
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run did not return after the subscription failed")
+	}
+}
+
+// blockingSubscribeStream never delivers; only a cap ends the pass.
+type blockingSubscribeStream struct {
+	grpc.ClientStream
+	ctx context.Context
+}
+
+func (s *blockingSubscribeStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	<-s.ctx.Done()
+	// A canceled gRPC stream returns a status code, not context.Canceled.
+	return nil, status.Error(codes.Canceled, "context canceled")
+}
+
+type blockingSubscribeClient struct {
+	filer_pb.SeaweedFilerClient
+}
+
+func (c *blockingSubscribeClient) SubscribeMetadata(ctx context.Context, _ *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_SubscribeMetadataClient, error) {
+	return &blockingSubscribeStream{ctx: ctx}, nil
+}
+
+// Counterweight to TestRun_ServerSideCancelFailsThePass: same
+// codes.Canceled from the reader, opposite verdict. Only intent
+// separates them, which is why the decision reads ctx not the error.
+func TestRun_CappedPassIsNotAFailure(t *testing.T) {
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 30},
+		}},
+	}, engine.CompileOptions{})
+
+	cfg := Config{
+		Shards:      []int{0, 1},
+		BucketsPath: "/buckets",
+		Engine:      e,
+		FilerClient: &blockingSubscribeClient{},
+		Client:      stubLifecycleClient{},
+		Persister:   newMemPersister(),
+		Lister:      stubSiblingLister{},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- Run(ctx, cfg) }()
+	select {
+	case err := <-done:
+		require.NoError(t, err, "a pass truncated by its own wall-clock cap is not a failure")
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run did not return after the pass was capped")
+	}
+}
+
+// Neither UntilNs nor keepalive reaches this: a healthy connection whose
+// filer has stopped producing.
+func TestRun_StalledSubscriptionTimesOutThePass(t *testing.T) {
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 30},
+		}},
+	}, engine.CompileOptions{})
+
+	cfg := Config{
+		Shards:                     []int{0, 1},
+		BucketsPath:                "/buckets",
+		Engine:                     e,
+		FilerClient:                &blockingSubscribeClient{},
+		Client:                     stubLifecycleClient{},
+		Persister:                  newMemPersister(),
+		Lister:                     stubSiblingLister{},
+		SubscriptionReceiveTimeout: 50 * time.Millisecond,
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- Run(context.Background(), cfg) }()
+	select {
+	case err := <-done:
+		require.ErrorIs(t, err, reader.ErrReceiveTimeout)
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run did not return on a stalled subscription")
+	}
+}
+
+func TestValidate_RejectsNegativeSubscriptionReceiveTimeout(t *testing.T) {
+	cfg := validatableConfig()
+	cfg.SubscriptionReceiveTimeout = -time.Second
+	require.ErrorContains(t, validate(cfg), "SubscriptionReceiveTimeout")
+	cfg.SubscriptionReceiveTimeout = 0
+	require.NoError(t, validate(cfg))
+}
+
+// slowSavePersister stretches teardown past the caller's deadline.
+type slowSavePersister struct {
+	*memPersister
+	delay time.Duration
+}
+
+func (p *slowSavePersister) Save(ctx context.Context, shardID int, c Cursor) error {
+	time.Sleep(p.delay)
+	return p.memPersister.Save(ctx, shardID, c)
+}
+
+// A reader that fails while the caller's deadline is still live, on a
+// pass whose teardown then outlives that deadline. Sampling ctx.Err()
+// after the drains and cursor saves would read the late deadline as
+// intent and report the truncated pass as a success.
+func TestRun_LateDeadlineDoesNotMaskReaderFailure(t *testing.T) {
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 30},
+		}},
+	}, engine.CompileOptions{})
+
+	cfg := Config{
+		Shards:      []int{0, 1},
+		BucketsPath: "/buckets",
+		Engine:      e,
+		FilerClient: &failingSubscribeClient{err: status.Error(codes.Unavailable, "filer down")},
+		Client:      stubLifecycleClient{},
+		Persister:   &slowSavePersister{memPersister: newMemPersister(), delay: 300 * time.Millisecond},
+		Lister:      stubSiblingLister{},
+	}
+
+	// Deadline lands during the cursor saves, well after the reader failed.
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- Run(ctx, cfg) }()
+	select {
+	case err := <-done:
+		require.Error(t, err, "a deadline expiring during teardown must not excuse an earlier reader failure")
+		require.NotNil(t, ctx.Err(), "test is meaningless unless the deadline did expire")
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run did not return")
+	}
+}
+
+// The case status-code classification could not express: the filer
+// cancels while the caller's context is untouched.
+func TestRun_ServerSideCancelFailsThePass(t *testing.T) {
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 30},
+		}},
+	}, engine.CompileOptions{})
+
+	cfg := Config{
+		Shards:      []int{0, 1},
+		BucketsPath: "/buckets",
+		Engine:      e,
+		FilerClient: &failingSubscribeClient{err: status.Error(codes.Canceled, "context canceled")},
+		Client:      stubLifecycleClient{},
+		Persister:   newMemPersister(),
+		Lister:      stubSiblingLister{},
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- Run(context.Background(), cfg) }()
+	select {
+	case err := <-done:
+		require.Error(t, err, "a peer-canceled stream truncates the pass and must not report success")
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run did not return after the subscription was canceled")
+	}
+}

weed/s3api/s3lifecycle/dailyrun/run_fanout_test.go

@@ -0,0 +1,148 @@
+package dailyrun
+
+import (
+	"context"
+	"io"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+// scriptedSubscribeStream replays a fixed list, then EOFs like a filer
+// honoring UntilNs.
+type scriptedSubscribeStream struct {
+	grpc.ClientStream
+	responses []*filer_pb.SubscribeMetadataResponse
+	next      int
+}
+
+func (s *scriptedSubscribeStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	if s.next >= len(s.responses) {
+		return nil, io.EOF
+	}
+	resp := s.responses[s.next]
+	s.next++
+	return resp, nil
+}
+
+type scriptedSubscribeClient struct {
+	filer_pb.SeaweedFilerClient
+	responses []*filer_pb.SubscribeMetadataResponse
+}
+
+func (c *scriptedSubscribeClient) SubscribeMetadata(_ context.Context, _ *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_SubscribeMetadataClient, error) {
+	return &scriptedSubscribeStream{responses: c.responses}, nil
+}
+
+func subscribeResponse(bucket, key string, mtime time.Time, tsNs int64) *filer_pb.SubscribeMetadataResponse {
+	return &filer_pb.SubscribeMetadataResponse{
+		TsNs: tsNs,
+		EventNotification: &filer_pb.EventNotification{
+			NewParentPath: "/buckets/" + bucket,
+			NewEntry: &filer_pb.Entry{
+				Name:       key,
+				Attributes: &filer_pb.FuseAttributes{Mtime: mtime.Unix(), FileSize: 1},
+			},
+		},
+	}
+}
+
+// keyForShard finds an object key in bucket that hashes to shardID.
+func keyForShard(t *testing.T, bucket string, shardID int) string {
+	t.Helper()
+	for i := 0; i < 10000; i++ {
+		key := "k" + strconv.Itoa(i)
+		if s3lifecycle.ShardID(bucket, key) == shardID {
+			return key
+		}
+	}
+	t.Fatalf("no key hashes to shard %d", shardID)
+	return ""
+}
+
+// The second wedge: a drain halted by a BLOCKED dispatch stops reading
+// its channel, the fan-out blocks once the 256 buffer fills, and every
+// other shard starves behind it.
+func TestRun_HaltedShardDoesNotStarveOthers(t *testing.T) {
+	const bucket = "bk"
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	prior := map[s3lifecycle.ActionKey]engine.PriorState{}
+	ruleHash := s3lifecycle.RuleHash(rule)
+	for _, k := range s3lifecycle.RuleActionKinds(rule) {
+		prior[s3lifecycle.ActionKey{Bucket: bucket, RuleHash: ruleHash, ActionKind: k}] = engine.PriorState{
+			BootstrapComplete: true,
+			Mode:              engine.ModeEventDriven,
+		}
+	}
+	e := engine.New()
+	snap := e.Compile([]engine.CompileInput{{Bucket: bucket, Rules: []*s3lifecycle.Rule{rule}}},
+		engine.CompileOptions{PriorStates: prior})
+
+	haltedShard, starvedShard := 0, 1
+	haltedKey := keyForShard(t, bucket, haltedShard)
+	starvedKey := keyForShard(t, bucket, starvedShard)
+
+	runNow := time.Now().UTC()
+	evTime := runNow.Add(-48 * time.Hour) // past the 1-day expiration
+	// Past the 256 buffer, so the fan-out blocks on the halted shard
+	// before reaching the starved shard's event.
+	var responses []*filer_pb.SubscribeMetadataResponse
+	for i := 0; i < 300; i++ {
+		responses = append(responses, subscribeResponse(bucket, haltedKey, evTime, evTime.UnixNano()+int64(i)))
+	}
+	responses = append(responses, subscribeResponse(bucket, starvedKey, evTime, runNow.Add(-time.Hour).UnixNano()))
+
+	// Pinned per object, not call index: the shards dispatch concurrently.
+	client := &recordingClient{outcomeByObject: map[string]s3_lifecycle_pb.LifecycleDeleteOutcome{
+		haltedKey: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED,
+	}}
+
+	// Resume from before the events; the cold-start floor would skip them.
+	// Hashes must match or runShard takes the recovery branch instead.
+	persister := newMemPersister()
+	seeded := Cursor{
+		TsNs:         runNow.Add(-72 * time.Hour).UnixNano(),
+		RuleSetHash:  engine.ReplayContentHash(snap),
+		PromotedHash: engine.PromotedHash(snap, engine.MaxEffectiveTTL(snap)),
+	}
+	for _, sh := range []int{haltedShard, starvedShard} {
+		require.NoError(t, persister.Save(context.Background(), sh, seeded))
+	}
+
+	cfg := Config{
+		Shards:      []int{haltedShard, starvedShard},
+		BucketsPath: "/buckets",
+		Engine:      e,
+		FilerClient: &scriptedSubscribeClient{responses: responses},
+		Client:      client,
+		Persister:   persister,
+		Lister:      stubSiblingLister{},
+		Now:         func() time.Time { return runNow },
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- Run(context.Background(), cfg) }()
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run wedged: the fan-out blocked on a shard that had stopped draining")
+	}
+
+	// Returning isn't enough: dropping the starved shard's events would
+	// return just as cleanly.
+	assert.Contains(t, client.seenObjects(), starvedKey,
+		"the shard behind the halted one must still get its events")
+	starvedCursor, found, err := persister.Load(context.Background(), starvedShard)
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.Greater(t, starvedCursor.TsNs, seeded.TsNs, "starved shard's cursor must advance")
+}

weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go

@@ -3,6 +3,7 @@ package dailyrun
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -15,19 +16,25 @@ import (
 // memPersister is a minimal in-memory CursorPersister. Phase 4b tests
 // drive runShard directly; the recovery-branch path returns before
 // drainShardEvents so the heavier filer/client/lister fakes aren't
-// needed here.
+// needed here. Locked because Run fans out a goroutine per shard and
+// they all load and save through the same persister.
 type memPersister struct {
+	mu    sync.Mutex
 	store map[int]Cursor
 }
 
 func newMemPersister() *memPersister { return &memPersister{store: map[int]Cursor{}} }
 
 func (p *memPersister) Load(_ context.Context, shardID int) (Cursor, bool, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	c, ok := p.store[shardID]
 	return c, ok, nil
 }
 
 func (p *memPersister) Save(_ context.Context, shardID int, c Cursor) error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	p.store[shardID] = c
 	return nil
 }

weed/s3api/s3lifecycle/reader/reader.go

@@ -75,17 +75,57 @@ type Reader struct {
 	StartTsNs int64
 	Events    chan<- *Event
 
+	// UntilTsNs ends the stream once the filer has delivered through this
+	// timestamp. Zero follows forever, which on an idle cluster parks a
+	// bounded caller in Recv until something unrelated is written.
+	UntilTsNs int64
+
 	// EventBudget caps how many events Run processes before returning nil.
 	// Zero = unbounded; the run continues until ctx cancellation or stream
 	// error. Used by the worker scheduler to bound a single READ task.
 	EventBudget int
 
+	// ReceiveTimeout bounds the wait for each response, covering a filer
+	// that stops producing while the transport still answers keepalives.
+	// Also opts into idle heartbeats so a caught-up stream stays alive.
+	// Keep above the filer's 15m maxGapStall — a subscriber parked on a
+	// gap sends nothing and is not stuck. Zero disables it.
+	ReceiveTimeout time.Duration
+
 	// bucketsPathSlash is BucketsPath with a guaranteed trailing slash,
 	// computed once on Run and reused per event to avoid recomputing the
 	// normalized prefix in extractBucketKey.
 	bucketsPathSlash string
 }
 
+// ErrReceiveTimeout: stream still open, but no events and no heartbeats.
+var ErrReceiveTimeout = errors.New("reader: metadata receive timeout")
+
+type receiveResult struct {
+	resp *filer_pb.SubscribeMetadataResponse
+	err  error
+}
+
+// awaitResponse waits for the next response under ReceiveTimeout. Started
+// per call, so it times the filer only — a slow Events consumer blocks in
+// dispatchOne, outside this window, and can't trip the watchdog.
+func (r *Reader) awaitResponse(ctx context.Context, received <-chan receiveResult) (*filer_pb.SubscribeMetadataResponse, error, bool) {
+	var timeout <-chan time.Time
+	if r.ReceiveTimeout > 0 {
+		timer := time.NewTimer(r.ReceiveTimeout)
+		defer timer.Stop()
+		timeout = timer.C
+	}
+	select {
+	case result := <-received:
+		return result.resp, result.err, false
+	case <-timeout:
+		return nil, nil, true
+	case <-ctx.Done():
+		return nil, ctx.Err(), false
+	}
+}
+
 // Run subscribes via SubscribeMetadata starting at the configured position,
 // filters to the configured shard set, and emits Events. Returns on
 // ctx.Done(), io.EOF, or stream error. Caller is responsible for closing
@@ -102,6 +142,9 @@ func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, cl
 	if r.BucketsPath == "" {
 		return errors.New("reader: empty BucketsPath")
 	}
+	if r.ReceiveTimeout < 0 {
+		return fmt.Errorf("reader: negative ReceiveTimeout %v", r.ReceiveTimeout)
+	}
 	r.bucketsPathSlash = r.BucketsPath
 	if !strings.HasSuffix(r.bucketsPathSlash, "/") {
 		r.bucketsPathSlash += "/"
@@ -111,20 +154,47 @@ func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, cl
 	if sinceNs == 0 && r.Cursor != nil {
 		sinceNs = r.Cursor.MinTsNs()
 	}
-	stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
+	// Own context: aborting the RPC is the only way to unblock Recv.
+	streamCtx, cancelStream := context.WithCancel(ctx)
+	defer cancelStream()
+	stream, err := client.SubscribeMetadata(streamCtx, &filer_pb.SubscribeMetadataRequest{
 		ClientName:             clientName,
 		PathPrefix:             r.BucketsPath,
 		SinceNs:                sinceNs,
+		UntilNs:                r.UntilTsNs,
 		ClientId:               clientID,
 		ClientSupportsBatching: true,
+		// dispatchOne drops these, but arriving at all is the point.
+		ClientSupportsIdleHeartbeat: r.ReceiveTimeout > 0,
 	})
 	if err != nil {
 		return fmt.Errorf("subscribe: %w", err)
 	}
 
+	// Buffered so a response landing as the watchdog fires doesn't strand
+	// this goroutine.
+	received := make(chan receiveResult, 1)
+	go func() {
+		for {
+			resp, recvErr := stream.Recv()
+			select {
+			case received <- receiveResult{resp: resp, err: recvErr}:
+			case <-streamCtx.Done():
+				return
+			}
+			if recvErr != nil {
+				return
+			}
+		}
+	}()
+
 	processed := 0
 	for {
-		resp, recvErr := stream.Recv()
+		resp, recvErr, timedOut := r.awaitResponse(streamCtx, received)
+		if timedOut {
+			cancelStream() // unwind the goroutine blocked in Recv
+			return fmt.Errorf("%w after %s", ErrReceiveTimeout, r.ReceiveTimeout)
+		}
 		if recvErr == io.EOF {
 			return nil
 		}

weed/s3api/s3lifecycle/reader/receive_timeout_test.go

@@ -0,0 +1,117 @@
+package reader
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+// stalledStream stays open and never delivers: keepalives answered, no
+// progress.
+type stalledStream struct {
+	grpc.ClientStream
+	ctx context.Context
+}
+
+func (s *stalledStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	<-s.ctx.Done()
+	return nil, s.ctx.Err()
+}
+
+type stalledClient struct {
+	filer_pb.SeaweedFilerClient
+	req *filer_pb.SubscribeMetadataRequest
+}
+
+func (c *stalledClient) SubscribeMetadata(ctx context.Context, req *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_SubscribeMetadataClient, error) {
+	c.req = req
+	return &stalledStream{ctx: ctx}, nil
+}
+
+func TestRun_ReceiveTimeoutOnStalledStream(t *testing.T) {
+	client := &stalledClient{}
+	r := &Reader{
+		ShardPredicate: func(int) bool { return true },
+		BucketsPath:    "/buckets",
+		Events:         make(chan *Event, 1),
+		ReceiveTimeout: 50 * time.Millisecond,
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- r.Run(context.Background(), client, "test", 1) }()
+	select {
+	case err := <-done:
+		require.ErrorIs(t, err, ErrReceiveTimeout)
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run did not time out on a stalled stream")
+	}
+	assert.True(t, client.req.ClientSupportsIdleHeartbeat,
+		"a reader with a receive timeout must ask for heartbeats, or a caught-up stream looks stalled")
+}
+
+// heartbeatStream delivers only heartbeats: a ts, no EventNotification.
+type heartbeatStream struct {
+	grpc.ClientStream
+	every time.Duration
+	sent  int
+	max   int
+}
+
+func (s *heartbeatStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	if s.sent >= s.max {
+		return nil, context.Canceled
+	}
+	time.Sleep(s.every)
+	s.sent++
+	return &filer_pb.SubscribeMetadataResponse{TsNs: int64(s.sent)}, nil
+}
+
+type heartbeatClient struct {
+	filer_pb.SeaweedFilerClient
+	stream *heartbeatStream
+}
+
+func (c *heartbeatClient) SubscribeMetadata(_ context.Context, _ *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_SubscribeMetadataClient, error) {
+	return c.stream, nil
+}
+
+// Why the timeout is safe to set: a caught-up stream keeps arriving, so
+// the watchdog fires on real silence, not on an idle cluster.
+func TestRun_HeartbeatsHoldTheStreamOpen(t *testing.T) {
+	stream := &heartbeatStream{every: 20 * time.Millisecond, max: 10}
+	r := &Reader{
+		ShardPredicate: func(int) bool { return true },
+		BucketsPath:    "/buckets",
+		Events:         make(chan *Event, 1),
+		ReceiveTimeout: 200 * time.Millisecond,
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- r.Run(context.Background(), &heartbeatClient{stream: stream}, "test", 1) }()
+	select {
+	case err := <-done:
+		assert.False(t, errors.Is(err, ErrReceiveTimeout),
+			"heartbeats spanning more than one timeout window must keep the stream alive")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run never returned")
+	}
+	assert.Equal(t, 10, stream.sent, "every heartbeat should have been consumed")
+}
+
+func TestRun_RejectsNegativeReceiveTimeout(t *testing.T) {
+	r := &Reader{
+		ShardPredicate: func(int) bool { return true },
+		BucketsPath:    "/buckets",
+		Events:         make(chan *Event, 1),
+		ReceiveTimeout: -time.Second,
+	}
+	err := r.Run(context.Background(), &stalledClient{}, "test", 1)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "ReceiveTimeout")
+}