fix(s3/lifecycle): align walker dispatch error label to RPC_ERROR

Mirrors the same gemini-flagged inconsistency on the walker side as
the Phase 6 PR fixed on the replay side: streaming uses "RPC_ERROR"
for transport-class failures, so the walker should too. Also folds
the nil-response case under RPC_ERROR rather than a new label —
operationally it's the same class of failure (server returned no
usable response), no point splitting the metric for it.

weed/s3api/s3lifecycle/bootstrap/walker.go

@@ -40,6 +40,13 @@ type Entry struct {
 
 	SuccessorModTime time.Time
 	NoncurrentIndex  *int
+
+	// VersionID is the S3 version id of this entry, empty for
+	// non-versioned buckets. Populated by the ListFunc adapter when
+	// the entry came from a `.versions/` directory; the walker itself
+	// doesn't use it, but the Dispatcher needs it to address the
+	// right version on LifecycleDelete.
+	VersionID string
 }
 
 // ListFunc must skip entries with Path <= start so kill-resume picks up

weed/s3api/s3lifecycle/dailyrun/filer_list_func.go

@@ -0,0 +1,320 @@
+package dailyrun
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"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/bootstrap"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// listPageSize is the page size for paginated directory listings. The
+// filer caps SeaweedList(..., limit=0) at DirListingLimit (1000 by
+// default) per call, so a single-page list would silently truncate
+// large directories. Atomic so tests can shrink it without racing.
+var listPageSize atomic.Uint32
+
+func init() { listPageSize.Store(1024) }
+
+// FilerListFunc returns a bootstrap.ListFunc that streams entries
+// under <bucketsPath>/<bucket>. Versioned siblings are expanded with
+// IsLatest / NumVersions / NoncurrentIndex / SuccessorModTime so the
+// walker's NoncurrentDays evaluation has the same per-version state
+// the streaming bootstrap injects via reader.Event.BootstrapVersion.
+// MPU init records at .uploads/<id> with ExtMultipartObjectKey set
+// are emitted whole with IsMPUInit=true and DestKey carrying the
+// user's intended path.
+func FilerListFunc(client filer_pb.SeaweedFilerClient, bucketsPath string) bootstrap.ListFunc {
+	return func(ctx context.Context, bucket, start string, cb func(*bootstrap.Entry) error) error {
+		if client == nil {
+			return fmt.Errorf("FilerListFunc: nil client")
+		}
+		root := strings.TrimSuffix(bucketsPath, "/") + "/" + bucket
+		return walkBucketTree(ctx, client, root, root, start, cb)
+	}
+}
+
+// walkBucketTree recurses through dir in two passes per level. Pass 1
+// expands `.versions/` dirs (populating skipBare with the bare null-
+// version keys that pass 2 must suppress). Pass 2 emits regular files
+// and recurses into non-special subdirectories.
+//
+// The two-pass shape mirrors scheduler/bootstrap.go's walkBucketDir
+// (see that file's comment for why `.versions/` has to be processed
+// before its bare sibling).
+func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot, start string, cb func(*bootstrap.Entry) error) error {
+	skipBare := map[string]bool{}
+
+	// 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 expandVersionsDir(ctx, client, bucketRoot, key, e, start, skipBare, cb)
+	}); err != nil {
+		return err
+	}
+
+	// Pass 2: everything else.
+	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) {
+				if start != "" && key <= start {
+					return nil
+				}
+				destKey := string(e.Extended[s3_constants.ExtMultipartObjectKey])
+				return cb(&bootstrap.Entry{
+					Path:      key,
+					DestKey:   destKey,
+					IsMPUInit: true,
+					ModTime:   time.Unix(e.Attributes.Mtime, int64(e.Attributes.MtimeNs)),
+					Size:      int64(e.Attributes.FileSize),
+				})
+			}
+			return walkBucketTree(ctx, client, full, bucketRoot, start, cb)
+		}
+		if skipBare[key] {
+			return nil
+		}
+		if start != "" && key <= start {
+			return nil
+		}
+		entry := &bootstrap.Entry{
+			Path:     key,
+			ModTime:  time.Unix(e.Attributes.Mtime, int64(e.Attributes.MtimeNs)),
+			Size:     int64(e.Attributes.FileSize),
+			IsLatest: true, // Non-versioned default.
+		}
+		return cb(entry)
+	})
+}
+
+// versionItem captures one sibling of a `.versions/` expansion. bareKey
+// is the bucket-relative path of the bare null-version entry when the
+// item represents it; for real version files it stays empty.
+type versionItem struct {
+	entry          *filer_pb.Entry
+	versionID      string
+	bareKey        string
+	isExplicitNull bool
+}
+
+// expandVersionsDir handles the `.versions/<key>` directory. Lists
+// version files, optionally appends the bare null-version sibling,
+// sorts newest-first, resolves the latest, and emits one Entry per
+// version with the sibling state walkEntry needs to evaluate
+// NoncurrentDays / NewerNoncurrent / ExpirationDays correctly.
+//
+// Ported from scheduler/bootstrap.go's same-named helper; both must
+// agree on sort, latest resolution, and successor derivation so the
+// streaming and walker paths reach the same verdict for the same
+// objects. Phase 5 deletes the scheduler copy.
+//
+// Resume note: every emitted sibling shares Path = logical key, so a
+// resume after a mid-expansion failure rewalks the whole sibling
+// group. Acceptable today because Phase 4b doesn't persist a
+// Checkpoint between runs (start is always "" via runShard).
+func expandVersionsDir(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketRoot, versionsKey string, versionsEntry *filer_pb.Entry, start string, skipBare map[string]bool, cb func(*bootstrap.Entry) error) error {
+	logical := strings.TrimSuffix(versionsKey, s3_constants.VersionsFolder)
+	if logical == "" {
+		return nil
+	}
+	versionsDir := bucketRoot + "/" + versionsKey
+	var children []*filer_pb.Entry
+	if err := listAll(ctx, client, versionsDir, func(e *filer_pb.Entry) error {
+		if e != nil && e.Attributes != nil && !e.IsDirectory {
+			children = append(children, e)
+		}
+		return nil
+	}); err != nil {
+		return 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. Treat it as a regular subdirectory so user-named
+		// files inside still surface.
+		return walkBucketTree(ctx, client, versionsDir, bucketRoot, start, cb)
+	}
+	if nullEntry, nullKey, explicit, ok := lookupNullVersion(ctx, client, bucketRoot, logical); ok {
+		items = append(items, versionItem{
+			entry:          nullEntry,
+			versionID:      "null",
+			bareKey:        nullKey,
+			isExplicitNull: explicit,
+		})
+	}
+
+	// Sort newest-first by mtime, ties broken by version_id (newer wins).
+	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:
+	//   1. Pointer names a real id -> that wins.
+	//   2. Pointer absent (or stale: set but no sibling carries it)
+	//      + items[0] is an EXPLICIT null -> null is latest.
+	//   3. Otherwise -> newest sibling (latestPos = 0 by default).
+	//
+	// A stale pointer falls through to the no-pointer fallback rather
+	// than silently leaving latestPos at 0 with no documented intent;
+	// the value happens to be the same today (newest sibling wins
+	// either way) but the explicit branching protects against future
+	// fallback refinements diverging by accident.
+	latestID := string(versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey])
+	latestPos := 0
+	pointerResolved := false
+	if latestID != "" {
+		for i, it := range items {
+			if it.versionID == latestID {
+				latestPos = i
+				pointerResolved = true
+				break
+			}
+		}
+	}
+	if !pointerResolved && len(items) > 0 && items[0].versionID == "null" && items[0].isExplicitNull {
+		latestPos = 0
+	}
+
+	if start != "" && logical <= start {
+		// All siblings share Path=logical, so the whole group is
+		// either above or below the resume marker.
+		return nil
+	}
+	for i, it := range items {
+		successor := s3lifecycle.SuccessorFromEntryStamp(it.entry)
+		if successor.IsZero() && i > 0 {
+			prev := items[i-1].entry.Attributes
+			successor = time.Unix(prev.Mtime, int64(prev.MtimeNs))
+		}
+		isLatest := i == latestPos
+		entry := &bootstrap.Entry{
+			Path:             logical,
+			VersionID:        it.versionID,
+			ModTime:          time.Unix(it.entry.Attributes.Mtime, int64(it.entry.Attributes.MtimeNs)),
+			Size:             int64(it.entry.Attributes.FileSize),
+			IsLatest:         isLatest,
+			IsDeleteMarker:   string(it.entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
+			NumVersions:      len(items),
+			SuccessorModTime: successor,
+		}
+		if !isLatest {
+			rank := i
+			if i > latestPos {
+				rank = i - 1
+			}
+			entry.NoncurrentIndex = &rank
+		}
+		if err := cb(entry); err != nil {
+			return err
+		}
+		if it.versionID == "null" && skipBare != nil {
+			skipBare[it.bareKey] = true
+		}
+	}
+	return nil
+}
+
+// lookupNullVersion returns the bare-key entry that represents the null
+// version of logical, if any. Both regular files and S3 directory-key
+// markers qualify. explicit is true when the entry carries
+// ExtVersionIdKey == "null" — the marker the suspended-versioning
+// write path applies; only an explicit-null bare can outrank a missing
+// `.versions/` pointer per the latest-resolution rules above.
+func lookupNullVersion(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketRoot, logical string) (*filer_pb.Entry, string, bool, bool) {
+	parent, name := util.NewFullPath(bucketRoot, logical).DirAndName()
+	resp, err := filer_pb.LookupEntry(ctx, client, &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
+	}
+	explicit := false
+	if id, hasID := e.Extended[s3_constants.ExtVersionIdKey]; hasID && string(id) == "null" {
+		explicit = true
+	}
+	return e, strings.TrimPrefix(parent+"/"+name, bucketRoot+"/"), explicit, true
+}
+
+// listAll issues paginated SeaweedList calls until exhausted. Ported
+// from scheduler/bootstrap.go's same-named helper; Phase 5 deletes
+// the scheduler copy when the streaming path is removed.
+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
+	}
+}
+
+func isVersionsDir(entry *filer_pb.Entry) bool {
+	return entry.IsDirectory && strings.HasSuffix(entry.Name, s3_constants.VersionsFolder)
+}
+
+// isMPUInitDir mirrors router.mpuInitInfo: a directory at
+// .uploads/<id> carrying the destination key in Extended is the MPU
+// init record. Verified shape + presence of ExtMultipartObjectKey;
+// directories at .uploads/<id> without the key are mid-write before
+// metadata landed and stay out of the dispatch path.
+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
+}

weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go

@@ -0,0 +1,442 @@
+package dailyrun
+
+import (
+	"context"
+	"io"
+	"sort"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+)
+
+// fakeFilerStream implements the ListEntries server-streaming client.
+type fakeFilerStream struct {
+	responses []*filer_pb.ListEntriesResponse
+	idx       int
+	ctx       context.Context
+}
+
+func (s *fakeFilerStream) Recv() (*filer_pb.ListEntriesResponse, error) {
+	if s.ctx != nil {
+		if err := s.ctx.Err(); err != nil {
+			return nil, err
+		}
+	}
+	if s.idx >= len(s.responses) {
+		return nil, io.EOF
+	}
+	r := s.responses[s.idx]
+	s.idx++
+	return r, nil
+}
+func (s *fakeFilerStream) Header() (metadata.MD, error) { return metadata.MD{}, nil }
+func (s *fakeFilerStream) Trailer() metadata.MD         { return metadata.MD{} }
+func (s *fakeFilerStream) CloseSend() error             { return nil }
+func (s *fakeFilerStream) Context() context.Context {
+	if s.ctx != nil {
+		return s.ctx
+	}
+	return context.Background()
+}
+func (s *fakeFilerStream) SendMsg(any) error { return nil }
+func (s *fakeFilerStream) RecvMsg(any) error { return nil }
+
+// fakeFiler maps directory paths to their immediate children. Only
+// ListEntries is implemented; other methods of SeaweedFilerClient are
+// inherited from the embedded interface and panic if called.
+type fakeFiler struct {
+	filer_pb.SeaweedFilerClient
+
+	mu   sync.Mutex
+	tree map[string][]*filer_pb.Entry
+}
+
+func (c *fakeFiler) 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 *fakeFiler) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	src := c.tree[in.Directory]
+	// Mirror the filer: sort by name, honor StartFromFileName exclusive,
+	// cap at Limit. listAll's pagination loop depends on these.
+	filtered := make([]*filer_pb.Entry, 0, len(src))
+	for _, e := range src {
+		if e == nil {
+			continue
+		}
+		if in.StartFromFileName != "" && !in.InclusiveStartFrom && 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 &fakeFilerStream{responses: resps, ctx: ctx}, nil
+}
+
+func file(name string, mtime time.Time, size int64) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name: name,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:    mtime.Unix(),
+			MtimeNs:  int32(mtime.Nanosecond()),
+			FileSize: uint64(size),
+		},
+	}
+}
+
+func dir(name string) *filer_pb.Entry {
+	return &filer_pb.Entry{Name: name, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}
+}
+
+func TestFilerListFunc_EmitsFlatFiles(t *testing.T) {
+	mtime := time.Now().Add(-7 * 24 * time.Hour)
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {file("a.txt", mtime, 10), file("b.txt", mtime, 20)},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got []string
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = append(got, e.Path)
+		return nil
+	}))
+	assert.Equal(t, []string{"a.txt", "b.txt"}, got)
+}
+
+func TestFilerListFunc_RecursesIntoSubdirs(t *testing.T) {
+	mtime := time.Now()
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt":         {dir("logs"), file("root.txt", mtime, 1)},
+		"/buckets/bkt/logs":    {dir("2026"), file("a.log", mtime, 5)},
+		"/buckets/bkt/logs/2026": {file("b.log", mtime, 7)},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var paths []string
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		paths = append(paths, e.Path)
+		return nil
+	}))
+	sort.Strings(paths)
+	assert.Equal(t, []string{"logs/2026/b.log", "logs/a.log", "root.txt"}, paths)
+}
+
+func TestFilerListFunc_MPUInitEmitsWithDestKey(t *testing.T) {
+	// .uploads/<id>/ with ExtMultipartObjectKey is the MPU init record.
+	// One Entry per init, IsMPUInit=true, DestKey = the user's path.
+	mtime := time.Now()
+	mpuDir := dir("upload-id-1")
+	mpuDir.Extended = map[string][]byte{
+		s3_constants.ExtMultipartObjectKey: []byte("user/path/object"),
+	}
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {
+			file("regular.txt", mtime, 1),
+			dir(s3_constants.MultipartUploadsFolder),
+		},
+		"/buckets/bkt/" + s3_constants.MultipartUploadsFolder: {
+			mpuDir,
+		},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got []*bootstrap.Entry
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = append(got, e)
+		return nil
+	}))
+	require.Len(t, got, 2)
+	byPath := map[string]*bootstrap.Entry{}
+	for _, e := range got {
+		byPath[e.Path] = e
+	}
+	require.NotNil(t, byPath["regular.txt"])
+	require.NotNil(t, byPath[s3_constants.MultipartUploadsFolder+"/upload-id-1"])
+	mpu := byPath[s3_constants.MultipartUploadsFolder+"/upload-id-1"]
+	assert.True(t, mpu.IsMPUInit)
+	assert.Equal(t, "user/path/object", mpu.DestKey)
+}
+
+func TestFilerListFunc_MPUInitWithoutDestKeyIsSkipped(t *testing.T) {
+	// A `.uploads/<id>` directory missing ExtMultipartObjectKey is
+	// mid-write before metadata landed; the dispatcher would error
+	// on empty DestKey. Skip it so the walk doesn't halt.
+	mtime := time.Now()
+	mpuDir := dir("upload-id-2") // no Extended
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {
+			file("regular.txt", mtime, 1),
+			dir(s3_constants.MultipartUploadsFolder),
+		},
+		"/buckets/bkt/" + s3_constants.MultipartUploadsFolder: {
+			mpuDir,
+		},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var paths []string
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		paths = append(paths, e.Path)
+		return nil
+	}))
+	assert.Equal(t, []string{"regular.txt"}, paths)
+}
+
+// fileWithExt is a versioned-file entry helper.
+func fileWithExt(name string, mtime time.Time, size int64, ext map[string][]byte) *filer_pb.Entry {
+	e := file(name, mtime, size)
+	e.Extended = ext
+	return e
+}
+
+func versionsDir(name string, latestID string) *filer_pb.Entry {
+	d := dir(name)
+	d.Extended = map[string][]byte{}
+	if latestID != "" {
+		d.Extended[s3_constants.ExtLatestVersionIdKey] = []byte(latestID)
+	}
+	return d
+}
+
+func TestFilerListFunc_VersionedExpansionMarksLatestByPointer(t *testing.T) {
+	// .versions/<key>/ with three real versions; parent's
+	// ExtLatestVersionIdKey points to v2 → IsLatest set on v2; the
+	// other two get NoncurrentIndex computed against the latest's
+	// position in the sorted (newest-first) list.
+	t1 := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	t2 := time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC)
+	t3 := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
+	versions := []*filer_pb.Entry{
+		fileWithExt("v1", t1, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}),
+		fileWithExt("v2", t2, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v2")}),
+		fileWithExt("v3", t3, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v3")}),
+	}
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v2")},
+		"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got []*bootstrap.Entry
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = append(got, e)
+		return nil
+	}))
+	require.Len(t, got, 3)
+
+	byID := map[string]*bootstrap.Entry{}
+	for _, e := range got {
+		byID[e.VersionID] = e
+		assert.Equal(t, "foo", e.Path, "every sibling's Path is the logical key")
+		assert.Equal(t, 3, e.NumVersions)
+	}
+	assert.True(t, byID["v2"].IsLatest, "pointer wins regardless of mtime order")
+	assert.False(t, byID["v1"].IsLatest)
+	assert.False(t, byID["v3"].IsLatest)
+	require.NotNil(t, byID["v3"].NoncurrentIndex, "noncurrent siblings get a rank")
+	require.NotNil(t, byID["v1"].NoncurrentIndex, "noncurrent siblings get a rank")
+}
+
+func TestFilerListFunc_VersionedExpansionStalePointerFallsBackToNewestSibling(t *testing.T) {
+	// ExtLatestVersionIdKey names a version that no sibling carries
+	// (stale pointer left behind by a write whose update was lost or
+	// raced). The fallback must NOT silently treat items[0] as latest
+	// via the default; structurally it must drop into the no-pointer
+	// path so the explicit-null bare check still runs.
+	tOld := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	tNew := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
+	versions := []*filer_pb.Entry{
+		fileWithExt("v_old", tOld, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vold")}),
+		fileWithExt("v_new", tNew, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vnew")}),
+	}
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v_GHOST")},
+		"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got []*bootstrap.Entry
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = append(got, e)
+		return nil
+	}))
+	require.Len(t, got, 2)
+	byID := map[string]*bootstrap.Entry{}
+	for _, e := range got {
+		byID[e.VersionID] = e
+	}
+	assert.True(t, byID["vnew"].IsLatest, "stale pointer must fall back to newest sibling")
+	assert.False(t, byID["vold"].IsLatest)
+}
+
+func TestFilerListFunc_VersionedExpansionNoPointerNewestSiblingWins(t *testing.T) {
+	// Parent has no ExtLatestVersionIdKey. With no explicit-null bare
+	// version, the newest-by-mtime sibling becomes latest.
+	tNew := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
+	tOld := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	versions := []*filer_pb.Entry{
+		fileWithExt("v_old", tOld, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vold")}),
+		fileWithExt("v_new", tNew, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vnew")}),
+	}
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "")},
+		"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got []*bootstrap.Entry
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = append(got, e)
+		return nil
+	}))
+	require.Len(t, got, 2)
+	byID := map[string]*bootstrap.Entry{}
+	for _, e := range got {
+		byID[e.VersionID] = e
+	}
+	assert.True(t, byID["vnew"].IsLatest, "newest sibling wins when pointer is absent")
+	assert.False(t, byID["vold"].IsLatest)
+}
+
+func TestFilerListFunc_VersionedExpansionExplicitNullIsLatestWhenPointerMissing(t *testing.T) {
+	// Suspended-versioning shape: bare object marked with
+	// ExtVersionIdKey="null", parent has no pointer. null is latest.
+	t1 := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	tNull := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC) // newest
+	bareNull := fileWithExt("foo", tNull, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("null")})
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {
+			bareNull,
+			versionsDir("foo"+s3_constants.VersionsFolder, ""),
+		},
+		"/buckets/bkt/foo" + s3_constants.VersionsFolder: {
+			fileWithExt("v1", t1, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}),
+		},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got []*bootstrap.Entry
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = append(got, e)
+		return nil
+	}))
+	// 2 sibling entries from expansion (null + v1). The bare "foo"
+	// MUST be suppressed in pass 2 via skipBare.
+	require.Len(t, got, 2)
+	byID := map[string]*bootstrap.Entry{}
+	for _, e := range got {
+		byID[e.VersionID] = e
+	}
+	require.NotNil(t, byID["null"])
+	require.NotNil(t, byID["v1"])
+	assert.True(t, byID["null"].IsLatest)
+	assert.False(t, byID["v1"].IsLatest)
+
+	// Walk again, verify no duplicate emission of the bare "foo".
+	count := 0
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		if e.Path == "foo" && e.VersionID == "" {
+			t.Errorf("bare foo should be suppressed by skipBare, got %+v", e)
+		}
+		count++
+		return nil
+	}))
+	assert.Equal(t, 2, count)
+}
+
+func TestFilerListFunc_VersionsDirWithoutMarkersRecursesAsRegular(t *testing.T) {
+	// A `.versions`-named folder whose children have no
+	// ExtVersionIdKey is a coincidence (user folder). Recurse into
+	// it; the file inside should surface as a regular entry.
+	mtime := time.Now()
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {versionsDir("looksLikeUserFolder"+s3_constants.VersionsFolder, "")},
+		"/buckets/bkt/looksLikeUserFolder" + s3_constants.VersionsFolder: {
+			file("inner.txt", mtime, 1),
+		},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var paths []string
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		paths = append(paths, e.Path)
+		return nil
+	}))
+	assert.Equal(t, []string{"looksLikeUserFolder" + s3_constants.VersionsFolder + "/inner.txt"}, paths)
+}
+
+func TestFilerListFunc_VersionedDeleteMarkerPropagates(t *testing.T) {
+	mtime := time.Now()
+	versions := []*filer_pb.Entry{
+		fileWithExt("v1", mtime, 0, map[string][]byte{
+			s3_constants.ExtVersionIdKey:    []byte("v1"),
+			s3_constants.ExtDeleteMarkerKey: []byte("true"),
+		}),
+	}
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v1")},
+		"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got *bootstrap.Entry
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = e
+		return nil
+	}))
+	require.NotNil(t, got)
+	assert.True(t, got.IsDeleteMarker, "ExtDeleteMarkerKey='true' must surface as IsDeleteMarker")
+}
+
+func TestFilerListFunc_HonorsStart(t *testing.T) {
+	// The walker's kill-resume contract: skip entries whose Path <= start.
+	mtime := time.Now()
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {file("a", mtime, 1), file("b", mtime, 1), file("c", mtime, 1)},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got []string
+	require.NoError(t, listFn(context.Background(), "bkt", "a", func(e *bootstrap.Entry) error {
+		got = append(got, e.Path)
+		return nil
+	}))
+	assert.Equal(t, []string{"b", "c"}, got)
+}
+
+func TestFilerListFunc_NilClient(t *testing.T) {
+	listFn := FilerListFunc(nil, "/buckets")
+	require.Error(t, listFn(context.Background(), "bkt", "", func(*bootstrap.Entry) error { return nil }))
+}
+
+func TestFilerListFunc_AttributesPropagate(t *testing.T) {
+	mtime := time.Date(2026, 5, 11, 12, 0, 0, 1234, time.UTC)
+	client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
+		"/buckets/bkt": {file("obj", mtime, 4096)},
+	}}
+	listFn := FilerListFunc(client, "/buckets")
+	var got *bootstrap.Entry
+	require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
+		got = e
+		return nil
+	}))
+	require.NotNil(t, got)
+	assert.Equal(t, "obj", got.Path)
+	assert.Equal(t, mtime.Unix(), got.ModTime.Unix())
+	assert.Equal(t, int64(1234), got.ModTime.UnixNano()-mtime.Unix()*int64(time.Second))
+	assert.Equal(t, int64(4096), got.Size)
+	assert.True(t, got.IsLatest)
+}

weed/s3api/s3lifecycle/dailyrun/replayability.go

@@ -1,69 +0,0 @@
-package dailyrun
-
-import (
-	"errors"
-	"fmt"
-
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
-)
-
-// UnsupportedRuleError fails the run loudly when the snapshot contains
-// a rule the Phase 2 replay path can't service. Surfaced verbatim to
-// the activity log so flipping algorithm=daily_replay on an
-// incompatible bucket isn't a silent dropped rule.
-type UnsupportedRuleError struct {
-	Bucket string
-	Kind   s3lifecycle.ActionKind
-	Reason string
-}
-
-func (e *UnsupportedRuleError) Error() string {
-	return fmt.Sprintf("daily_replay: unsupported action kind %s on bucket %q: %s", e.Kind, e.Bucket, e.Reason)
-}
-
-func IsUnsupportedRule(err error) bool {
-	var u *UnsupportedRuleError
-	return errors.As(err, &u)
-}
-
-func isReplayEligibleKind(k s3lifecycle.ActionKind) bool {
-	switch k {
-	case s3lifecycle.ActionKindExpirationDays,
-		s3lifecycle.ActionKindNoncurrentDays,
-		s3lifecycle.ActionKindAbortMPU:
-		return true
-	}
-	return false
-}
-
-// checkSnapshotForUnsupported rejects (a) walker-bound action kinds and
-// (b) replay-kind actions in any Mode other than ModeEventDriven.
-// router.Route silently drops non-ModeEventDriven actions; rejecting
-// them here turns the silent drop into a loud failure. Phase 4
-// partitions these into walk-bound actions and removes the gate.
-func checkSnapshotForUnsupported(snap *engine.Snapshot) *UnsupportedRuleError {
-	if snap == nil {
-		return nil
-	}
-	for _, a := range snap.AllActions() {
-		if a == nil || !a.IsActive() {
-			continue
-		}
-		if !isReplayEligibleKind(a.Key.ActionKind) {
-			return &UnsupportedRuleError{
-				Bucket: a.Bucket,
-				Kind:   a.Key.ActionKind,
-				Reason: "Phase 2 only routes ExpirationDays / NoncurrentDays / AbortMPU; ExpirationDate, ExpiredDeleteMarker, NewerNoncurrent land in Phase 4",
-			}
-		}
-		if a.Mode != engine.ModeEventDriven {
-			return &UnsupportedRuleError{
-				Bucket: a.Bucket,
-				Kind:   a.Key.ActionKind,
-				Reason: fmt.Sprintf("action is in Mode=%v (router.Route only dispatches ModeEventDriven); scan_only promotions land in Phase 4", a.Mode),
-			}
-		}
-	}
-	return nil
-}

weed/s3api/s3lifecycle/dailyrun/replayability_test.go

@@ -1,117 +0,0 @@
-package dailyrun
-
-import (
-	"testing"
-	"time"
-
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
-	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func newSnapshotWith(t *testing.T, inputs []engine.CompileInput) *engine.Snapshot {
-	t.Helper()
-	e := engine.New()
-	e.Compile(inputs, engine.CompileOptions{})
-	snap := e.Snapshot()
-	for _, a := range snap.AllActions() {
-		snap.MarkActive(a.Key)
-	}
-	return snap
-}
-
-func ruleExpirationDays(days int) *s3lifecycle.Rule {
-	return &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days}
-}
-
-func ruleExpirationDate(t time.Time) *s3lifecycle.Rule {
-	return &s3lifecycle.Rule{ID: "r-date", Status: s3lifecycle.StatusEnabled, ExpirationDate: t}
-}
-
-func ruleNoncurrentDays(days int) *s3lifecycle.Rule {
-	return &s3lifecycle.Rule{ID: "r-nc", Status: s3lifecycle.StatusEnabled, NoncurrentVersionExpirationDays: days}
-}
-
-func ruleAbortMPU(days int) *s3lifecycle.Rule {
-	return &s3lifecycle.Rule{ID: "r-mpu", Status: s3lifecycle.StatusEnabled, AbortMPUDaysAfterInitiation: days}
-}
-
-func ruleNewerNoncurrent(n int) *s3lifecycle.Rule {
-	return &s3lifecycle.Rule{ID: "r-newer", Status: s3lifecycle.StatusEnabled, NewerNoncurrentVersions: n}
-}
-
-func ruleExpiredDeleteMarker() *s3lifecycle.Rule {
-	return &s3lifecycle.Rule{ID: "r-edm", Status: s3lifecycle.StatusEnabled, ExpiredObjectDeleteMarker: true}
-}
-
-func TestCheckSnapshotForUnsupported_AllReplayKindsAccepted(t *testing.T) {
-	snap := newSnapshotWith(t, []engine.CompileInput{
-		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
-			ruleExpirationDays(30),
-			ruleNoncurrentDays(7),
-			ruleAbortMPU(7),
-		}},
-	})
-	require.Nil(t, checkSnapshotForUnsupported(snap))
-}
-
-func TestCheckSnapshotForUnsupported_ExpirationDateRejected(t *testing.T) {
-	snap := newSnapshotWith(t, []engine.CompileInput{
-		{Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpirationDate(time.Now().Add(48 * time.Hour))}},
-	})
-	err := checkSnapshotForUnsupported(snap)
-	require.NotNil(t, err)
-	assert.Equal(t, s3lifecycle.ActionKindExpirationDate, err.Kind)
-	assert.Equal(t, "b1", err.Bucket)
-}
-
-func TestCheckSnapshotForUnsupported_NewerNoncurrentRejected(t *testing.T) {
-	snap := newSnapshotWith(t, []engine.CompileInput{
-		{Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleNewerNoncurrent(2)}},
-	})
-	err := checkSnapshotForUnsupported(snap)
-	require.NotNil(t, err)
-	assert.Equal(t, s3lifecycle.ActionKindNewerNoncurrent, err.Kind)
-}
-
-func TestCheckSnapshotForUnsupported_ExpiredDeleteMarkerRejected(t *testing.T) {
-	snap := newSnapshotWith(t, []engine.CompileInput{
-		{Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpiredDeleteMarker()}},
-	})
-	err := checkSnapshotForUnsupported(snap)
-	require.NotNil(t, err)
-	assert.Equal(t, s3lifecycle.ActionKindExpiredDeleteMarker, err.Kind)
-}
-
-func TestCheckSnapshotForUnsupported_NonEventDrivenModeRejected(t *testing.T) {
-	// router.Route silently drops non-ModeEventDriven actions; gate
-	// must reject loudly.
-	snap := newSnapshotWith(t, []engine.CompileInput{
-		{Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpirationDays(30)}},
-	})
-	for _, a := range snap.AllActions() {
-		if a.Key.ActionKind == s3lifecycle.ActionKindExpirationDays {
-			a.Mode = engine.ModeScanOnly
-		}
-	}
-	err := checkSnapshotForUnsupported(snap)
-	require.NotNil(t, err)
-	assert.Equal(t, s3lifecycle.ActionKindExpirationDays, err.Kind)
-	assert.Contains(t, err.Reason, "ModeEventDriven")
-}
-
-func TestIsUnsupportedRule_TypeCheck(t *testing.T) {
-	var u error = &UnsupportedRuleError{Bucket: "b", Kind: s3lifecycle.ActionKindExpirationDate, Reason: "x"}
-	assert.True(t, IsUnsupportedRule(u))
-	assert.False(t, IsUnsupportedRule(nil))
-	assert.False(t, IsUnsupportedRule(assertNonNilError()))
-}
-
-func assertNonNilError() error { return errPlain }
-
-type plainErr struct{}
-
-func (plainErr) Error() string { return "plain" }
-
-var errPlain = plainErr{}

weed/s3api/s3lifecycle/dailyrun/run.go

@@ -25,6 +25,12 @@ type LifecycleClient interface {
 	LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error)
 }
 
+// WalkerFunc handles the per-shard bucket walk for a given engine view.
+// Phase 4b uses it on the recovery branch (rule-content edit / partition
+// flip) so already-due objects across the rewritten rule set get caught
+// before the cursor rewinds.
+type WalkerFunc func(ctx context.Context, view *engine.Snapshot, shardID int) error
+
 type Config struct {
 	Shards      []int
 	BucketsPath string
@@ -40,6 +46,21 @@ type Config struct {
 	// nil -> no rate limit. Shared across all shard goroutines.
 	Limiter *rate.Limiter
 
+	// Meta-log retention boundary. Rules whose effective TTL exceeds
+	// this can't be serviced by replay alone and get partitioned into
+	// the walk view (engine.PromotedHash). 0 falls back to maxTTL,
+	// which keeps PromotedHash empty and the partition-flip recovery
+	// trigger dormant.
+	RetentionWindow time.Duration
+
+	// Walker is invoked on the recovery branch (rule-content edit or
+	// partition flip) before the cursor rewind. It receives the
+	// engine.RecoveryView so only the rules that need bulk re-evaluation
+	// are walked, and the per-shard ID so the implementation can filter
+	// entries. nil disables walker invocation entirely — the cursor
+	// still rewinds, matching Phase 4a behavior.
+	Walker WalkerFunc
+
 	ClientName string
 	// 0 -> randomized per-run.
 	ClientID int32
@@ -68,9 +89,6 @@ func Run(ctx context.Context, cfg Config) error {
 
 	// Capture once so a mid-run Compile can't make shards disagree.
 	snap := cfg.Engine.Snapshot()
-	if unsupported := checkSnapshotForUnsupported(snap); unsupported != nil {
-		return unsupported
-	}
 
 	workers := cfg.Workers
 	if workers <= 0 {
@@ -136,10 +154,12 @@ func validate(cfg Config) error {
 }
 
 // runShard executes one daily-replay pass; see DESIGN.md for algorithm.
-// Phase 2: no walker on rule-change / cold-start; PromotedHash trigger
-// is dormant until Phase 4b wires real retention.
-// checkSnapshotForUnsupported already rejected walker-bound and
-// scan_only rules.
+// Two walker invocations under cfg.Walker (when set):
+//   - recovery branch: RecoveryView, so already-due objects across the
+//     rewritten rule set fire before the cursor rewinds.
+//   - steady state: RulesForShard's walk view, so walker-bound and
+//     scan_only-promoted rules fire every day even when replay rules
+//     are unchanged.
 func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int) error {
 	persisted, found, err := cfg.Persister.Load(ctx, shardID)
 	if err != nil {
@@ -152,9 +172,21 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim
 	// one-time rewind to runNow - maxTTL, self-healing on save.
 	rsh := engine.ReplayContentHash(snap)
 	maxTTL := engine.MaxEffectiveTTL(snap)
-	// retentionWindow=maxTTL keeps promoted empty (no rule's TTL
-	// exceeds the max). Phase 4b plumbs real meta-log retention.
-	promoted := engine.PromotedHash(snap, maxTTL)
+	// Operator-supplied retention falls back to maxTTL. In steady
+	// state every active replay rule has TTL <= maxTTL by construction,
+	// so promoted is empty and the partition-flip trigger is dormant.
+	// During bootstrap (rules compiled but not yet active) maxTTL is
+	// 0, retentionWindow is 0, and every rule with TTL > 0 lands in
+	// the walk partition; the resulting non-empty promoted forces a
+	// recovery walk on the first run after rules activate, which is
+	// the intended bootstrap behavior. Once the handler plumbs the
+	// real meta-log retention here, PromotedHash starts catching
+	// retention-driven partition flips in addition.
+	retentionWindow := cfg.RetentionWindow
+	if retentionWindow <= 0 {
+		retentionWindow = maxTTL
+	}
+	promoted := engine.PromotedHash(snap, retentionWindow)
 
 	if rsh == [32]byte{} {
 		return cfg.Persister.Save(ctx, shardID, Cursor{
@@ -165,10 +197,15 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim
 	}
 
 	// Recovery: rule-content edit (RuleSetHash mismatch) or partition
-	// flip (PromotedHash mismatch — dormant until real retention).
-	// Phase 4b adds the walker here; until then we rewind and let the
-	// sliding meta-log replay catch up.
+	// flip (PromotedHash mismatch). Walk the rewritten rule set so
+	// already-due objects fire before the cursor rewinds; then rewind
+	// and let the sliding meta-log replay catch up steady state.
 	if found && (persisted.RuleSetHash != rsh || persisted.PromotedHash != promoted) {
+		if cfg.Walker != nil {
+			if werr := cfg.Walker(ctx, engine.RecoveryView(snap), shardID); werr != nil {
+				return fmt.Errorf("shard=%d: recovery walk: %w", shardID, werr)
+			}
+		}
 		next := Cursor{
 			TsNs:         runNow.Add(-maxTTL).UnixNano(),
 			RuleSetHash:  rsh,
@@ -177,6 +214,20 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim
 		return cfg.Persister.Save(ctx, shardID, next)
 	}
 
+	// Steady-state walker for walker-bound and scan_only-promoted rules.
+	// RulesForShard splits the snapshot using the same retentionWindow
+	// PromotedHash used, so the walk view is exactly the partition the
+	// hash already accounted for. Empty walk view (no rules need walking
+	// today) skips the call so non-versioned, replay-only deployments
+	// don't pay an O(N) bucket-walk per run.
+	if cfg.Walker != nil {
+		if _, walkView := snap.RulesForShard(shardID, retentionWindow); walkView != nil && len(walkView.AllActions()) > 0 {
+			if werr := cfg.Walker(ctx, walkView, shardID); werr != nil {
+				return fmt.Errorf("shard=%d: steady walk: %w", shardID, werr)
+			}
+		}
+	}
+
 	// Cold start: scan from now-maxTTL so already-due objects within
 	// meta-log retention still expire.
 	startTsNs := persisted.TsNs

weed/s3api/s3lifecycle/dailyrun/walk_buckets.go

@@ -0,0 +1,76 @@
+package dailyrun
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+)
+
+// WalkBuckets runs bootstrap.Walk for each bucket using a per-bucket
+// ListFunc wrapped to drop entries whose ShardID doesn't belong to
+// shardID. Returns the first bucket error; remaining buckets log and
+// continue so one bucket's filer error doesn't kill the whole walk.
+//
+// Composable shape: callers (the worker handler) supply a real
+// filer-backed ListFunc and a real Dispatcher (WalkerDispatcher).
+// Tests pass bootstrap.EntryCallback and a stub Dispatcher.
+func WalkBuckets(ctx context.Context, view *engine.Snapshot, shardID int, buckets []string, list bootstrap.ListFunc, dispatch bootstrap.Dispatcher) error {
+	if view == nil {
+		return errors.New("WalkBuckets: nil view")
+	}
+	if list == nil {
+		return errors.New("WalkBuckets: nil list")
+	}
+	if dispatch == nil {
+		return errors.New("WalkBuckets: nil dispatch")
+	}
+	filtered := perShardListFunc(list, shardID)
+	var firstErr error
+	for _, b := range buckets {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		if _, err := bootstrap.Walk(ctx, view, b, filtered, dispatch, bootstrap.WalkOptions{}); err != nil {
+			if firstErr == nil {
+				firstErr = fmt.Errorf("walk %s: %w", b, err)
+			} else {
+				glog.V(1).Infof("walker: additional bucket %s: %v", b, err)
+			}
+		}
+	}
+	return firstErr
+}
+
+// perShardListFunc wraps an inner ListFunc so only entries whose
+// logical-key shard matches shardID reach the callback. The walker
+// emits one entry per logical object — versioned siblings share the
+// same logical key so they're either all in-shard or all out — and
+// MPU init records carry the user's destination in DestKey, so the
+// shard predicate must use DestKey there to match what the dispatcher
+// will send.
+func perShardListFunc(inner bootstrap.ListFunc, shardID int) bootstrap.ListFunc {
+	return func(ctx context.Context, bucket, start string, cb func(*bootstrap.Entry) error) error {
+		return inner(ctx, bucket, start, func(e *bootstrap.Entry) error {
+			if e == nil {
+				return nil
+			}
+			if entryShardID(bucket, e) != shardID {
+				return nil
+			}
+			return cb(e)
+		})
+	}
+}
+
+func entryShardID(bucket string, e *bootstrap.Entry) int {
+	key := e.Path
+	if e.IsMPUInit && e.DestKey != "" {
+		key = e.DestKey
+	}
+	return s3lifecycle.ShardID(bucket, key)
+}

weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go

@@ -0,0 +1,175 @@
+package dailyrun
+
+import (
+	"context"
+	"errors"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// recordingDispatcher captures every (action, entry) pair the walker
+// emits so tests can assert dispatch order and shard filtering.
+type recordingDispatcher struct {
+	calls []recordedDispatch
+	err   error
+}
+
+type recordedDispatch struct {
+	bucket string
+	path   string
+}
+
+func (d *recordingDispatcher) Delete(_ context.Context, action *engine.CompiledAction, entry *bootstrap.Entry) error {
+	d.calls = append(d.calls, recordedDispatch{bucket: action.Bucket, path: entry.Path})
+	return d.err
+}
+
+// findShardForPath returns the shard ID for an entry with given path
+// in given bucket. Helper for tests that want to force entries onto
+// a known shard.
+func findShardForPath(bucket, path string) int {
+	return s3lifecycle.ShardID(bucket, path)
+}
+
+// fixedShardEntries returns a slice of bootstrap.Entries whose paths
+// all live in the same shard. Used so a per-shard filter test has
+// deterministic expectations.
+func fixedShardEntries(t *testing.T, bucket string, shardID int, count int) []*bootstrap.Entry {
+	t.Helper()
+	var out []*bootstrap.Entry
+	for i := 0; tries(i, count); i++ {
+		path := "obj-" + intToStr(i)
+		if findShardForPath(bucket, path) == shardID {
+			out = append(out, &bootstrap.Entry{
+				Path: path,
+				// Old enough to expire under a 7-day ExpirationDays rule.
+				ModTime:  time.Now().Add(-90 * 24 * time.Hour),
+				Size:     1,
+				IsLatest: true,
+			})
+			if len(out) >= count {
+				return out
+			}
+		}
+	}
+	t.Fatalf("could not find %d entries in shard %d for bucket %s after 4096 attempts", count, shardID, bucket)
+	return nil
+}
+
+func intToStr(i int) string {
+	const hex = "0123456789abcdef"
+	if i == 0 {
+		return "0"
+	}
+	var out []byte
+	for n := i; n > 0; n /= 16 {
+		out = append([]byte{hex[n%16]}, out...)
+	}
+	return string(out)
+}
+
+func tries(i, count int) bool { return i < count*256+256 }
+
+func snapshotForBucketRule(t *testing.T, bucket string, days int) *engine.Snapshot {
+	t.Helper()
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: bucket, Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days},
+		}},
+	}, engine.CompileOptions{})
+	snap := e.Snapshot()
+	for _, a := range snap.AllActions() {
+		snap.MarkActive(a.Key)
+	}
+	return snap
+}
+
+func TestWalkBuckets_DispatchesOnlyShardMatchingEntries(t *testing.T) {
+	bucket := "b1"
+	const targetShard = 0
+	// Build entries: half in shard 0, half elsewhere. The walker's
+	// per-shard filter must drop the non-matching ones.
+	inShard := fixedShardEntries(t, bucket, targetShard, 3)
+	otherShard := fixedShardEntries(t, bucket, (targetShard+1)%16, 3)
+	all := append([]*bootstrap.Entry{}, inShard...)
+	all = append(all, otherShard...)
+
+	snap := snapshotForBucketRule(t, bucket, 7) // 30-day-old objects expire
+	d := &recordingDispatcher{}
+	err := WalkBuckets(context.Background(), snap, targetShard, []string{bucket},
+		bootstrap.EntryCallback(all), d)
+	require.NoError(t, err)
+
+	gotPaths := make([]string, 0, len(d.calls))
+	for _, c := range d.calls {
+		gotPaths = append(gotPaths, c.path)
+		assert.Equal(t, bucket, c.bucket)
+	}
+	sort.Strings(gotPaths)
+
+	wantPaths := make([]string, 0, len(inShard))
+	for _, e := range inShard {
+		wantPaths = append(wantPaths, e.Path)
+	}
+	sort.Strings(wantPaths)
+	assert.Equal(t, wantPaths, gotPaths, "walker must dispatch exactly the in-shard entries")
+}
+
+func TestWalkBuckets_NilGuards(t *testing.T) {
+	require.Error(t, WalkBuckets(context.Background(), nil, 0, nil, bootstrap.EntryCallback(nil), &recordingDispatcher{}))
+	require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), 0, nil, nil, &recordingDispatcher{}))
+	require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), 0, nil, bootstrap.EntryCallback(nil), nil))
+}
+
+func TestWalkBuckets_OneBucketErrorDoesNotStopOthers(t *testing.T) {
+	// Two buckets, first ListFunc errors. WalkBuckets should still
+	// process the second bucket but return the first bucket's error.
+	listErr := errors.New("filer flake")
+	list := func(_ context.Context, bucket, _ string, _ func(*bootstrap.Entry) error) error {
+		if bucket == "bad" {
+			return listErr
+		}
+		return nil
+	}
+
+	snap := snapshotForBucketRule(t, "good", 7)
+	err := WalkBuckets(context.Background(), snap, 0, []string{"bad", "good"}, list, &recordingDispatcher{})
+	require.Error(t, err)
+	assert.ErrorIs(t, err, listErr)
+}
+
+func TestWalkBuckets_HonorsContextCancellationBetweenBuckets(t *testing.T) {
+	// Pre-cancel ctx before invocation. Even an empty bucket list
+	// must surface ctx.Err early so a long walk in progress can
+	// short-circuit.
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	err := WalkBuckets(ctx, snapshotForBucketRule(t, "b", 7), 0, []string{"b"},
+		bootstrap.EntryCallback(nil), &recordingDispatcher{})
+	assert.ErrorIs(t, err, context.Canceled)
+}
+
+func TestEntryShardID_MPUUsesDestKey(t *testing.T) {
+	bucket := "bkt"
+	// Force a (Path, DestKey) pair that maps to different shards
+	// so we can prove the filter picks the right one.
+	for i := 0; i < 4096; i++ {
+		path := ".uploads/" + intToStr(i)
+		destKey := "user/" + intToStr(i)
+		if s3lifecycle.ShardID(bucket, path) != s3lifecycle.ShardID(bucket, destKey) {
+			e := &bootstrap.Entry{Path: path, DestKey: destKey, IsMPUInit: true}
+			assert.Equal(t, s3lifecycle.ShardID(bucket, destKey), entryShardID(bucket, e),
+				"MPU init must use DestKey for shard, not the .uploads/<id> path")
+			return
+		}
+	}
+	t.Fatal("could not find an MPU path/destkey pair with diverging shards in 4096 attempts")
+}

weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go

@@ -0,0 +1,87 @@
+package dailyrun
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+)
+
+// WalkerDispatcher adapts LifecycleClient to bootstrap.Dispatcher so
+// the Phase 4b walker can drive the same LifecycleDelete RPC the
+// meta-log replay path uses. No CAS witness is supplied; the server's
+// identityMatches treats nil ExpectedIdentity as "bootstrap call, skip
+// witness" (see weed/s3api/s3api_internal_lifecycle.go), which is the
+// right contract for a full-tree walk that has just observed the entry.
+type WalkerDispatcher struct {
+	Client LifecycleClient
+}
+
+// Compile-time check.
+var _ bootstrap.Dispatcher = (*WalkerDispatcher)(nil)
+
+// Delete classifies non-DONE server outcomes as errors so the walker
+// halts and the caller persists progress under the bootstrap
+// checkpoint rather than silently skipping objects.
+func (d *WalkerDispatcher) Delete(ctx context.Context, action *engine.CompiledAction, entry *bootstrap.Entry) error {
+	if d == nil || d.Client == nil {
+		return fmt.Errorf("walker dispatch: nil client")
+	}
+	if action == nil || entry == nil {
+		return fmt.Errorf("walker dispatch: nil action or entry")
+	}
+	objectPath := entry.Path
+	if entry.IsMPUInit {
+		// Rule-prefix matching used DestKey; the server takes the
+		// canonical object path for the LifecycleDelete RPC, which
+		// is also DestKey. The walker hits the .uploads/<id>
+		// directory itself only when ActionKind=ABORT_MPU, and the
+		// server resolves the upload from (bucket, object_path) +
+		// the init record's metadata.
+		if entry.DestKey == "" {
+			return fmt.Errorf("walker dispatch: MPU init entry with empty DestKey: %s", entry.Path)
+		}
+		objectPath = entry.DestKey
+	}
+	rh := action.Key.RuleHash
+	req := &s3_lifecycle_pb.LifecycleDeleteRequest{
+		Bucket:     action.Bucket,
+		ObjectPath: objectPath,
+		VersionId:  entry.VersionID,
+		RuleHash:   rh[:],
+		ActionKind: toProtoActionKind(action.Key.ActionKind),
+		// ExpectedIdentity intentionally nil; server bootstraps from
+		// the live entry on this code path.
+	}
+	kindLabel := action.Key.ActionKind.String()
+	resp, err := d.Client.LifecycleDelete(ctx, req)
+	if err != nil {
+		// "RPC_ERROR" matches the streaming dispatcher and processMatches
+		// labels so transport-class failures aggregate under one key.
+		stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, "RPC_ERROR").Inc()
+		return fmt.Errorf("walker dispatch %s/%s %s: %w", action.Bucket, objectPath, action.Key.ActionKind, err)
+	}
+	if resp == nil {
+		// A misbehaving server stub returning (nil, nil) would panic on
+		// the switch below. Bucket under RPC_ERROR rather than a new
+		// label — operationally it's the same class of failure.
+		stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, "RPC_ERROR").Inc()
+		return fmt.Errorf("walker dispatch %s/%s %s: nil response", action.Bucket, objectPath, action.Key.ActionKind)
+	}
+	stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, 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:
+		return nil
+	default:
+		// RETRY_LATER / BLOCKED / UNSPECIFIED: surface as error so the
+		// walk halts at this entry and resumes from
+		// Checkpoint.LastScannedPath on the next run.
+		return fmt.Errorf("walker dispatch %s/%s %s: outcome=%s reason=%s",
+			action.Bucket, objectPath, action.Key.ActionKind, resp.Outcome, resp.Reason)
+	}
+}

weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go

@@ -0,0 +1,159 @@
+package dailyrun
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// walkerStubClient captures the last LifecycleDeleteRequest so tests
+// can assert on the request shape produced by WalkerDispatcher.
+type walkerStubClient struct {
+	lastReq *s3_lifecycle_pb.LifecycleDeleteRequest
+	outcome s3_lifecycle_pb.LifecycleDeleteOutcome
+	err     error
+	reason  string
+	nilResp bool // return (nil, nil) — pin the dispatcher's defensive guard
+}
+
+func (c *walkerStubClient) LifecycleDelete(_ context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
+	c.lastReq = req
+	if c.err != nil {
+		return nil, c.err
+	}
+	if c.nilResp {
+		return nil, nil
+	}
+	return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: c.outcome, Reason: c.reason}, nil
+}
+
+func sampleAction(t *testing.T, kind s3lifecycle.ActionKind) *engine.CompiledAction {
+	t.Helper()
+	var rh [8]byte
+	for i := range rh {
+		rh[i] = byte(0xa0 + i)
+	}
+	return &engine.CompiledAction{
+		Bucket: "bkt",
+		Key:    s3lifecycle.ActionKey{Bucket: "bkt", ActionKind: kind, RuleHash: rh},
+		Mode:   engine.ModeEventDriven,
+	}
+}
+
+func TestWalkerDispatcher_NonVersionedSendsExpectedRequest(t *testing.T) {
+	c := &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
+	d := &WalkerDispatcher{Client: c}
+	a := sampleAction(t, s3lifecycle.ActionKindExpirationDays)
+	err := d.Delete(context.Background(), a, &bootstrap.Entry{Path: "obj"})
+	require.NoError(t, err)
+	require.NotNil(t, c.lastReq)
+	assert.Equal(t, "bkt", c.lastReq.Bucket)
+	assert.Equal(t, "obj", c.lastReq.ObjectPath)
+	assert.Equal(t, "", c.lastReq.VersionId)
+	assert.Equal(t, a.Key.RuleHash[:], c.lastReq.RuleHash)
+	assert.Equal(t, s3_lifecycle_pb.ActionKind_EXPIRATION_DAYS, c.lastReq.ActionKind)
+	// Bootstrap-style call: server skips CAS witness when nil.
+	assert.Nil(t, c.lastReq.ExpectedIdentity)
+}
+
+func TestWalkerDispatcher_VersionedPassesVersionID(t *testing.T) {
+	c := &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
+	d := &WalkerDispatcher{Client: c}
+	a := sampleAction(t, s3lifecycle.ActionKindNoncurrentDays)
+	err := d.Delete(context.Background(), a, &bootstrap.Entry{Path: "obj", VersionID: "v-abc"})
+	require.NoError(t, err)
+	assert.Equal(t, "v-abc", c.lastReq.VersionId)
+}
+
+func TestWalkerDispatcher_MPUInitUsesDestKey(t *testing.T) {
+	// Rule-prefix matching used DestKey; the RPC ObjectPath must match
+	// so the server resolves the upload from the user's intended key.
+	c := &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
+	d := &WalkerDispatcher{Client: c}
+	a := sampleAction(t, s3lifecycle.ActionKindAbortMPU)
+	err := d.Delete(context.Background(), a, &bootstrap.Entry{
+		Path:      ".uploads/abc123",
+		DestKey:   "user/path/object",
+		IsMPUInit: true,
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "user/path/object", c.lastReq.ObjectPath)
+}
+
+func TestWalkerDispatcher_MPUInitEmptyDestKeyErrors(t *testing.T) {
+	// An MPU init record with no DestKey is mid-write before metadata
+	// landed; skipping silently in the walker is fine, but the
+	// dispatcher must not invent a path.
+	c := &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}
+	d := &WalkerDispatcher{Client: c}
+	a := sampleAction(t, s3lifecycle.ActionKindAbortMPU)
+	err := d.Delete(context.Background(), a, &bootstrap.Entry{
+		Path:      ".uploads/abc123",
+		IsMPUInit: true,
+	})
+	require.Error(t, err)
+	assert.Nil(t, c.lastReq, "no RPC should be sent when DestKey is empty")
+}
+
+func TestWalkerDispatcher_AcceptsAllResolvedOutcomes(t *testing.T) {
+	for _, oc := range []s3_lifecycle_pb.LifecycleDeleteOutcome{
+		s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
+		s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED,
+		s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK,
+	} {
+		c := &walkerStubClient{outcome: oc}
+		d := &WalkerDispatcher{Client: c}
+		err := d.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"})
+		assert.NoError(t, err, "outcome %s must be treated as resolved", oc)
+	}
+}
+
+func TestWalkerDispatcher_UnresolvedOutcomeReturnsError(t *testing.T) {
+	// RETRY_LATER, BLOCKED, and UNSPECIFIED all halt the walk so it
+	// resumes from Checkpoint.LastScannedPath on the next run.
+	for _, oc := range []s3_lifecycle_pb.LifecycleDeleteOutcome{
+		s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER,
+		s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED,
+		s3_lifecycle_pb.LifecycleDeleteOutcome_LIFECYCLE_DELETE_OUTCOME_UNSPECIFIED,
+	} {
+		c := &walkerStubClient{outcome: oc, reason: "server said so"}
+		d := &WalkerDispatcher{Client: c}
+		err := d.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"})
+		require.Error(t, err, "outcome %s must halt the walk", oc)
+		assert.Contains(t, err.Error(), oc.String())
+	}
+}
+
+func TestWalkerDispatcher_TransportErrorReturnsWrappedError(t *testing.T) {
+	c := &walkerStubClient{err: errors.New("transport boom")}
+	d := &WalkerDispatcher{Client: c}
+	err := d.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "transport boom")
+}
+
+func TestWalkerDispatcher_NilResponseReturnsError(t *testing.T) {
+	// A server returning (nil, nil) would otherwise panic on the
+	// outcome switch.
+	c := &walkerStubClient{nilResp: true}
+	d := &WalkerDispatcher{Client: c}
+	err := d.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "nil response")
+}
+
+func TestWalkerDispatcher_NilGuardsReturnError(t *testing.T) {
+	d := &WalkerDispatcher{Client: &walkerStubClient{}}
+	require.Error(t, d.Delete(context.Background(), nil, &bootstrap.Entry{Path: "obj"}))
+	require.Error(t, d.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), nil))
+
+	nilClient := &WalkerDispatcher{}
+	require.Error(t, nilClient.Delete(context.Background(), sampleAction(t, s3lifecycle.ActionKindExpirationDays), &bootstrap.Entry{Path: "obj"}))
+}

weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go

@@ -0,0 +1,143 @@
+package dailyrun
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// memPersister is a minimal in-memory CursorPersister. Phase 4b tests
+// drive runShard directly; the recovery-branch path returns before
+// drainShardEvents so the heavier filer/client/lister fakes aren't
+// needed here.
+type memPersister struct {
+	store map[int]Cursor
+}
+
+func newMemPersister() *memPersister { return &memPersister{store: map[int]Cursor{}} }
+
+func (p *memPersister) Load(_ context.Context, shardID int) (Cursor, bool, error) {
+	c, ok := p.store[shardID]
+	return c, ok, nil
+}
+
+func (p *memPersister) Save(_ context.Context, shardID int, c Cursor) error {
+	p.store[shardID] = c
+	return nil
+}
+
+func snapshotWithRule(t *testing.T, days int) *engine.Snapshot {
+	t.Helper()
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days},
+		}},
+	}, engine.CompileOptions{})
+	snap := e.Snapshot()
+	for _, a := range snap.AllActions() {
+		snap.MarkActive(a.Key)
+	}
+	return snap
+}
+
+func TestRunShard_WalkerInvokedOnRecoveryBranch(t *testing.T) {
+	snap := snapshotWithRule(t, 30)
+	p := newMemPersister()
+	// Seed a persisted cursor whose RuleSetHash differs from snap's, so
+	// the rule-change branch fires.
+	var stale [32]byte
+	for i := range stale {
+		stale[i] = 0xAA
+	}
+	require.NoError(t, p.Save(context.Background(), 3, Cursor{TsNs: 1234, RuleSetHash: stale}))
+
+	var gotView *engine.Snapshot
+	var gotShard int
+	calls := 0
+	cfg := Config{
+		Persister: p,
+		Walker: func(_ context.Context, view *engine.Snapshot, shardID int) error {
+			calls++
+			gotView = view
+			gotShard = shardID
+			return nil
+		},
+	}
+	runNow := time.Unix(1_700_000_000, 0).UTC()
+	require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 3))
+
+	assert.Equal(t, 1, calls, "walker must fire exactly once on recovery")
+	require.NotNil(t, gotView, "walker received the RecoveryView")
+	assert.Equal(t, 3, gotShard)
+
+	// Cursor rewound to runNow - maxTTL with the new hashes persisted.
+	got, ok, err := p.Load(context.Background(), 3)
+	require.NoError(t, err)
+	require.True(t, ok)
+	expectedFloor := runNow.Add(-engine.MaxEffectiveTTL(snap)).UnixNano()
+	assert.Equal(t, expectedFloor, got.TsNs, "cursor must rewind to runNow - maxTTL after recovery walk")
+	assert.NotEqual(t, stale, got.RuleSetHash, "stale hash must be replaced")
+}
+
+func TestRunShard_NilWalkerOnRecoveryIsNoop(t *testing.T) {
+	// Phase 4a behavior: a nil Walker must not crash and must still
+	// rewind the cursor.
+	snap := snapshotWithRule(t, 30)
+	p := newMemPersister()
+	var stale [32]byte
+	for i := range stale {
+		stale[i] = 0xBB
+	}
+	require.NoError(t, p.Save(context.Background(), 0, Cursor{TsNs: 9999, RuleSetHash: stale}))
+
+	cfg := Config{Persister: p} // Walker nil
+	runNow := time.Unix(1_700_000_000, 0).UTC()
+	require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0))
+
+	got, ok, err := p.Load(context.Background(), 0)
+	require.NoError(t, err)
+	require.True(t, ok)
+	assert.Equal(t, runNow.Add(-engine.MaxEffectiveTTL(snap)).UnixNano(), got.TsNs)
+}
+
+func TestRunShard_WalkerErrorPropagates(t *testing.T) {
+	// A walker failure must surface as the runShard error so the daily
+	// run treats the shard as interrupted, and must NOT advance the
+	// cursor (the seeded cursor stays).
+	snap := snapshotWithRule(t, 30)
+	p := newMemPersister()
+	var stale [32]byte
+	for i := range stale {
+		stale[i] = 0xCC
+	}
+	require.NoError(t, p.Save(context.Background(), 7, Cursor{TsNs: 42, RuleSetHash: stale}))
+
+	cfg := Config{
+		Persister: p,
+		Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error {
+			return errors.New("walker boom")
+		},
+	}
+	runNow := time.Unix(1_700_000_000, 0).UTC()
+	err := runShard(context.Background(), cfg, snap, runNow, 7)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "walker boom")
+
+	// Cursor untouched.
+	got, _, _ := p.Load(context.Background(), 7)
+	assert.Equal(t, int64(42), got.TsNs, "walker failure must leave cursor unchanged")
+	assert.Equal(t, stale, got.RuleSetHash)
+}
+
+// The "matching cursor doesn't invoke walker" case is implicit: the
+// walker call is inside the rule-change / partition-flip `if` and
+// can't be reached otherwise. Exercising it end-to-end requires the
+// full filer + lister + client harness; covered by the integration
+// tests once Phase 4b is wired into the handler.

weed/worker/tasks/s3_lifecycle/handler.go

@@ -319,10 +319,29 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe
 
 	limiter, limiterDesc := buildLimiterFromClusterContext(request.GetClusterContext())
 
+	// Reuse one LifecycleClient across the replay drain and the
+	// walker's per-entry dispatch.
+	client := lifecycleRPCAdapter{c: rpc}
+
+	// Bucket list for the walker — derived from inputs so it matches
+	// the snapshot the engine compiled.
+	buckets := make([]string, 0, len(inputs))
+	for _, in := range inputs {
+		if in.Bucket != "" {
+			buckets = append(buckets, in.Bucket)
+		}
+	}
+	walkerListFn := dailyrun.FilerListFunc(filerClient, bucketsPath)
+	walkerDispatch := &dailyrun.WalkerDispatcher{Client: client}
+	walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shardID int) error {
+		return dailyrun.WalkBuckets(walkCtx, view, shardID, buckets, walkerListFn, walkerDispatch)
+	})
+
 	_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
 		JobId: request.Job.JobId, JobType: jobType,
 		State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "starting",
-		Message: fmt.Sprintf("daily_replay shards=%d workers=%d runtime=%s rate=%s", len(shards), cfg.Workers, cfg.MaxRuntime, limiterDesc),
+		Message: fmt.Sprintf("daily_replay shards=%d workers=%d runtime=%s rate=%s buckets=%d walker=on",
+			len(shards), cfg.Workers, cfg.MaxRuntime, limiterDesc, len(buckets)),
 	})
 
 	runErr := dailyrun.Run(ctx, dailyrun.Config{
@@ -330,19 +349,14 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe
 		BucketsPath: bucketsPath,
 		Engine:      eng,
 		FilerClient: filerClient,
-		Client:      lifecycleRPCAdapter{c: rpc},
+		Client:      client,
 		Persister:   &dailyrun.FilerCursorPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
 		Lister:      dispatcher.NewFilerSiblingLister(filerClient, bucketsPath),
 		Workers:     cfg.Workers,
 		Limiter:     limiter,
+		Walker:      walker,
 		ClientName:  "worker-s3-lifecycle-daily",
 	})
-	if dailyrun.IsUnsupportedRule(runErr) {
-		// Surface the typed error verbatim so admin marks the run as
-		// failed with the user-facing reason in the activity log.
-		glog.Warningf("daily_replay: %v", runErr)
-		return runErr
-	}
 	if runErr != nil {
 		glog.Warningf("daily_replay: %v", runErr)
 		return runErr