mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-30 00:40:44 +01:00
fix(s3/lifecycle): bound each runPass ctx + refresh in runLifecycleShard
Two CI bugs surfaced after PR #9466 deleted the streaming worker: 1. The shell command's -refresh loop never fires. runPass used the outer ctx (full -runtime), so dailyrun.Run blocked for the entire 1800s s3tests window — the background worker only ran one pass and never re-loaded configs that tests created mid-run. test_lifecycle_expiration sees 6 objects when expecting 4 because expire1/* never reaches the worker's snapshot. Cap each pass to cadence+5s when cadence>0; one-shot (cadence=0) keeps the full ctx. 2. TestLifecycleExpiredDeleteMarkerCleanup's docstring says "pass 1 cleans v1; pass 2 removes the now-orphaned marker," but runLifecycleShard invoked with no -refresh — only one pass ran. The marker rule can't fire in the same pass that dispatches v1's delete because v1 is still in .versions/. Add -refresh 1s so the 10s runtime gets multiple passes.
This commit is contained in:
1 parent
de9dcf0b84
commit
13c731ee6c
2 files changed
+21
-3
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user