diff --git a/weed/s3api/s3lifecycle/bootstrap/walker.go b/weed/s3api/s3lifecycle/bootstrap/walker.go index adedb4fcd..ac3679872 100644 --- a/weed/s3api/s3lifecycle/bootstrap/walker.go +++ b/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 diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go new file mode 100644 index 000000000..39c5192c2 --- /dev/null +++ b/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 /. 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/ 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/` 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/ carrying the destination key in Extended is the MPU +// init record. Verified shape + presence of ExtMultipartObjectKey; +// directories at .uploads/ 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 +} diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go new file mode 100644 index 000000000..47ec0514a --- /dev/null +++ b/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// 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/` 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// 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) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/replayability.go b/weed/s3api/s3lifecycle/dailyrun/replayability.go deleted file mode 100644 index 5f72b5962..000000000 --- a/weed/s3api/s3lifecycle/dailyrun/replayability.go +++ /dev/null @@ -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 -} diff --git a/weed/s3api/s3lifecycle/dailyrun/replayability_test.go b/weed/s3api/s3lifecycle/dailyrun/replayability_test.go deleted file mode 100644 index b79ad840a..000000000 --- a/weed/s3api/s3lifecycle/dailyrun/replayability_test.go +++ /dev/null @@ -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{} diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index b684d7fba..d334fc8fc 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/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 diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go new file mode 100644 index 000000000..bfd63afae --- /dev/null +++ b/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) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go new file mode 100644 index 000000000..4f3f5b75c --- /dev/null +++ b/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/ path") + return + } + } + t.Fatal("could not find an MPU path/destkey pair with diverging shards in 4096 attempts") +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go new file mode 100644 index 000000000..2dede7c82 --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go @@ -0,0 +1,85 @@ +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/ + // 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 { + stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, "TRANSPORT_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. Surface as an error so the walk halts at + // this entry, preserving the in-flight cursor's correctness. + stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, "NIL_RESPONSE").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) + } +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go new file mode 100644 index 000000000..a826ee51b --- /dev/null +++ b/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"})) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go new file mode 100644 index 000000000..aba8ed877 --- /dev/null +++ b/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. diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 98c2404fd..7e927e9db 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/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