From ad77362be35d46f92afa62821e0a4fa370b7db06 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 9 May 2026 22:02:09 -0700 Subject: [PATCH] test(s3/lifecycle): bundle reader + scheduler helper coverage (#9412) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(s3/lifecycle): bundle reader + scheduler helper coverage Bundles direct tests for previously-uncovered helpers in two packages. Bumps reader 73.2% → 79.2% and scheduler 71.6% → 73.6%. Reader Event predicates (4): - IsCreate: NewEntry-only event classifies as create - IsDelete: OldEntry-only event classifies as delete - both entries (update): neither IsCreate nor IsDelete (strict exclusivity so router routes updates through their own path) - no entries (degenerate): neither (so a metadata-only filer event with no payload doesn't trigger spurious dispatches) Reader LogStartup (4): exercises both shape branches (single-shard ShardID vs ShardPredicate), the explicit-StartTsNs override path, and the Cursor.MinTsNs fallback when StartTsNs=0. Side-effect-only function; tests pin compile-time shape and visit each code path. Scheduler pipelineFanout.InjectEvent (5): - nil event silently absorbed (no follow-up panic in receiving pipeline) - unknown shard returns nil (forward-compat for future shard-mapping gaps) - known shard succeeds - ctx cancellation propagates when underlying pipeline's buffer fills - routes to the correct pipeline among multiple, with cross-pipeline isolation proven via per-pipeline buffer state * test(s3/lifecycle): rename canceled to canceledCtx in fanout test Per gemini review on #9412: a bare 'canceled' identifier reads like a bool. Rename to canceledCtx so the type is obvious at the call site. --- .../reader/event_predicates_test.go | 47 +++++++++++ .../s3lifecycle/reader/log_startup_test.go | 44 +++++++++++ .../scheduler/pipeline_fanout_test.go | 77 +++++++++++++++++++ 3 files changed, 168 insertions(+) create mode 100644 weed/s3api/s3lifecycle/reader/event_predicates_test.go create mode 100644 weed/s3api/s3lifecycle/reader/log_startup_test.go create mode 100644 weed/s3api/s3lifecycle/scheduler/pipeline_fanout_test.go diff --git a/weed/s3api/s3lifecycle/reader/event_predicates_test.go b/weed/s3api/s3lifecycle/reader/event_predicates_test.go new file mode 100644 index 000000000..2c086f61a --- /dev/null +++ b/weed/s3api/s3lifecycle/reader/event_predicates_test.go @@ -0,0 +1,47 @@ +package reader + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/stretchr/testify/assert" +) + +// IsCreate / IsDelete are the small but routing-critical predicates +// the dispatcher uses to decide which match path applies. Both were +// previously exercised only through end-to-end Tick/Match tests; pin +// them directly here. + +func TestEventIsCreate_PopulatedNewEntryNoOldEntry(t *testing.T) { + e := &Event{NewEntry: &filer_pb.Entry{Name: "k"}} + assert.True(t, e.IsCreate()) + assert.False(t, e.IsDelete()) +} + +func TestEventIsDelete_PopulatedOldEntryNoNewEntry(t *testing.T) { + e := &Event{OldEntry: &filer_pb.Entry{Name: "k"}} + assert.True(t, e.IsDelete()) + assert.False(t, e.IsCreate()) +} + +func TestEventBothEntries_NeitherCreateNorDelete(t *testing.T) { + // An update event carries both old and new; the predicates are + // strict (one or the other, not both) so the router can route + // updates through the dedicated path rather than treating them + // as creates or deletes. + e := &Event{ + OldEntry: &filer_pb.Entry{Name: "k"}, + NewEntry: &filer_pb.Entry{Name: "k"}, + } + assert.False(t, e.IsCreate(), "update event must not classify as create") + assert.False(t, e.IsDelete(), "update event must not classify as delete") +} + +func TestEventNoEntries_NeitherCreateNorDelete(t *testing.T) { + // A degenerate event with neither side populated must not classify + // as either; otherwise a metadata-only filer event with no bucket + // payload could trigger spurious dispatches. + e := &Event{} + assert.False(t, e.IsCreate()) + assert.False(t, e.IsDelete()) +} diff --git a/weed/s3api/s3lifecycle/reader/log_startup_test.go b/weed/s3api/s3lifecycle/reader/log_startup_test.go new file mode 100644 index 000000000..061a0a5e8 --- /dev/null +++ b/weed/s3api/s3lifecycle/reader/log_startup_test.go @@ -0,0 +1,44 @@ +package reader + +import ( + "testing" +) + +// LogStartup writes a single glog line summarising the reader's +// resume position; the only behavioral output is a side effect on the +// log sink, but exercising both branches still pins compile-time +// shape (e.g. that ShardPredicate-set readers don't trip on a missing +// ShardID and vice versa) and lets coverage actually visit the code. + +func TestLogStartup_ShardIDOnly(t *testing.T) { + // Single-shard configuration: ShardID is set, ShardPredicate is nil, + // no Cursor, no StartTsNs. The function must run without panic. + r := &Reader{ShardID: 7, EventBudget: 100} + r.LogStartup() +} + +func TestLogStartup_ShardPredicate(t *testing.T) { + // ShardPredicate-set readers take a different log branch; pinning + // the call here catches a regression that returns or panics. + r := &Reader{ + ShardPredicate: func(int) bool { return true }, + EventBudget: 100, + } + r.LogStartup() +} + +func TestLogStartup_StartTsNsOverridesCursor(t *testing.T) { + // Explicit StartTsNs takes precedence over Cursor.MinTsNs; this + // branch is otherwise only hit when a worker is replaying a + // specific position. Run it through to make sure the override is + // honored without consulting the Cursor. + r := &Reader{ShardID: 0, StartTsNs: 1700000000_000_000_000, Cursor: NewCursor()} + r.LogStartup() +} + +func TestLogStartup_CursorMinFallback(t *testing.T) { + // StartTsNs=0 with a non-nil Cursor falls back to Cursor.MinTsNs. + c := NewCursor() + r := &Reader{ShardID: 0, Cursor: c} + r.LogStartup() +} diff --git a/weed/s3api/s3lifecycle/scheduler/pipeline_fanout_test.go b/weed/s3api/s3lifecycle/scheduler/pipeline_fanout_test.go new file mode 100644 index 000000000..a97b6c363 --- /dev/null +++ b/weed/s3api/s3lifecycle/scheduler/pipeline_fanout_test.go @@ -0,0 +1,77 @@ +package scheduler + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// pipelineFanout is the thin shard-routing layer the bootstrapper sees +// as an EventInjector. It maps Event.ShardID to the pipeline that owns +// that shard. Was previously at 0% coverage. + +func TestPipelineFanout_NilEventIsNoOp(t *testing.T) { + // A nil event must be silently absorbed; otherwise a follow-up + // panic in the receiving pipeline would crash the bootstrapper. + f := pipelineFanout{} + assert.NoError(t, f.InjectEvent(context.Background(), nil)) +} + +func TestPipelineFanout_UnknownShardIsNoOp(t *testing.T) { + // A shard not covered by any pipeline in the fanout returns nil + // rather than erroring; the comment in scheduler.go documents this + // as forward-compat for future shard-mapping changes that might + // introduce gaps. + f := pipelineFanout{0: &dispatcher.Pipeline{EventBuffer: 1}} + assert.NoError(t, f.InjectEvent(context.Background(), &reader.Event{ShardID: 99})) +} + +func TestPipelineFanout_KnownShardSucceeds(t *testing.T) { + // A matching shard reaches the pipeline's InjectEvent, which writes + // to its (buffered) events channel and returns nil. + f := pipelineFanout{0: &dispatcher.Pipeline{EventBuffer: 1}} + assert.NoError(t, f.InjectEvent(context.Background(), &reader.Event{ShardID: 0})) +} + +func TestPipelineFanout_PropagatesContextCancellation(t *testing.T) { + // When the underlying pipeline's InjectEvent blocks on a full + // buffer and the ctx is canceled, the fanout must propagate the + // ctx error. Pre-fill the pipeline's buffer (size 1) so the second + // send blocks long enough for the cancellation to win the select. + p := &dispatcher.Pipeline{EventBuffer: 1} + require.NoError(t, p.InjectEvent(context.Background(), &reader.Event{})) + f := pipelineFanout{0: p} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := f.InjectEvent(ctx, &reader.Event{ShardID: 0}) + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled) +} + +func TestPipelineFanout_RoutesToCorrectPipeline(t *testing.T) { + // Two pipelines, each with buffer=1: an event for shard 7 must + // fill pipeline B's buffer (proven by the second send to that + // pipeline blocking with canceled ctx) without affecting pipeline + // A's buffer (proven by the third send still succeeding to A + // because A's buffer is still empty). + pA := &dispatcher.Pipeline{EventBuffer: 1} + pB := &dispatcher.Pipeline{EventBuffer: 1} + f := pipelineFanout{0: pA, 7: pB} + + require.NoError(t, f.InjectEvent(context.Background(), &reader.Event{ShardID: 7})) + + // Second send to shard 7 would block on the full buffer; use a + // pre-canceled ctx to detect the buffer-full state without hanging. + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + err := f.InjectEvent(canceledCtx, &reader.Event{ShardID: 7}) + require.Error(t, err, "B's buffer should be full so a canceled-ctx send returns ctx.Err") + + // Send to shard 0 still succeeds because A's buffer is untouched. + require.NoError(t, f.InjectEvent(context.Background(), &reader.Event{ShardID: 0})) +}