diff --git a/test/s3/lifecycle/s3_lifecycle_versioning_test.go b/test/s3/lifecycle/s3_lifecycle_versioning_test.go index 39ea89e019..c21a37e04b 100644 --- a/test/s3/lifecycle/s3_lifecycle_versioning_test.go +++ b/test/s3/lifecycle/s3_lifecycle_versioning_test.go @@ -86,8 +86,12 @@ func backdateVersionedMtime(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket func runLifecycleShard(t *testing.T) string { t.Helper() + // -refresh 1s drives an inter-pass loop inside the 10s runtime so + // rules that need two passes to settle (e.g. + // TestLifecycleExpiredDeleteMarkerCleanup: pass 1 removes v1, pass 2 + // removes the now-orphaned marker) get the second pass. out := runShellCommand(t, fmt.Sprintf( - "s3.lifecycle.run-shard -shards 0-15 -s3 %s -events 0 -dispatch 200ms -checkpoint 5s -runtime 10s", + "s3.lifecycle.run-shard -shards 0-15 -s3 %s -events 0 -refresh 1s -runtime 10s", envOr("S3_GRPC_ENDPOINT", defaultS3GrpcEndpoint), )) require.NotContains(t, out, "FATAL", "shell output:\n%s", out) diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go index baf694a967..e53b56c53c 100644 --- a/weed/shell/command_s3_lifecycle_run_shard.go +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -124,8 +124,22 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i announcedLoad := false runPass := func() error { + // When -refresh is set we MUST cap each pass; otherwise + // drainShardEvents blocks until the outer runtime ctx + // expires and the loop's ticker never fires. With cadence=0 + // (one-shot) the pass uses the full ctx and returns when + // the runtime cap hits. + passCtx := ctx + if *cadence > 0 { + var passCancel context.CancelFunc + // Pass budget = cadence + grace. Grace gives the + // drain time to actually process events that arrived + // during the cadence window. + passCtx, passCancel = context.WithTimeout(ctx, *cadence+5*time.Second) + defer passCancel() + } eng := engine.New() - inputs, parseErrors, err := scheduler.LoadCompileInputs(ctx, filerClient, bucketsPath) + inputs, parseErrors, err := scheduler.LoadCompileInputs(passCtx, filerClient, bucketsPath) if err != nil { return fmt.Errorf("load lifecycle configs: %w", err) } @@ -154,7 +168,7 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shardID int) error { return dailyrun.WalkBuckets(walkCtx, view, shardID, buckets, listFn, walkerDispatch) }) - return dailyrun.Run(ctx, dailyrun.Config{ + return dailyrun.Run(passCtx, dailyrun.Config{ Shards: shards, BucketsPath: bucketsPath, Engine: eng,