From f954781169002cd3f16280d1b7b9f85dd86a99b2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 12 May 2026 11:39:15 -0700 Subject: [PATCH] =?UTF-8?q?feat(s3/lifecycle):=20Phase=204b=20=E2=80=94=20?= =?UTF-8?q?daily=20walker=20for=20recovery=20and=20steady=20state=20(#9459?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(s3/lifecycle): plumb RetentionWindow into dailyrun.Config Adds a Config.RetentionWindow field that runShard threads into engine.PromotedHash. Zero (the default) falls back to maxTTL, which matches Phase 4a behavior — PromotedHash stays empty and the partition-flip recovery trigger stays dormant. Pure plumbing. The handler still passes zero so nothing changes at runtime. The walker work (Phase 4b proper) sets a real retention from the meta-log boundary and the partition-flip trigger starts firing. * feat(s3/lifecycle): WalkerDispatcher adapter for the daily-run walker Phase 4b prep. Implements bootstrap.Dispatcher on top of LifecycleClient so the same LifecycleDelete RPC drives both the meta-log replay path and the walker. No CAS witness — the server's identityMatches treats nil ExpectedIdentity as a bootstrap call and rebuilds the witness from the live entry, which is the right contract for a full-tree walk. Adds VersionID to bootstrap.Entry so versioned-bucket walks address the right version. MPU init uses DestKey for ObjectPath (matching the prefix-match contract); rejecting empty DestKey keeps malformed init records out of the dispatch path. Not wired yet — runShard still doesn't invoke the walker. Follow-up commits add the ListFunc adapter and the recovery-branch wiring. * feat(s3/lifecycle): wire Walker hook into runShard's recovery branch Adds a Config.Walker callback that fires on rule-content edit / partition flip BEFORE the cursor rewinds, so already-due objects across the rewritten rule set get caught instead of waiting on meta-log replay alone. The callback receives engine.RecoveryView(snap) and the per-shard ID; nil disables it (Phase 4a behavior preserved). Decoupling the wiring from the implementation: the handler-side WalkerFunc that drives bootstrap.Walk via the filer is the follow-up commit, and tests can stub the callback without standing up the full filer/client/lister harness. Tests pin: walker fires exactly once on hash mismatch, walker error propagates and leaves the cursor unchanged, nil Walker is a no-op. * feat(s3/lifecycle): WalkBuckets composes ListFunc + Dispatcher per shard Adds dailyrun.WalkBuckets — the composable driver the handler-side WalkerFunc will call. Iterates a bucket list, wraps the supplied bootstrap.ListFunc with a per-shard filter (Path for non-MPU, DestKey for MPU init), and runs bootstrap.Walk per bucket using the supplied Dispatcher. First bucket error wins; remaining buckets log and run to completion so one filer flake doesn't kill the shard. Composable rather than monolithic so callers and tests can swap parts: production uses a filer-backed ListFunc + WalkerDispatcher; tests use bootstrap.EntryCallback + a stub. The filer-backed ListFunc is the next commit. Tests pin: shard filter routes only matching entries, MPU shard uses DestKey not the .uploads/ path, single-bucket error propagates while other buckets still run, ctx cancellation short-circuits between buckets, nil guards on view/list/dispatch. * feat(s3/lifecycle): filer-backed ListFunc for the daily-run walker Phase 4b: dailyrun.FilerListFunc returns a bootstrap.ListFunc that streams entries under / by paginated SeaweedList. Recurses into regular directories; .versions/ and .uploads/ are skipped at this stage so they don't surface as raw children — the sibling expansion (versioned NoncurrentDays state, MPU init dispatch) lands in the next commit. listAll and isVersionsDir are ported from scheduler/bootstrap.go's same-named helpers. Phase 5 deletes the scheduler copies along with the streaming path. Tests pin: flat listing, recursion through nested directories, .versions/ and .uploads/ skipped, kill-resume via the start path contract, nil-client error, attribute propagation (mtime / size / IsLatest default). * feat(s3/lifecycle): versioned-sibling expansion in FilerListFunc 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. * feat(s3/lifecycle): emit MPU init records from FilerListFunc Last gap in the filer-backed ListFunc. A directory at .uploads/ carrying ExtMultipartObjectKey is the MPU init record; emit one bootstrap.Entry with IsMPUInit=true and DestKey set to the user's intended path. The walker's MatchPath uses DestKey for prefix matching; the WalkerDispatcher uses it for the LifecycleDelete RPC's ObjectPath. .uploads/ directories without the extended key are mid-write before metadata landed and stay skipped. isMPUInitDir is upgraded from the path-shape-only stub to the full shape + extended-attr check that mirrors router.mpuInitInfo and scheduler/bootstrap.go's same-named helper. Tests pin: valid init record emits with the right DestKey, missing ExtMultipartObjectKey skips the directory. * feat(s3/lifecycle): wire walker into executeDailyReplay Activates the recovery-branch walker. The handler composes the three Phase 4b building blocks — FilerListFunc + WalkerDispatcher + WalkBuckets — into a dailyrun.WalkerFunc and passes it via Config.Walker. The bucket list is derived from the compiled inputs so it matches the engine snapshot exactly. Effect on master behavior: when a worker observes a RuleSetHash or PromotedHash mismatch on its persisted cursor (rule content edited / partition flip), runShard now walks the live filer tree under the RecoveryView before rewinding the cursor. Already-due objects across the rewritten rule set fire immediately instead of waiting on the sliding meta-log replay. Still scoped to replay-eligible action kinds because checkSnapshotForUnsupported continues to reject walker-bound rules (ExpirationDate / ExpiredDeleteMarker / NewerNoncurrent) and scan_only-promoted rules at the top of Run. The follow-up commit relaxes the gate once the steady-state walker over RulesForShard's walk view is wired so those rules fire every day, not just on rule edits. * feat(s3/lifecycle): steady-state walker + drop unsupported-rule gate Adds the second walker invocation in runShard. After the recovery check passes, runShard derives the walk view via snap.RulesForShard (using the same retentionWindow PromotedHash used, so the partition is consistent) and runs the walker over it. The view holds walker-bound action kinds (ExpirationDate / ExpiredDeleteMarker / NewerNoncurrent) plus any replay-eligible rules promoted to walk by retention shortage; an empty view skips the call so non-versioned, replay-only deployments don't pay an O(N) bucket walk per run. With the walker now servicing every rule kind, checkSnapshotForUnsupported and its UnsupportedRuleError type are obsolete. router.Route gates replay on Mode == ModeEventDriven, so walker-bound and scan_only rules are silently dropped by replay and picked up by the walker instead — no double-dispatch. Drop the gate, delete replayability.go + replayability_test.go, and remove the handler's redundant IsUnsupportedRule branch. * fix(s3/lifecycle): walker dispatcher nil-response guard + retention-comment Two PR-review fixes on 9459: 1. WalkerDispatcher.Delete used to panic on a (nil, nil) RPC return — add a defensive nil-response check so the walk halts cleanly instead. Spotted by coderabbit. 2. The retentionWindow=maxTTL comment in runShard claimed PromotedHash "stays empty" in fallback mode, which gemini correctly pointed out is only true once rules are active. During bootstrap (rules compiled but IsActive=false) MaxEffectiveTTL is 0 while PromotedHash counts every non-disabled rule, so promoted becomes non-empty and the next post-activation run hits the recovery branch. That's the intended bootstrap walk — rewrite the comment to explain it rather than misstate the invariant. Test: pins nil-response → error path on WalkerDispatcher. * fix(s3/lifecycle): explicit stale-pointer fallback in versioned expansion Reviewer caught a structural bug in expandVersionsDir's latest resolution: when ExtLatestVersionIdKey was set but no scanned sibling carried that id (stale pointer), the code left latestPos at the default 0 without ever entering the no-pointer fallback. Today the two paths yield the same value (newest sibling wins), but the implicit fall-through makes the intent unclear and would break silently if the no-pointer branch ever did anything more than latestPos=0. Track a pointerResolved flag explicitly so the no-pointer branch (including the explicit-null-bare check) re-runs on a stale pointer. Behavior unchanged today. Test pins: stale pointer + two real versions falls back to newest-sibling (vnew, not vold). * feat(s3/lifecycle): walker-side dispatch metrics in WalkerDispatcher Mirrors the Phase 6 instrumentation already on the replay side (processMatches) onto the walker's Delete dispatch. Every walker dispatch now bumps S3LifecycleDispatchCounter with the resolved outcome (or TRANSPORT_ERROR / NIL_RESPONSE for the failure paths) so streaming, daily_replay's replay drain, and daily_replay's walker share a single per-(bucket, kind, outcome) counter view. Lands together with the rest of Phase 4b — no new metric, just an extra observation site for the existing one. --- weed/s3api/s3lifecycle/bootstrap/walker.go | 7 + .../s3lifecycle/dailyrun/filer_list_func.go | 320 +++++++++++++ .../dailyrun/filer_list_func_test.go | 442 ++++++++++++++++++ .../s3lifecycle/dailyrun/replayability.go | 69 --- .../dailyrun/replayability_test.go | 117 ----- weed/s3api/s3lifecycle/dailyrun/run.go | 77 ++- .../s3lifecycle/dailyrun/walk_buckets.go | 76 +++ .../s3lifecycle/dailyrun/walk_buckets_test.go | 175 +++++++ .../s3lifecycle/dailyrun/walker_dispatcher.go | 85 ++++ .../dailyrun/walker_dispatcher_test.go | 159 +++++++ .../dailyrun/walker_recovery_test.go | 143 ++++++ weed/worker/tasks/s3_lifecycle/handler.go | 30 +- 12 files changed, 1493 insertions(+), 207 deletions(-) create mode 100644 weed/s3api/s3lifecycle/dailyrun/filer_list_func.go create mode 100644 weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go delete mode 100644 weed/s3api/s3lifecycle/dailyrun/replayability.go delete mode 100644 weed/s3api/s3lifecycle/dailyrun/replayability_test.go create mode 100644 weed/s3api/s3lifecycle/dailyrun/walk_buckets.go create mode 100644 weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go create mode 100644 weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go create mode 100644 weed/s3api/s3lifecycle/dailyrun/walker_dispatcher_test.go create mode 100644 weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go 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