From 37e505b8fddee557c4279d97238b0c008e9151ed Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 13 May 2026 02:13:11 -0700 Subject: [PATCH] refactor(s3/lifecycle): one meta-log subscription per dailyrun.Run pass (#9481) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(s3/lifecycle): one meta-log subscription per dailyrun.Run pass Per-shard Reader subscriptions multiplied filer load by len(cfg.Shards) even though the same gRPC stream could serve every shard in a worker process. Replace with one SubscribeMetadata stream covering all shards in cfg.Shards: the Reader's ShardPredicate accepts the shard set, and a fan-out goroutine routes events to per-shard channels by ev.ShardID. drainShardEvents now reads from a passed-in channel; shards whose persisted cursor is fresher than the global floor (runNow - maxTTL) filter ev.TsNs <= startTsNs locally. The fan-out cancels the reader when the first ev.TsNs > runNow arrives — meta-log order means the rest of the stream is past the pass boundary too. cfg.Workers no longer gates shard concurrency: with the shared subscription, every shard goroutine must be live to drain its channel, or the fan-out stalls. The field is retained for back-compat and ignored. Dispatch throttling still goes through cfg.Limiter. Filer load: 16x -> 1x SubscribeMetadata streams per pass. * fix(s3/lifecycle): shared subscription floor is min(per-shard cursor) The shared subscription used runNow - maxTTL as its starting TsNs, but that's the cold-start floor. For shards whose persisted cursor sits below the floor — exactly the case a rule with TTL == maxTTL produces, where a pending event's PUT TsNs ends up at runNow - maxTTL — events that the per-shard drain still needs are filtered out before the Reader even forwards them. Same regression I fixed in 6796ab6db for the per-shard subscription; now applied at the shared level. computeGlobalStartTsNs loads every shard's cursor and picks the minimum, falling back to the cold-start floor only for shards with no persisted cursor. --- weed/s3api/s3lifecycle/dailyrun/run.go | 246 +++++++++++++----- .../dailyrun/walker_recovery_test.go | 6 +- 2 files changed, 190 insertions(+), 62 deletions(-) diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 82fed0624..624ac547d 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -77,6 +77,12 @@ type Config struct { // concurrently. Returns the first shard error; the rest log and run to // completion so one shard's transient failure doesn't lose other shards' // progress. +// +// All cfg.Shards share one meta-log subscription. The Reader's +// ShardPredicate accepts any shard in cfg.Shards, and a fan-out +// goroutine routes events to per-shard channels by ev.ShardID. This +// replaces the earlier per-shard Reader that opened 16 SubscribeMetadata +// streams to filer per pass. func Run(ctx context.Context, cfg Config) error { if err := validate(cfg); err != nil { return err @@ -91,32 +97,67 @@ func Run(ctx context.Context, cfg Config) error { // Capture once so a mid-run Compile can't make shards disagree. snap := cfg.Engine.Snapshot() + rsh := engine.ReplayContentHash(snap) + maxTTL := engine.MaxEffectiveTTL(snap) - workers := cfg.Workers - if workers <= 0 { - workers = 1 + // Since all shards share one subscription, every shard goroutine + // must be live to drain its channel — capping concurrency would + // stall the fan-out. cfg.Workers no longer gates shard concurrency + // (dispatch is throttled via cfg.Limiter); accepted for backwards + // compatibility but otherwise inert. + _ = cfg.Workers + + // rsh==[32]byte{} means no replay-eligible rules — runShard's + // walker-only branch fires and saves; no subscription needed. + var ( + shardEvents map[int]chan *reader.Event + readerDone chan error + fanoutDone chan struct{} + cancelRead context.CancelFunc + globalStartTsNs int64 + ) + if rsh != [32]byte{} { + // Pre-load all per-shard cursors so the shared subscription + // can start from min(per-shard startTsNs). Using the cold-start + // floor (runNow - maxTTL) globally would skip past pending + // events on shards whose cursor is older than the floor — + // 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) + defer cancelRead() } - if workers > len(cfg.Shards) { - workers = len(cfg.Shards) - } - sem := make(chan struct{}, workers) var wg sync.WaitGroup errCh := make(chan error, len(cfg.Shards)) for _, sh := range cfg.Shards { sh := sh wg.Add(1) - sem <- struct{}{} go func() { defer wg.Done() - defer func() { <-sem }() - if err := runShard(ctx, cfg, snap, runNow, sh); err != nil { + var ch <-chan *reader.Event + if shardEvents != nil { + ch = shardEvents[sh] + } + if err := runShard(ctx, cfg, snap, runNow, sh, ch); err != nil { errCh <- err } }() } wg.Wait() close(errCh) + + // 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. + 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) + } + } + var first error errCount := 0 for err := range errCh { @@ -136,6 +177,123 @@ func Run(ctx context.Context, cfg Config) error { return first } +// computeGlobalStartTsNs scans every shard's persisted cursor and +// returns the minimum startTsNs the shared subscription must cover. +// For each shard the start point is the persisted cursor (steady state) +// or runNow - maxTTL (cold start, when no cursor exists). Load errors +// downgrade to the cold-start floor for that shard — failing closed +// would be worse than re-scanning maxTTL of events. +func computeGlobalStartTsNs(ctx context.Context, cfg Config, runNow time.Time, maxTTL time.Duration) int64 { + floor := runNow.Add(-maxTTL).UnixNano() + min := int64(0) + first := true + for _, sh := range cfg.Shards { + var start int64 + if persisted, found, err := cfg.Persister.Load(ctx, sh); err == nil && found { + start = persisted.TsNs + } else { + start = floor + } + if first || start < min { + min = start + first = false + } + } + if first { + // No shards (validate() should have caught this, but defensive). + return floor + } + return min +} + +// startSharedSubscription opens one SubscribeMetadata stream covering +// every shard in cfg.Shards and fans events out to per-shard channels. +// 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. +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)) + for _, sh := range cfg.Shards { + shardSet[sh] = true + // Per-shard channel size has to absorb event bursts between + // drains' iterations without backpressuring the fan-out. + // 256 covers a busy test bucket; production tuning can lift + // this further. + shardEvents[sh] = make(chan *reader.Event, 256) + } + + clientName := cfg.ClientName + if clientName == "" { + clientName = "worker-s3-lifecycle-daily" + } + clientID := cfg.ClientID + if clientID == 0 { + clientID = int32(util.RandomInt32()) + } + + 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, + Events: events, + EventBudget: cfg.EventBudget, + } + + readerCtx, cancelReader := context.WithCancel(ctx) + readerDone := make(chan error, 1) + go func() { + readerDone <- rd.Run(readerCtx, cfg.FilerClient, clientName, clientID) + }() + + runUpTo := runNow.UnixNano() + fanoutDone := make(chan struct{}) + go func() { + defer close(fanoutDone) + defer func() { + for _, ch := range shardEvents { + close(ch) + } + }() + for { + select { + case <-readerCtx.Done(): + return + case ev, ok := <-events: + if !ok { + return + } + if ev == nil { + continue + } + // Meta-log events arrive in TsNs order; the first + // event past runUpTo means everything after is past + // too. Cancel the reader so subsequent passes don't + // pay for stream tail we'd drop anyway. + if ev.TsNs > runUpTo { + cancelReader() + return + } + ch := shardEvents[ev.ShardID] + if ch == nil { + continue + } + select { + case <-readerCtx.Done(): + return + case ch <- ev: + } + } + } + }() + + return shardEvents, readerDone, fanoutDone, cancelReader +} + func validate(cfg Config) error { if cfg.Engine == nil { return errors.New("daily_run: nil Engine") @@ -170,7 +328,7 @@ func validate(cfg Config) error { // - steady state: RulesForShard's walk view, so walker-bound and // scan_only-promoted rules fire every day even when replay rules // are unchanged. -func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int) error { +func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int, events <-chan *reader.Event) error { shardLabel := strconv.Itoa(shardID) shardStart := time.Now() defer func() { @@ -285,7 +443,7 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim startTsNs = runNow.Add(-maxTTL).UnixNano() } - lastOK, _, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, snap, startTsNs) + 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 @@ -311,42 +469,18 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim }) } -// drainShardEvents subscribes to the meta-log from startTsNs and -// dispatches matches whose due_time is past runNow. cursorAdvanceTo -// stops growing at the first event with a not-yet-due match so that -// event is re-scanned in a later run. halted=true marks an unresolved -// dispatch outcome. -func drainShardEvents(ctx context.Context, cfg Config, runNow time.Time, shardID int, snap *engine.Snapshot, startTsNs int64) (int64, bool, error) { - clientName := cfg.ClientName - if clientName == "" { - clientName = "worker-s3-lifecycle-daily" - } - clientID := cfg.ClientID - if clientID == 0 { - clientID = int32(util.RandomInt32()) - } - runUpTo := runNow.UnixNano() - if startTsNs >= runUpTo { - return startTsNs, false, nil - } - - events := make(chan *reader.Event, 64) - rd := &reader.Reader{ - ShardID: shardID, - BucketsPath: cfg.BucketsPath, - StartTsNs: startTsNs, - Events: events, - EventBudget: cfg.EventBudget, - } - - readerCtx, cancelReader := context.WithCancel(ctx) - defer cancelReader() - - readerDone := make(chan error, 1) - go func() { - readerDone <- rd.Run(readerCtx, cfg.FilerClient, clientName, clientID) - }() - +// drainShardEvents reads pre-fanned-out events for this shard from the +// shared meta-log subscription and dispatches matches whose due_time is +// past runNow. cursorAdvanceTo stops growing at the first event with a +// not-yet-due match so that event is re-scanned in a later run. +// halted=true marks an unresolved dispatch outcome. +// +// The subscription itself lives at the Run() level — one filer stream +// fans out to per-shard channels via shardEventDispatcher. Each shard +// drains its own channel and advances its own cursor; future-TsNs +// events (relative to this pass's runNow) are filtered out by the +// dispatcher before they reach this channel. +func drainShardEvents(ctx context.Context, cfg Config, runNow time.Time, shardID int, snap *engine.Snapshot, startTsNs int64, events <-chan *reader.Event) (int64, bool, error) { cursorAdvanceTo := startTsNs stuck := false halted := false @@ -355,8 +489,6 @@ drain: for { select { case <-ctx.Done(): - cancelReader() - <-readerDone return cursorAdvanceTo, true, ctx.Err() case ev, ok := <-events: if !ok { @@ -365,16 +497,16 @@ drain: if ev == nil { continue } - if ev.TsNs > runUpTo { - cancelReader() - break drain + // The global subscription starts at min(per-shard startTsNs), + // so this shard may receive events that are already past its + // own cursor. Skip them rather than re-dispatching. + if ev.TsNs <= startTsNs { + continue } stats.S3LifecycleDailyRunEventsScanned.WithLabelValues(strconv.Itoa(shardID)).Inc() matches := router.Route(ctx, snap, ev, runNow, cfg.Lister) eventSkipped, eventHalted, eventErr := processMatches(ctx, cfg, runNow, ev, matches) if eventErr != nil { - cancelReader() - <-readerDone return cursorAdvanceTo, true, eventErr } if eventHalted { @@ -391,10 +523,6 @@ drain: } } - cancelReader() - if rerr := <-readerDone; rerr != nil && !errors.Is(rerr, context.Canceled) { - glog.V(2).Infof("daily_run shard=%d: reader returned: %v", shardID, rerr) - } return cursorAdvanceTo, halted, nil } diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go index aba8ed877..21e599598 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go @@ -71,7 +71,7 @@ func TestRunShard_WalkerInvokedOnRecoveryBranch(t *testing.T) { }, } runNow := time.Unix(1_700_000_000, 0).UTC() - require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 3)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 3, nil)) assert.Equal(t, 1, calls, "walker must fire exactly once on recovery") require.NotNil(t, gotView, "walker received the RecoveryView") @@ -99,7 +99,7 @@ func TestRunShard_NilWalkerOnRecoveryIsNoop(t *testing.T) { cfg := Config{Persister: p} // Walker nil runNow := time.Unix(1_700_000_000, 0).UTC() - require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, nil)) got, ok, err := p.Load(context.Background(), 0) require.NoError(t, err) @@ -126,7 +126,7 @@ func TestRunShard_WalkerErrorPropagates(t *testing.T) { }, } runNow := time.Unix(1_700_000_000, 0).UTC() - err := runShard(context.Background(), cfg, snap, runNow, 7) + err := runShard(context.Background(), cfg, snap, runNow, 7, nil) require.Error(t, err) assert.Contains(t, err.Error(), "walker boom")