feat(s3/lifecycle): filer-backed cursor Persister + drop BlockerStore (#9358)

* feat(s3/lifecycle): filer-backed cursor Persister

FilerPersister persists per-shard cursor maps as JSON to
/etc/s3/lifecycle/cursors/shard-NN.json via filer.SaveInsideFiler.
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. Pipeline.Run loads on start; the periodic checkpoint and
graceful-shutdown save go through this implementation.

A small FilerStore interface wraps the SeaweedFilerClient surface
the persister needs, so tests inject an in-memory fake instead of
mocking the full gRPC client.

* refactor(s3/lifecycle): drop BlockerStore — durable cursor IS the block

A frozen cursor doesn't advance, so the durable cursor (FilerPersister)
encodes the blocked state on its own. On worker restart the reader
re-encounters the poison event at MinTsNs, the dispatcher walks the
same retry budget to BLOCKED, and the cursor freezes at the same
EventTs. Other in-flight events between freeze tsNs and prior cursor
positions self-resolve via NOOP_RESOLVED (STALE_IDENTITY) since the
underlying objects were already deleted on the prior pass.

Removed:
  - BlockerStore interface + InMemoryBlockerStore + BlockerRecord
  - Dispatcher.Blockers + Dispatcher.ReplayBlockers
  - the BlockerStore.Put call in handleBlocked
  - Pipeline.Blockers field + the ReplayBlockers call on startup

Added a TestDispatchRestartReFreezesNaturally that pins the
self-recovery property: a fresh Dispatcher with a fresh Cursor, fed
the same poison event, reaches the same frozen state at the same
EventTs without any durable blocker store.

Operator visibility: a cursor whose MinTsNs hasn't advanced is the
signal — surfaced via the durable cursor file.

* refactor(filer): SaveInsideFiler accepts ctx

ReadInsideFiler already takes ctx; SaveInsideFiler used context.Background()
internally and silently dropped the caller's ctx. Symmetric API now;
cancellation/deadlines propagate through LookupEntry / CreateEntry /
UpdateEntry. Mechanical update of all callers — most pass
context.Background() since the existing call sites have no ctx in scope.

* fix(s3/lifecycle): deterministic order in cursor save

Iterating Go maps yields random order, so json.Encode produced a different
byte sequence on each save even when the state hadn't changed. Sort
entries by (Bucket, ActionKind, RuleHash) before encoding so the on-disk
file diffs cleanly. New test pins byte-identical output across two saves
of the same map.

* fix(s3/lifecycle): log reason when freezing cursor in handleBlocked

handleBlocked dropped the reason via _ = reason with a comment claiming
the caller logged it; none of the three callers do. A frozen cursor is
the only surface where the operator finds out something stuck, so the
reason has to land somewhere. glog.Warningf with shard, key, eventTs,
and the original reason — same shape the rest of the package uses.

weed/credential/filer_etc/filer_etc_group.go

@@ -78,7 +78,7 @@ func (store *FilerEtcStore) saveGroup(ctx context.Context, group *iam_pb.Group)
 		if err != nil {
 			return err
 		}
-		return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamGroupsDirectory, group.Name+".json", data)
+		return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamGroupsDirectory, group.Name+".json", data)
 	})
 }
 

weed/credential/filer_etc/filer_etc_identity.go

@@ -484,7 +484,7 @@ func (store *FilerEtcStore) saveIdentity(ctx context.Context, identity *iam_pb.I
 		if err != nil {
 			return err
 		}
-		return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamIdentitiesDirectory, identity.Name+".json", data)
+		return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamIdentitiesDirectory, identity.Name+".json", data)
 	})
 }
 

weed/credential/filer_etc/filer_etc_policy.go

@@ -64,7 +64,7 @@ func (store *FilerEtcStore) saveLegacyPoliciesCollection(ctx context.Context, po
 		if err != nil {
 			return err
 		}
-		return filer.SaveInsideFiler(client, filer.IamConfigDirectory, filer.IamPoliciesFile, content)
+		return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory, filer.IamPoliciesFile, content)
 	})
 }
 
@@ -216,7 +216,7 @@ func (store *FilerEtcStore) savePolicy(ctx context.Context, name string, documen
 		if err != nil {
 			return err
 		}
-		return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamPoliciesDirectory, name+".json", data)
+		return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamPoliciesDirectory, name+".json", data)
 	})
 }
 

weed/credential/filer_etc/filer_etc_service_account.go

@@ -71,7 +71,7 @@ func (store *FilerEtcStore) saveServiceAccount(ctx context.Context, sa *iam_pb.S
 		if err != nil {
 			return err
 		}
-		return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamServiceAccountsDirectory, sa.Id+".json", data)
+		return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamServiceAccountsDirectory, sa.Id+".json", data)
 	})
 }
 

weed/filer/read_write.go

@@ -41,15 +41,15 @@ func ReadInsideFiler(ctx context.Context, filerClient filer_pb.SeaweedFilerClien
 	return
 }
 
-func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, content []byte) error {
+func SaveInsideFiler(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, name string, content []byte) error {
 
-	resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
+	resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
 		Directory: dir,
 		Name:      name,
 	})
 
 	if err == filer_pb.ErrNotFound {
-		err = filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
+		err = filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{
 			Directory: dir,
 			Entry: &filer_pb.Entry{
 				Name:        name,
@@ -69,7 +69,7 @@ func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, conte
 		entry.Content = content
 		entry.Attributes.Mtime = time.Now().Unix()
 		entry.Attributes.FileSize = uint64(len(content))
-		err = filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
+		err = filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{
 			Directory: dir,
 			Entry:     entry,
 		})

weed/filer/remote_mapping.go

@@ -52,7 +52,7 @@ func InsertMountMapping(filerClient filer_pb.FilerClient, dir string, remoteStor
 
 	// save back
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return SaveInsideFiler(client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent)
+		return SaveInsideFiler(context.Background(), client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent)
 	})
 	if err != nil {
 		return fmt.Errorf("save mapping: %w", err)
@@ -83,7 +83,7 @@ func DeleteMountMapping(filerClient filer_pb.FilerClient, dir string) (err error
 
 	// save back
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return SaveInsideFiler(client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent)
+		return SaveInsideFiler(context.Background(), client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent)
 	})
 	if err != nil {
 		return fmt.Errorf("save mapping: %w", err)

weed/iamapi/iamapi_server.go

@@ -193,7 +193,7 @@ func (iama *IamS3ApiConfigure) PutS3ApiConfigurationToFiler(s3cfg *iam_pb.S3ApiC
 	}
 	return pb.WithOneOfGrpcFilerClients(false, iama.option.Filers, iama.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 		err = util.Retry("saveIamIdentity", func() error {
-			return filer.SaveInsideFiler(client, filer.IamConfigDirectory, filer.IamIdentityFile, buf.Bytes())
+			return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory, filer.IamIdentityFile, buf.Bytes())
 		})
 		if err != nil {
 			return err
@@ -229,7 +229,7 @@ func (iama *IamS3ApiConfigure) PutPolicies(policies *Policies) (err error) {
 		return err
 	}
 	return pb.WithOneOfGrpcFilerClients(false, iama.option.Filers, iama.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
-		if err := filer.SaveInsideFiler(client, filer.IamConfigDirectory, filer.IamPoliciesFile, b); err != nil {
+		if err := filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory, filer.IamPoliciesFile, b); err != nil {
 			return err
 		}
 		return nil

weed/mq/kafka/gateway/coordinator_registry.go

@@ -662,7 +662,7 @@ func (cr *CoordinatorRegistry) saveCoordinatorAssignment(consumerGroup string, a
 
 		// Save to individual file: /topics/kafka/.meta/coordinators/<consumer-group>_assignments.json
 		fileName := fmt.Sprintf("%s_assignments.json", consumerGroup)
-		return filer.SaveInsideFiler(client, CoordinatorAssignmentsDir, fileName, assignmentData)
+		return filer.SaveInsideFiler(context.Background(), client, CoordinatorAssignmentsDir, fileName, assignmentData)
 	})
 }
 

weed/mq/offset/consumer_group_storage.go

@@ -86,7 +86,7 @@ func (f *FilerConsumerGroupOffsetStorage) SaveConsumerGroupPosition(t topic.Topi
 	}
 
 	return f.filerClientAccessor.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return filer.SaveInsideFiler(client, consumersDir, offsetFileName, jsonBytes)
+		return filer.SaveInsideFiler(context.Background(), client, consumersDir, offsetFileName, jsonBytes)
 	})
 }
 

weed/mq/offset/filer_storage.go

@@ -38,7 +38,7 @@ func (f *FilerOffsetStorage) SaveCheckpoint(namespace, topicName string, partiti
 	util.Uint64toBytes(offsetBytes, uint64(offset))
 
 	return f.filerClientAccessor.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return filer.SaveInsideFiler(client, partitionDir, fileName, offsetBytes)
+		return filer.SaveInsideFiler(context.Background(), client, partitionDir, fileName, offsetBytes)
 	})
 }
 

weed/mq/topic/topic.go

@@ -100,7 +100,7 @@ func (t Topic) ReadConfFileWithMetadata(client filer_pb.SeaweedFilerClient) (*mq
 func (t Topic) WriteConfFile(client filer_pb.SeaweedFilerClient, conf *mq_pb.ConfigureTopicResponse) error {
 	var buf bytes.Buffer
 	filer.ProtoToText(&buf, conf)
-	if err := filer.SaveInsideFiler(client, t.Dir(), filer.TopicConfFile, buf.Bytes()); err != nil {
+	if err := filer.SaveInsideFiler(context.Background(), client, t.Dir(), filer.TopicConfFile, buf.Bytes()); err != nil {
 		return fmt.Errorf("save topic %v conf: %w", t, err)
 	}
 	return nil

weed/s3api/s3api_bucket_handlers.go

@@ -1051,7 +1051,7 @@ func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWr
 			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		}
 		if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
+			return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
 		}); err != nil {
 			glog.Errorf("PutBucketLifecycleConfigurationHandler save config inside filer: %s", err)
 			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -1106,7 +1106,7 @@ func (s3a *S3ApiServer) DeleteBucketLifecycleHandler(w http.ResponseWriter, r *h
 			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		}
 		if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
+			return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
 		}); err != nil {
 			glog.Errorf("DeleteBucketLifecycleHandler save config inside filer: %s", err)
 			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)

weed/s3api/s3lifecycle/dispatcher/blocker.go

@@ -1,77 +0,0 @@
-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/<shard>/<rule_hash>/<kind>.
-//
-// 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
-}

weed/s3api/s3lifecycle/dispatcher/dispatcher.go

@@ -20,19 +20,22 @@ type LifecycleClient interface {
 }
 
 // Dispatcher consumes due Matches, calls LifecycleDelete, and routes the
-// outcome back to the per-shard cursor and the blocker store.
+// 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) -> escalate to BLOCKED
-//   BLOCKED                      -> BlockerStore.Put + Cursor.Freeze
-//   FATAL_EVENT_ERROR / unknown  -> treat as BLOCKED
+//   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
-	Blockers BlockerStore
 	Schedule *router.Schedule
 
 	// RetryBudget caps RETRY_LATER attempts before escalating to BLOCKED.
@@ -44,8 +47,8 @@ type Dispatcher struct {
 	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.
+	// 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
 }
 
@@ -169,33 +172,11 @@ func (d *Dispatcher) handleRetryLater(ctx context.Context, m router.Match, reaso
 
 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)
-	}
+	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())
 }
 
-// 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:

weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go

@@ -38,7 +38,6 @@ func newDispatcher(client LifecycleClient) (*Dispatcher, *router.Schedule) {
 		ShardID:      0,
 		Client:       client,
 		Cursor:       reader.NewCursor(),
-		Blockers:     NewInMemoryBlockerStore(),
 		Schedule:     sched,
 		RetryBudget:  3,
 		RetryBackoff: time.Millisecond,
@@ -142,12 +141,8 @@ func TestDispatchRetryBudgetEscalatesToBlocked(t *testing.T) {
 	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")
+	if d.Cursor.Get(m.Key) != t0.UnixNano() {
+		t.Fatal("frozen cursor should be pinned at event ts")
 	}
 }
 
@@ -241,27 +236,33 @@ func TestDispatchSkipsFrozenCursor(t *testing.T) {
 	}
 }
 
-func TestReplayBlockersRefreezes(t *testing.T) {
-	store := NewInMemoryBlockerStore()
+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()
-	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(),
+	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")
 	}
-	if err := d.ReplayBlockers(context.Background()); err != nil {
-		t.Fatalf("ReplayBlockers: %v", err)
+
+	// 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 !d.Cursor.IsFrozen(key) {
-		t.Fatal("ReplayBlockers should refreeze cursor")
+	if d2.Cursor.Get(m.Key) != t0.UnixNano() {
+		t.Fatal("re-freeze cursor not pinned at event ts")
 	}
 }

weed/s3api/s3lifecycle/dispatcher/filer_persister.go

@@ -0,0 +1,141 @@
+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.
+type FilerStore interface {
+	Read(ctx context.Context, dir, name string) ([]byte, error)
+	Save(ctx context.Context, dir, name string, content []byte) error
+}
+
+// NewFilerStoreClient adapts a SeaweedFilerClient into a FilerStore.
+func NewFilerStoreClient(client filer_pb.SeaweedFilerClient) FilerStore {
+	return &filerStoreClient{client: client}
+}
+
+type filerStoreClient struct {
+	client filer_pb.SeaweedFilerClient
+}
+
+func (s *filerStoreClient) Read(ctx context.Context, dir, name string) ([]byte, error) {
+	return filer.ReadInsideFiler(ctx, s.client, dir, name)
+}
+
+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)

weed/s3api/s3lifecycle/dispatcher/filer_persister_test.go

@@ -0,0 +1,159 @@
+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")
+	}
+}

weed/s3api/s3lifecycle/dispatcher/pipeline.go

@@ -23,7 +23,6 @@ type Pipeline struct {
 	Engine    *engine.Engine
 	Cursor    *reader.Cursor
 	Persister reader.Persister
-	Blockers  BlockerStore
 	Client    LifecycleClient
 
 	FilerClient filer_pb.SeaweedFilerClient
@@ -55,14 +54,16 @@ const (
 // 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 {
+		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.
+	// Restore cursor; freezes re-arm naturally when the reader re-encounters
+	// the poison event at MinTsNs and the dispatch state machine drives it
+	// back to BLOCKED.
 	state, err := p.Persister.Load(ctx, p.ShardID)
 	if err != nil {
 		return fmt.Errorf("cursor load: %w", err)
@@ -73,12 +74,8 @@ func (p *Pipeline) Run(ctx context.Context) error {
 		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

weed/shell/command_fs_configure.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"bytes"
+	"context"
 	"flag"
 	"fmt"
 	"io"
@@ -142,7 +143,7 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
 	if *apply {
 
 		if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes())
+			return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes())
 		}); err != nil && err != filer_pb.ErrNotFound {
 			return err
 		}

weed/shell/command_remote_configure.go

@@ -205,7 +205,7 @@ func (c *commandRemoteConfigure) saveRemoteStorage(commandEnv *CommandEnv, write
 	}
 
 	if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return filer.SaveInsideFiler(client, filer.DirectoryEtcRemote, conf.Name+filer.REMOTE_STORAGE_CONF_SUFFIX, data)
+		return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcRemote, conf.Name+filer.REMOTE_STORAGE_CONF_SUFFIX, data)
 	}); err != nil && err != filer_pb.ErrNotFound {
 		return err
 	}

weed/shell/command_s3_bucket_quota_check.go

@@ -92,7 +92,7 @@ func (c *commandS3BucketQuotaEnforce) Do(args []string, commandEnv *CommandEnv,
 		fc.ToText(&buf2)
 
 		if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes())
+			return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes())
 		}); err != nil && err != filer_pb.ErrNotFound {
 			return err
 		}

weed/shell/command_s3_circuitbreaker.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"bytes"
+	"context"
 	"flag"
 	"fmt"
 	"io"
@@ -186,7 +187,7 @@ func (c *commandS3CircuitBreaker) Do(args []string, commandEnv *CommandEnv, writ
 
 	if *apply {
 		if err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			return filer.SaveInsideFiler(client, dir, file, buf.Bytes())
+			return filer.SaveInsideFiler(context.Background(), client, dir, file, buf.Bytes())
 		}); err != nil {
 			return err
 		}