diff --git a/weed/s3api/s3lifecycle/scheduler/bootstrap.go b/weed/s3api/s3lifecycle/scheduler/bootstrap.go index ce17f64d4..af85c91ad 100644 --- a/weed/s3api/s3lifecycle/scheduler/bootstrap.go +++ b/weed/s3api/s3lifecycle/scheduler/bootstrap.go @@ -78,33 +78,70 @@ func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string // Synthesized events carry TsNs=0 so dispatcher.advance is a no-op for // them — the reader still resumes from its persisted cursor on restart. // -// Bucket completion is in-memory per process; a fresh worker run walks -// each bucket once on first refresh. +// Walk state is tracked in-memory per process via two maps: +// - lastCompleted: time of the last successful walk, drives the +// BootstrapInterval cadence; zero interval means "walk once per +// process". +// - inFlight: buckets whose walk goroutines are still running. +// Skipped regardless of cadence so a walk that takes longer than +// BootstrapInterval can't trigger a duplicate goroutine on the +// next refresh. type BucketBootstrapper struct { FilerClient filer_pb.SeaweedFilerClient BucketsPath string Injector EventInjector - mu sync.Mutex - known map[string]bool + // BootstrapInterval gates re-walks. Zero means "walk once per + // process". Non-zero means "walk again once it's been at least + // this long since the last completed walk" — the cadence scan_only + // actions rely on, since they can only fire from bootstrap. + BootstrapInterval time.Duration + + // Now overrides time.Now for tests. + Now func() time.Time + + mu sync.Mutex + lastCompleted map[string]time.Time + inFlight map[string]bool +} + +func (b *BucketBootstrapper) now() time.Time { + if b.Now != nil { + return b.Now() + } + return time.Now() } // KickOffNew launches a one-shot walker goroutine for every bucket -// in `buckets` that hasn't been seen before. +// that's not currently in flight and either has never completed a walk +// or whose last successful walk finished more than BootstrapInterval ago. func (b *BucketBootstrapper) KickOffNew(ctx context.Context, buckets []string) { if b.Injector == nil { return } + now := b.now() b.mu.Lock() - if b.known == nil { - b.known = map[string]bool{} + if b.lastCompleted == nil { + b.lastCompleted = map[string]time.Time{} + } + if b.inFlight == nil { + b.inFlight = map[string]bool{} } fresh := make([]string, 0, len(buckets)) for _, bucket := range buckets { - if b.known[bucket] { + if b.inFlight[bucket] { + // Walk still running from an earlier KickOffNew — never + // double up regardless of cadence. A large bucket that + // takes longer than BootstrapInterval would otherwise + // have a fresh goroutine fire on every refresh tick. continue } - b.known[bucket] = true + if last, ok := b.lastCompleted[bucket]; ok { + if b.BootstrapInterval <= 0 || now.Sub(last) < b.BootstrapInterval { + continue + } + } + b.inFlight[bucket] = true fresh = append(fresh, bucket) } b.mu.Unlock() @@ -152,9 +189,22 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) { count++ return b.Injector.InjectEvent(ctx, ev) } - if err := walkBucketDir(ctx, b.FilerClient, root, root, cb); err != nil { + walkErr := walkBucketDir(ctx, b.FilerClient, root, root, cb) + b.mu.Lock() + delete(b.inFlight, bucket) + if walkErr == nil { + // Stamp completion so BootstrapInterval cadence measures from + // end-of-walk. Failures leave lastCompleted alone, so the next + // KickOffNew sees no record and walks the bucket again. + if b.lastCompleted == nil { + b.lastCompleted = map[string]time.Time{} + } + b.lastCompleted[bucket] = b.now() + } + b.mu.Unlock() + if walkErr != nil { if ctx.Err() == nil { - glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, err) + glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, walkErr) } return } diff --git a/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go index 2f4b1b450..620398f90 100644 --- a/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go +++ b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go @@ -396,9 +396,11 @@ func TestBucketBootstrapper_KickOffNew_LaunchesPerBucket(t *testing.T) { b.mu.Lock() defer b.mu.Unlock() - assert.True(t, b.known["bucketA"]) - assert.True(t, b.known["bucketB"]) - assert.Len(t, b.known, 2) + _, hasA := b.lastCompleted["bucketA"] + _, hasB := b.lastCompleted["bucketB"] + assert.True(t, hasA) + assert.True(t, hasB) + assert.Len(t, b.lastCompleted, 2) } func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) { @@ -442,8 +444,9 @@ func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) { b.mu.Lock() defer b.mu.Unlock() - assert.Len(t, b.known, 3) - assert.True(t, b.known["bucketC"]) + assert.Len(t, b.lastCompleted, 3) + _, hasC := b.lastCompleted["bucketC"] + assert.True(t, hasC) } func TestBucketBootstrapper_KickOffNew_NilInjectorIsNoop(t *testing.T) { @@ -456,12 +459,13 @@ func TestBucketBootstrapper_KickOffNew_NilInjectorIsNoop(t *testing.T) { require.NotPanics(t, func() { b.KickOffNew(context.Background(), []string{"bucketA"}) }) - // No walks must have been kicked off, and known must remain empty. + // No walks must have been kicked off, and the lastWalk map must + // remain empty. time.Sleep(20 * time.Millisecond) assert.Equal(t, int32(0), atomic.LoadInt32(&client.listedN)) b.mu.Lock() defer b.mu.Unlock() - assert.Empty(t, b.known) + assert.Empty(t, b.lastCompleted) } func TestBucketBootstrapper_KickOffNew_EmptyBucketListIsNoop(t *testing.T) { @@ -1100,3 +1104,193 @@ func TestWalkBucketDir_PaginatesBeyondListingLimit(t *testing.T) { // entries at page 2 = 3 paginated calls per pass = 6 total. assert.Equal(t, 6, calls) } + +// fakeClock is a thread-safe time source for tests that need to fast- +// forward across a BootstrapInterval boundary. Bootstrap goroutines +// read it concurrently with the test advancing it, so a plain +// `clock := time.Now()` plus closure write would race under -race. +type fakeClock struct { + mu sync.Mutex + t time.Time +} + +func newFakeClock() *fakeClock { return &fakeClock{t: time.Now()} } + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + c.t = c.t.Add(d) + c.mu.Unlock() +} + +// waitForCompleted blocks until the bootstrapper has stamped a +// lastCompleted entry for the given bucket. Polling listedN is not +// enough — that fires once both list passes have started, but +// walkBucket stamps lastCompleted only after walkBucketDir returns, +// so a clock.Advance between those events would record the stamp at +// post-advance time and skew BootstrapInterval cadence assertions. +func waitForCompleted(t *testing.T, b *BucketBootstrapper, bucket string) { + t.Helper() + waitFor(t, func() bool { + b.mu.Lock() + _, ok := b.lastCompleted[bucket] + b.mu.Unlock() + return ok + }, "lastCompleted stamp for "+bucket) +} + +func TestBucketBootstrapper_KickOffNew_BootstrapIntervalRevisitsBucket(t *testing.T) { + // scan_only actions only fire from bootstrap, so a long-running + // worker has to revisit each bucket on a cadence. With + // BootstrapInterval set, KickOffNew re-walks once enough wall-clock + // has passed since the last completed walk. + client := newEmptyFilerClient() + inj := &recordingInjector{} + clock := newFakeClock() + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + BootstrapInterval: time.Hour, + Now: clock.Now, + } + + // First wave: walks once. Wait for the goroutine to actually stamp + // lastCompleted before advancing the clock — otherwise the stamp + // could land at clock+30m instead of T0 and the cadence assertion + // would race. + b.KickOffNew(context.Background(), []string{"bucketA"}) + waitForCompleted(t, b, "bucketA") + firstCount := atomic.LoadInt32(&client.listedN) + + // Inside the interval: skip. + clock.Advance(30 * time.Minute) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := atomic.LoadInt32(&client.listedN); got != firstCount { + t.Fatalf("inside interval, must not re-walk; listedN=%d, want %d", got, firstCount) + } + + // Past the interval: re-walk. + clock.Advance(45 * time.Minute) // total elapsed > 1h + b.KickOffNew(context.Background(), []string{"bucketA"}) + waitFor(t, func() bool { return atomic.LoadInt32(&client.listedN) >= firstCount+2 }, "re-walk after interval") +} + +func TestBucketBootstrapper_KickOffNew_ZeroIntervalLegacyOnceOnly(t *testing.T) { + // BootstrapInterval == 0 preserves the original "walk once per + // process" behavior so existing deployments don't get a different + // cadence by default. + client := newEmptyFilerClient() + inj := &recordingInjector{} + clock := newFakeClock() + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + Now: clock.Now, + } + + b.KickOffNew(context.Background(), []string{"bucketA"}) + waitForCompleted(t, b, "bucketA") + firstCount := atomic.LoadInt32(&client.listedN) + + // Even after 100 hours, KickOffNew skips the bucket. + clock.Advance(100 * time.Hour) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := atomic.LoadInt32(&client.listedN); got != firstCount { + t.Fatalf("zero interval must keep the once-per-process behavior; listedN=%d, want %d", got, firstCount) + } +} + +// blockingInjector lets the test pin a walk in progress until it +// signals release. Useful for asserting in-flight debounce. +type blockingInjector struct { + mu sync.Mutex + events []*reader.Event + released chan struct{} +} + +func newBlockingInjector() *blockingInjector { + return &blockingInjector{released: make(chan struct{})} +} + +func (b *blockingInjector) InjectEvent(ctx context.Context, ev *reader.Event) error { + select { + case <-b.released: + case <-ctx.Done(): + return ctx.Err() + } + b.mu.Lock() + b.events = append(b.events, ev) + b.mu.Unlock() + return nil +} + +func (b *blockingInjector) release() { close(b.released) } + +func (b *blockingInjector) eventCount() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.events) +} + +func TestBucketBootstrapper_KickOffNew_InFlightDebounceBlocksDuplicate(t *testing.T) { + // A walk that takes longer than BootstrapInterval would otherwise + // have a fresh KickOffNew start a duplicate goroutine on the next + // refresh tick. The inFlight set prevents that. Verify by: + // 1) Pinning a walk in progress via a blockingInjector, + // 2) Advancing the clock past BootstrapInterval, + // 3) Confirming the second KickOffNew is a no-op while the first + // is still running, + // 4) Releasing the first walk and asserting only one walk + // completed (one bucket-root listing pair). + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + "/buckets/bucketA": {fileEntry("a.txt")}, + }, + } + inj := newBlockingInjector() + clock := newFakeClock() + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + BootstrapInterval: time.Hour, + Now: clock.Now, + } + + b.KickOffNew(context.Background(), []string{"bucketA"}) + // Wait for the walk goroutine to actually start listing (pass 1 + // fires a ListEntries before InjectEvent). + waitFor(t, func() bool { return atomic.LoadInt32(&client.listedN) >= 1 }, "first walk to begin listing") + + // Advance past the interval — a stale-state KickOffNew would now + // see the lastCompleted as expired and try again. + clock.Advance(2 * time.Hour) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := inj.eventCount(); got != 0 { + t.Fatalf("first walk still blocked, got %d injected events from a phantom second walk", got) + } + + // Release: the first walk completes. eventCount goes to 1. + inj.release() + waitFor(t, func() bool { return inj.eventCount() == 1 }, "first walk to drain") + // Even after another KickOffNew at the same simulated time, the + // in-flight is now cleared and lastCompleted is fresh — second + // KickOffNew within interval is a no-op. + prevListed := atomic.LoadInt32(&client.listedN) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := atomic.LoadInt32(&client.listedN); got != prevListed { + t.Fatalf("post-release KickOffNew within interval must be a no-op; listedN=%d, want %d", got, prevListed) + } +} + diff --git a/weed/s3api/s3lifecycle/scheduler/scheduler.go b/weed/s3api/s3lifecycle/scheduler/scheduler.go index c4571b49b..fc697cc94 100644 --- a/weed/s3api/s3lifecycle/scheduler/scheduler.go +++ b/weed/s3api/s3lifecycle/scheduler/scheduler.go @@ -33,11 +33,12 @@ type Scheduler struct { ClientID int32 ClientName string - Workers int - DispatchTick time.Duration - CheckpointTick time.Duration - RefreshInterval time.Duration - RetryBackoff time.Duration + Workers int + DispatchTick time.Duration + CheckpointTick time.Duration + RefreshInterval time.Duration + BootstrapInterval time.Duration + RetryBackoff time.Duration } // Run blocks until ctx is canceled. Spawns Workers + 1 goroutines: one @@ -100,9 +101,10 @@ func (s *Scheduler) Run(ctx context.Context) error { } bs := &BucketBootstrapper{ - FilerClient: s.FilerClient, - BucketsPath: s.BucketsPath, - Injector: pipelineFanout(pipelinesByShard), + FilerClient: s.FilerClient, + BucketsPath: s.BucketsPath, + Injector: pipelineFanout(pipelinesByShard), + BootstrapInterval: s.BootstrapInterval, } s.refreshEngine(ctx, bs) diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go index ace7f489f..1f84b6b15 100644 --- a/weed/shell/command_s3_lifecycle_run_shard.go +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -68,6 +68,7 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i dispatchTick := fs.Duration("dispatch", 5*time.Second, "dispatcher tick cadence") checkpointTick := fs.Duration("checkpoint", 30*time.Second, "cursor checkpoint cadence") refreshInterval := fs.Duration("refresh", 5*time.Minute, "interval for rebuilding the engine snapshot from filer-backed bucket configs; 0 = compile once at startup") + bootstrapInterval := fs.Duration("bootstrap-interval", 0, "cadence for revisiting each bucket's bootstrap walk; 0 = walk once per process. scan_only actions only fire from bootstrap, so a long-running worker needs a non-zero value to handle their retention horizon") runtime := fs.Duration("runtime", 0, "wall-clock cap on the run; 0 = no timeout. -events alone can hang on quiet shards") if err := fs.Parse(args); err != nil { return err @@ -119,9 +120,10 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i } bsr := &scheduler.BucketBootstrapper{ - FilerClient: filerClient, - BucketsPath: bucketsPath, - Injector: pipeline, + FilerClient: filerClient, + BucketsPath: bucketsPath, + Injector: pipeline, + BootstrapInterval: *bootstrapInterval, } bootstrapCtx, bootstrapCancel := context.WithCancel(context.Background()) defer bootstrapCancel() diff --git a/weed/worker/tasks/s3_lifecycle/config.go b/weed/worker/tasks/s3_lifecycle/config.go index 07d8288f4..f39b8b06e 100644 --- a/weed/worker/tasks/s3_lifecycle/config.go +++ b/weed/worker/tasks/s3_lifecycle/config.go @@ -9,31 +9,34 @@ import ( const ( jobType = "s3_lifecycle" - defaultWorkers = 1 - defaultDispatchTickMinutes = int64(1) - defaultCheckpointTickSeconds = int64(30) - defaultRefreshIntervalMinutes = int64(5) - defaultMaxRuntimeMinutes = int64(60) + defaultWorkers = 1 + defaultDispatchTickMinutes = int64(1) + defaultCheckpointTickSeconds = int64(30) + defaultRefreshIntervalMinutes = int64(5) + defaultMaxRuntimeMinutes = int64(60) + defaultBootstrapIntervalMinutes = int64(0) // 0 = walk once per process ) // Config is the parsed AdminConfigForm + WorkerConfigForm view. type Config struct { - Workers int - DispatchTick time.Duration - CheckpointTick time.Duration - RefreshInterval time.Duration - MaxRuntime time.Duration + Workers int + DispatchTick time.Duration + CheckpointTick time.Duration + RefreshInterval time.Duration + BootstrapInterval time.Duration + MaxRuntime time.Duration } // ParseConfig pulls the lifecycle Handler config from the merged // admin+worker config values. Missing fields fall back to defaults. func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Config { cfg := Config{ - Workers: int(readInt64(adminValues, "workers", defaultWorkers)), - DispatchTick: time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute, - CheckpointTick: time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second, - RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute, - MaxRuntime: time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute, + Workers: int(readInt64(adminValues, "workers", defaultWorkers)), + DispatchTick: time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute, + CheckpointTick: time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second, + RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute, + BootstrapInterval: time.Duration(readInt64(workerValues, "bootstrap_interval_minutes", defaultBootstrapIntervalMinutes)) * time.Minute, + MaxRuntime: time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute, } if cfg.Workers <= 0 { cfg.Workers = defaultWorkers @@ -47,6 +50,13 @@ func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Co if cfg.RefreshInterval <= 0 { cfg.RefreshInterval = time.Duration(defaultRefreshIntervalMinutes) * time.Minute } + // BootstrapInterval is intentionally NOT clamped — zero means + // "walk once per process", which is the legacy default for any + // deployment that hasn't opted into a cadence yet. Negative values + // fall back to zero. + if cfg.BootstrapInterval < 0 { + cfg.BootstrapInterval = 0 + } if cfg.MaxRuntime <= 0 { cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute } diff --git a/weed/worker/tasks/s3_lifecycle/config_test.go b/weed/worker/tasks/s3_lifecycle/config_test.go index d2afae189..840648dbd 100644 --- a/weed/worker/tasks/s3_lifecycle/config_test.go +++ b/weed/worker/tasks/s3_lifecycle/config_test.go @@ -24,6 +24,9 @@ func TestParseConfigDefaults(t *testing.T) { if cfg.MaxRuntime != 60*time.Minute { t.Errorf("MaxRuntime default=%v, want 60m", cfg.MaxRuntime) } + if cfg.BootstrapInterval != 0 { + t.Errorf("BootstrapInterval default=%v, want 0 (walk-once-per-process)", cfg.BootstrapInterval) + } } func TestParseConfigOverrides(t *testing.T) { @@ -31,10 +34,11 @@ func TestParseConfigOverrides(t *testing.T) { "workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 4}}, } worker := map[string]*plugin_pb.ConfigValue{ - "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}}, - "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}}, - "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}}, - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}}, + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, } cfg := ParseConfig(admin, worker) if cfg.Workers != 4 { @@ -49,6 +53,9 @@ func TestParseConfigOverrides(t *testing.T) { if cfg.RefreshInterval != 10*time.Minute { t.Errorf("RefreshInterval=%v, want 10m", cfg.RefreshInterval) } + if cfg.BootstrapInterval != 120*time.Minute { + t.Errorf("BootstrapInterval=%v, want 120m", cfg.BootstrapInterval) + } if cfg.MaxRuntime != 120*time.Minute { t.Errorf("MaxRuntime=%v, want 120m", cfg.MaxRuntime) } @@ -56,10 +63,11 @@ func TestParseConfigOverrides(t *testing.T) { func TestParseConfigClampsZeroAndNegative(t *testing.T) { worker := map[string]*plugin_pb.ConfigValue{ - "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, } cfg := ParseConfig(nil, worker) if cfg.DispatchTick != 1*time.Minute { @@ -71,7 +79,23 @@ func TestParseConfigClampsZeroAndNegative(t *testing.T) { if cfg.RefreshInterval != 5*time.Minute { t.Errorf("zero RefreshInterval should clamp to default, got %v", cfg.RefreshInterval) } + if cfg.BootstrapInterval != 0 { + t.Errorf("negative BootstrapInterval should clamp to 0, got %v", cfg.BootstrapInterval) + } if cfg.MaxRuntime != 60*time.Minute { t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime) } } + +func TestParseConfigBootstrapIntervalZeroIsLegacy(t *testing.T) { + // Explicit zero stays zero (walk-once-per-process). Existing + // deployments that don't set bootstrap_interval_minutes get the + // legacy behavior unchanged. + worker := map[string]*plugin_pb.ConfigValue{ + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + } + cfg := ParseConfig(nil, worker) + if cfg.BootstrapInterval != 0 { + t.Errorf("zero BootstrapInterval must stay zero, got %v", cfg.BootstrapInterval) + } +} diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 247616b77..d86f32f62 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -122,6 +122,14 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}}, }, + { + Name: "bootstrap_interval_minutes", + Label: "Bootstrap Re-walk Interval (minutes)", + Description: "How often each bucket is re-walked. scan_only rules — those whose retention horizon exceeds meta-log retention — only fire from the bootstrap walk, so a non-zero value is required to enforce them on a long-running worker. 0 keeps the legacy walk-once-per-process behavior.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + }, { Name: "max_runtime_minutes", Label: "Max Runtime (minutes)", @@ -134,10 +142,11 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { }, }, DefaultValues: map[string]*plugin_pb.ConfigValue{ - "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}}, - "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}}, - "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}}, - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}}, + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}}, + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultBootstrapIntervalMinutes}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}}, }, }, AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{ @@ -240,17 +249,18 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn) sched := &scheduler.Scheduler{ - BucketsPath: bucketsPath, - Engine: engine.New(), - Persister: &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, - Client: lifecycleRPCAdapter{c: rpc}, - FilerClient: filerClient, - ClientID: util.RandomInt32(), - ClientName: "worker-s3-lifecycle", - Workers: cfg.Workers, - DispatchTick: cfg.DispatchTick, - CheckpointTick: cfg.CheckpointTick, - RefreshInterval: cfg.RefreshInterval, + BucketsPath: bucketsPath, + Engine: engine.New(), + Persister: &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, + Client: lifecycleRPCAdapter{c: rpc}, + FilerClient: filerClient, + ClientID: util.RandomInt32(), + ClientName: "worker-s3-lifecycle", + Workers: cfg.Workers, + DispatchTick: cfg.DispatchTick, + CheckpointTick: cfg.CheckpointTick, + RefreshInterval: cfg.RefreshInterval, + BootstrapInterval: cfg.BootstrapInterval, } if err := sched.Run(runCtx); err != nil { glog.Warningf("s3 lifecycle execute: %v", err)