s3_lifecycle: bound shared metadata subscriptions

weed/s3api/s3lifecycle/dailyrun/run.go

@@ -88,8 +88,16 @@ type Config struct {
 
 	// 0 -> unbounded.
 	EventBudget int
+
+	// SubscriptionReceiveTimeout bounds the gap between responses on the
+	// shared metadata stream. 0 uses the production default. The filer sends
+	// idle heartbeats once the reader catches up, while the generous default
+	// still permits the filer's bounded metadata-gap recovery to run.
+	SubscriptionReceiveTimeout time.Duration
 }
 
+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'
@@ -167,11 +175,12 @@ 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 subscriptionErr 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)
+			subscriptionErr = fmt.Errorf("daily_run: shared subscription: %w", rerr)
 		}
 	}
 
@@ -185,6 +194,14 @@ func Run(ctx context.Context, cfg Config) error {
 			glog.V(1).Infof("daily_run: additional shard error: %v", err)
 		}
 	}
+	if subscriptionErr != nil {
+		errCount++
+		if first == nil {
+			first = subscriptionErr
+		} else {
+			glog.V(1).Infof("daily_run: additional subscription error: %v", subscriptionErr)
+		}
+	}
 	status := "ok"
 	if first != nil {
 		status = "error"
@@ -277,9 +294,9 @@ func computeGlobalStartTsNs(ctx context.Context, cfg Config, runNow time.Time, m
 // Subscription floor is the caller-supplied globalStartTsNs (typically
 // min over per-shard cursors). Shards whose own startTsNs is fresher
 // filter out already-past events themselves inside drainShardEvents.
-// 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.
+// The filer-side UntilNs bound normally ends the stream at runNow. Events
+// arriving with TsNs > runUpTo are retained as a defensive fallback: they
+// cause the fan-out to cancel the reader and close all per-shard channels.
 func startSharedSubscription(ctx context.Context, cfg Config, runNow time.Time, globalStartTsNs int64) (map[int]chan *reader.Event, chan error, chan struct{}, context.CancelFunc) {
 	shardSet := make(map[int]bool, len(cfg.Shards))
 	shardEvents := make(map[int]chan *reader.Event, len(cfg.Shards))
@@ -302,18 +319,28 @@ func startSharedSubscription(ctx context.Context, cfg Config, runNow time.Time,
 	}
 
 	events := make(chan *reader.Event, 4*len(cfg.Shards))
+	receiveTimeout := cfg.SubscriptionReceiveTimeout
+	if receiveTimeout == 0 {
+		receiveTimeout = defaultSubscriptionReceiveTimeout
+	}
 	rd := &reader.Reader{
 		ShardPredicate: func(id int) bool { return shardSet[id] },
 		BucketsPath:    cfg.BucketsPath,
 		StartTsNs:      globalStartTsNs,
+		UntilTsNs:      runNow.UnixNano(),
 		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)
+		// EOF, a transport error, or the receive watchdog must all release
+		// the fan-out so it closes every shard channel. Without this signal,
+		// runShard waits forever even though no reader can produce more events.
+		cancelReader()
 	}()
 
 	runUpTo := runNow.UnixNano()
@@ -389,6 +416,9 @@ func validate(cfg Config) error {
 	if cfg.WalkerInterval < 0 {
 		return fmt.Errorf("daily_run: negative WalkerInterval %v (0 = unthrottled, positive values throttle)", cfg.WalkerInterval)
 	}
+	if cfg.SubscriptionReceiveTimeout < 0 {
+		return fmt.Errorf("daily_run: negative SubscriptionReceiveTimeout %v", cfg.SubscriptionReceiveTimeout)
+	}
 	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)

weed/s3api/s3lifecycle/dailyrun/subscription_test.go

@@ -0,0 +1,115 @@
+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"
+)
+
+type eofSubscribeClient struct {
+	filer_pb.SeaweedFilerClient
+	request *filer_pb.SubscribeMetadataRequest
+}
+
+func (c *eofSubscribeClient) SubscribeMetadata(_ context.Context, req *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.SubscribeMetadataResponse], error) {
+	c.request = req
+	return &eofSubscribeStream{}, nil
+}
+
+type eofSubscribeStream struct {
+	grpc.ServerStreamingClient[filer_pb.SubscribeMetadataResponse]
+}
+
+func (*eofSubscribeStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	return nil, io.EOF
+}
+
+func TestSharedSubscriptionEOFStopsFanout(t *testing.T) {
+	client := &eofSubscribeClient{}
+	runNow := time.Unix(0, 987654321)
+	_, readerDone, fanoutDone, cancel := startSharedSubscription(context.Background(), Config{
+		Shards:                     []int{0},
+		BucketsPath:                "/buckets",
+		FilerClient:                client,
+		SubscriptionReceiveTimeout: time.Second,
+	}, runNow, 123)
+	defer cancel()
+
+	select {
+	case <-fanoutDone:
+	case <-time.After(time.Second):
+		t.Fatal("fan-out stayed blocked after the metadata reader reached EOF")
+	}
+	require.NoError(t, <-readerDone)
+	require.NotNil(t, client.request)
+	assert.Equal(t, runNow.UnixNano(), client.request.UntilNs)
+	assert.Equal(t, int64(123), client.request.SinceNs)
+	assert.True(t, client.request.ClientSupportsIdleHeartbeat)
+}
+
+func TestSharedSubscriptionReceiveTimeoutStopsFanout(t *testing.T) {
+	client := &blockingDailySubscribeClient{}
+	_, readerDone, fanoutDone, cancel := startSharedSubscription(context.Background(), Config{
+		Shards:                     []int{0},
+		BucketsPath:                "/buckets",
+		FilerClient:                client,
+		SubscriptionReceiveTimeout: 50 * time.Millisecond,
+	}, time.Now(), 123)
+	defer cancel()
+
+	select {
+	case <-fanoutDone:
+	case <-time.After(time.Second):
+		t.Fatal("fan-out stayed blocked after the metadata receive timeout")
+	}
+	require.ErrorIs(t, <-readerDone, reader.ErrReceiveTimeout)
+}
+
+func TestRunReturnsSharedSubscriptionReceiveTimeout(t *testing.T) {
+	eng := engine.New()
+	eng.Compile([]engine.CompileInput{{
+		Bucket: "bucket",
+		Rules: []*s3lifecycle.Rule{{
+			ID: "expire", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1,
+		}},
+	}}, engine.CompileOptions{})
+	for _, action := range eng.Snapshot().AllActions() {
+		eng.Snapshot().MarkActive(action.Key)
+	}
+
+	cfg := validatableConfig()
+	cfg.Shards = []int{0}
+	cfg.Engine = eng
+	cfg.FilerClient = &blockingDailySubscribeClient{}
+	cfg.SubscriptionReceiveTimeout = 50 * time.Millisecond
+
+	err := Run(context.Background(), cfg)
+	require.ErrorIs(t, err, reader.ErrReceiveTimeout)
+}
+
+type blockingDailySubscribeClient struct {
+	filer_pb.SeaweedFilerClient
+}
+
+func (*blockingDailySubscribeClient) SubscribeMetadata(ctx context.Context, _ *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.SubscribeMetadataResponse], error) {
+	return &blockingDailySubscribeStream{ctx: ctx}, nil
+}
+
+type blockingDailySubscribeStream struct {
+	grpc.ServerStreamingClient[filer_pb.SubscribeMetadataResponse]
+	ctx context.Context
+}
+
+func (s *blockingDailySubscribeStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	<-s.ctx.Done()
+	return nil, s.ctx.Err()
+}

weed/s3api/s3lifecycle/reader/reader.go

@@ -73,8 +73,17 @@ type Reader struct {
 	// then means "subscribe from the start of the meta-log".
 	Cursor    *Cursor
 	StartTsNs int64
+	// UntilTsNs bounds the subscription at an inclusive metadata timestamp.
+	// Zero leaves the stream unbounded.
+	UntilTsNs int64
 	Events    chan<- *Event
 
+	// ReceiveTimeout bounds the time spent waiting for the next stream
+	// response. The filer sends idle heartbeats to readers that opt in, so a
+	// caught-up but healthy stream remains active while a half-open stream
+	// eventually fails. Zero disables the timeout.
+	ReceiveTimeout time.Duration
+
 	// 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.
@@ -102,6 +111,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 +123,70 @@ 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{
+	streamCtx, cancelStream := context.WithCancel(ctx)
+	defer cancelStream()
+	stream, err := client.SubscribeMetadata(streamCtx, &filer_pb.SubscribeMetadataRequest{
 		ClientName:             clientName,
 		PathPrefix:             r.BucketsPath,
 		SinceNs:                sinceNs,
 		ClientId:               clientID,
+		UntilNs:                r.UntilTsNs,
 		ClientSupportsBatching: true,
+		// Heartbeats provide application-level proof that the response path is
+		// alive. They are consumed below like any other response but filtered
+		// out by dispatchOne because they carry no EventNotification.
+		ClientSupportsIdleHeartbeat: r.ReceiveTimeout > 0,
 	})
 	if err != nil {
 		return fmt.Errorf("subscribe: %w", err)
 	}
 
+	type receiveResult struct {
+		response *filer_pb.SubscribeMetadataResponse
+		err      error
+	}
+	receiveCh := make(chan receiveResult, 1)
+	go func() {
+		for {
+			resp, recvErr := stream.Recv()
+			select {
+			case receiveCh <- receiveResult{response: resp, err: recvErr}:
+			case <-streamCtx.Done():
+				return
+			}
+			if recvErr != nil {
+				return
+			}
+		}
+	}()
+
 	processed := 0
 	for {
-		resp, recvErr := stream.Recv()
+		var (
+			resp    *filer_pb.SubscribeMetadataResponse
+			recvErr error
+			timer   *time.Timer
+			timeout <-chan time.Time
+		)
+		if r.ReceiveTimeout > 0 {
+			timer = time.NewTimer(r.ReceiveTimeout)
+			timeout = timer.C
+		}
+		select {
+		case result := <-receiveCh:
+			if timer != nil {
+				timer.Stop()
+			}
+			resp, recvErr = result.response, result.err
+		case <-timeout:
+			cancelStream()
+			return fmt.Errorf("%w after %s", ErrReceiveTimeout, r.ReceiveTimeout)
+		case <-streamCtx.Done():
+			if timer != nil {
+				timer.Stop()
+			}
+			return streamCtx.Err()
+		}
 		if recvErr == io.EOF {
 			return nil
 		}
@@ -147,6 +209,10 @@ func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, cl
 	}
 }
 
+// ErrReceiveTimeout reports that an otherwise open subscription stopped
+// delivering both metadata events and negotiated idle heartbeats.
+var ErrReceiveTimeout = errors.New("reader: metadata receive timeout")
+
 func (r *Reader) dispatchOne(ctx context.Context, resp *filer_pb.SubscribeMetadataResponse, processed *int) error {
 	if resp == nil || resp.EventNotification == nil {
 		return nil

weed/s3api/s3lifecycle/reader/receive_timeout_test.go

@@ -0,0 +1,51 @@
+package reader
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+type blockingSubscribeClient struct {
+	filer_pb.SeaweedFilerClient
+	request *filer_pb.SubscribeMetadataRequest
+}
+
+func (c *blockingSubscribeClient) SubscribeMetadata(ctx context.Context, req *filer_pb.SubscribeMetadataRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.SubscribeMetadataResponse], error) {
+	c.request = req
+	return &blockingSubscribeStream{ctx: ctx}, nil
+}
+
+type blockingSubscribeStream struct {
+	grpc.ServerStreamingClient[filer_pb.SubscribeMetadataResponse]
+	ctx context.Context
+}
+
+func (s *blockingSubscribeStream) Recv() (*filer_pb.SubscribeMetadataResponse, error) {
+	<-s.ctx.Done()
+	return nil, s.ctx.Err()
+}
+
+func TestRunTimesOutHalfOpenSubscription(t *testing.T) {
+	client := &blockingSubscribeClient{}
+	r := &Reader{
+		ShardID:        0,
+		BucketsPath:    "/buckets",
+		UntilTsNs:      12345,
+		Events:         make(chan *Event),
+		ReceiveTimeout: 50 * time.Millisecond,
+	}
+
+	started := time.Now()
+	err := r.Run(context.Background(), client, "test-lifecycle", 7)
+	require.ErrorIs(t, err, ErrReceiveTimeout)
+	assert.Less(t, time.Since(started), time.Second)
+	require.NotNil(t, client.request)
+	assert.Equal(t, int64(12345), client.request.UntilNs)
+	assert.True(t, client.request.ClientSupportsIdleHeartbeat)
+}