test(s3/lifecycle): bundle reader + scheduler helper coverage (#9412)

* 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.
This commit is contained in:
Chris Lu
2026-05-09 22:02:09 -07:00
committed by GitHub
parent 7996dc1d67
commit ad77362be3
3 changed files with 168 additions and 0 deletions
@@ -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())
}
@@ -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()
}
@@ -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}))
}