From 2cc8a56f8609a21a95b437e91b585d34f37a364d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 11 May 2026 23:10:43 -0700 Subject: [PATCH] feat(s3/lifecycle): versioned-sibling expansion in FilerListFunc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the .versions// expansion to the daily-run's filer-backed ListFunc. Each call emits one bootstrap.Entry per sibling (real version files + the bare null version, when found) with the same sibling state the streaming bootstrap injects via reader.Event: - Path = logical key (not the .versions/ physical path), so bootstrap.Walk's MatchPath uses the user's intended path. - VersionID per sibling (version_id or "null"). - IsLatest resolved via parent's ExtLatestVersionIdKey, falling back to explicit-null-bare, falling back to newest-by-mtime. - NoncurrentIndex rank computed against the latest's position. - SuccessorModTime: SuccessorFromEntryStamp if stamped, else the previous-newer sibling's mtime (legacy derivation). - IsDeleteMarker from ExtDeleteMarkerKey. - NumVersions = len(siblings). Two-pass walk so .versions/ dirs run before regular files; the bare null-version path is recorded in skipBare so pass 2 doesn't emit it twice. expandVersionsDir and lookupNullVersion are ported from scheduler/bootstrap.go. Sort order, latest resolution, and successor derivation must agree with that path verbatim so streaming and walker reach the same verdict on the same objects. Phase 5 deletes the scheduler copy. MPU init (.uploads/) remains skipped — the dedicated commit emits it with IsMPUInit and DestKey. Tests pin: pointer-wins latest resolution, no-pointer newest-sibling fallback, explicit-null-is-latest with skipBare suppression of the bare emission, coincidentally-named .versions folder recursing as a regular subdir, delete-marker propagation. --- .../s3lifecycle/dailyrun/filer_list_func.go | 210 ++++++++++++++++-- .../dailyrun/filer_list_func_test.go | 193 +++++++++++++++- 2 files changed, 379 insertions(+), 24 deletions(-) diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go index c7c676c88..e20659707 100644 --- a/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go +++ b/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go @@ -3,13 +3,16 @@ 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 @@ -21,12 +24,13 @@ var listPageSize atomic.Uint32 func init() { listPageSize.Store(1024) } // FilerListFunc returns a bootstrap.ListFunc that streams entries -// under / for use by the daily-run walker. +// 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. // -// Phase 4b scope: non-versioned, non-MPU entries only. Versioned -// `.versions/` directories and MPU init records at `.uploads/` -// are skipped at this stage; the follow-up commit adds the sibling -// expansion needed for noncurrent retention math and MPU dispatch. +// MPU init records at .uploads/ are skipped here; the follow-up +// commit emits one Entry per init with IsMPUInit and DestKey set. 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 { @@ -37,29 +41,52 @@ func FilerListFunc(client filer_pb.SeaweedFilerClient, bucketsPath string) boots } } -// walkBucketTree recursively lists dir, emitting each file as a -// bootstrap.Entry whose Path is bucket-relative. Subdirectories -// recurse; `.versions/` and `.uploads/` directories are skipped for -// now (Phase 4b follow-up). +// 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 isVersionsDir(e) { - // TODO(phase4b): expand into per-version Entries. - return nil - } if isMPUInitDirShape(key) { // TODO(phase4b): emit MPU init as a single Entry. return nil } return walkBucketTree(ctx, client, full, bucketRoot, start, cb) } - // Resume contract: skip entries Path <= start. + if skipBare[key] { + return nil + } if start != "" && key <= start { return nil } @@ -67,12 +94,165 @@ func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir Path: key, ModTime: time.Unix(e.Attributes.Mtime, int64(e.Attributes.MtimeNs)), Size: int64(e.Attributes.FileSize), - IsLatest: true, // Non-versioned default; versioned expansion overrides. + 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 + items[0] is an EXPLICIT null -> null is latest. + // 3. Pointer absent otherwise -> newest sibling. + latestID := string(versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey]) + latestPos := 0 + if latestID != "" { + for i, it := range items { + if it.versionID == latestID { + latestPos = i + break + } + } + } else if len(items) > 0 && items[0].versionID == "null" && items[0].isExplicitNull { + latestPos = 0 + } + + 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. diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go index 2845cf9e8..5479d2c50 100644 --- a/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go @@ -59,6 +59,17 @@ type fakeFiler struct { 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() @@ -132,20 +143,15 @@ func TestFilerListFunc_RecursesIntoSubdirs(t *testing.T) { assert.Equal(t, []string{"logs/2026/b.log", "logs/a.log", "root.txt"}, paths) } -func TestFilerListFunc_SkipsVersionsAndUploadsDirsForNow(t *testing.T) { - // Phase 4b-pre: `.versions//` and `.uploads//` are not yet - // expanded. Pin that they don't leak raw children into the dispatch - // path; the follow-up commit adds the proper sibling/MPU expansion. +func TestFilerListFunc_SkipsUploadsDirsForNow(t *testing.T) { + // Phase 4b-pre: `.uploads//` MPU init records are skipped + // here. The follow-up commit emits one IsMPUInit Entry per init. mtime := time.Now() client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ "/buckets/bkt": { file("regular.txt", mtime, 1), - dir("foo" + s3_constants.VersionsFolder), dir(s3_constants.MultipartUploadsFolder), }, - "/buckets/bkt/foo" + s3_constants.VersionsFolder: { - file("v_001", mtime, 1), - }, "/buckets/bkt/" + s3_constants.MultipartUploadsFolder: { dir("upload-id-1"), }, @@ -156,7 +162,176 @@ func TestFilerListFunc_SkipsVersionsAndUploadsDirsForNow(t *testing.T) { paths = append(paths, e.Path) return nil })) - assert.Equal(t, []string{"regular.txt"}, paths, ".versions/ and .uploads/ must not surface raw children") + assert.Equal(t, []string{"regular.txt"}, paths, ".uploads/ must not surface raw children") +} + +// 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_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) {