From 5c991f38f583e386e36d28083b32a89373e26185 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 7 May 2026 15:44:09 -0700 Subject: [PATCH] feat(s3/lifecycle): dispatcher + per-shard pipeline (Phase 3 PR-D) (#9356) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit feat(s3/lifecycle): dispatcher + blocker store + per-shard pipeline Dispatcher consumes due Matches from the schedule, calls LifecycleDelete, and routes outcomes: DONE / NOOP_RESOLVED / SKIPPED_OBJECT_LOCK -> Cursor.Advance RETRY_LATER (within budget) -> re-schedule with backoff RETRY_LATER (budget exhausted) / BLOCKED -> BlockerStore.Put + Freeze BlockerStore is a small interface with InMemoryBlockerStore for tests; the filer-backed impl follows when the worker task registration lands. Pipeline composes Reader + Router + Dispatcher into a single Run loop keyed by shard. Cursor is restored on start, blockers are replayed as freezes, checkpoints write at a configurable cadence, and a final save fires on shutdown. The meta-log itself is the durable buffer for in-flight schedule entries — restart re-derives them from the cursor's MinTsNs. --- weed/s3api/s3lifecycle/dispatcher/blocker.go | 77 ++++++ .../s3lifecycle/dispatcher/dispatcher.go | 217 ++++++++++++++++ .../s3lifecycle/dispatcher/dispatcher_test.go | 240 ++++++++++++++++++ weed/s3api/s3lifecycle/dispatcher/pipeline.go | 172 +++++++++++++ .../s3lifecycle/dispatcher/pipeline_test.go | 84 ++++++ 5 files changed, 790 insertions(+) create mode 100644 weed/s3api/s3lifecycle/dispatcher/blocker.go create mode 100644 weed/s3api/s3lifecycle/dispatcher/dispatcher.go create mode 100644 weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go create mode 100644 weed/s3api/s3lifecycle/dispatcher/pipeline.go create mode 100644 weed/s3api/s3lifecycle/dispatcher/pipeline_test.go diff --git a/weed/s3api/s3lifecycle/dispatcher/blocker.go b/weed/s3api/s3lifecycle/dispatcher/blocker.go new file mode 100644 index 000000000..c189353db --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/blocker.go @@ -0,0 +1,77 @@ +package dispatcher + +import ( + "context" + "sync" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// BlockerRecord persists the state needed to re-freeze a cursor on worker +// restart. Stored under /etc/s3/lifecycle/blockers///. +// +// Operator action via the blocker-resolve flow either: +// - Quarantine: keep the freeze; mark the (object, version) skipped via a +// side-table so subsequent re-evaluations don't trip the same blocker. +// - Retry: clear the record and Unfreeze; the cursor advances on the next +// successful dispatch. +type BlockerRecord struct { + ShardID int + Key s3lifecycle.ActionKey + FrozenAtNs int64 // tsNs of the event that tripped the blocker + Reason string // RPC outcome reason + CreatedAt time.Time +} + +// BlockerStore persists BlockerRecords. The contract: +// - Put replaces any existing record for (ShardID, Key) atomically. +// - Delete is idempotent (no error on missing). +// - List returns all records for ShardID for restart-time freeze replay. +type BlockerStore interface { + Put(ctx context.Context, rec BlockerRecord) error + Delete(ctx context.Context, shardID int, key s3lifecycle.ActionKey) error + List(ctx context.Context, shardID int) ([]BlockerRecord, error) +} + +// InMemoryBlockerStore is a BlockerStore for tests. +type InMemoryBlockerStore struct { + mu sync.Mutex + records map[int]map[s3lifecycle.ActionKey]BlockerRecord +} + +func NewInMemoryBlockerStore() *InMemoryBlockerStore { + return &InMemoryBlockerStore{records: map[int]map[s3lifecycle.ActionKey]BlockerRecord{}} +} + +func (s *InMemoryBlockerStore) Put(ctx context.Context, rec BlockerRecord) error { + s.mu.Lock() + defer s.mu.Unlock() + m, ok := s.records[rec.ShardID] + if !ok { + m = map[s3lifecycle.ActionKey]BlockerRecord{} + s.records[rec.ShardID] = m + } + m[rec.Key] = rec + return nil +} + +func (s *InMemoryBlockerStore) Delete(ctx context.Context, shardID int, key s3lifecycle.ActionKey) error { + s.mu.Lock() + defer s.mu.Unlock() + if m, ok := s.records[shardID]; ok { + delete(m, key) + } + return nil +} + +func (s *InMemoryBlockerStore) List(ctx context.Context, shardID int) ([]BlockerRecord, error) { + s.mu.Lock() + defer s.mu.Unlock() + m := s.records[shardID] + out := make([]BlockerRecord, 0, len(m)) + for _, rec := range m { + out = append(out, rec) + } + return out, nil +} diff --git a/weed/s3api/s3lifecycle/dispatcher/dispatcher.go b/weed/s3api/s3lifecycle/dispatcher/dispatcher.go new file mode 100644 index 000000000..bc3589e40 --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/dispatcher.go @@ -0,0 +1,217 @@ +package dispatcher + +import ( + "context" + "fmt" + "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" +) + +// 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 and the blocker store. +// +// 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) -> escalate to BLOCKED +// BLOCKED -> BlockerStore.Put + Cursor.Freeze +// FATAL_EVENT_ERROR / unknown -> treat as BLOCKED +type Dispatcher struct { + ShardID int + Client LifecycleClient + Cursor *reader.Cursor + Blockers BlockerStore + 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 BLOCKED is + // durable and the cursor is durable. + 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 _, m := range due { + if err := ctx.Err(); err != nil { + // Re-queue and return; the caller is shutting down. + d.Schedule.Add(m) + 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 { + // 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. + d.handleRetryLater(ctx, m, fmt.Sprintf("RPC: %v", err), now) + return + } + 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)) + } +} + +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)) + rec := BlockerRecord{ + ShardID: d.ShardID, + Key: m.Key, + FrozenAtNs: m.EventTs.UnixNano(), + Reason: reason, + CreatedAt: time.Now(), + } + if err := d.Blockers.Put(ctx, rec); err != nil { + // Persistence failure: retry; cursor stays frozen in-memory. + glog.Errorf("lifecycle blocker persist: shard=%d key=%v: %v", d.ShardID, m.Key, err) + } + d.Cursor.Freeze(m.Key, m.EventTs.UnixNano()) +} + +// ReplayBlockers re-applies in-memory freezes from the durable BlockerStore. +// Call once on Pipeline startup before the reader begins emitting. +func (d *Dispatcher) ReplayBlockers(ctx context.Context) error { + recs, err := d.Blockers.List(ctx, d.ShardID) + if err != nil { + return fmt.Errorf("blocker list: %w", err) + } + for _, r := range recs { + d.Cursor.Freeze(r.Key, r.FrozenAtNs) + } + return nil +} + +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, + } +} diff --git a/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go b/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go new file mode 100644 index 000000000..8c812e871 --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go @@ -0,0 +1,240 @@ +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/reader" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router" +) + +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(), + Blockers: NewInMemoryBlockerStore(), + 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") + } + recs, _ := d.Blockers.List(context.Background(), 0) + if len(recs) != 1 { + t.Fatalf("expected 1 blocker record, got %d", len(recs)) + } + if recs[0].Reason == "" { + t.Fatal("blocker record missing reason") + } +} + +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 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 TestReplayBlockersRefreezes(t *testing.T) { + store := NewInMemoryBlockerStore() + t0 := time.Now() + key := s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays} + store.Put(context.Background(), BlockerRecord{ + ShardID: 0, + Key: key, + FrozenAtNs: t0.UnixNano(), + Reason: "prior run", + CreatedAt: t0, + }) + d := &Dispatcher{ + ShardID: 0, + Cursor: reader.NewCursor(), + Blockers: store, + Schedule: router.NewSchedule(), + } + if err := d.ReplayBlockers(context.Background()); err != nil { + t.Fatalf("ReplayBlockers: %v", err) + } + if !d.Cursor.IsFrozen(key) { + t.Fatal("ReplayBlockers should refreeze cursor") + } +} diff --git a/weed/s3api/s3lifecycle/dispatcher/pipeline.go b/weed/s3api/s3lifecycle/dispatcher/pipeline.go new file mode 100644 index 000000000..d7735c42f --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/pipeline.go @@ -0,0 +1,172 @@ +package dispatcher + +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/engine" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router" +) + +// Pipeline composes the per-shard reader, router, dispatcher, and cursor +// checkpoint into a single Run loop. One Pipeline per (worker, shard). +type Pipeline struct { + ShardID int + BucketsPath string + + Engine *engine.Engine + Cursor *reader.Cursor + Persister reader.Persister + Blockers BlockerStore + 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 +} + +const ( + defaultDispatchTick = 5 * time.Second + defaultCheckpointTick = 30 * time.Second + defaultEventBuffer = 1024 +) + +// Run blocks until ctx is canceled or a fatal error occurs. On exit, the +// 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.Cursor == nil || p.Persister == nil || + p.Blockers == nil || p.Client == nil || p.FilerClient == nil { + return errors.New("pipeline: missing required dependency") + } + if p.BucketsPath == "" { + return errors.New("pipeline: BucketsPath required") + } + + // 1. Restore cursor + replay blocker freezes. + state, err := p.Persister.Load(ctx, p.ShardID) + if err != nil { + return fmt.Errorf("cursor load: %w", err) + } + p.Cursor.Restore(state) + + dispatch := &Dispatcher{ + ShardID: p.ShardID, + Client: p.Client, + Cursor: p.Cursor, + Blockers: p.Blockers, + Schedule: router.NewSchedule(), + } + if err := dispatch.ReplayBlockers(ctx); err != nil { + return fmt.Errorf("blocker replay: %w", err) + } + + // 2. Wire reader -> router -> schedule via a buffered channel. + bufSize := p.EventBuffer + if bufSize <= 0 { + bufSize = defaultEventBuffer + } + events := make(chan *reader.Event, bufSize) + rd := &reader.Reader{ + ShardID: p.ShardID, + BucketsPath: p.BucketsPath, + Cursor: p.Cursor, + Events: events, + EventBudget: p.EventBudget, + } + rd.LogStartup() + + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + + var wg sync.WaitGroup + var readerErr error + + // Reader goroutine. + wg.Add(1) + go func() { + defer wg.Done() + defer close(events) + readerErr = rd.Run(runCtx, p.FilerClient, p.ClientName, p.ClientID) + if readerErr != nil && !errors.Is(readerErr, context.Canceled) { + glog.Errorf("lifecycle reader: shard=%d: %v", p.ShardID, readerErr) + } + cancel() // wake the dispatcher goroutine to drain & exit + }() + + // Router/dispatcher goroutine: pulls events, routes them, ticks schedule. + 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() + snap := p.Engine.Snapshot() + + for { + select { + case <-runCtx.Done(): + dispatch.Tick(context.Background(), time.Now()) + return + case ev, ok := <-events: + if !ok { + dispatch.Tick(context.Background(), time.Now()) + return + } + if snap == nil { + snap = p.Engine.Snapshot() + } + for _, m := range router.Route(snap, ev, time.Now()) { + dispatch.Schedule.Add(m) + } + case <-dt.C: + snap = p.Engine.Snapshot() // refresh between tick boundaries + dispatch.Tick(runCtx, time.Now()) + case <-ct.C: + if err := p.Persister.Save(runCtx, p.ShardID, p.Cursor.Snapshot()); err != nil { + glog.Warningf("lifecycle cursor checkpoint: shard=%d: %v", p.ShardID, err) + } + } + } + }() + + wg.Wait() + + // Final cursor checkpoint on graceful shutdown. + if err := p.Persister.Save(context.Background(), p.ShardID, p.Cursor.Snapshot()); err != nil { + glog.Warningf("lifecycle cursor final save: shard=%d: %v", p.ShardID, err) + } + + if readerErr != nil && !errors.Is(readerErr, context.Canceled) { + return readerErr + } + return nil +} diff --git a/weed/s3api/s3lifecycle/dispatcher/pipeline_test.go b/weed/s3api/s3lifecycle/dispatcher/pipeline_test.go new file mode 100644 index 000000000..aeb5ed1b8 --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/pipeline_test.go @@ -0,0 +1,84 @@ +package dispatcher + +import ( + "context" + "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(snap, ev, now) + 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") + } +}