feat(s3/lifecycle): delete streaming algorithm path (Phase 5b) (#9466)

* feat(s3/lifecycle): delete streaming algorithm path (Phase 5b)

Phase 5a (PR #9465) retired the algorithm flag and made daily_replay
the only execution path. The streaming-side code (scheduler.Scheduler,
scheduler.BucketBootstrapper, dispatcher.Pipeline, dispatcher.Dispatcher,
dispatcher.FilerPersister, and their tests) has had no in-tree caller
since then. This PR deletes it.

Net change: ~4800 lines removed, ~130 added (the scheduler/configload
tests' helper file the deleted bootstrap_test.go used to host).

Removed:
  - weed/s3api/s3lifecycle/scheduler/{bootstrap,bootstrap_test,
    scheduler,scheduler_test,pipeline_fanout_test,
    refresh_default,refresh_s3tests}.go
  - weed/s3api/s3lifecycle/dispatcher/{dispatcher,dispatcher_test,
    dispatcher_helpers_test,edge_cases_test,multi_shard_test,
    pipeline,pipeline_test,pipeline_helpers_test,toproto_test,
    dispatch_ticks_default,dispatch_ticks_s3tests}.go
  - weed/s3api/s3lifecycle/dispatcher/filer_persister_test.go
    (FilerPersister deleted; FilerStore tests don't need their own
    file)
  - weed/shell/command_s3_lifecycle_run_shard{,_test}.go
    (debug-only shell command that only ever wrapped the streaming
    pipeline; the production worker now exercises the same path
    every daily run)

Trimmed:
  - dispatcher/filer_persister.go down to FilerStore +
    NewFilerStoreClient — the small interface daily_replay's cursor
    persister (dailyrun.FilerCursorPersister) plugs into.

Kept (still consumed by daily_replay):
  - scheduler/configload.{go,_test.go} (LoadCompileInputs,
    AllActivePriorStates)
  - dispatcher/sibling_lister.{go,_test.go} (NewFilerSiblingLister,
    FilerSiblingLister)
  - dispatcher/filer_persister.go (FilerStore, NewFilerStoreClient)

scheduler/testhelpers_test.go restores fakeFilerClient, fakeListStream,
dirEntry, fileEntry — helpers the configload tests used to share with
the deleted bootstrap_test.go.

Updates the handler-package doc strings and one reader-package
comment that still named the streaming pipeline.

* fix(s3/lifecycle): hold lock through tree read in test filer client

gemini caught an inconsistency in scheduler/testhelpers_test.go:
LookupDirectoryEntry reads c.tree under c.mu, but ListEntries was
releasing the lock before reading c.tree. The map is effectively
static during tests so there's no actual race today, but matching
the convention keeps the helper safe if a future test mutates the
tree mid-run.
This commit is contained in:
Chris Lu
2026-05-12 12:54:52 -07:00
committed by GitHub
parent 745e864bda
commit 5004b4e542
25 changed files with 131 additions and 4804 deletions
@@ -1,10 +0,0 @@
//go:build !s3tests
package dispatcher
import "time"
const (
defaultDispatchTick = 5 * time.Second
defaultCheckpointTick = 30 * time.Second
)
@@ -1,14 +0,0 @@
//go:build s3tests
package dispatcher
import "time"
// Under the s3tests build tag the engine treats one "Day" as 10 seconds
// (util.LifeCycleInterval), so the dispatcher must run far below that to
// notice a freshly-due action inside the upstream s3-tests 30s polling
// window. Production timings live in dispatch_ticks_default.go.
const (
defaultDispatchTick = 500 * time.Millisecond
defaultCheckpointTick = 2 * time.Second
)
@@ -1,224 +0,0 @@
package dispatcher
import (
"context"
"errors"
"fmt"
"strconv"
"time"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
"github.com/seaweedfs/seaweedfs/weed/stats"
)
// LifecycleClient abstracts the LifecycleDelete RPC so the dispatcher is
// testable without a live S3 server.
type LifecycleClient interface {
LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error)
}
// Dispatcher consumes due Matches, calls LifecycleDelete, and routes the
// outcome back to the per-shard cursor.
//
// State machine for one Match:
// DONE / NOOP_RESOLVED / SKIPPED_OBJECT_LOCK -> Cursor.Advance
// RETRY_LATER (within budget) -> back into the schedule with backoff
// RETRY_LATER (budget exhausted) / BLOCKED -> Cursor.Freeze in-memory
// FATAL_EVENT_ERROR / unknown -> treat as BLOCKED
//
// A frozen cursor doesn't advance, so the durable cursor is the durable
// "stuck" state on its own: a worker restart re-encounters the poison
// event at MinTsNs and re-freezes after the same retry cycle. No separate
// blocker store is needed.
type Dispatcher struct {
ShardID int
Client LifecycleClient
Cursor *reader.Cursor
Schedule *router.Schedule
// RetryBudget caps RETRY_LATER attempts before escalating to BLOCKED.
// Zero defaults to defaultRetryBudget.
RetryBudget int
// RetryBackoff is the wait between RETRY_LATER re-schedules. Zero
// defaults to defaultRetryBackoff.
RetryBackoff time.Duration
// retries[Match.Key+ObjectKey] = attempts so far. In-memory only:
// worker restart resets the budget, which is fine because the cursor
// is durable and the same poison event will land us here again.
retries map[retryKey]int
}
const (
defaultRetryBudget = 5
defaultRetryBackoff = 30 * time.Second
)
type retryKey struct {
bucket string
ruleHash [8]byte
kind s3lifecycle.ActionKind
objectKey string
versionID string
}
func keyOf(m router.Match) retryKey {
return retryKey{
bucket: m.Key.Bucket,
ruleHash: m.Key.RuleHash,
kind: m.Key.ActionKind,
objectKey: m.ObjectKey,
versionID: m.VersionID,
}
}
func (d *Dispatcher) budget() int {
if d.RetryBudget > 0 {
return d.RetryBudget
}
return defaultRetryBudget
}
func (d *Dispatcher) backoff() time.Duration {
if d.RetryBackoff > 0 {
return d.RetryBackoff
}
return defaultRetryBackoff
}
// Tick drains the schedule for due Matches and dispatches each. Returns the
// count of Matches processed. Safe to call repeatedly.
func (d *Dispatcher) Tick(ctx context.Context, now time.Time) int {
if d.retries == nil {
d.retries = map[retryKey]int{}
}
due := d.Schedule.Drain(now)
for i, m := range due {
if err := ctx.Err(); err != nil {
// Caller is shutting down. Re-queue every remaining drained
// Match (current and any not yet visited) so they're not
// lost across the worker restart — the schedule already
// popped them all out, so the dispatcher owns putting them
// back.
for _, rem := range due[i:] {
d.Schedule.Add(rem)
}
return 0
}
d.dispatchOne(ctx, m, now)
}
return len(due)
}
func (d *Dispatcher) dispatchOne(ctx context.Context, m router.Match, now time.Time) {
// A frozen cursor means a prior BLOCKED is still active for this
// (shard, ActionKey); skip until operator clears it.
if d.Cursor.IsFrozen(m.Key) {
return
}
ruleHash := m.Key.RuleHash
req := &s3_lifecycle_pb.LifecycleDeleteRequest{
Bucket: m.Bucket,
ObjectPath: m.ObjectKey,
VersionId: m.VersionID,
RuleHash: ruleHash[:],
ActionKind: toProtoActionKind(m.Key.ActionKind),
ExpectedIdentity: toProtoIdentity(m.Identity),
}
resp, err := d.Client.LifecycleDelete(ctx, req)
if err != nil {
// Context cancellation is shutdown, not a transport failure: put
// the Match back on the schedule untouched so the next worker
// run picks it up at its original DueTime, with no retry-budget
// burn.
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
d.Schedule.Add(m)
return
}
// Transport error: classify as RETRY_LATER. The remote handler
// already classifies its own filer-side errors; the only path
// that hits this branch is the RPC itself failing.
stats.S3LifecycleDispatchCounter.WithLabelValues(m.Bucket, m.Key.ActionKind.String(), "RPC_ERROR").Inc()
d.handleRetryLater(ctx, m, fmt.Sprintf("RPC: %v", err), now)
return
}
stats.S3LifecycleDispatchCounter.WithLabelValues(m.Bucket, m.Key.ActionKind.String(), resp.Outcome.String()).Inc()
switch resp.Outcome {
case s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED,
s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK:
d.advance(m)
case s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER:
d.handleRetryLater(ctx, m, resp.Reason, now)
case s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED:
d.handleBlocked(ctx, m, resp.Reason)
default:
d.handleBlocked(ctx, m, fmt.Sprintf("unknown outcome %v: %s", resp.Outcome, resp.Reason))
}
}
// observeScheduleDepth publishes the current schedule depth gauge for
// this shard. Called from Pipeline on each tick.
func (d *Dispatcher) observeScheduleDepth() {
stats.S3LifecycleScheduleDepthGauge.WithLabelValues(strconv.Itoa(d.ShardID)).Set(float64(d.Schedule.Len()))
}
func (d *Dispatcher) advance(m router.Match) {
delete(d.retries, keyOf(m))
d.Cursor.Advance(m.Key, m.EventTs.UnixNano())
}
func (d *Dispatcher) handleRetryLater(ctx context.Context, m router.Match, reason string, now time.Time) {
rk := keyOf(m)
d.retries[rk]++
if d.retries[rk] > d.budget() {
d.handleBlocked(ctx, m, fmt.Sprintf("retry budget exhausted: %s", reason))
return
}
// Re-schedule with backoff; same Match, new DueTime.
m.DueTime = now.Add(d.backoff())
d.Schedule.Add(m)
}
func (d *Dispatcher) handleBlocked(ctx context.Context, m router.Match, reason string) {
delete(d.retries, keyOf(m))
glog.Warningf("lifecycle: cursor frozen shard=%d key=%+v eventTs=%s reason=%s",
d.ShardID, m.Key, m.EventTs.UTC().Format(time.RFC3339Nano), reason)
d.Cursor.Freeze(m.Key, m.EventTs.UnixNano())
}
func toProtoActionKind(k s3lifecycle.ActionKind) s3_lifecycle_pb.ActionKind {
switch k {
case s3lifecycle.ActionKindExpirationDays:
return s3_lifecycle_pb.ActionKind_EXPIRATION_DAYS
case s3lifecycle.ActionKindExpirationDate:
return s3_lifecycle_pb.ActionKind_EXPIRATION_DATE
case s3lifecycle.ActionKindNoncurrentDays:
return s3_lifecycle_pb.ActionKind_NONCURRENT_DAYS
case s3lifecycle.ActionKindNewerNoncurrent:
return s3_lifecycle_pb.ActionKind_NEWER_NONCURRENT
case s3lifecycle.ActionKindAbortMPU:
return s3_lifecycle_pb.ActionKind_ABORT_MPU
case s3lifecycle.ActionKindExpiredDeleteMarker:
return s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER
}
return s3_lifecycle_pb.ActionKind_ACTION_KIND_UNSPECIFIED
}
func toProtoIdentity(id *router.EntryIdentity) *s3_lifecycle_pb.EntryIdentity {
if id == nil {
return nil
}
return &s3_lifecycle_pb.EntryIdentity{
MtimeNs: id.MtimeNs,
Size: id.Size,
HeadFid: id.HeadFid,
ExtendedHash: id.ExtendedHash,
}
}
@@ -1,109 +0,0 @@
package dispatcher
import (
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
"github.com/stretchr/testify/assert"
)
// Direct coverage for dispatcher pure helpers (keyOf, budget, backoff)
// and the retryKey identity that drives the retry-budget map. The
// existing dispatcher tests exercise these only through Tick; pinning
// each one separately makes a regression in the helper itself fail at
// the helper level.
func TestKeyOf_DerivesIdentityFromMatch(t *testing.T) {
hash := [8]byte{0xde, 0xad, 0xbe, 0xef, 1, 2, 3, 4}
m := router.Match{
Key: s3lifecycle.ActionKey{
Bucket: "bk",
RuleHash: hash,
ActionKind: s3lifecycle.ActionKindExpirationDays,
},
ObjectKey: "obj.txt",
VersionID: "v_abc",
}
got := keyOf(m)
assert.Equal(t, "bk", got.bucket)
assert.Equal(t, hash, got.ruleHash)
assert.Equal(t, s3lifecycle.ActionKindExpirationDays, got.kind)
assert.Equal(t, "obj.txt", got.objectKey)
assert.Equal(t, "v_abc", got.versionID)
}
func TestKeyOf_EqualMatchesProduceEqualKeys(t *testing.T) {
// retryKey is used as a map key in the retry budget; equality must
// hold between two Match values with identical fields so the second
// dispatch finds the first's retry counter.
hash := [8]byte{0xde, 0xad, 0xbe, 0xef}
m1 := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", RuleHash: hash, ActionKind: s3lifecycle.ActionKindExpirationDays},
ObjectKey: "obj",
VersionID: "v_x",
}
m2 := m1
assert.Equal(t, keyOf(m1), keyOf(m2))
}
func TestKeyOf_DistinctVersionIDsProduceDistinctKeys(t *testing.T) {
// Two versions of the same logical object must NOT share a retry
// budget; otherwise a noisy version could starve a healthy one.
base := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindNoncurrentDays},
ObjectKey: "obj",
VersionID: "v_a",
}
other := base
other.VersionID = "v_b"
assert.NotEqual(t, keyOf(base), keyOf(other))
}
func TestKeyOf_DistinctActionKindsProduceDistinctKeys(t *testing.T) {
// The same (bucket, object, version) hit by two different action
// kinds must each have their own retry budget.
base := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
ObjectKey: "obj",
}
other := base
other.Key.ActionKind = s3lifecycle.ActionKindNoncurrentDays
assert.NotEqual(t, keyOf(base), keyOf(other))
}
func TestDispatcherBudget_ReturnsConfiguredValueWhenSet(t *testing.T) {
d := &Dispatcher{RetryBudget: 9}
assert.Equal(t, 9, d.budget())
}
func TestDispatcherBudget_FallsBackToDefaultWhenZero(t *testing.T) {
// Operators leaving the budget at zero opt into the documented
// default. A regression that returns 0 would NOOP every retry.
d := &Dispatcher{}
assert.Equal(t, defaultRetryBudget, d.budget())
}
func TestDispatcherBudget_NegativeFallsBackToDefault(t *testing.T) {
// budget() guards on > 0, so a negative value falls back rather
// than producing nonsense. Pin the contract so a refactor that
// flips the comparison is caught.
d := &Dispatcher{RetryBudget: -1}
assert.Equal(t, defaultRetryBudget, d.budget())
}
func TestDispatcherBackoff_ReturnsConfiguredValueWhenSet(t *testing.T) {
d := &Dispatcher{RetryBackoff: 5 * time.Second}
assert.Equal(t, 5*time.Second, d.backoff())
}
func TestDispatcherBackoff_FallsBackToDefaultWhenZero(t *testing.T) {
d := &Dispatcher{}
assert.Equal(t, defaultRetryBackoff, d.backoff())
}
func TestDispatcherBackoff_NegativeFallsBackToDefault(t *testing.T) {
d := &Dispatcher{RetryBackoff: -time.Second}
assert.Equal(t, defaultRetryBackoff, d.backoff())
}
@@ -1,318 +0,0 @@
package dispatcher
import (
"context"
"errors"
"testing"
"time"
dto "github.com/prometheus/client_model/go"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
"github.com/seaweedfs/seaweedfs/weed/stats"
)
type fakeClient struct {
calls int
respond func(call int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error)
}
func (f *fakeClient) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
f.calls++
return f.respond(f.calls)
}
func mkMatch(eventTs time.Time, due time.Time, key string) router.Match {
return router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
EventTs: eventTs,
DueTime: due,
Bucket: "bk",
ObjectKey: key,
}
}
func newDispatcher(client LifecycleClient) (*Dispatcher, *router.Schedule) {
sched := router.NewSchedule()
d := &Dispatcher{
ShardID: 0,
Client: client,
Cursor: reader.NewCursor(),
Schedule: sched,
RetryBudget: 3,
RetryBackoff: time.Millisecond,
}
return d, sched
}
func TestDispatchDoneAdvancesCursor(t *testing.T) {
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
}, nil
},
}
d, sched := newDispatcher(client)
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
got := d.Tick(context.Background(), t0)
if got != 1 {
t.Fatalf("Tick processed=%d, want 1", got)
}
if d.Cursor.Get(m.Key) != t0.UnixNano() {
t.Fatalf("cursor not advanced: %d", d.Cursor.Get(m.Key))
}
if d.Cursor.IsFrozen(m.Key) {
t.Fatal("cursor should not be frozen")
}
}
func TestDispatchNoopAdvancesCursor(t *testing.T) {
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED,
Reason: "STALE_IDENTITY",
}, nil
},
}
d, sched := newDispatcher(client)
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
d.Tick(context.Background(), t0)
if d.Cursor.Get(m.Key) != t0.UnixNano() {
t.Fatal("NOOP_RESOLVED should advance cursor")
}
}
func TestDispatchRetryLaterReSchedules(t *testing.T) {
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER,
Reason: "TRANSPORT_ERROR",
}, nil
},
}
d, sched := newDispatcher(client)
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
// First tick: dispatched, retry-budget = 1, re-scheduled.
d.Tick(context.Background(), t0)
if sched.Len() != 1 {
t.Fatalf("expected re-schedule on RETRY_LATER, sched.Len=%d", sched.Len())
}
if d.Cursor.Get(m.Key) != 0 {
t.Fatal("cursor must not advance on RETRY_LATER")
}
if d.Cursor.IsFrozen(m.Key) {
t.Fatal("cursor must not freeze within budget")
}
}
func TestDispatchRetryBudgetEscalatesToBlocked(t *testing.T) {
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER,
Reason: "stuck",
}, nil
},
}
d, sched := newDispatcher(client)
d.RetryBudget = 2
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
// Tick repeatedly; each pushes the re-scheduled entry forward by backoff,
// so we advance "now" past each backoff to drain it.
now := t0
for i := 0; i < 5 && sched.Len() > 0; i++ {
now = now.Add(d.RetryBackoff + time.Millisecond)
d.Tick(context.Background(), now)
}
if !d.Cursor.IsFrozen(m.Key) {
t.Fatal("expected freeze after budget exhausted")
}
if d.Cursor.Get(m.Key) != t0.UnixNano() {
t.Fatal("frozen cursor should be pinned at event ts")
}
}
func TestDispatchBlockedFreezesCursor(t *testing.T) {
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED,
Reason: "FATAL_EVENT_ERROR: empty bucket",
}, nil
},
}
d, sched := newDispatcher(client)
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
d.Tick(context.Background(), t0)
if !d.Cursor.IsFrozen(m.Key) {
t.Fatal("BLOCKED must freeze cursor")
}
if d.Cursor.Get(m.Key) != t0.UnixNano() {
t.Fatal("frozen cursor should be pinned at event ts")
}
}
func TestDispatchContextCancelDoesNotBurnBudget(t *testing.T) {
// Worker shutdown causes the in-flight RPC to return context.Canceled.
// The Match should go back on the schedule untouched; no retry-budget
// burn means a quick restart can't escalate it to BLOCKED.
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return nil, context.Canceled
},
}
d, sched := newDispatcher(client)
d.RetryBudget = 1
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
d.Tick(context.Background(), t0)
if sched.Len() != 1 {
t.Fatalf("expected re-queue on ctx cancel, sched.Len=%d", sched.Len())
}
if d.Cursor.IsFrozen(m.Key) {
t.Fatal("ctx cancel must not freeze cursor")
}
if got := d.retries[keyOf(m)]; got != 0 {
t.Fatalf("ctx cancel must not burn retry budget, retries=%d", got)
}
}
func TestDispatchTransportErrorRetries(t *testing.T) {
// gRPC error: classified as RETRY_LATER. After the budget the cursor freezes.
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return nil, errors.New("connection refused")
},
}
d, sched := newDispatcher(client)
d.RetryBudget = 1
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
now := t0
for i := 0; i < 5 && sched.Len() > 0 && !d.Cursor.IsFrozen(m.Key); i++ {
now = now.Add(d.RetryBackoff + time.Millisecond)
d.Tick(context.Background(), now)
}
if !d.Cursor.IsFrozen(m.Key) {
t.Fatal("transport-error retries should escalate to BLOCKED past budget")
}
}
func TestDispatchSkipsFrozenCursor(t *testing.T) {
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
t.Fatal("frozen cursor should not call RPC")
return nil, nil
},
}
d, sched := newDispatcher(client)
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
d.Cursor.Freeze(m.Key, t0.UnixNano())
sched.Add(m)
d.Tick(context.Background(), t0)
if client.calls != 0 {
t.Fatalf("expected 0 RPC calls, got %d", client.calls)
}
}
func TestDispatchRestartReFreezesNaturally(t *testing.T) {
// No durable blocker store: the durable cursor + a deterministic poison
// event self-recover to the blocked state on a fresh Dispatcher. After
// the budget burns, the new cursor freezes at the same EventTs.
respond := func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED,
Reason: "deterministic poison",
}, nil
}
d, sched := newDispatcher(&fakeClient{respond: respond})
t0 := time.Now()
m := mkMatch(t0, t0, "obj")
sched.Add(m)
d.Tick(context.Background(), t0)
if !d.Cursor.IsFrozen(m.Key) {
t.Fatal("first run should freeze")
}
// Simulate restart: brand-new Dispatcher and Cursor, same poison event.
d2, sched2 := newDispatcher(&fakeClient{respond: respond})
sched2.Add(m)
d2.Tick(context.Background(), t0)
if !d2.Cursor.IsFrozen(m.Key) {
t.Fatal("restart should re-freeze without a durable blocker store")
}
if d2.Cursor.Get(m.Key) != t0.UnixNano() {
t.Fatal("re-freeze cursor not pinned at event ts")
}
}
func TestDispatchEmitsOutcomeMetric(t *testing.T) {
// The Prometheus counter is the operator's signal that lifecycle
// is doing real work; it must increment for every dispatch path
// (success, retry, transport error). Use the shared label tuple
// (bucket, kind, outcome) and read the counter delta.
client := &fakeClient{
respond: func(call int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
if call == 1 {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
}, nil
}
return nil, errors.New("network down")
},
}
d, sched := newDispatcher(client)
t0 := time.Now()
bucket := "bk"
kind := s3lifecycle.ActionKindExpirationDays.String()
startDone := counterValue(t, bucket, kind, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE.String())
startRpc := counterValue(t, bucket, kind, "RPC_ERROR")
sched.Add(mkMatch(t0, t0, "obj-done"))
d.Tick(context.Background(), t0)
// Second match exercises the transport-error path.
sched.Add(mkMatch(t0, t0, "obj-fail"))
d.Tick(context.Background(), t0)
if got := counterValue(t, bucket, kind, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE.String()) - startDone; got != 1 {
t.Errorf("DONE counter delta=%v, want 1", got)
}
if got := counterValue(t, bucket, kind, "RPC_ERROR") - startRpc; got != 1 {
t.Errorf("RPC_ERROR counter delta=%v, want 1", got)
}
}
func counterValue(t *testing.T, bucket, kind, outcome string) float64 {
t.Helper()
m := &dto.Metric{}
if err := stats.S3LifecycleDispatchCounter.WithLabelValues(bucket, kind, outcome).Write(m); err != nil {
t.Fatalf("read counter: %v", err)
}
return m.GetCounter().GetValue()
}
@@ -1,134 +0,0 @@
package dispatcher
import (
"context"
"errors"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// errFilerStore satisfies FilerStore with a forced Read error so the
// Load path's "non-NotFound transport error" branch is reachable.
// Save mirrors the pattern; tests for the nil-Store branch don't need
// a stub at all.
type errFilerStore struct {
readErr error
saveErr error
readData []byte
}
func (e *errFilerStore) Read(_ context.Context, _, _ string) ([]byte, error) {
if e.readErr != nil {
return nil, e.readErr
}
return e.readData, nil
}
func (e *errFilerStore) Save(_ context.Context, _, _ string, _ []byte) error {
return e.saveErr
}
// ---------- FilerPersister edges ----------
func TestFilerPersisterLoad_NilStoreErrors(t *testing.T) {
// The persister's Store dependency is required; loading with a nil
// Store must error rather than panic on the Read call.
p := &FilerPersister{}
_, err := p.Load(context.Background(), 0)
require.Error(t, err)
assert.Contains(t, err.Error(), "nil Store")
}
func TestFilerPersisterSave_NilStoreErrors(t *testing.T) {
p := &FilerPersister{}
err := p.Save(context.Background(), 0, map[s3lifecycle.ActionKey]int64{})
require.Error(t, err)
assert.Contains(t, err.Error(), "nil Store")
}
func TestFilerPersisterLoad_NonNotFoundErrorIsWrapped(t *testing.T) {
// A transport-level error (anything that isn't ErrNotFound) wraps
// with the shard ID context so operators can attribute it. Pin
// both that the error surfaces and that the original is recoverable
// via errors.Is so a caller can distinguish.
want := errors.New("filer-side bang")
p := &FilerPersister{Store: &errFilerStore{readErr: want}}
_, err := p.Load(context.Background(), 3)
require.Error(t, err)
assert.ErrorIs(t, err, want)
assert.Contains(t, err.Error(), "shard=3")
}
func TestFilerPersisterLoad_EmptyContentReturnsEmptyMap(t *testing.T) {
// Read returning a successful empty []byte means "file exists but
// is zero length"; the persister must treat this as "no entries"
// rather than try to JSON-decode an empty slice and return an
// error.
p := &FilerPersister{Store: &errFilerStore{readData: []byte{}}}
got, err := p.Load(context.Background(), 0)
require.NoError(t, err)
assert.NotNil(t, got)
assert.Empty(t, got)
}
// ---------- Tick edges ----------
func TestTick_InitializesRetriesMapOnFirstCall(t *testing.T) {
// retries is a lazy map; a Dispatcher constructed without it must
// not panic on the first Tick. Pin that handleRetryLater can write
// into the map afterwards (proven by the dispatcher returning an
// integer rather than crashing).
d, sched := newDispatcher(&fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, nil
},
})
require.Nil(t, d.retries, "preconditions: retries map starts nil")
t0 := time.Now()
sched.Add(router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
ObjectKey: "k",
DueTime: t0,
})
processed := d.Tick(context.Background(), t0.Add(time.Hour))
assert.Equal(t, 1, processed)
assert.NotNil(t, d.retries, "Tick must initialize the retries map")
}
func TestTick_CtxShutdownMidLoopRequeuesAndReturnsZero(t *testing.T) {
// If ctx is canceled before Tick starts dispatching, the entire
// drained batch must be re-queued. Drain pops ALL due matches at
// once, so a naive "re-add the current Match only" would silently
// lose every Match past the cancellation point. Three matches
// here exercises that the loop re-queues the current AND every
// remaining drained entry.
calls := 0
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
calls++
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, nil
},
}
d, sched := newDispatcher(client)
t0 := time.Now()
for _, k := range []string{"k1", "k2", "k3"} {
sched.Add(router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
ObjectKey: k,
DueTime: t0,
})
}
ctx, cancel := context.WithCancel(context.Background())
cancel() // shutdown before Tick even starts dispatching
processed := d.Tick(ctx, t0.Add(time.Hour))
assert.Equal(t, 0, processed, "shutdown must report zero processed")
assert.Equal(t, 0, calls, "shutdown must skip every dispatch call")
assert.Equal(t, 3, sched.Len(), "every drained Match must be re-queued — none lost")
}
@@ -1,22 +1,15 @@
package dispatcher
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
)
// FilerStore is the small subset of filer-client operations the persistence
// layer needs. The default implementation calls filer.ReadInsideFiler /
// filer.SaveInsideFiler; tests inject an in-memory fake.
// FilerStore is the small subset of filer-client operations the
// daily-replay cursor persister needs (dailyrun.FilerCursorPersister).
// Tests inject an in-memory fake.
type FilerStore interface {
Read(ctx context.Context, dir, name string) ([]byte, error)
Save(ctx context.Context, dir, name string, content []byte) error
@@ -38,104 +31,3 @@ func (s *filerStoreClient) Read(ctx context.Context, dir, name string) ([]byte,
func (s *filerStoreClient) Save(ctx context.Context, dir, name string, content []byte) error {
return filer.SaveInsideFiler(ctx, s.client, dir, name, content)
}
// CursorDir is the filer directory holding per-shard cursor files.
const CursorDir = "/etc/s3/lifecycle/cursors"
// FilerPersister persists per-shard cursor maps to /etc/s3/lifecycle/cursors/
// as JSON. One file per shard keeps Save atomic — the filer writes the entry
// in a single mutation, so a crash mid-write doesn't leak partial state.
type FilerPersister struct {
Store FilerStore
}
// cursorFile is the on-disk JSON shape. cursorFileEntry repeats the
// ActionKey fields explicitly so the format stays human-readable and stable
// against Go-side struct rearrangements.
type cursorFile struct {
Version int `json:"version"`
ShardID int `json:"shard_id"`
Entries []cursorFileEntry `json:"entries"`
}
type cursorFileEntry struct {
Bucket string `json:"bucket"`
RuleHash []byte `json:"rule_hash"` // base64 in JSON
ActionKind int `json:"action_kind"`
TsNs int64 `json:"ts_ns"`
}
const cursorFileVersion = 1
func cursorFileName(shardID int) string {
return fmt.Sprintf("shard-%02d.json", shardID)
}
func (p *FilerPersister) Load(ctx context.Context, shardID int) (map[s3lifecycle.ActionKey]int64, error) {
if p.Store == nil {
return nil, errors.New("FilerPersister: nil Store")
}
content, err := p.Store.Read(ctx, CursorDir, cursorFileName(shardID))
if err != nil {
if errors.Is(err, filer_pb.ErrNotFound) {
return map[s3lifecycle.ActionKey]int64{}, nil
}
return nil, fmt.Errorf("cursor read shard=%d: %w", shardID, err)
}
if len(content) == 0 {
return map[s3lifecycle.ActionKey]int64{}, nil
}
var cf cursorFile
if err := json.Unmarshal(content, &cf); err != nil {
return nil, fmt.Errorf("cursor decode shard=%d: %w", shardID, err)
}
out := make(map[s3lifecycle.ActionKey]int64, len(cf.Entries))
for _, e := range cf.Entries {
k := s3lifecycle.ActionKey{
Bucket: e.Bucket,
ActionKind: s3lifecycle.ActionKind(e.ActionKind),
}
copy(k.RuleHash[:], e.RuleHash)
out[k] = e.TsNs
}
return out, nil
}
func (p *FilerPersister) Save(ctx context.Context, shardID int, state map[s3lifecycle.ActionKey]int64) error {
if p.Store == nil {
return errors.New("FilerPersister: nil Store")
}
cf := cursorFile{Version: cursorFileVersion, ShardID: shardID}
cf.Entries = make([]cursorFileEntry, 0, len(state))
for k, v := range state {
hash := k.RuleHash
cf.Entries = append(cf.Entries, cursorFileEntry{
Bucket: k.Bucket,
RuleHash: hash[:],
ActionKind: int(k.ActionKind),
TsNs: v,
})
}
// Stable order so the on-disk file diffs cleanly across saves.
sort.Slice(cf.Entries, func(i, j int) bool {
a, b := cf.Entries[i], cf.Entries[j]
if a.Bucket != b.Bucket {
return a.Bucket < b.Bucket
}
if a.ActionKind != b.ActionKind {
return a.ActionKind < b.ActionKind
}
return bytes.Compare(a.RuleHash, b.RuleHash) < 0
})
var buf bytes.Buffer
if err := json.NewEncoder(&buf).Encode(cf); err != nil {
return fmt.Errorf("cursor encode shard=%d: %w", shardID, err)
}
if err := p.Store.Save(ctx, CursorDir, cursorFileName(shardID), buf.Bytes()); err != nil {
return fmt.Errorf("cursor save shard=%d: %w", shardID, err)
}
return nil
}
// Compile-time interface check.
var _ reader.Persister = (*FilerPersister)(nil)
@@ -1,159 +0,0 @@
package dispatcher
import (
"context"
"sync"
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
)
// fakeFilerStore is an in-memory FilerStore for tests.
type fakeFilerStore struct {
mu sync.Mutex
files map[string][]byte
}
func newFakeFilerStore() *fakeFilerStore {
return &fakeFilerStore{files: map[string][]byte{}}
}
func (f *fakeFilerStore) key(dir, name string) string { return dir + "/" + name }
func (f *fakeFilerStore) Read(ctx context.Context, dir, name string) ([]byte, error) {
f.mu.Lock()
defer f.mu.Unlock()
v, ok := f.files[f.key(dir, name)]
if !ok {
return nil, filer_pb.ErrNotFound
}
return append([]byte(nil), v...), nil
}
func (f *fakeFilerStore) Save(ctx context.Context, dir, name string, content []byte) error {
f.mu.Lock()
defer f.mu.Unlock()
f.files[f.key(dir, name)] = append([]byte(nil), content...)
return nil
}
func mkKey(bucket string, kind s3lifecycle.ActionKind, hashByte byte) s3lifecycle.ActionKey {
k := s3lifecycle.ActionKey{Bucket: bucket, ActionKind: kind}
for i := range k.RuleHash {
k.RuleHash[i] = hashByte
}
return k
}
func TestFilerPersisterEmptyLoadReturnsEmptyMap(t *testing.T) {
p := &FilerPersister{Store: newFakeFilerStore()}
state, err := p.Load(context.Background(), 0)
if err != nil {
t.Fatalf("Load on empty: %v", err)
}
if len(state) != 0 {
t.Fatalf("expected empty map, got %d entries", len(state))
}
}
func TestFilerPersisterSaveLoadRoundTrip(t *testing.T) {
p := &FilerPersister{Store: newFakeFilerStore()}
ctx := context.Background()
in := map[s3lifecycle.ActionKey]int64{
mkKey("bucket-a", s3lifecycle.ActionKindExpirationDays, 0xAA): 12345,
mkKey("bucket-b", s3lifecycle.ActionKindAbortMPU, 0xBB): 67890,
mkKey("bucket-a", s3lifecycle.ActionKindNoncurrentDays, 0xCC): 54321,
}
if err := p.Save(ctx, 3, in); err != nil {
t.Fatalf("Save: %v", err)
}
out, err := p.Load(ctx, 3)
if err != nil {
t.Fatalf("Load: %v", err)
}
if len(out) != len(in) {
t.Fatalf("Load count=%d, want %d", len(out), len(in))
}
for k, v := range in {
if got := out[k]; got != v {
t.Fatalf("Load[%v]=%d, want %d", k, got, v)
}
}
}
func TestFilerPersisterIsolatesShards(t *testing.T) {
p := &FilerPersister{Store: newFakeFilerStore()}
ctx := context.Background()
stateA := map[s3lifecycle.ActionKey]int64{mkKey("a", s3lifecycle.ActionKindExpirationDays, 1): 100}
stateB := map[s3lifecycle.ActionKey]int64{mkKey("a", s3lifecycle.ActionKindExpirationDays, 1): 200}
if err := p.Save(ctx, 0, stateA); err != nil {
t.Fatalf("Save 0: %v", err)
}
if err := p.Save(ctx, 1, stateB); err != nil {
t.Fatalf("Save 1: %v", err)
}
loadA, _ := p.Load(ctx, 0)
loadB, _ := p.Load(ctx, 1)
if loadA[mkKey("a", s3lifecycle.ActionKindExpirationDays, 1)] != 100 {
t.Fatalf("shard 0 leaked from shard 1: %v", loadA)
}
if loadB[mkKey("a", s3lifecycle.ActionKindExpirationDays, 1)] != 200 {
t.Fatalf("shard 1 reads stale: %v", loadB)
}
}
func TestFilerPersisterSaveOverwrites(t *testing.T) {
p := &FilerPersister{Store: newFakeFilerStore()}
ctx := context.Background()
k := mkKey("b", s3lifecycle.ActionKindExpirationDays, 0xAA)
if err := p.Save(ctx, 0, map[s3lifecycle.ActionKey]int64{k: 100}); err != nil {
t.Fatalf("Save 1: %v", err)
}
if err := p.Save(ctx, 0, map[s3lifecycle.ActionKey]int64{k: 200}); err != nil {
t.Fatalf("Save 2: %v", err)
}
out, _ := p.Load(ctx, 0)
if out[k] != 200 {
t.Fatalf("overwrite not applied, got %d", out[k])
}
}
func TestFilerPersisterSaveIsDeterministic(t *testing.T) {
// Saving the same map twice must produce byte-identical content so the
// on-disk file diffs cleanly when the state hasn't changed.
store := newFakeFilerStore()
p := &FilerPersister{Store: store}
ctx := context.Background()
in := map[s3lifecycle.ActionKey]int64{
mkKey("zeta", s3lifecycle.ActionKindNoncurrentDays, 0xCC): 300,
mkKey("alpha", s3lifecycle.ActionKindExpirationDays, 0xAA): 100,
mkKey("alpha", s3lifecycle.ActionKindAbortMPU, 0xBB): 200,
}
if err := p.Save(ctx, 0, in); err != nil {
t.Fatalf("Save 1: %v", err)
}
first := append([]byte(nil), store.files[store.key(CursorDir, cursorFileName(0))]...)
if err := p.Save(ctx, 0, in); err != nil {
t.Fatalf("Save 2: %v", err)
}
second := store.files[store.key(CursorDir, cursorFileName(0))]
if string(first) != string(second) {
t.Fatalf("non-deterministic save:\n first=%s\nsecond=%s", first, second)
}
}
func TestFilerPersisterCorruptDataReturnsError(t *testing.T) {
store := newFakeFilerStore()
store.Save(context.Background(), CursorDir, "shard-00.json", []byte("not json"))
p := &FilerPersister{Store: store}
if _, err := p.Load(context.Background(), 0); err == nil {
t.Fatal("expected decode error, got nil")
}
}
@@ -1,275 +0,0 @@
package dispatcher
import (
"context"
"sync"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
)
// Layer 2 component tests for the dispatcher pipeline mechanics:
// the per-shard composition that Pipeline.Run wires up at runtime
// (cursors, schedules, dispatchers, persister), exercised directly so
// the tests stay independent of a live filer connection. Pipeline.Run
// itself can't run here because it builds a real reader.Reader; these
// tests instead drive Tick directly while sharing the same fakeClient
// that dispatcher_test.go uses.
// shardKit bundles the per-shard state Pipeline.Run constructs in
// production: a Cursor, a Schedule, and a Dispatcher hooked to both.
type shardKit struct {
id int
cursor *reader.Cursor
dispatch *Dispatcher
}
func newShardKit(t *testing.T, id int, client LifecycleClient) *shardKit {
t.Helper()
c := reader.NewCursor()
return &shardKit{
id: id,
cursor: c,
dispatch: &Dispatcher{
ShardID: id,
Client: client,
Cursor: c,
Schedule: router.NewSchedule(),
RetryBudget: 3,
RetryBackoff: time.Millisecond,
},
}
}
// recordingClient lets Layer 2 tests assert exactly which RPCs the
// dispatcher made. Goroutine-safe because Tick is called from test
// goroutines but inspections happen from the test goroutine.
type recordingClient struct {
mu sync.Mutex
requests []*s3_lifecycle_pb.LifecycleDeleteRequest
respond func(req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error)
}
func (r *recordingClient) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
r.mu.Lock()
r.requests = append(r.requests, req)
r.mu.Unlock()
return r.respond(req)
}
func TestPipelineMultiShardFanOutKeepsCursorsIsolated(t *testing.T) {
// Two events for two different shards land in different schedules
// and dispatch independently. Each shard's cursor advances only
// for the event(s) targeting that shard.
client := &recordingClient{
respond: func(*s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, nil
},
}
shards := map[int]*shardKit{
0: newShardKit(t, 0, client),
1: newShardKit(t, 1, client),
}
t0 := time.Now()
mA := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
EventTs: t0,
DueTime: t0,
Bucket: "bk",
ObjectKey: "alpha",
}
// Use a different action kind on shard 1 so the cursor key differs
// from shard 0 — shared keys would write to the same cursor entry.
mB := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindNoncurrentDays},
EventTs: t0.Add(time.Second),
DueTime: t0.Add(time.Second),
Bucket: "bk",
ObjectKey: "beta",
}
shards[0].dispatch.Schedule.Add(mA)
shards[1].dispatch.Schedule.Add(mB)
// Tick shard 0 first; shard 1's schedule must still hold its match.
if got := shards[0].dispatch.Tick(context.Background(), t0.Add(time.Hour)); got != 1 {
t.Fatalf("shard 0 Tick processed=%d, want 1", got)
}
if got := shards[1].dispatch.Schedule.Len(); got != 1 {
t.Fatalf("shard 1 schedule must not be drained by shard 0 Tick: len=%d", got)
}
// Now tick shard 1.
if got := shards[1].dispatch.Tick(context.Background(), t0.Add(time.Hour)); got != 1 {
t.Fatalf("shard 1 Tick processed=%d, want 1", got)
}
if shards[0].cursor.Get(mA.Key) != mA.EventTs.UnixNano() {
t.Fatalf("shard 0 cursor not advanced: got %d", shards[0].cursor.Get(mA.Key))
}
if shards[1].cursor.Get(mB.Key) != mB.EventTs.UnixNano() {
t.Fatalf("shard 1 cursor not advanced: got %d", shards[1].cursor.Get(mB.Key))
}
// Cursors must NOT cross-contaminate.
if shards[0].cursor.Get(mB.Key) != 0 {
t.Fatalf("shard 0 cursor saw shard 1's key: got %d", shards[0].cursor.Get(mB.Key))
}
}
func TestPipelineCursorFreezeOnOneShardDoesntBlockOthers(t *testing.T) {
// Shard 0 dispatches a poison event that returns BLOCKED — its
// cursor freezes at the event TsNs. Shard 1's progress must be
// independent: it dispatches successfully and advances.
client := &recordingClient{
respond: func(req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
if req.Bucket == "bk" && req.ObjectPath == "poison" {
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED}, nil
}
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, nil
},
}
shards := map[int]*shardKit{
0: newShardKit(t, 0, client),
1: newShardKit(t, 1, client),
}
t0 := time.Now()
poison := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
EventTs: t0,
DueTime: t0,
Bucket: "bk",
ObjectKey: "poison",
}
good := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindNoncurrentDays},
EventTs: t0.Add(time.Second),
DueTime: t0.Add(time.Second),
Bucket: "bk",
ObjectKey: "good",
}
shards[0].dispatch.Schedule.Add(poison)
shards[1].dispatch.Schedule.Add(good)
shards[0].dispatch.Tick(context.Background(), t0.Add(time.Hour))
shards[1].dispatch.Tick(context.Background(), t0.Add(time.Hour))
if !shards[0].cursor.IsFrozen(poison.Key) {
t.Fatalf("shard 0 cursor must freeze on BLOCKED outcome")
}
if shards[1].cursor.IsFrozen(good.Key) {
t.Fatalf("shard 1 cursor must not freeze when shard 0 fails")
}
if shards[1].cursor.Get(good.Key) != good.EventTs.UnixNano() {
t.Fatalf("shard 1 cursor must advance independently: got %d", shards[1].cursor.Get(good.Key))
}
}
func TestPipelinePersisterCheckpointRoundTripsEveryShard(t *testing.T) {
// A cursor checkpoint persists the per-shard map; a fresh dispatcher
// can Restore from that map and resume at the same TsNs. Pipeline.Run
// drives this on a ticker; tests exercise the contract directly.
client := &recordingClient{
respond: func(*s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, nil
},
}
shards := map[int]*shardKit{
0: newShardKit(t, 0, client),
1: newShardKit(t, 1, client),
}
t0 := time.Now()
addAndTick := func(s *shardKit, obj string, kind s3lifecycle.ActionKind, ts time.Time) {
s.dispatch.Schedule.Add(router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: kind},
EventTs: ts,
DueTime: ts,
Bucket: "bk",
ObjectKey: obj,
})
s.dispatch.Tick(context.Background(), t0.Add(time.Hour))
}
addAndTick(shards[0], "alpha", s3lifecycle.ActionKindExpirationDays, t0)
addAndTick(shards[1], "beta", s3lifecycle.ActionKindNoncurrentDays, t0.Add(time.Second))
pers := reader.NewInMemoryPersister()
for id, s := range shards {
if err := pers.Save(context.Background(), id, s.cursor.Snapshot()); err != nil {
t.Fatalf("save shard %d: %v", id, err)
}
}
// Restore into fresh cursors, verify every shard's TsNs round-trips.
for id, s := range shards {
state, err := pers.Load(context.Background(), id)
if err != nil {
t.Fatalf("load shard %d: %v", id, err)
}
c := reader.NewCursor()
c.Restore(state)
want := s.cursor.Snapshot()
got := c.Snapshot()
if len(got) != len(want) {
t.Fatalf("shard %d snapshot len: got %d, want %d", id, len(got), len(want))
}
for k, v := range want {
if got[k] != v {
t.Fatalf("shard %d cursor key %v: got %d, want %d", id, k, got[k], v)
}
}
}
}
func TestPipelineRetryLaterRespectsDispatchTickCadence(t *testing.T) {
// RETRY_LATER schedules the match for a later DueTime. A Tick at
// the present moment must NOT re-dispatch it; only a Tick past
// the new DueTime triggers a second RPC. This pins the cadence
// contract against premature retries from later refresh ticks.
calls := 0
client := &recordingClient{
respond: func(*s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
calls++
if calls == 1 {
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER}, nil
}
return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, nil
},
}
s := newShardKit(t, 0, client)
s.dispatch.RetryBackoff = time.Hour // long backoff so the second Tick at t0+1m doesn't fire
t0 := time.Now()
m := router.Match{
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
EventTs: t0,
DueTime: t0,
Bucket: "bk",
ObjectKey: "obj",
}
s.dispatch.Schedule.Add(m)
if got := s.dispatch.Tick(context.Background(), t0); got != 1 {
t.Fatalf("first Tick processed=%d, want 1 (RETRY_LATER counts as processed)", got)
}
// Within the backoff window: nothing fires.
if got := s.dispatch.Tick(context.Background(), t0.Add(time.Minute)); got != 0 {
t.Fatalf("Tick within backoff must skip: processed=%d", got)
}
if calls != 1 {
t.Fatalf("RPC must not retry within backoff: calls=%d", calls)
}
// Past backoff: re-dispatches.
if got := s.dispatch.Tick(context.Background(), t0.Add(2*time.Hour)); got != 1 {
t.Fatalf("Tick past backoff must re-dispatch: processed=%d", got)
}
if calls != 2 {
t.Fatalf("expected exactly 2 calls (initial + one retry), got %d", calls)
}
}
@@ -1,314 +0,0 @@
package dispatcher
import (
"context"
"errors"
"fmt"
"strconv"
"sync"
"time"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
"github.com/seaweedfs/seaweedfs/weed/stats"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// Pipeline composes the reader, router, dispatcher, and cursor checkpoint
// into a single Run loop. One Pipeline can handle a contiguous shard span
// or any explicit set of shards via Shards; ShardID still works for the
// single-shard case (and is preferred for short-form configuration).
//
// Internally there is exactly one filer subscription regardless of how
// many shards Shards contains; events are filtered by the reader's
// ShardPredicate and routed to the matching shard's Cursor + Schedule
// inside the existing dispatch goroutine — no per-shard goroutines.
type Pipeline struct {
ShardID int // used when Shards is empty
Shards []int // overrides ShardID when non-empty
BucketsPath string
Engine *engine.Engine
// Cursor is consulted only when len(Shards) <= 1. Range mode allocates
// a fresh Cursor per shard internally.
Cursor *reader.Cursor
Persister reader.Persister
Client LifecycleClient
FilerClient filer_pb.SeaweedFilerClient
ClientID int32
ClientName string
// Tick cadence for the dispatcher. Zero = defaultDispatchTick.
DispatchTick time.Duration
// Cadence for cursor checkpoint writes. Zero = defaultCheckpointTick.
CheckpointTick time.Duration
// EventBudget caps reader events per Run; zero = unbounded (run until
// ctx cancellation). Used by the worker scheduler to bound a single
// READ task.
EventBudget int
// EventBuffer sets the channel capacity between reader and router
// goroutines. Zero = defaultEventBuffer.
EventBuffer int
// events is the input channel for the dispatch goroutine. The reader
// is the primary writer; InjectEvent allows external code (the bucket
// bootstrapper) to push synthesized events through the same router
// path. Initialized lazily by InjectEvent and Run; ready signals
// readiness to InjectEvent callers that arrive before Run.
eventsOnce sync.Once
events chan *reader.Event
eventsReady chan struct{}
}
// ensureEventsChan lazily creates the events channel and the readiness
// signal so InjectEvent works whether it's called before or after Run.
func (p *Pipeline) ensureEventsChan() {
p.eventsOnce.Do(func() {
bufSize := p.EventBuffer
if bufSize <= 0 {
bufSize = defaultEventBuffer
}
p.events = make(chan *reader.Event, bufSize)
p.eventsReady = make(chan struct{})
close(p.eventsReady)
})
}
// InjectEvent pushes a synthesized event onto the same input the reader
// feeds. Used by the bucket bootstrapper to backfill pre-rule entries:
// each entry becomes one *reader.Event, flows through router.Route, and
// schedules a Match with DueTime=mtime+delay. Set TsNs=0 so the cursor
// doesn't advance — the reader still resumes from its persisted position
// on restart.
func (p *Pipeline) InjectEvent(ctx context.Context, ev *reader.Event) error {
p.ensureEventsChan()
select {
case <-ctx.Done():
return ctx.Err()
case p.events <- ev:
return nil
}
}
// Tick defaults live in dispatch_ticks_*.go so the s3tests build can shrink
// them without touching production timings. defaultDispatchTick and
// defaultCheckpointTick are the only knobs that change per build tag.
const (
defaultEventBuffer = 1024
shutdownDrainTimeout = 30 * time.Second
shutdownSaveTimeout = 5 * time.Second
)
// shardState bundles per-shard mutable state so the single dispatch
// goroutine can route an event to the right cursor + schedule by lookup.
type shardState struct {
cursor *reader.Cursor
dispatch *Dispatcher
}
// Run blocks until ctx is canceled or a fatal error occurs. On exit, every
// shard's cursor is persisted; in-flight schedule entries are dropped
// (the meta-log is the durable buffer, so a restart re-derives them).
func (p *Pipeline) Run(ctx context.Context) error {
if p.Engine == nil || p.Persister == nil || p.Client == nil || p.FilerClient == nil {
return errors.New("pipeline: missing required dependency")
}
if p.BucketsPath == "" {
return errors.New("pipeline: BucketsPath required")
}
// Resolve the active shard set. Single-shard configurations populate
// either Shards=[N] or Shards=nil with ShardID=N (latter is the legacy
// path that also supplies a Cursor); both feed the same range model.
shardIDs := p.Shards
if len(shardIDs) == 0 {
shardIDs = []int{p.ShardID}
}
shardSet := make(map[int]struct{}, len(shardIDs))
for _, s := range shardIDs {
if s < 0 || s >= s3lifecycle.ShardCount {
return fmt.Errorf("pipeline: shard %d out of [0,%d)", s, s3lifecycle.ShardCount)
}
shardSet[s] = struct{}{}
}
// Per-shard cursor + dispatcher. Cursors restore from the durable
// store; freezes re-arm naturally when the reader re-encounters the
// poison event at MinTsNs and the dispatch state machine drives it
// back to BLOCKED.
states := make(map[int]*shardState, len(shardIDs))
var minStartTsNs int64 = -1
for _, shardID := range shardIDs {
c := p.Cursor
if len(shardIDs) != 1 || c == nil {
c = reader.NewCursor()
}
state, err := p.Persister.Load(ctx, shardID)
if err != nil {
return fmt.Errorf("cursor load shard=%d: %w", shardID, err)
}
c.Restore(state)
states[shardID] = &shardState{
cursor: c,
dispatch: &Dispatcher{
ShardID: shardID,
Client: p.Client,
Cursor: c,
Schedule: router.NewSchedule(),
},
}
if mt := c.MinTsNs(); mt > 0 && (minStartTsNs < 0 || mt < minStartTsNs) {
minStartTsNs = mt
}
}
if minStartTsNs < 0 {
minStartTsNs = 0
}
p.ensureEventsChan()
events := p.events
rd := &reader.Reader{
BucketsPath: p.BucketsPath,
ShardPredicate: func(s int) bool {
_, ok := shardSet[s]
return ok
},
StartTsNs: minStartTsNs,
Events: events,
EventBudget: p.EventBudget,
}
rd.LogStartup()
runCtx, cancel := context.WithCancel(ctx)
defer cancel()
var wg sync.WaitGroup
var readerErr error
// Reader goroutine. Doesn't close the events channel because external
// callers (BucketBootstrapper) also write to it; the dispatcher exits
// on runCtx cancellation rather than channel close.
wg.Add(1)
go func() {
defer wg.Done()
readerErr = rd.Run(runCtx, p.FilerClient, p.ClientName, p.ClientID)
if readerErr != nil && !isCtxShutdown(readerErr) {
glog.Errorf("lifecycle reader: shards=%v: %v", shardIDs, readerErr)
}
cancel() // wake the dispatcher goroutine to drain & exit
}()
lister := &filerSiblingLister{client: p.FilerClient, bucketsPath: p.BucketsPath}
// Router/dispatcher goroutine: pulls events, routes them to per-shard
// schedules, ticks every shard's dispatcher on the same cadence, and
// checkpoints every shard's cursor on the checkpoint cadence. One
// goroutine handles all shards — there is no fan-out per shard.
wg.Add(1)
go func() {
defer wg.Done()
dispatchTick := p.DispatchTick
if dispatchTick <= 0 {
dispatchTick = defaultDispatchTick
}
checkpointTick := p.CheckpointTick
if checkpointTick <= 0 {
checkpointTick = defaultCheckpointTick
}
dt := time.NewTicker(dispatchTick)
defer dt.Stop()
ct := time.NewTicker(checkpointTick)
defer ct.Stop()
drainAll := func() {
drainCtx, drainCancel := context.WithTimeout(context.Background(), shutdownDrainTimeout)
defer drainCancel()
now := time.Now()
for _, st := range states {
st.dispatch.Tick(drainCtx, now)
}
}
for {
select {
case <-runCtx.Done():
drainAll()
return
case ev, ok := <-events:
if !ok {
drainAll()
return
}
st := states[ev.ShardID]
if st == nil {
continue
}
stats.S3LifecycleEventCounter.WithLabelValues(strconv.Itoa(ev.ShardID)).Inc()
// Always re-fetch the snapshot — caching it across events
// means an event arriving between dispatch ticks routes
// against a stale snap. With bootstrap injection, events
// often land within the same dispatch interval as the
// engine.Compile that introduced their bucket; routing
// against the prior (empty) snapshot would silently drop
// every match. Engine.Snapshot is an atomic Load.
snap := p.Engine.Snapshot()
for _, m := range router.Route(runCtx, snap, ev, time.Now(), lister) {
st.dispatch.Schedule.Add(m)
}
case <-dt.C:
now := time.Now()
for _, st := range states {
st.dispatch.Tick(runCtx, now)
st.dispatch.observeScheduleDepth()
}
case <-ct.C:
for shardID, st := range states {
if err := p.Persister.Save(runCtx, shardID, st.cursor.Snapshot()); err != nil {
glog.Warningf("lifecycle cursor checkpoint: shard=%d: %v", shardID, err)
}
stats.S3LifecycleCursorMinTsNs.WithLabelValues(strconv.Itoa(shardID)).Set(float64(st.cursor.MinTsNs()))
}
}
}
}()
wg.Wait()
// Final cursor checkpoint on graceful shutdown.
for shardID, st := range states {
saveCtx, saveCancel := context.WithTimeout(context.Background(), shutdownSaveTimeout)
err := p.Persister.Save(saveCtx, shardID, st.cursor.Snapshot())
saveCancel()
if err != nil {
glog.Warningf("lifecycle cursor final save: shard=%d: %v", shardID, err)
}
}
if readerErr != nil && !isCtxShutdown(readerErr) {
return readerErr
}
return nil
}
// isCtxShutdown reports whether err is a graceful ctx-driven shutdown
// (Canceled or DeadlineExceeded), including the gRPC status forms that
// don't unwrap to the std-lib ctx errors.
func isCtxShutdown(err error) bool {
if err == nil {
return false
}
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return true
}
if code := status.Code(err); code == codes.Canceled || code == codes.DeadlineExceeded {
return true
}
return false
}
@@ -1,189 +0,0 @@
package dispatcher
import (
"context"
"errors"
"strconv"
"sync"
"testing"
"time"
"github.com/prometheus/client_golang/prometheus/testutil"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// Direct tests for dispatcher / pipeline pure helpers and their side
// effects on the Prometheus surface. The Tick + Run integration tests
// drive these only as one slice of a larger flow; pinning each helper
// individually makes a regression in the helper itself fail at the
// helper level.
// ---------- observeScheduleDepth ----------
func TestObserveScheduleDepth_ReportsScheduleLen(t *testing.T) {
// observeScheduleDepth sets the gauge to the current Schedule.Len.
// The dispatcher calls it on every Tick so operators can see how
// far behind schedule the worker is. Use a distinct shard label so
// other tests sharing the global gauge don't bleed in.
const shard = 137
d := &Dispatcher{ShardID: shard, Schedule: router.NewSchedule()}
g := stats_collect.S3LifecycleScheduleDepthGauge.WithLabelValues(strconv.Itoa(shard))
// Drop the label series at end of test so the global registry
// doesn't accumulate stale entries across test runs in the same
// process.
defer stats_collect.S3LifecycleScheduleDepthGauge.DeleteLabelValues(strconv.Itoa(shard))
// Empty schedule -> 0.
d.observeScheduleDepth()
assert.Equal(t, float64(0), testutil.ToFloat64(g))
// Adding two matches lifts the gauge to 2 on the next observe.
t0 := time.Now()
d.Schedule.Add(router.Match{DueTime: t0})
d.Schedule.Add(router.Match{DueTime: t0.Add(time.Second)})
d.observeScheduleDepth()
assert.Equal(t, float64(2), testutil.ToFloat64(g))
// Draining drops it back to 0; Tick would re-call but observe is a
// pure setter so we drive it manually here.
_ = d.Schedule.Drain(t0.Add(2 * time.Second))
d.observeScheduleDepth()
assert.Equal(t, float64(0), testutil.ToFloat64(g))
}
// ---------- ensureEventsChan ----------
func TestEnsureEventsChan_HonorsEventBuffer(t *testing.T) {
p := &Pipeline{EventBuffer: 7}
p.ensureEventsChan()
require.NotNil(t, p.events)
assert.Equal(t, 7, cap(p.events), "EventBuffer must size the channel")
require.NotNil(t, p.eventsReady)
}
func TestEnsureEventsChan_FallsBackToDefault(t *testing.T) {
// EventBuffer=0 means "use the default"; non-positive must not
// produce a zero-sized channel that would block every InjectEvent.
p := &Pipeline{}
p.ensureEventsChan()
assert.Equal(t, defaultEventBuffer, cap(p.events))
}
func TestEnsureEventsChan_NegativeFallsBackToDefault(t *testing.T) {
p := &Pipeline{EventBuffer: -1}
p.ensureEventsChan()
assert.Equal(t, defaultEventBuffer, cap(p.events))
}
func TestEnsureEventsChan_IdempotentAcrossConcurrentCallers(t *testing.T) {
// sync.Once must guarantee a single channel allocation under
// concurrent calls; otherwise a races would create two channels
// and lose events on the discarded one.
p := &Pipeline{EventBuffer: 4}
const N = 32
var wg sync.WaitGroup
wg.Add(N)
for i := 0; i < N; i++ {
go func() {
defer wg.Done()
p.ensureEventsChan()
}()
}
wg.Wait()
require.NotNil(t, p.events)
assert.Equal(t, 4, cap(p.events))
}
// ---------- InjectEvent ----------
func TestInjectEvent_DeliversToEventsChannel(t *testing.T) {
p := &Pipeline{EventBuffer: 1}
ev := &reader.Event{Bucket: "bk", Key: "k"}
require.NoError(t, p.InjectEvent(context.Background(), ev))
select {
case got := <-p.events:
assert.Same(t, ev, got, "InjectEvent must deliver the same pointer")
default:
t.Fatal("event was not enqueued")
}
}
func TestInjectEvent_ReturnsCtxErrWhenCanceled(t *testing.T) {
// A canceled context must propagate the cancellation; otherwise
// the bootstrap walker could keep injecting after the worker is
// shutting down. ensureEventsChan clamps EventBuffer<=0 to the
// default, so to force the select to actually block we pre-fill
// the buffer (size 1) and then send into the canceled context.
p := &Pipeline{EventBuffer: 1}
require.NoError(t, p.InjectEvent(context.Background(), &reader.Event{}))
ctx, cancel := context.WithCancel(context.Background())
cancel()
err := p.InjectEvent(ctx, &reader.Event{})
require.Error(t, err)
assert.ErrorIs(t, err, context.Canceled)
}
func TestInjectEvent_EnsureEventsChanCalledIfFirstCall(t *testing.T) {
// InjectEvent must call ensureEventsChan internally so a caller can
// inject before Run starts (e.g. bootstrap walks that fire before
// the reader is up). Pin that the channel exists post-call.
p := &Pipeline{}
require.Nil(t, p.events, "preconditions: events not yet allocated")
require.NoError(t, p.InjectEvent(context.Background(), &reader.Event{}))
require.NotNil(t, p.events, "InjectEvent must allocate the channel via ensureEventsChan")
}
// ---------- isCtxShutdown ----------
func TestIsCtxShutdown_NilIsNotShutdown(t *testing.T) {
assert.False(t, isCtxShutdown(nil))
}
func TestIsCtxShutdown_ContextCanceledIsShutdown(t *testing.T) {
assert.True(t, isCtxShutdown(context.Canceled))
assert.True(t, isCtxShutdown(context.DeadlineExceeded))
}
func TestIsCtxShutdown_GRPCCanceledStatusIsShutdown(t *testing.T) {
// Filer / S3 RPCs can come back wrapped as gRPC status codes; the
// helper must recognize Canceled / DeadlineExceeded by code so a
// shutdown doesn't get misclassified as a transport failure.
assert.True(t, isCtxShutdown(status.Error(codes.Canceled, "client closed")))
assert.True(t, isCtxShutdown(status.Error(codes.DeadlineExceeded, "stream timeout")))
}
func TestIsCtxShutdown_OtherErrorsAreNotShutdown(t *testing.T) {
// A garden-variety transport error must NOT classify as shutdown;
// otherwise the worker would silently swallow real failures.
assert.False(t, isCtxShutdown(errors.New("connection refused")))
assert.False(t, isCtxShutdown(status.Error(codes.Unavailable, "filer down")))
assert.False(t, isCtxShutdown(status.Error(codes.Internal, "boom")))
}
// ---------- cursorFileName ----------
func TestCursorFileName_PadsShardIDToTwoDigits(t *testing.T) {
// The persister stores per-shard cursor files at predictable paths
// so a manual operator inspection finds them sorted by shard. Pin
// the zero-padded form so a refactor that drops %02d doesn't break
// existing on-disk filenames.
cases := map[int]string{
0: "shard-00.json",
1: "shard-01.json",
9: "shard-09.json",
10: "shard-10.json",
15: "shard-15.json",
}
for shard, want := range cases {
t.Run(want, func(t *testing.T) {
assert.Equal(t, want, cursorFileName(shard))
})
}
}
@@ -1,141 +0,0 @@
package dispatcher
import (
"context"
"strings"
"testing"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
)
// TestPipelineIntegrationInMemory exercises router + dispatcher end-to-end
// without the Reader (which requires a live filer client). The Reader's
// behavior is covered separately in the reader package; this test pins the
// composition: an event flows through Route -> Schedule -> Dispatcher and
// the cursor advances on a successful RPC.
func TestPipelineIntegrationInMemory(t *testing.T) {
rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
hash := s3lifecycle.RuleHash(rule)
prior := map[s3lifecycle.ActionKey]engine.PriorState{}
for _, k := range s3lifecycle.RuleActionKinds(rule) {
prior[s3lifecycle.ActionKey{Bucket: "bk", RuleHash: hash, ActionKind: k}] = engine.PriorState{
BootstrapComplete: true,
Mode: engine.ModeEventDriven,
}
}
e := engine.New()
snap := e.Compile([]engine.CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{rule}}},
engine.CompileOptions{PriorStates: prior})
now := time.Now()
old := now.Add(-48 * time.Hour)
ev := &reader.Event{
TsNs: old.UnixNano(),
Bucket: "bk",
Key: "obj.txt",
NewEntry: &filer_pb.Entry{
Name: "obj.txt",
Attributes: &filer_pb.FuseAttributes{
Mtime: old.Unix(),
FileSize: 1,
},
},
}
matches := router.Route(context.Background(), snap, ev, now, nil)
if len(matches) != 1 {
t.Fatalf("expected 1 match, got %v", matches)
}
client := &fakeClient{
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return &s3_lifecycle_pb.LifecycleDeleteResponse{
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
}, nil
},
}
d, sched := newDispatcher(client)
d.ShardID = s3lifecycle.ShardID("bk", "obj.txt")
for _, m := range matches {
sched.Add(m)
}
processed := d.Tick(context.Background(), now.Add(time.Hour)) // past DueTime
if processed != 1 {
t.Fatalf("Tick processed=%d, want 1", processed)
}
if d.Cursor.Get(matches[0].Key) != ev.TsNs {
t.Fatalf("cursor not at event TsNs: %d", d.Cursor.Get(matches[0].Key))
}
}
func TestPipelineRunRequiresDependencies(t *testing.T) {
p := &Pipeline{}
err := p.Run(context.Background())
if err == nil {
t.Fatal("expected error for empty Pipeline")
}
}
// stubFilerClient satisfies filer_pb.SeaweedFilerClient just enough to
// pass the nil-check in Pipeline.Run; methods would panic if called,
// but the validation tests below all return before any RPC.
type stubFilerClient struct {
filer_pb.SeaweedFilerClient
}
// fullPipeline assembles a Pipeline whose dependencies all pass the
// nil-check, so individual tests can knock out one piece at a time
// to exercise specific validation branches.
func fullPipeline() *Pipeline {
return &Pipeline{
Engine: engine.New(),
Persister: reader.NewInMemoryPersister(),
Client: &fakeClient{},
FilerClient: &stubFilerClient{},
BucketsPath: "/buckets",
ShardID: 0,
}
}
func TestPipelineRunValidation(t *testing.T) {
// Each case mutates one piece of a fullPipeline() and asserts the
// expected error fragment. Per-dependency cases pin that the nil
// check exercises every required field individually; the Buckets-
// Path case asserts the distinct error message; the shard cases
// pin the half-open [0, ShardCount) range and that any one bad
// entry refuses the whole multi-shard run.
cases := []struct {
name string
mutate func(*Pipeline)
wantErr string
}{
{"missing Engine", func(p *Pipeline) { p.Engine = nil }, "missing required dependency"},
{"missing Persister", func(p *Pipeline) { p.Persister = nil }, "missing required dependency"},
{"missing Client", func(p *Pipeline) { p.Client = nil }, "missing required dependency"},
{"missing FilerClient", func(p *Pipeline) { p.FilerClient = nil }, "missing required dependency"},
{"missing BucketsPath", func(p *Pipeline) { p.BucketsPath = "" }, "BucketsPath required"},
{"negative ShardID", func(p *Pipeline) { p.ShardID = -1 }, "out of"},
{"ShardID at boundary", func(p *Pipeline) { p.ShardID = s3lifecycle.ShardCount }, "out of"},
{"multi-shard out of range", func(p *Pipeline) {
p.Shards = []int{0, 1, s3lifecycle.ShardCount + 1}
}, "out of"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
p := fullPipeline()
tc.mutate(p)
err := p.Run(context.Background())
if err == nil || !strings.Contains(err.Error(), tc.wantErr) {
t.Fatalf("expected error containing %q, got %v", tc.wantErr, err)
}
})
}
}
@@ -1,85 +0,0 @@
package dispatcher
import (
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
"github.com/stretchr/testify/assert"
)
// toProtoActionKind / toProtoIdentity are the worker-side CAS-witness
// converters; getting them wrong silently breaks the LifecycleDelete
// flow because the server's identity check always fails. These tests
// pin the full mapping so a new ActionKind added in s3lifecycle but
// forgotten here can't slip through as ACTION_KIND_UNSPECIFIED.
func TestToProtoActionKind_AllKnownKindsMap(t *testing.T) {
cases := []struct {
in s3lifecycle.ActionKind
want s3_lifecycle_pb.ActionKind
}{
{s3lifecycle.ActionKindExpirationDays, s3_lifecycle_pb.ActionKind_EXPIRATION_DAYS},
{s3lifecycle.ActionKindExpirationDate, s3_lifecycle_pb.ActionKind_EXPIRATION_DATE},
{s3lifecycle.ActionKindNoncurrentDays, s3_lifecycle_pb.ActionKind_NONCURRENT_DAYS},
{s3lifecycle.ActionKindNewerNoncurrent, s3_lifecycle_pb.ActionKind_NEWER_NONCURRENT},
{s3lifecycle.ActionKindAbortMPU, s3_lifecycle_pb.ActionKind_ABORT_MPU},
{s3lifecycle.ActionKindExpiredDeleteMarker, s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER},
}
for _, c := range cases {
t.Run(c.in.String(), func(t *testing.T) {
assert.Equal(t, c.want, toProtoActionKind(c.in))
})
}
}
func TestToProtoActionKind_UnspecifiedAndUnknownFallToUnspecified(t *testing.T) {
// Both the in-package zero value and a future kind not listed in
// the switch must collapse to ACTION_KIND_UNSPECIFIED so the server
// can reject with FATAL rather than silently dispatch a wrong kind.
assert.Equal(t, s3_lifecycle_pb.ActionKind_ACTION_KIND_UNSPECIFIED,
toProtoActionKind(s3lifecycle.ActionKindUnspecified))
assert.Equal(t, s3_lifecycle_pb.ActionKind_ACTION_KIND_UNSPECIFIED,
toProtoActionKind(s3lifecycle.ActionKind(999)))
}
func TestToProtoIdentity_NilReturnsNil(t *testing.T) {
// LifecycleDelete treats a nil ExpectedIdentity as "no CAS"; the
// converter must preserve that signal rather than emit an empty
// non-nil identity that would force a re-fetch + comparison.
assert.Nil(t, toProtoIdentity(nil))
}
func TestToProtoIdentity_AllFieldsCopied(t *testing.T) {
in := &router.EntryIdentity{
MtimeNs: 1700000000_000_000_123,
Size: 4096,
HeadFid: "1,abc",
ExtendedHash: []byte{0xde, 0xad, 0xbe, 0xef},
}
out := toProtoIdentity(in)
if out == nil {
t.Fatal("non-nil input must produce non-nil output")
}
assert.Equal(t, in.MtimeNs, out.MtimeNs)
assert.Equal(t, in.Size, out.Size)
assert.Equal(t, in.HeadFid, out.HeadFid)
assert.Equal(t, in.ExtendedHash, out.ExtendedHash)
}
func TestToProtoIdentity_EmptyFieldsSurvive(t *testing.T) {
// A bootstrap-fresh entry can land at the dispatcher with zero
// MtimeNs / Size / no HeadFid — the converter must still produce a
// non-nil identity so the server treats it as a real CAS witness
// rather than the no-CAS sentinel.
in := &router.EntryIdentity{}
out := toProtoIdentity(in)
if out == nil {
t.Fatal("zero-valued identity must still produce non-nil")
}
assert.Zero(t, out.MtimeNs)
assert.Zero(t, out.Size)
assert.Empty(t, out.HeadFid)
assert.Empty(t, out.ExtendedHash)
}
@@ -10,11 +10,11 @@ import (
"github.com/stretchr/testify/require"
)
// Layer 2 contracts the dispatcher pipeline relies on. These augment
// Layer 2 contracts the daily-replay drain relies on. These augment
// reader_test.go (extract + dispatch) and cursor_test.go (Advance +
// Freeze) by pinning the cursor composition surface that
// dispatcher.Pipeline depends on for resume-point selection,
// checkpoint persistence, and Run-time input validation.
// Freeze) by pinning the cursor composition surface that the daily-run
// drain depends on for resume-point selection, checkpoint persistence,
// and Run-time input validation.
func TestCursorMinTsNsWithFrozenKeysIncluded(t *testing.T) {
// MinTsNs is the resume point Reader.Run feeds to SubscribeMetadata
@@ -1,487 +0,0 @@
package scheduler
import (
"context"
"fmt"
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
"github.com/seaweedfs/seaweedfs/weed/util"
)
// EventInjector is the bootstrap-side hook into the dispatcher pipeline.
// One implementation routes events to the right per-shard pipeline; the
// shell's single-pipeline path passes pipeline.InjectEvent directly.
type EventInjector interface {
InjectEvent(ctx context.Context, ev *reader.Event) error
}
// listPageSize is the page size for paginated directory listings during
// the bucket walk. The filer caps SeaweedList(..., limit=0) at
// DirListingLimit (1000 by default) per call, so a single-page list
// would silently truncate large directories — a correctness bug for
// noncurrent retention since older versions past the page boundary
// would never reach the rank/sort math. Atomic so tests can shrink it
// without racing the async bootstrap goroutines other tests leave
// behind (KickOffNew dispatches walks via `go b.walkBucket(...)`,
// and a fresh test's Cleanup might land before those goroutines exit).
var listPageSize atomic.Uint32
func init() {
listPageSize.Store(1024)
}
// listAll issues paginated SeaweedList calls until the listing is
// exhausted, invoking fn for every entry. Pagination uses
// startFrom = lastEntryName (exclusive) to advance.
func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, fn func(*filer_pb.Entry) error) error {
pageSize := listPageSize.Load()
startFrom := ""
for {
var pageCount uint32
var lastName string
if err := filer_pb.SeaweedList(ctx, client, dir, "", func(e *filer_pb.Entry, _ bool) error {
pageCount++
if e != nil {
lastName = e.Name
}
return fn(e)
}, startFrom, false, pageSize); err != nil {
return err
}
if pageCount < pageSize {
return nil
}
startFrom = lastName
}
}
// BucketBootstrapper backfills already-existing entries when a freshly-PUT
// rule's bucket appears in the engine. The reader-driven path only sees
// meta-log events created after the rule lands; without this walk,
// objects PUT before the rule would never expire.
//
// Per bucket: one one-shot goroutine that lists every entry under
// /buckets/<bucket> and synthesizes a *reader.Event for each one. The
// pipeline's existing router.Route + Schedule machinery handles the rest:
// currently-due matches fire on the next dispatch tick, and not-yet-due
// matches sit in the per-shard schedule until their DueTime arrives.
//
// Synthesized events carry TsNs=0 so dispatcher.advance is a no-op for
// them — the reader still resumes from its persisted cursor on restart.
//
// 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
// 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
// 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.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.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
}
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()
for _, bucket := range fresh {
bucket := bucket
go b.walkBucket(ctx, bucket)
}
}
func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) {
root := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket
glog.V(0).Infof("lifecycle bootstrap: starting walk for bucket %s (root=%s)", bucket, root)
count := 0
// skipBare records bucket-relative bare-key paths that
// expandVersionsDir already routed as the null version. Without it
// the walker's regular emission would also fire for the bare entry
// — in a versioned bucket buildObjectInfo classifies it as
// IsLatest=true, NumVersions=0, and ExpirationDays would create a
// stray delete marker that hides the real latest.
skipBare := map[string]bool{}
var cb func(entry *filer_pb.Entry, key string) error
cb = func(entry *filer_pb.Entry, key string) error {
if isVersionsDir(entry) {
n, err := b.expandVersionsDir(ctx, bucket, root, key, entry, cb, skipBare)
count += n
return err
}
if !entry.IsDirectory && skipBare[key] {
return nil
}
if entry.IsDirectoryKeyObject() && skipBare[key] {
return nil
}
ev := &reader.Event{
// TsNs=0 sentinel: dispatcher.advance treats <=0 as no-op,
// so the reader's persisted cursor isn't ratcheted forward
// past meta-log events that haven't been processed yet.
TsNs: 0,
Bucket: bucket,
Key: key,
ShardID: s3lifecycle.ShardID(bucket, key),
NewEntry: entry,
}
count++
return b.Injector.InjectEvent(ctx, ev)
}
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, walkErr)
}
return
}
glog.V(0).Infof("lifecycle bootstrap: bucket %s injected %d entries", bucket, count)
}
// versionItem is the per-sibling state expandVersionsDir builds: the
// filer entry plus its version_id (or "null" for the bare null version
// living outside .versions/). isExplicitNull marks bare entries the
// suspended-versioning write path tagged with ExtVersionIdKey="null"
// (s3api_object_handlers_put.go); we trust those as latest when the
// .versions/ pointer is missing. A pre-versioning bare object has no
// such marker, so a missing pointer there is a race window with a new
// version write and we keep the newest-sibling fallback.
type versionItem struct {
entry *filer_pb.Entry
versionID string
bareKey string // bucket-relative path; non-empty only for the null version
isExplicitNull bool
}
// expandVersionsDir lists <root>/<key>/ and, when the children look like
// SeaweedFS version files, injects one reader.Event per version with
// BootstrapVersion populated. The bare logical key (the "null" version,
// living outside .versions/) is included as a sibling so:
// - pre-versioning objects with a newer .versions/ history fire
// NoncurrentDays as id="null"
// - suspended-bucket writes (which clear the .versions/ latest pointer)
// correctly classify null as the current version while every
// .versions/ child becomes noncurrent
//
// versionsKey is the bucket-relative path of the .versions/ directory
// (e.g. "logs/foo.versions"). When the bare null version is included,
// its bucket-relative path is added to skipBare so the walker's regular
// emission for the same entry is suppressed.
//
// When no child has ExtVersionIdKey the directory is a coincidentally-
// named user folder; recurse via fallback (the bucket walk's own cb).
func (b *BucketBootstrapper) expandVersionsDir(ctx context.Context, bucket, root, versionsKey string, versionsEntry *filer_pb.Entry, fallback func(*filer_pb.Entry, string) error, skipBare map[string]bool) (int, error) {
logical := strings.TrimSuffix(versionsKey, s3_constants.VersionsFolder)
if logical == "" {
return 0, nil
}
versionsDir := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket + "/" + versionsKey
// Collect file children only. Subdirectories under .versions/ would
// corrupt sort/rank math; the disambiguation pass below also wants
// to see only file-shaped children. Paginate so a hot key with
// thousands of versions doesn't truncate at DirListingLimit.
var children []*filer_pb.Entry
if err := listAll(ctx, b.FilerClient, versionsDir, func(e *filer_pb.Entry) error {
if e != nil && e.Attributes != nil && !e.IsDirectory {
children = append(children, e)
}
return nil
}); err != nil {
return 0, fmt.Errorf("list %s: %w", versionsDir, err)
}
items := make([]versionItem, 0, len(children)+1)
for _, e := range children {
if id, ok := e.Extended[s3_constants.ExtVersionIdKey]; ok && len(id) > 0 {
items = append(items, versionItem{entry: e, versionID: string(id)})
}
}
if len(items) == 0 {
// Coincidentally-named user folder (or an empty .versions
// container). fallback is the bucket walk's own cb so nested
// .versions/ entries inside still expand.
if fallback == nil {
return 0, nil
}
if err := walkBucketDir(ctx, b.FilerClient, versionsDir, root, fallback); err != nil {
return 0, err
}
return 0, nil
}
// Look up the bare null version. SeaweedFS keeps it at the logical
// path for pre-versioning objects and for suspended-bucket writes.
// Both shapes count: regular file (PUT'd object) and explicit S3
// directory-key marker (object name ends in /).
if nullEntry, nullKey, explicit, ok := b.lookupNullVersion(ctx, bucket, logical); ok {
items = append(items, versionItem{
entry: nullEntry,
versionID: "null",
bareKey: nullKey,
isExplicitNull: explicit,
})
}
// Sort newest-first: primary by mtime ns, fallback by version_id
// (CompareVersionIds returns <0 when first arg is newer). PUTs only
// set second-level Mtime, so collisions in the same second are
// resolved by the canonical version-id ordering used elsewhere.
sort.SliceStable(items, func(i, j int) bool {
mi := items[i].entry.Attributes.Mtime*int64(1e9) + int64(items[i].entry.Attributes.MtimeNs)
mj := items[j].entry.Attributes.Mtime*int64(1e9) + int64(items[j].entry.Attributes.MtimeNs)
if mi != mj {
return mi > mj
}
return s3lifecycle.CompareVersionIds(items[i].versionID, items[j].versionID) < 0
})
// Resolve latest position.
// 1. Pointer names a real id -> that wins (in-order or backdated).
// 2. Pointer absent + items[0] is an EXPLICIT null (suspended write
// cleared the pointer and tagged the bare object as null, AND
// the bare object is newest by mtime) -> null is latest.
// 3. Pointer absent in any other shape: fall back to newest
// sibling. Catches the post-suspended re-enable race window —
// a fresh .versions/<v1> write whose pointer update hasn't
// landed yet outranks the older suspended-null bare object.
latestID := string(versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey])
latestPos := 0
if latestID != "" {
for i, it := range items {
if it.versionID == latestID {
latestPos = i
break
}
}
} else if len(items) > 0 && items[0].versionID == "null" && items[0].isExplicitNull {
latestPos = 0
}
count := 0
for i, it := range items {
// Prefer the explicit demotion stamp written by the S3 PUT
// handler. Falling back to the next-newer sibling's mtime is
// the legacy derivation and stays in place for entries written
// before the stamp was introduced.
successor := s3lifecycle.SuccessorFromEntryStamp(it.entry)
if successor.IsZero() && i > 0 {
prev := items[i-1].entry.Attributes
successor = time.Unix(prev.Mtime, int64(prev.MtimeNs))
}
bv := &reader.BootstrapVersion{
LogicalKey: logical,
VersionID: it.versionID,
IsLatest: i == latestPos,
IsDeleteMarker: string(it.entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
NumVersions: len(items),
SuccessorModTime: successor,
}
if !bv.IsLatest {
rank := i
if i > latestPos {
rank = i - 1
}
bv.NoncurrentIndex = rank
}
// Event Key for bookkeeping: real version files keep the
// .versions/<file> path; the null version uses its bare path
// so the dispatcher's identity check resolves to the same
// entry the walker would have emitted.
evKey := versionsKey + "/" + it.entry.Name
if it.versionID == "null" {
evKey = it.bareKey
}
ev := &reader.Event{
TsNs: 0,
Bucket: bucket,
Key: evKey,
ShardID: s3lifecycle.ShardID(bucket, logical),
NewEntry: it.entry,
BootstrapVersion: bv,
}
if err := b.Injector.InjectEvent(ctx, ev); err != nil {
return count, err
}
if it.versionID == "null" && skipBare != nil {
skipBare[it.bareKey] = true
}
count++
}
return count, nil
}
// lookupNullVersion returns the bare-key entry that represents the null
// version of logical, if any. Both regular files and S3 directory-key
// markers (an empty directory entry with Mime set) qualify. The
// explicit return reports whether the entry's Extended map carries
// ExtVersionIdKey == "null" — the marker the suspended-versioning
// write path applies (s3api_object_handlers_put.go). bucketRelKey is
// the bucket-relative path the walker would otherwise emit, so the
// caller can suppress the duplicate.
func (b *BucketBootstrapper) lookupNullVersion(ctx context.Context, bucket, logical string) (entry *filer_pb.Entry, bucketRelKey string, explicit bool, ok bool) {
bucketPath := strings.TrimSuffix(b.BucketsPath, "/") + "/" + bucket
parent, name := util.NewFullPath(bucketPath, logical).DirAndName()
resp, err := filer_pb.LookupEntry(ctx, b.FilerClient, &filer_pb.LookupDirectoryEntryRequest{
Directory: parent,
Name: name,
})
if err != nil || resp == nil || resp.Entry == nil {
return nil, "", false, false
}
e := resp.Entry
if e.IsDirectory && !e.IsDirectoryKeyObject() {
return nil, "", false, false
}
if id, hasID := e.Extended[s3_constants.ExtVersionIdKey]; hasID && string(id) == "null" {
explicit = true
}
return e, strings.TrimPrefix(parent+"/"+name, bucketPath+"/"), explicit, true
}
// walkBucketDir streams entries under dir and invokes cb. Two kinds of
// directories are emitted whole rather than recursed into:
// - .uploads/<id> MPU init dirs (router fires ABORT_MPU off the dir entry)
// - <key>.versions/ directories (caller expands them into per-version
// events; recursing here would emit individual version files without
// the sibling state needed for NoncurrentDays / NewerNoncurrent)
//
// .versions/ dirs are processed before everything else at each level so
// the cb's expandVersionsDir call can record the bare null-version key
// in the walk-shared skip set before the same level emits the bare entry.
// Two streaming passes (rather than buffering the whole directory) trade
// a second listing for bounded memory on flat buckets with millions of
// entries.
func walkBucketDir(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot string, cb func(entry *filer_pb.Entry, key string) error) error {
// Pass 1: .versions/ dirs only.
if err := listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
if e == nil || e.Attributes == nil {
return nil
}
if !e.IsDirectory || !isVersionsDir(e) {
return nil
}
full := dir + "/" + e.Name
key := strings.TrimPrefix(full, bucketRoot+"/")
return cb(e, key)
}); err != nil {
return fmt.Errorf("list %s: %w", dir, err)
}
// Pass 2: everything else. Bare entries whose name was claimed by
// a sibling .versions/ expansion are dropped by the cb's skip-set.
return listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
if e == nil || e.Attributes == nil {
return nil
}
if e.IsDirectory && isVersionsDir(e) {
return nil
}
full := dir + "/" + e.Name
key := strings.TrimPrefix(full, bucketRoot+"/")
if e.IsDirectory {
if isMPUInitDir(key, e) {
return cb(e, key)
}
return walkBucketDir(ctx, client, full, bucketRoot, cb)
}
return cb(e, key)
})
}
// isMPUInitDir mirrors router.mpuInitInfo: a directory at .uploads/<id>
// carrying the destination key in Extended is the MPU init record. The
// router helper is package-private so this is duplicated rather than
// adding a public extraction API just for this caller.
func isMPUInitDir(key string, entry *filer_pb.Entry) bool {
uploadsPrefix := s3_constants.MultipartUploadsFolder + "/"
if !strings.HasPrefix(key, uploadsPrefix) {
return false
}
rest := key[len(uploadsPrefix):]
if rest == "" || strings.ContainsRune(rest, '/') {
return false
}
v, ok := entry.Extended[s3_constants.ExtMultipartObjectKey]
return ok && len(v) > 0
}
// isVersionsDir matches `<x>.versions/` by name suffix. We can't gate on
// ExtLatestVersionIdKey here: createDeleteMarker writes the version file
// before updating the parent's Extended pointer, so a walk that races
// with that update would see the directory without the pointer and
// recurse into raw version files, losing the sibling state needed for
// noncurrent rules. expandVersionsDir handles disambiguation by
// inspecting children for ExtVersionIdKey; coincidentally-named
// directories that aren't real .versions storage fall through to a
// regular recursion.
func isVersionsDir(entry *filer_pb.Entry) bool {
return entry.IsDirectory && strings.HasSuffix(entry.Name, s3_constants.VersionsFolder)
}
File diff suppressed because it is too large Load Diff
@@ -1,77 +0,0 @@
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}))
}
@@ -1,7 +0,0 @@
//go:build !s3tests
package scheduler
import "time"
const defaultRefreshInterval = 5 * time.Minute
@@ -1,9 +0,0 @@
//go:build s3tests
package scheduler
import "time"
// Under the s3tests build tag rules are PUT and expected to fire within
// seconds, so the engine snapshot is rebuilt aggressively.
const defaultRefreshInterval = 2 * time.Second
@@ -1,222 +0,0 @@
package scheduler
import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
)
// defaultRefreshInterval lives in refresh_*.go so the s3tests build can
// shrink the engine-rebuild cadence to seconds.
const (
defaultRetryBackoff = 5 * time.Second
)
// Scheduler runs N pipeline goroutines, one per worker, each owning a
// contiguous shard slice of [0, ShardCount). It periodically rebuilds the
// engine snapshot from the filer's bucket configs.
type Scheduler struct {
BucketsPath string
Engine *engine.Engine
Persister reader.Persister
Client dispatcher.LifecycleClient
FilerClient filer_pb.SeaweedFilerClient
ClientID int32
ClientName string
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
// engine-refresh ticker plus one Pipeline runner per worker.
func (s *Scheduler) Run(ctx context.Context) error {
if s.Engine == nil || s.Persister == nil || s.Client == nil || s.FilerClient == nil {
return errors.New("scheduler: missing required dependency")
}
if s.BucketsPath == "" {
return errors.New("scheduler: BucketsPath required")
}
workers := s.Workers
if workers <= 0 {
workers = 1
}
refresh := s.RefreshInterval
if refresh <= 0 {
refresh = defaultRefreshInterval
}
retry := s.RetryBackoff
if retry <= 0 {
retry = defaultRetryBackoff
}
// Build all pipelines up front and remember which shard each owns.
// Doing this before Run lets the bootstrap injector route an event
// to the right pipeline by shard, and lets InjectEvent's lazy events
// channel be primed before the per-bucket walker starts pushing.
type pipelineSlot struct {
pipeline *dispatcher.Pipeline
shards []int
}
var slots []*pipelineSlot
pipelinesByShard := make(map[int]*dispatcher.Pipeline)
for i := 0; i < workers; i++ {
shardSet := AssignShards(i, workers)
if len(shardSet) == 0 {
continue
}
slot := &pipelineSlot{
pipeline: &dispatcher.Pipeline{
Shards: shardSet,
BucketsPath: s.BucketsPath,
Engine: s.Engine,
Persister: s.Persister,
Client: s.Client,
FilerClient: s.FilerClient,
ClientID: s.ClientID,
ClientName: fmt.Sprintf("%s-w%02d", s.ClientName, i),
DispatchTick: s.DispatchTick,
CheckpointTick: s.CheckpointTick,
},
shards: shardSet,
}
slots = append(slots, slot)
for _, sh := range shardSet {
pipelinesByShard[sh] = slot.pipeline
}
}
bs := &BucketBootstrapper{
FilerClient: s.FilerClient,
BucketsPath: s.BucketsPath,
Injector: pipelineFanout(pipelinesByShard),
BootstrapInterval: s.BootstrapInterval,
}
s.refreshEngine(ctx, bs)
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
t := time.NewTicker(refresh)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
s.refreshEngine(ctx, bs)
}
}
}()
for _, slot := range slots {
slot := slot
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
default:
}
if err := slot.pipeline.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
glog.Warningf("lifecycle scheduler: shards=%v: %v", slot.shards, err)
}
select {
case <-ctx.Done():
return
case <-time.After(retry):
}
}
}()
}
wg.Wait()
return nil
}
func (s *Scheduler) refreshEngine(ctx context.Context, bs *BucketBootstrapper) {
inputs, parseErrors, err := LoadCompileInputs(ctx, s.FilerClient, s.BucketsPath)
if err != nil {
glog.Warningf("lifecycle scheduler: refresh engine: %v", err)
return
}
for _, pe := range parseErrors {
glog.Warningf("lifecycle scheduler: malformed config in bucket %s: %v", pe.Bucket, pe.Err)
}
s.Engine.Compile(inputs, engine.CompileOptions{PriorStates: AllActivePriorStates(inputs)})
if bs != nil {
buckets := make([]string, 0, len(inputs))
for _, in := range inputs {
buckets = append(buckets, in.Bucket)
}
bs.KickOffNew(ctx, buckets)
}
}
// pipelineFanout routes a synthesized event to the pipeline that owns
// its shard. Returned as an EventInjector so the bootstrapper doesn't
// know about pipelines.
type pipelineFanout map[int]*dispatcher.Pipeline
func (f pipelineFanout) InjectEvent(ctx context.Context, ev *reader.Event) error {
if ev == nil {
return nil
}
p, ok := f[ev.ShardID]
if !ok {
// Shard isn't covered by any pipeline in this scheduler — nothing
// to do. This shouldn't happen with AssignShards covering [0, ShardCount),
// but stay tolerant if a future change introduces gaps.
return nil
}
return p.InjectEvent(ctx, ev)
}
// AssignShards returns the contiguous shard slice for worker idx of total.
// Shards distribute as evenly as possible: with ShardCount=16 and total=3,
// workers receive [0..5], [6..10], [11..15].
func AssignShards(idx, total int) []int {
if total <= 0 || idx < 0 || idx >= total {
return nil
}
shardCount := s3lifecycle.ShardCount
base := shardCount / total
extra := shardCount % total
var lo, hi int
if idx < extra {
lo = idx * (base + 1)
hi = lo + base + 1
} else {
lo = idx*base + extra
hi = lo + base
}
if hi > shardCount {
hi = shardCount
}
if lo >= hi {
return nil
}
out := make([]int, 0, hi-lo)
for i := lo; i < hi; i++ {
out = append(out, i)
}
return out
}
@@ -1,86 +0,0 @@
package scheduler
import (
"reflect"
"testing"
)
func TestAssignShardsSingleWorker(t *testing.T) {
got := AssignShards(0, 1)
want := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
}
func TestAssignShardsEvenSplit(t *testing.T) {
a := AssignShards(0, 2)
b := AssignShards(1, 2)
wantA := []int{0, 1, 2, 3, 4, 5, 6, 7}
wantB := []int{8, 9, 10, 11, 12, 13, 14, 15}
if !reflect.DeepEqual(a, wantA) {
t.Fatalf("worker 0: got %v, want %v", a, wantA)
}
if !reflect.DeepEqual(b, wantB) {
t.Fatalf("worker 1: got %v, want %v", b, wantB)
}
}
func TestAssignShardsUnevenSplit(t *testing.T) {
// 16 shards / 3 workers = 5,5,5 + 1 remainder. Earlier workers absorb
// the remainder so the slices stay contiguous: 6,5,5.
w0 := AssignShards(0, 3)
w1 := AssignShards(1, 3)
w2 := AssignShards(2, 3)
if len(w0) != 6 || w0[0] != 0 || w0[len(w0)-1] != 5 {
t.Fatalf("w0 expected [0..5], got %v", w0)
}
if len(w1) != 5 || w1[0] != 6 || w1[len(w1)-1] != 10 {
t.Fatalf("w1 expected [6..10], got %v", w1)
}
if len(w2) != 5 || w2[0] != 11 || w2[len(w2)-1] != 15 {
t.Fatalf("w2 expected [11..15], got %v", w2)
}
// All shards covered, exactly once.
covered := map[int]int{}
for _, w := range [][]int{w0, w1, w2} {
for _, s := range w {
covered[s]++
}
}
if len(covered) != 16 {
t.Fatalf("expected 16 unique shards, got %d", len(covered))
}
for s, c := range covered {
if c != 1 {
t.Fatalf("shard %d covered %d times, expected 1", s, c)
}
}
}
func TestAssignShardsMoreWorkersThanShards(t *testing.T) {
// 20 workers, 16 shards: first 16 get one shard each, last 4 get nothing.
for i := 0; i < 16; i++ {
got := AssignShards(i, 20)
if len(got) != 1 || got[0] != i {
t.Fatalf("worker %d: got %v, want [%d]", i, got, i)
}
}
for i := 16; i < 20; i++ {
if got := AssignShards(i, 20); got != nil {
t.Fatalf("worker %d should get nil, got %v", i, got)
}
}
}
func TestAssignShardsBounds(t *testing.T) {
if got := AssignShards(-1, 4); got != nil {
t.Fatalf("idx -1: got %v, want nil", got)
}
if got := AssignShards(4, 4); got != nil {
t.Fatalf("idx==total: got %v, want nil", got)
}
if got := AssignShards(0, 0); got != nil {
t.Fatalf("total 0: got %v, want nil", got)
}
}
@@ -0,0 +1,119 @@
package scheduler
import (
"context"
"io"
"sort"
"sync"
"sync/atomic"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
// fakeListStream implements grpc.ServerStreamingClient[filer_pb.ListEntriesResponse]
// for configload tests.
type fakeListStream struct {
responses []*filer_pb.ListEntriesResponse
index int
ctx context.Context
}
func (s *fakeListStream) Recv() (*filer_pb.ListEntriesResponse, error) {
if s.ctx != nil {
if err := s.ctx.Err(); err != nil {
return nil, err
}
}
if s.index >= len(s.responses) {
return nil, io.EOF
}
r := s.responses[s.index]
s.index++
return r, nil
}
func (s *fakeListStream) Header() (metadata.MD, error) { return metadata.MD{}, nil }
func (s *fakeListStream) Trailer() metadata.MD { return metadata.MD{} }
func (s *fakeListStream) CloseSend() error { return nil }
func (s *fakeListStream) Context() context.Context {
if s.ctx != nil {
return s.ctx
}
return context.Background()
}
func (s *fakeListStream) SendMsg(any) error { return nil }
func (s *fakeListStream) RecvMsg(any) error { return nil }
// fakeFilerClient is the in-memory filer used by configload tests.
type fakeFilerClient struct {
filer_pb.SeaweedFilerClient
mu sync.Mutex
tree map[string][]*filer_pb.Entry
listed []string
listedN int32
}
func (c *fakeFilerClient) LookupDirectoryEntry(_ context.Context, in *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
c.mu.Lock()
defer c.mu.Unlock()
for _, e := range c.tree[in.Directory] {
if e != nil && e.Name == in.Name {
return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil
}
}
return nil, filer_pb.ErrNotFound
}
func (c *fakeFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
c.mu.Lock()
c.listed = append(c.listed, in.Directory)
src := c.tree[in.Directory]
c.mu.Unlock()
atomic.AddInt32(&c.listedN, 1)
filtered := make([]*filer_pb.Entry, 0, len(src))
for _, e := range src {
if e == nil {
continue
}
if in.StartFromFileName != "" {
if in.InclusiveStartFrom {
if e.Name < in.StartFromFileName {
continue
}
} else if e.Name <= in.StartFromFileName {
continue
}
}
filtered = append(filtered, e)
}
sort.SliceStable(filtered, func(i, j int) bool { return filtered[i].Name < filtered[j].Name })
if in.Limit > 0 && uint32(len(filtered)) > in.Limit {
filtered = filtered[:in.Limit]
}
resps := make([]*filer_pb.ListEntriesResponse, 0, len(filtered))
for _, e := range filtered {
resps = append(resps, &filer_pb.ListEntriesResponse{Entry: e})
}
return &fakeListStream{responses: resps, ctx: ctx}, nil
}
func dirEntry(name string, extended map[string][]byte) *filer_pb.Entry {
return &filer_pb.Entry{
Name: name,
IsDirectory: true,
Attributes: &filer_pb.FuseAttributes{},
Extended: extended,
}
}
func fileEntry(name string) *filer_pb.Entry {
return &filer_pb.Entry{
Name: name,
IsDirectory: false,
Attributes: &filer_pb.FuseAttributes{},
}
}
@@ -1,329 +0,0 @@
package shell
import (
"context"
"flag"
"fmt"
"io"
"sort"
"strconv"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler"
"github.com/seaweedfs/seaweedfs/weed/util"
)
func init() {
Commands = append(Commands, &commandS3LifecycleRunShard{})
}
type commandS3LifecycleRunShard struct{}
func (c *commandS3LifecycleRunShard) Name() string {
return "s3.lifecycle.run-shard"
}
func (c *commandS3LifecycleRunShard) Help() string {
return `manually run one or more shards of the event-driven S3 lifecycle worker
Subscribes once to the filer meta-log, filters events to the configured
(bucket, key-prefix-hash) shards, routes them through the compiled lifecycle
engine, and dispatches due actions to the S3 server's LifecycleDelete RPC.
Persists each shard's cursor to /etc/s3/lifecycle/cursors/shard-NN.json so
subsequent runs resume.
The -shards form covers a range or set; one filer subscription handles the
whole set, with no per-shard goroutine fan-out. Provide either -shard or
-shards, not both.
# single shard
s3.lifecycle.run-shard -shard 0 -s3 localhost:8333 -events 100
# contiguous range, all 16 shards via one subscription
s3.lifecycle.run-shard -shards 0-15 -s3 localhost:8333 -events 5000
# explicit set
s3.lifecycle.run-shard -shards 0,3,7 -s3 localhost:8333
# custom cadence
s3.lifecycle.run-shard -shards 0-15 -s3 s3-host:8333 -dispatch 1s -checkpoint 10s
`
}
func (c *commandS3LifecycleRunShard) HasTag(CommandTag) bool { return false }
func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer io.Writer) error {
fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
shard := fs.Int("shard", -1, "single shard id in [0, 16); use -shards for a range or set")
shardsSpec := fs.String("shards", "", "shard range \"lo-hi\" or comma list \"a,b,c\"; mutually exclusive with -shard")
s3Endpoint := fs.String("s3", "", "s3 server gRPC endpoint, host:port")
eventBudget := fs.Int("events", 1000, "max in-shard events to process before returning (0 = unbounded; counts only events that pass the shard filter)")
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
}
shards, err := resolveShardSelection(*shard, *shardsSpec)
if err != nil {
return err
}
if *s3Endpoint == "" {
return fmt.Errorf("-s3 required (host:port of s3 server gRPC)")
}
if *eventBudget < 0 {
return fmt.Errorf("-events must be >= 0 (0 = unbounded)")
}
bucketsPath, err := resolveBucketsPath(env)
if err != nil {
return fmt.Errorf("resolve buckets path: %w", err)
}
fmt.Fprintf(writer, "buckets path: %s\n", bucketsPath)
dialCtx, dialCancel := context.WithTimeout(context.Background(), 30*time.Second)
conn, err := pb.GrpcDial(dialCtx, *s3Endpoint, false, env.option.GrpcDialOption)
dialCancel()
if err != nil {
return fmt.Errorf("dial s3 %s: %w", *s3Endpoint, err)
}
defer conn.Close()
rpcClient := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(conn)
// Run the whole pipeline inside one WithFilerClient so the reader's
// SubscribeMetadata stream and the persister share a single connection.
return env.WithFilerClient(true, func(filerClient filer_pb.SeaweedFilerClient) error {
eng := engine.New()
pipeline := &dispatcher.Pipeline{
Shards: shards,
BucketsPath: bucketsPath,
Engine: eng,
Persister: &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
Client: &lifecycleClientCallable{c: rpcClient},
FilerClient: filerClient,
ClientID: util.RandomInt32(),
ClientName: fmt.Sprintf("shell-lifecycle-%s", formatShardLabel(shards)),
DispatchTick: *dispatchTick,
CheckpointTick: *checkpointTick,
EventBudget: *eventBudget,
}
bsr := &scheduler.BucketBootstrapper{
FilerClient: filerClient,
BucketsPath: bucketsPath,
Injector: pipeline,
BootstrapInterval: *bootstrapInterval,
}
bootstrapCtx, bootstrapCancel := context.WithCancel(context.Background())
defer bootstrapCancel()
compile := func(initial bool) {
inputs, parseErrors, err := scheduler.LoadCompileInputs(context.Background(), filerClient, bucketsPath)
if err != nil {
if initial {
fmt.Fprintf(writer, "warning: load lifecycle configs: %v\n", err)
}
return
}
if initial {
for i, pe := range parseErrors {
if i < 3 {
fmt.Fprintf(writer, "warning: %s: %v\n", pe.Bucket, pe.Err)
}
}
if extra := len(parseErrors) - 3; extra > 0 {
fmt.Fprintf(writer, "warning: %d additional bucket(s) had malformed lifecycle config\n", extra)
}
if len(inputs) == 0 {
fmt.Fprintln(writer, "no buckets with enabled lifecycle rules at startup; will refresh and pick them up as they're added")
} else {
fmt.Fprintf(writer, "loaded lifecycle for %d bucket(s)\n", len(inputs))
}
}
eng.Compile(inputs, engine.CompileOptions{PriorStates: scheduler.AllActivePriorStates(inputs)})
// First time we see a bucket, spin up a one-shot walker that
// lists existing entries and synthesizes events. The reader-
// driven path only sees events created after the rule lands,
// so without this backfill objects PUT before the rule (the
// s3-tests scenario) would never expire.
buckets := make([]string, 0, len(inputs))
for _, in := range inputs {
buckets = append(buckets, in.Bucket)
}
bsr.KickOffNew(bootstrapCtx, buckets)
}
compile(true)
var ctx context.Context
var cancel context.CancelFunc
if *runtime > 0 {
ctx, cancel = context.WithTimeout(context.Background(), *runtime)
} else {
ctx, cancel = context.WithCancel(context.Background())
}
defer cancel()
// Periodic engine rebuild so rules added (or disabled) after startup
// land in the snapshot the dispatcher reads on its next tick.
if *refreshInterval > 0 {
go func() {
t := time.NewTicker(*refreshInterval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
compile(false)
}
}
}()
}
fmt.Fprintf(writer, "running shards %s (event budget=%d, runtime=%s, refresh=%s)…\n", formatShardLabel(shards), *eventBudget, *runtime, *refreshInterval)
if err := pipeline.Run(ctx); err != nil {
return fmt.Errorf("pipeline: %w", err)
}
fmt.Fprintf(writer, "shards %s complete; cursors checkpointed\n", formatShardLabel(shards))
return nil
})
}
// resolveShardSelection turns the -shard / -shards flags into a sorted,
// deduplicated []int. Exactly one form must be specified.
func resolveShardSelection(singleShard int, shardsSpec string) ([]int, error) {
if singleShard >= 0 && shardsSpec != "" {
return nil, fmt.Errorf("-shard and -shards are mutually exclusive")
}
if singleShard < 0 && shardsSpec == "" {
return nil, fmt.Errorf("specify -shard <id> or -shards <range|set>")
}
if singleShard >= 0 {
if singleShard >= s3lifecycle.ShardCount {
return nil, fmt.Errorf("-shard %d out of [0,%d)", singleShard, s3lifecycle.ShardCount)
}
return []int{singleShard}, nil
}
return parseShardsSpec(shardsSpec)
}
// parseShardsSpec accepts "lo-hi" (inclusive) or "a,b,c" and returns a
// sorted, deduplicated, in-range []int.
func parseShardsSpec(spec string) ([]int, error) {
spec = strings.TrimSpace(spec)
seen := map[int]struct{}{}
add := func(v int) error {
if v < 0 || v >= s3lifecycle.ShardCount {
return fmt.Errorf("shard %d out of [0,%d)", v, s3lifecycle.ShardCount)
}
seen[v] = struct{}{}
return nil
}
if strings.Contains(spec, "-") && !strings.Contains(spec, ",") {
parts := strings.SplitN(spec, "-", 2)
lo, err := strconv.Atoi(strings.TrimSpace(parts[0]))
if err != nil {
return nil, fmt.Errorf("range lo: %w", err)
}
hi, err := strconv.Atoi(strings.TrimSpace(parts[1]))
if err != nil {
return nil, fmt.Errorf("range hi: %w", err)
}
if lo > hi {
return nil, fmt.Errorf("range lo %d > hi %d", lo, hi)
}
for v := lo; v <= hi; v++ {
if err := add(v); err != nil {
return nil, err
}
}
} else {
for _, part := range strings.Split(spec, ",") {
part = strings.TrimSpace(part)
if part == "" {
continue
}
v, err := strconv.Atoi(part)
if err != nil {
return nil, fmt.Errorf("shard list: %w", err)
}
if err := add(v); err != nil {
return nil, err
}
}
}
if len(seen) == 0 {
return nil, fmt.Errorf("empty shard set")
}
out := make([]int, 0, len(seen))
for v := range seen {
out = append(out, v)
}
sort.Ints(out)
return out, nil
}
func formatShardLabel(shards []int) string {
if len(shards) == 1 {
return fmt.Sprintf("%d", shards[0])
}
// Detect contiguous range.
contiguous := true
for i := 1; i < len(shards); i++ {
if shards[i] != shards[i-1]+1 {
contiguous = false
break
}
}
if contiguous {
return fmt.Sprintf("%d-%d", shards[0], shards[len(shards)-1])
}
parts := make([]string, len(shards))
for i, v := range shards {
parts[i] = strconv.Itoa(v)
}
return strings.Join(parts, ",")
}
// lifecycleClientCallable adapts the generated grpc client (variadic
// CallOption tail) to the dispatcher.LifecycleClient interface.
type lifecycleClientCallable struct {
c s3_lifecycle_pb.SeaweedS3LifecycleInternalClient
}
func (l *lifecycleClientCallable) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
return l.c.LifecycleDelete(ctx, req)
}
// resolveBucketsPath fetches the filer's configured buckets directory.
// Falls back to /buckets when the filer doesn't return one.
func resolveBucketsPath(env *CommandEnv) (string, error) {
var path string
err := env.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return err
}
path = resp.GetDirBuckets()
return nil
})
if err != nil {
return "", err
}
if path == "" {
path = "/buckets"
}
return path, nil
}
@@ -1,92 +0,0 @@
package shell
import (
"reflect"
"testing"
)
func TestParseShardsSpec_Range(t *testing.T) {
got, err := parseShardsSpec("3-7")
if err != nil {
t.Fatalf("err: %v", err)
}
want := []int{3, 4, 5, 6, 7}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
}
func TestParseShardsSpec_Set(t *testing.T) {
got, err := parseShardsSpec("0,3,7")
if err != nil {
t.Fatalf("err: %v", err)
}
want := []int{0, 3, 7}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
}
func TestParseShardsSpec_DedupSort(t *testing.T) {
got, err := parseShardsSpec("7,3,3,0")
if err != nil {
t.Fatalf("err: %v", err)
}
want := []int{0, 3, 7}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
}
func TestParseShardsSpec_OutOfRange(t *testing.T) {
if _, err := parseShardsSpec("16"); err == nil {
t.Fatal("expected out-of-range error for 16")
}
if _, err := parseShardsSpec("-1"); err == nil {
t.Fatal("expected out-of-range error for -1")
}
if _, err := parseShardsSpec("0-16"); err == nil {
t.Fatal("expected out-of-range error for 0-16")
}
}
func TestParseShardsSpec_BadRange(t *testing.T) {
if _, err := parseShardsSpec("7-3"); err == nil {
t.Fatal("expected lo>hi error")
}
}
func TestResolveShardSelection_Mutex(t *testing.T) {
if _, err := resolveShardSelection(0, "1,2"); err == nil {
t.Fatal("expected mutex error when both -shard and -shards set")
}
if _, err := resolveShardSelection(-1, ""); err == nil {
t.Fatal("expected error when neither set")
}
}
func TestResolveShardSelection_SingleShard(t *testing.T) {
got, err := resolveShardSelection(5, "")
if err != nil {
t.Fatalf("err: %v", err)
}
if !reflect.DeepEqual(got, []int{5}) {
t.Fatalf("got %v, want [5]", got)
}
}
func TestFormatShardLabel(t *testing.T) {
cases := []struct {
in []int
want string
}{
{[]int{5}, "5"},
{[]int{0, 1, 2, 3}, "0-3"},
{[]int{0, 2, 5}, "0,2,5"},
}
for _, tc := range cases {
if got := formatShardLabel(tc.in); got != tc.want {
t.Errorf("formatShardLabel(%v)=%q, want %q", tc.in, got, tc.want)
}
}
}
+5 -6
View File
@@ -33,8 +33,8 @@ func init() {
}
// Handler is the worker-side runner for S3 object lifecycle expiration.
// One Execute call drives a long-running scheduler.Scheduler against the
// S3 endpoints discovered from the master; admin caps concurrency at one
// One Execute call drives one bounded dailyrun.Run pass against the S3
// endpoints discovered from the master; admin caps concurrency at one
// job per worker so a fresh proposal only spawns a new run after the
// prior one exits.
type Handler struct {
@@ -226,10 +226,9 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ
return h.executeDailyReplay(runCtx, request, bucketsPath, filerClient, rpc, cfg, sender)
}
// executeDailyReplay runs the bounded daily-replay path. Reuses the
// streaming path's filer / s3 / engine setup but routes the per-shard
// loop through dailyrun.Run instead of scheduler.Scheduler. Phase 2:
// replay-only, refuses walker-bound action kinds with a typed error.
// executeDailyReplay runs one bounded daily-replay pass via
// dailyrun.Run. The walker fires inside runShard on rule-content edits
// and against the steady-state walk view; all rule kinds are serviced.
func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.ExecuteJobRequest, bucketsPath string, filerClient filer_pb.SeaweedFilerClient, rpc s3_lifecycle_pb.SeaweedS3LifecycleInternalClient, cfg Config, sender pluginworker.ExecutionSender) error {
eng := engine.New()
inputs, parseErrors, err := scheduler.LoadCompileInputs(ctx, filerClient, bucketsPath)