diff --git a/test/s3/lifecycle/s3_lifecycle_admin_dispatch_test.go b/test/s3/lifecycle/s3_lifecycle_admin_dispatch_test.go index 58c521f87..857f812de 100644 --- a/test/s3/lifecycle/s3_lifecycle_admin_dispatch_test.go +++ b/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 { diff --git a/weed/s3api/s3lifecycle/DESIGN.md b/weed/s3api/s3lifecycle/DESIGN.md index 6df726481..c0cd9e434 100644 --- a/weed/s3api/s3lifecycle/DESIGN.md +++ b/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. diff --git a/weed/s3api/s3lifecycle/dailyrun/process_matches_test.go b/weed/s3api/s3lifecycle/dailyrun/process_matches_test.go index c61330f96..82dd662de 100644 --- a/weed/s3api/s3lifecycle/dailyrun/process_matches_test.go +++ b/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] diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 0c845cacc..f459b1bdd 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/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 } diff --git a/weed/s3api/s3lifecycle/dailyrun/run_bounded_test.go b/weed/s3api/s3lifecycle/dailyrun/run_bounded_test.go new file mode 100644 index 000000000..cbf4b1560 --- /dev/null +++ b/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") + } +} diff --git a/weed/s3api/s3lifecycle/dailyrun/run_fanout_test.go b/weed/s3api/s3lifecycle/dailyrun/run_fanout_test.go new file mode 100644 index 000000000..59c849654 --- /dev/null +++ b/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") +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go index 21e599598..1dbe068f1 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go +++ b/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 } diff --git a/weed/s3api/s3lifecycle/reader/reader.go b/weed/s3api/s3lifecycle/reader/reader.go index a82015d96..c88badd78 100644 --- a/weed/s3api/s3lifecycle/reader/reader.go +++ b/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 } 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..544131ce1 --- /dev/null +++ b/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") +}