From 55aa4174ff925083dd12df27c5798b0ee69322bd Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 4 Aug 2026 22:56:06 -0700 Subject: [PATCH] s3_lifecycle: bound shared metadata subscriptions --- weed/s3api/s3lifecycle/dailyrun/run.go | 38 +++++- .../s3lifecycle/dailyrun/subscription_test.go | 115 ++++++++++++++++++ weed/s3api/s3lifecycle/reader/reader.go | 70 ++++++++++- .../reader/receive_timeout_test.go | 51 ++++++++ 4 files changed, 268 insertions(+), 6 deletions(-) create mode 100644 weed/s3api/s3lifecycle/dailyrun/subscription_test.go create mode 100644 weed/s3api/s3lifecycle/reader/receive_timeout_test.go diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 0c845cacc..cf46976ce 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/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) diff --git a/weed/s3api/s3lifecycle/dailyrun/subscription_test.go b/weed/s3api/s3lifecycle/dailyrun/subscription_test.go new file mode 100644 index 000000000..8ce3b4732 --- /dev/null +++ b/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() +} diff --git a/weed/s3api/s3lifecycle/reader/reader.go b/weed/s3api/s3lifecycle/reader/reader.go index a82015d96..3c016db45 100644 --- a/weed/s3api/s3lifecycle/reader/reader.go +++ b/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 diff --git a/weed/s3api/s3lifecycle/reader/receive_timeout_test.go b/weed/s3api/s3lifecycle/reader/receive_timeout_test.go new file mode 100644 index 000000000..5bbc14ea0 --- /dev/null +++ b/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) +}