feat(s3/lifecycle): versioned-sibling expansion in FilerListFunc

Adds the .versions/<key>/ 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/<file> 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/<id>) 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.
This commit is contained in:
Chris Lu
2026-05-11 23:10:43 -07:00
parent 3fdfa8113f
commit 2cc8a56f86
2 changed files with 379 additions and 24 deletions
@@ -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 <bucketsPath>/<bucket> for use by the daily-run walker.
// 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.
//
// Phase 4b scope: non-versioned, non-MPU entries only. Versioned
// `.versions/` directories and MPU init records at `.uploads/<id>`
// 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/<id> 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/<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 + 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.
@@ -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/<key>/` and `.uploads/<id>/` 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/<id>/` 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/<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_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) {