mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-16 20:26:45 +00:00
feat(s3/lifecycle): Phase 4b — daily walker for recovery and steady state (#9459)
* 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/<id> 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 <bucketsPath>/<bucket> 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/<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.
* feat(s3/lifecycle): emit MPU init records from FilerListFunc
Last gap in the filer-backed ListFunc. A directory at .uploads/<id>
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/<id> 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.
This commit is contained in:
Chris Lu
@@ -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
|
||||
|
||||
@@ -0,0 +1,320 @@
|
||||
package dailyrun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
// listPageSize is the page size for paginated directory listings. The
|
||||
// filer caps SeaweedList(..., limit=0) at DirListingLimit (1000 by
|
||||
// default) per call, so a single-page list would silently truncate
|
||||
// large directories. Atomic so tests can shrink it without racing.
|
||||
var listPageSize atomic.Uint32
|
||||
|
||||
func init() { listPageSize.Store(1024) }
|
||||
|
||||
// FilerListFunc returns a bootstrap.ListFunc that streams entries
|
||||
// under <bucketsPath>/<bucket>. Versioned siblings are expanded with
|
||||
// IsLatest / NumVersions / NoncurrentIndex / SuccessorModTime so the
|
||||
// walker's NoncurrentDays evaluation has the same per-version state
|
||||
// the streaming bootstrap injects via reader.Event.BootstrapVersion.
|
||||
// MPU init records at .uploads/<id> with ExtMultipartObjectKey set
|
||||
// are emitted whole with IsMPUInit=true and DestKey carrying the
|
||||
// user's intended path.
|
||||
func FilerListFunc(client filer_pb.SeaweedFilerClient, bucketsPath string) bootstrap.ListFunc {
|
||||
return func(ctx context.Context, bucket, start string, cb func(*bootstrap.Entry) error) error {
|
||||
if client == nil {
|
||||
return fmt.Errorf("FilerListFunc: nil client")
|
||||
}
|
||||
root := strings.TrimSuffix(bucketsPath, "/") + "/" + bucket
|
||||
return walkBucketTree(ctx, client, root, root, start, cb)
|
||||
}
|
||||
}
|
||||
|
||||
// walkBucketTree recurses through dir in two passes per level. Pass 1
|
||||
// expands `.versions/` dirs (populating skipBare with the bare null-
|
||||
// version keys that pass 2 must suppress). Pass 2 emits regular files
|
||||
// and recurses into non-special subdirectories.
|
||||
//
|
||||
// The two-pass shape mirrors scheduler/bootstrap.go's walkBucketDir
|
||||
// (see that file's comment for why `.versions/` has to be processed
|
||||
// before its bare sibling).
|
||||
func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot, start string, cb func(*bootstrap.Entry) error) error {
|
||||
skipBare := map[string]bool{}
|
||||
|
||||
// Pass 1: .versions/ dirs only.
|
||||
if err := listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
|
||||
if e == nil || e.Attributes == nil {
|
||||
return nil
|
||||
}
|
||||
if !e.IsDirectory || !isVersionsDir(e) {
|
||||
return nil
|
||||
}
|
||||
full := dir + "/" + e.Name
|
||||
key := strings.TrimPrefix(full, bucketRoot+"/")
|
||||
return expandVersionsDir(ctx, client, bucketRoot, key, e, start, skipBare, cb)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Pass 2: everything else.
|
||||
return listAll(ctx, client, dir, func(e *filer_pb.Entry) error {
|
||||
if e == nil || e.Attributes == nil {
|
||||
return nil
|
||||
}
|
||||
if e.IsDirectory && isVersionsDir(e) {
|
||||
return nil
|
||||
}
|
||||
full := dir + "/" + e.Name
|
||||
key := strings.TrimPrefix(full, bucketRoot+"/")
|
||||
if e.IsDirectory {
|
||||
if isMPUInitDir(key, e) {
|
||||
if start != "" && key <= start {
|
||||
return nil
|
||||
}
|
||||
destKey := string(e.Extended[s3_constants.ExtMultipartObjectKey])
|
||||
return cb(&bootstrap.Entry{
|
||||
Path: key,
|
||||
DestKey: destKey,
|
||||
IsMPUInit: true,
|
||||
ModTime: time.Unix(e.Attributes.Mtime, int64(e.Attributes.MtimeNs)),
|
||||
Size: int64(e.Attributes.FileSize),
|
||||
})
|
||||
}
|
||||
return walkBucketTree(ctx, client, full, bucketRoot, start, cb)
|
||||
}
|
||||
if skipBare[key] {
|
||||
return nil
|
||||
}
|
||||
if start != "" && key <= start {
|
||||
return nil
|
||||
}
|
||||
entry := &bootstrap.Entry{
|
||||
Path: key,
|
||||
ModTime: time.Unix(e.Attributes.Mtime, int64(e.Attributes.MtimeNs)),
|
||||
Size: int64(e.Attributes.FileSize),
|
||||
IsLatest: true, // Non-versioned default.
|
||||
}
|
||||
return cb(entry)
|
||||
})
|
||||
}
|
||||
|
||||
// versionItem captures one sibling of a `.versions/` expansion. bareKey
|
||||
// is the bucket-relative path of the bare null-version entry when the
|
||||
// item represents it; for real version files it stays empty.
|
||||
type versionItem struct {
|
||||
entry *filer_pb.Entry
|
||||
versionID string
|
||||
bareKey string
|
||||
isExplicitNull bool
|
||||
}
|
||||
|
||||
// expandVersionsDir handles the `.versions/<key>` directory. Lists
|
||||
// version files, optionally appends the bare null-version sibling,
|
||||
// sorts newest-first, resolves the latest, and emits one Entry per
|
||||
// version with the sibling state walkEntry needs to evaluate
|
||||
// NoncurrentDays / NewerNoncurrent / ExpirationDays correctly.
|
||||
//
|
||||
// Ported from scheduler/bootstrap.go's same-named helper; both must
|
||||
// agree on sort, latest resolution, and successor derivation so the
|
||||
// streaming and walker paths reach the same verdict for the same
|
||||
// objects. Phase 5 deletes the scheduler copy.
|
||||
//
|
||||
// Resume note: every emitted sibling shares Path = logical key, so a
|
||||
// resume after a mid-expansion failure rewalks the whole sibling
|
||||
// group. Acceptable today because Phase 4b doesn't persist a
|
||||
// Checkpoint between runs (start is always "" via runShard).
|
||||
func expandVersionsDir(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketRoot, versionsKey string, versionsEntry *filer_pb.Entry, start string, skipBare map[string]bool, cb func(*bootstrap.Entry) error) error {
|
||||
logical := strings.TrimSuffix(versionsKey, s3_constants.VersionsFolder)
|
||||
if logical == "" {
|
||||
return nil
|
||||
}
|
||||
versionsDir := bucketRoot + "/" + versionsKey
|
||||
var children []*filer_pb.Entry
|
||||
if err := listAll(ctx, client, versionsDir, func(e *filer_pb.Entry) error {
|
||||
if e != nil && e.Attributes != nil && !e.IsDirectory {
|
||||
children = append(children, e)
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return fmt.Errorf("list %s: %w", versionsDir, err)
|
||||
}
|
||||
items := make([]versionItem, 0, len(children)+1)
|
||||
for _, e := range children {
|
||||
if id, ok := e.Extended[s3_constants.ExtVersionIdKey]; ok && len(id) > 0 {
|
||||
items = append(items, versionItem{entry: e, versionID: string(id)})
|
||||
}
|
||||
}
|
||||
if len(items) == 0 {
|
||||
// Coincidentally-named user folder, or an empty `.versions`
|
||||
// container. Treat it as a regular subdirectory so user-named
|
||||
// files inside still surface.
|
||||
return walkBucketTree(ctx, client, versionsDir, bucketRoot, start, cb)
|
||||
}
|
||||
if nullEntry, nullKey, explicit, ok := lookupNullVersion(ctx, client, bucketRoot, logical); ok {
|
||||
items = append(items, versionItem{
|
||||
entry: nullEntry,
|
||||
versionID: "null",
|
||||
bareKey: nullKey,
|
||||
isExplicitNull: explicit,
|
||||
})
|
||||
}
|
||||
|
||||
// Sort newest-first by mtime, ties broken by version_id (newer wins).
|
||||
sort.SliceStable(items, func(i, j int) bool {
|
||||
mi := items[i].entry.Attributes.Mtime*int64(1e9) + int64(items[i].entry.Attributes.MtimeNs)
|
||||
mj := items[j].entry.Attributes.Mtime*int64(1e9) + int64(items[j].entry.Attributes.MtimeNs)
|
||||
if mi != mj {
|
||||
return mi > mj
|
||||
}
|
||||
return s3lifecycle.CompareVersionIds(items[i].versionID, items[j].versionID) < 0
|
||||
})
|
||||
|
||||
// Resolve latest:
|
||||
// 1. Pointer names a real id -> that wins.
|
||||
// 2. Pointer absent (or stale: set but no sibling carries it)
|
||||
// + items[0] is an EXPLICIT null -> null is latest.
|
||||
// 3. Otherwise -> newest sibling (latestPos = 0 by default).
|
||||
//
|
||||
// A stale pointer falls through to the no-pointer fallback rather
|
||||
// than silently leaving latestPos at 0 with no documented intent;
|
||||
// the value happens to be the same today (newest sibling wins
|
||||
// either way) but the explicit branching protects against future
|
||||
// fallback refinements diverging by accident.
|
||||
latestID := string(versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey])
|
||||
latestPos := 0
|
||||
pointerResolved := false
|
||||
if latestID != "" {
|
||||
for i, it := range items {
|
||||
if it.versionID == latestID {
|
||||
latestPos = i
|
||||
pointerResolved = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if !pointerResolved && len(items) > 0 && items[0].versionID == "null" && items[0].isExplicitNull {
|
||||
latestPos = 0
|
||||
}
|
||||
|
||||
if start != "" && logical <= start {
|
||||
// All siblings share Path=logical, so the whole group is
|
||||
// either above or below the resume marker.
|
||||
return nil
|
||||
}
|
||||
for i, it := range items {
|
||||
successor := s3lifecycle.SuccessorFromEntryStamp(it.entry)
|
||||
if successor.IsZero() && i > 0 {
|
||||
prev := items[i-1].entry.Attributes
|
||||
successor = time.Unix(prev.Mtime, int64(prev.MtimeNs))
|
||||
}
|
||||
isLatest := i == latestPos
|
||||
entry := &bootstrap.Entry{
|
||||
Path: logical,
|
||||
VersionID: it.versionID,
|
||||
ModTime: time.Unix(it.entry.Attributes.Mtime, int64(it.entry.Attributes.MtimeNs)),
|
||||
Size: int64(it.entry.Attributes.FileSize),
|
||||
IsLatest: isLatest,
|
||||
IsDeleteMarker: string(it.entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true",
|
||||
NumVersions: len(items),
|
||||
SuccessorModTime: successor,
|
||||
}
|
||||
if !isLatest {
|
||||
rank := i
|
||||
if i > latestPos {
|
||||
rank = i - 1
|
||||
}
|
||||
entry.NoncurrentIndex = &rank
|
||||
}
|
||||
if err := cb(entry); err != nil {
|
||||
return err
|
||||
}
|
||||
if it.versionID == "null" && skipBare != nil {
|
||||
skipBare[it.bareKey] = true
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// lookupNullVersion returns the bare-key entry that represents the null
|
||||
// version of logical, if any. Both regular files and S3 directory-key
|
||||
// markers qualify. explicit is true when the entry carries
|
||||
// ExtVersionIdKey == "null" — the marker the suspended-versioning
|
||||
// write path applies; only an explicit-null bare can outrank a missing
|
||||
// `.versions/` pointer per the latest-resolution rules above.
|
||||
func lookupNullVersion(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketRoot, logical string) (*filer_pb.Entry, string, bool, bool) {
|
||||
parent, name := util.NewFullPath(bucketRoot, logical).DirAndName()
|
||||
resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: parent,
|
||||
Name: name,
|
||||
})
|
||||
if err != nil || resp == nil || resp.Entry == nil {
|
||||
return nil, "", false, false
|
||||
}
|
||||
e := resp.Entry
|
||||
if e.IsDirectory && !e.IsDirectoryKeyObject() {
|
||||
return nil, "", false, false
|
||||
}
|
||||
explicit := false
|
||||
if id, hasID := e.Extended[s3_constants.ExtVersionIdKey]; hasID && string(id) == "null" {
|
||||
explicit = true
|
||||
}
|
||||
return e, strings.TrimPrefix(parent+"/"+name, bucketRoot+"/"), explicit, true
|
||||
}
|
||||
|
||||
// listAll issues paginated SeaweedList calls until exhausted. Ported
|
||||
// from scheduler/bootstrap.go's same-named helper; Phase 5 deletes
|
||||
// the scheduler copy when the streaming path is removed.
|
||||
func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, fn func(*filer_pb.Entry) error) error {
|
||||
pageSize := listPageSize.Load()
|
||||
startFrom := ""
|
||||
for {
|
||||
var pageCount uint32
|
||||
var lastName string
|
||||
if err := filer_pb.SeaweedList(ctx, client, dir, "", func(e *filer_pb.Entry, _ bool) error {
|
||||
pageCount++
|
||||
if e != nil {
|
||||
lastName = e.Name
|
||||
}
|
||||
return fn(e)
|
||||
}, startFrom, false, pageSize); err != nil {
|
||||
return err
|
||||
}
|
||||
if pageCount < pageSize {
|
||||
return nil
|
||||
}
|
||||
startFrom = lastName
|
||||
}
|
||||
}
|
||||
|
||||
func isVersionsDir(entry *filer_pb.Entry) bool {
|
||||
return entry.IsDirectory && strings.HasSuffix(entry.Name, s3_constants.VersionsFolder)
|
||||
}
|
||||
|
||||
// isMPUInitDir mirrors router.mpuInitInfo: a directory at
|
||||
// .uploads/<id> carrying the destination key in Extended is the MPU
|
||||
// init record. Verified shape + presence of ExtMultipartObjectKey;
|
||||
// directories at .uploads/<id> without the key are mid-write before
|
||||
// metadata landed and stay out of the dispatch path.
|
||||
func isMPUInitDir(key string, entry *filer_pb.Entry) bool {
|
||||
uploadsPrefix := s3_constants.MultipartUploadsFolder + "/"
|
||||
if !strings.HasPrefix(key, uploadsPrefix) {
|
||||
return false
|
||||
}
|
||||
rest := key[len(uploadsPrefix):]
|
||||
if rest == "" || strings.ContainsRune(rest, '/') {
|
||||
return false
|
||||
}
|
||||
v, ok := entry.Extended[s3_constants.ExtMultipartObjectKey]
|
||||
return ok && len(v) > 0
|
||||
}
|
||||
@@ -0,0 +1,442 @@
|
||||
package dailyrun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"sort"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/metadata"
|
||||
)
|
||||
|
||||
// fakeFilerStream implements the ListEntries server-streaming client.
|
||||
type fakeFilerStream struct {
|
||||
responses []*filer_pb.ListEntriesResponse
|
||||
idx int
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s *fakeFilerStream) Recv() (*filer_pb.ListEntriesResponse, error) {
|
||||
if s.ctx != nil {
|
||||
if err := s.ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if s.idx >= len(s.responses) {
|
||||
return nil, io.EOF
|
||||
}
|
||||
r := s.responses[s.idx]
|
||||
s.idx++
|
||||
return r, nil
|
||||
}
|
||||
func (s *fakeFilerStream) Header() (metadata.MD, error) { return metadata.MD{}, nil }
|
||||
func (s *fakeFilerStream) Trailer() metadata.MD { return metadata.MD{} }
|
||||
func (s *fakeFilerStream) CloseSend() error { return nil }
|
||||
func (s *fakeFilerStream) Context() context.Context {
|
||||
if s.ctx != nil {
|
||||
return s.ctx
|
||||
}
|
||||
return context.Background()
|
||||
}
|
||||
func (s *fakeFilerStream) SendMsg(any) error { return nil }
|
||||
func (s *fakeFilerStream) RecvMsg(any) error { return nil }
|
||||
|
||||
// fakeFiler maps directory paths to their immediate children. Only
|
||||
// ListEntries is implemented; other methods of SeaweedFilerClient are
|
||||
// inherited from the embedded interface and panic if called.
|
||||
type fakeFiler struct {
|
||||
filer_pb.SeaweedFilerClient
|
||||
|
||||
mu sync.Mutex
|
||||
tree map[string][]*filer_pb.Entry
|
||||
}
|
||||
|
||||
func (c *fakeFiler) LookupDirectoryEntry(_ context.Context, in *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
for _, e := range c.tree[in.Directory] {
|
||||
if e != nil && e.Name == in.Name {
|
||||
return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil
|
||||
}
|
||||
}
|
||||
return nil, filer_pb.ErrNotFound
|
||||
}
|
||||
|
||||
func (c *fakeFiler) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
src := c.tree[in.Directory]
|
||||
// Mirror the filer: sort by name, honor StartFromFileName exclusive,
|
||||
// cap at Limit. listAll's pagination loop depends on these.
|
||||
filtered := make([]*filer_pb.Entry, 0, len(src))
|
||||
for _, e := range src {
|
||||
if e == nil {
|
||||
continue
|
||||
}
|
||||
if in.StartFromFileName != "" && !in.InclusiveStartFrom && e.Name <= in.StartFromFileName {
|
||||
continue
|
||||
}
|
||||
filtered = append(filtered, e)
|
||||
}
|
||||
sort.SliceStable(filtered, func(i, j int) bool { return filtered[i].Name < filtered[j].Name })
|
||||
if in.Limit > 0 && uint32(len(filtered)) > in.Limit {
|
||||
filtered = filtered[:in.Limit]
|
||||
}
|
||||
resps := make([]*filer_pb.ListEntriesResponse, 0, len(filtered))
|
||||
for _, e := range filtered {
|
||||
resps = append(resps, &filer_pb.ListEntriesResponse{Entry: e})
|
||||
}
|
||||
return &fakeFilerStream{responses: resps, ctx: ctx}, nil
|
||||
}
|
||||
|
||||
func file(name string, mtime time.Time, size int64) *filer_pb.Entry {
|
||||
return &filer_pb.Entry{
|
||||
Name: name,
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
Mtime: mtime.Unix(),
|
||||
MtimeNs: int32(mtime.Nanosecond()),
|
||||
FileSize: uint64(size),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func dir(name string) *filer_pb.Entry {
|
||||
return &filer_pb.Entry{Name: name, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}
|
||||
}
|
||||
|
||||
func TestFilerListFunc_EmitsFlatFiles(t *testing.T) {
|
||||
mtime := time.Now().Add(-7 * 24 * time.Hour)
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {file("a.txt", mtime, 10), file("b.txt", mtime, 20)},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got []string
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = append(got, e.Path)
|
||||
return nil
|
||||
}))
|
||||
assert.Equal(t, []string{"a.txt", "b.txt"}, got)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_RecursesIntoSubdirs(t *testing.T) {
|
||||
mtime := time.Now()
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {dir("logs"), file("root.txt", mtime, 1)},
|
||||
"/buckets/bkt/logs": {dir("2026"), file("a.log", mtime, 5)},
|
||||
"/buckets/bkt/logs/2026": {file("b.log", mtime, 7)},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var paths []string
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
paths = append(paths, e.Path)
|
||||
return nil
|
||||
}))
|
||||
sort.Strings(paths)
|
||||
assert.Equal(t, []string{"logs/2026/b.log", "logs/a.log", "root.txt"}, paths)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_MPUInitEmitsWithDestKey(t *testing.T) {
|
||||
// .uploads/<id>/ with ExtMultipartObjectKey is the MPU init record.
|
||||
// One Entry per init, IsMPUInit=true, DestKey = the user's path.
|
||||
mtime := time.Now()
|
||||
mpuDir := dir("upload-id-1")
|
||||
mpuDir.Extended = map[string][]byte{
|
||||
s3_constants.ExtMultipartObjectKey: []byte("user/path/object"),
|
||||
}
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {
|
||||
file("regular.txt", mtime, 1),
|
||||
dir(s3_constants.MultipartUploadsFolder),
|
||||
},
|
||||
"/buckets/bkt/" + s3_constants.MultipartUploadsFolder: {
|
||||
mpuDir,
|
||||
},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got []*bootstrap.Entry
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = append(got, e)
|
||||
return nil
|
||||
}))
|
||||
require.Len(t, got, 2)
|
||||
byPath := map[string]*bootstrap.Entry{}
|
||||
for _, e := range got {
|
||||
byPath[e.Path] = e
|
||||
}
|
||||
require.NotNil(t, byPath["regular.txt"])
|
||||
require.NotNil(t, byPath[s3_constants.MultipartUploadsFolder+"/upload-id-1"])
|
||||
mpu := byPath[s3_constants.MultipartUploadsFolder+"/upload-id-1"]
|
||||
assert.True(t, mpu.IsMPUInit)
|
||||
assert.Equal(t, "user/path/object", mpu.DestKey)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_MPUInitWithoutDestKeyIsSkipped(t *testing.T) {
|
||||
// A `.uploads/<id>` directory missing ExtMultipartObjectKey is
|
||||
// mid-write before metadata landed; the dispatcher would error
|
||||
// on empty DestKey. Skip it so the walk doesn't halt.
|
||||
mtime := time.Now()
|
||||
mpuDir := dir("upload-id-2") // no Extended
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {
|
||||
file("regular.txt", mtime, 1),
|
||||
dir(s3_constants.MultipartUploadsFolder),
|
||||
},
|
||||
"/buckets/bkt/" + s3_constants.MultipartUploadsFolder: {
|
||||
mpuDir,
|
||||
},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var paths []string
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
paths = append(paths, e.Path)
|
||||
return nil
|
||||
}))
|
||||
assert.Equal(t, []string{"regular.txt"}, paths)
|
||||
}
|
||||
|
||||
// fileWithExt is a versioned-file entry helper.
|
||||
func fileWithExt(name string, mtime time.Time, size int64, ext map[string][]byte) *filer_pb.Entry {
|
||||
e := file(name, mtime, size)
|
||||
e.Extended = ext
|
||||
return e
|
||||
}
|
||||
|
||||
func versionsDir(name string, latestID string) *filer_pb.Entry {
|
||||
d := dir(name)
|
||||
d.Extended = map[string][]byte{}
|
||||
if latestID != "" {
|
||||
d.Extended[s3_constants.ExtLatestVersionIdKey] = []byte(latestID)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func TestFilerListFunc_VersionedExpansionMarksLatestByPointer(t *testing.T) {
|
||||
// .versions/<key>/ with three real versions; parent's
|
||||
// ExtLatestVersionIdKey points to v2 → IsLatest set on v2; the
|
||||
// other two get NoncurrentIndex computed against the latest's
|
||||
// position in the sorted (newest-first) list.
|
||||
t1 := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
t2 := time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC)
|
||||
t3 := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
|
||||
versions := []*filer_pb.Entry{
|
||||
fileWithExt("v1", t1, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}),
|
||||
fileWithExt("v2", t2, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v2")}),
|
||||
fileWithExt("v3", t3, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v3")}),
|
||||
}
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v2")},
|
||||
"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got []*bootstrap.Entry
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = append(got, e)
|
||||
return nil
|
||||
}))
|
||||
require.Len(t, got, 3)
|
||||
|
||||
byID := map[string]*bootstrap.Entry{}
|
||||
for _, e := range got {
|
||||
byID[e.VersionID] = e
|
||||
assert.Equal(t, "foo", e.Path, "every sibling's Path is the logical key")
|
||||
assert.Equal(t, 3, e.NumVersions)
|
||||
}
|
||||
assert.True(t, byID["v2"].IsLatest, "pointer wins regardless of mtime order")
|
||||
assert.False(t, byID["v1"].IsLatest)
|
||||
assert.False(t, byID["v3"].IsLatest)
|
||||
require.NotNil(t, byID["v3"].NoncurrentIndex, "noncurrent siblings get a rank")
|
||||
require.NotNil(t, byID["v1"].NoncurrentIndex, "noncurrent siblings get a rank")
|
||||
}
|
||||
|
||||
func TestFilerListFunc_VersionedExpansionStalePointerFallsBackToNewestSibling(t *testing.T) {
|
||||
// ExtLatestVersionIdKey names a version that no sibling carries
|
||||
// (stale pointer left behind by a write whose update was lost or
|
||||
// raced). The fallback must NOT silently treat items[0] as latest
|
||||
// via the default; structurally it must drop into the no-pointer
|
||||
// path so the explicit-null bare check still runs.
|
||||
tOld := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
tNew := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
|
||||
versions := []*filer_pb.Entry{
|
||||
fileWithExt("v_old", tOld, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vold")}),
|
||||
fileWithExt("v_new", tNew, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vnew")}),
|
||||
}
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v_GHOST")},
|
||||
"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got []*bootstrap.Entry
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = append(got, e)
|
||||
return nil
|
||||
}))
|
||||
require.Len(t, got, 2)
|
||||
byID := map[string]*bootstrap.Entry{}
|
||||
for _, e := range got {
|
||||
byID[e.VersionID] = e
|
||||
}
|
||||
assert.True(t, byID["vnew"].IsLatest, "stale pointer must fall back to newest sibling")
|
||||
assert.False(t, byID["vold"].IsLatest)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_VersionedExpansionNoPointerNewestSiblingWins(t *testing.T) {
|
||||
// Parent has no ExtLatestVersionIdKey. With no explicit-null bare
|
||||
// version, the newest-by-mtime sibling becomes latest.
|
||||
tNew := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
|
||||
tOld := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
versions := []*filer_pb.Entry{
|
||||
fileWithExt("v_old", tOld, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vold")}),
|
||||
fileWithExt("v_new", tNew, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("vnew")}),
|
||||
}
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "")},
|
||||
"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got []*bootstrap.Entry
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = append(got, e)
|
||||
return nil
|
||||
}))
|
||||
require.Len(t, got, 2)
|
||||
byID := map[string]*bootstrap.Entry{}
|
||||
for _, e := range got {
|
||||
byID[e.VersionID] = e
|
||||
}
|
||||
assert.True(t, byID["vnew"].IsLatest, "newest sibling wins when pointer is absent")
|
||||
assert.False(t, byID["vold"].IsLatest)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_VersionedExpansionExplicitNullIsLatestWhenPointerMissing(t *testing.T) {
|
||||
// Suspended-versioning shape: bare object marked with
|
||||
// ExtVersionIdKey="null", parent has no pointer. null is latest.
|
||||
t1 := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
tNull := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC) // newest
|
||||
bareNull := fileWithExt("foo", tNull, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("null")})
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {
|
||||
bareNull,
|
||||
versionsDir("foo"+s3_constants.VersionsFolder, ""),
|
||||
},
|
||||
"/buckets/bkt/foo" + s3_constants.VersionsFolder: {
|
||||
fileWithExt("v1", t1, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}),
|
||||
},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got []*bootstrap.Entry
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = append(got, e)
|
||||
return nil
|
||||
}))
|
||||
// 2 sibling entries from expansion (null + v1). The bare "foo"
|
||||
// MUST be suppressed in pass 2 via skipBare.
|
||||
require.Len(t, got, 2)
|
||||
byID := map[string]*bootstrap.Entry{}
|
||||
for _, e := range got {
|
||||
byID[e.VersionID] = e
|
||||
}
|
||||
require.NotNil(t, byID["null"])
|
||||
require.NotNil(t, byID["v1"])
|
||||
assert.True(t, byID["null"].IsLatest)
|
||||
assert.False(t, byID["v1"].IsLatest)
|
||||
|
||||
// Walk again, verify no duplicate emission of the bare "foo".
|
||||
count := 0
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
if e.Path == "foo" && e.VersionID == "" {
|
||||
t.Errorf("bare foo should be suppressed by skipBare, got %+v", e)
|
||||
}
|
||||
count++
|
||||
return nil
|
||||
}))
|
||||
assert.Equal(t, 2, count)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_VersionsDirWithoutMarkersRecursesAsRegular(t *testing.T) {
|
||||
// A `.versions`-named folder whose children have no
|
||||
// ExtVersionIdKey is a coincidence (user folder). Recurse into
|
||||
// it; the file inside should surface as a regular entry.
|
||||
mtime := time.Now()
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {versionsDir("looksLikeUserFolder"+s3_constants.VersionsFolder, "")},
|
||||
"/buckets/bkt/looksLikeUserFolder" + s3_constants.VersionsFolder: {
|
||||
file("inner.txt", mtime, 1),
|
||||
},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var paths []string
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
paths = append(paths, e.Path)
|
||||
return nil
|
||||
}))
|
||||
assert.Equal(t, []string{"looksLikeUserFolder" + s3_constants.VersionsFolder + "/inner.txt"}, paths)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_VersionedDeleteMarkerPropagates(t *testing.T) {
|
||||
mtime := time.Now()
|
||||
versions := []*filer_pb.Entry{
|
||||
fileWithExt("v1", mtime, 0, map[string][]byte{
|
||||
s3_constants.ExtVersionIdKey: []byte("v1"),
|
||||
s3_constants.ExtDeleteMarkerKey: []byte("true"),
|
||||
}),
|
||||
}
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v1")},
|
||||
"/buckets/bkt/foo" + s3_constants.VersionsFolder: versions,
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got *bootstrap.Entry
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = e
|
||||
return nil
|
||||
}))
|
||||
require.NotNil(t, got)
|
||||
assert.True(t, got.IsDeleteMarker, "ExtDeleteMarkerKey='true' must surface as IsDeleteMarker")
|
||||
}
|
||||
|
||||
func TestFilerListFunc_HonorsStart(t *testing.T) {
|
||||
// The walker's kill-resume contract: skip entries whose Path <= start.
|
||||
mtime := time.Now()
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {file("a", mtime, 1), file("b", mtime, 1), file("c", mtime, 1)},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got []string
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "a", func(e *bootstrap.Entry) error {
|
||||
got = append(got, e.Path)
|
||||
return nil
|
||||
}))
|
||||
assert.Equal(t, []string{"b", "c"}, got)
|
||||
}
|
||||
|
||||
func TestFilerListFunc_NilClient(t *testing.T) {
|
||||
listFn := FilerListFunc(nil, "/buckets")
|
||||
require.Error(t, listFn(context.Background(), "bkt", "", func(*bootstrap.Entry) error { return nil }))
|
||||
}
|
||||
|
||||
func TestFilerListFunc_AttributesPropagate(t *testing.T) {
|
||||
mtime := time.Date(2026, 5, 11, 12, 0, 0, 1234, time.UTC)
|
||||
client := &fakeFiler{tree: map[string][]*filer_pb.Entry{
|
||||
"/buckets/bkt": {file("obj", mtime, 4096)},
|
||||
}}
|
||||
listFn := FilerListFunc(client, "/buckets")
|
||||
var got *bootstrap.Entry
|
||||
require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error {
|
||||
got = e
|
||||
return nil
|
||||
}))
|
||||
require.NotNil(t, got)
|
||||
assert.Equal(t, "obj", got.Path)
|
||||
assert.Equal(t, mtime.Unix(), got.ModTime.Unix())
|
||||
assert.Equal(t, int64(1234), got.ModTime.UnixNano()-mtime.Unix()*int64(time.Second))
|
||||
assert.Equal(t, int64(4096), got.Size)
|
||||
assert.True(t, got.IsLatest)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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{}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -0,0 +1,175 @@
|
||||
package dailyrun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sort"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// recordingDispatcher captures every (action, entry) pair the walker
|
||||
// emits so tests can assert dispatch order and shard filtering.
|
||||
type recordingDispatcher struct {
|
||||
calls []recordedDispatch
|
||||
err error
|
||||
}
|
||||
|
||||
type recordedDispatch struct {
|
||||
bucket string
|
||||
path string
|
||||
}
|
||||
|
||||
func (d *recordingDispatcher) Delete(_ context.Context, action *engine.CompiledAction, entry *bootstrap.Entry) error {
|
||||
d.calls = append(d.calls, recordedDispatch{bucket: action.Bucket, path: entry.Path})
|
||||
return d.err
|
||||
}
|
||||
|
||||
// findShardForPath returns the shard ID for an entry with given path
|
||||
// in given bucket. Helper for tests that want to force entries onto
|
||||
// a known shard.
|
||||
func findShardForPath(bucket, path string) int {
|
||||
return s3lifecycle.ShardID(bucket, path)
|
||||
}
|
||||
|
||||
// fixedShardEntries returns a slice of bootstrap.Entries whose paths
|
||||
// all live in the same shard. Used so a per-shard filter test has
|
||||
// deterministic expectations.
|
||||
func fixedShardEntries(t *testing.T, bucket string, shardID int, count int) []*bootstrap.Entry {
|
||||
t.Helper()
|
||||
var out []*bootstrap.Entry
|
||||
for i := 0; tries(i, count); i++ {
|
||||
path := "obj-" + intToStr(i)
|
||||
if findShardForPath(bucket, path) == shardID {
|
||||
out = append(out, &bootstrap.Entry{
|
||||
Path: path,
|
||||
// Old enough to expire under a 7-day ExpirationDays rule.
|
||||
ModTime: time.Now().Add(-90 * 24 * time.Hour),
|
||||
Size: 1,
|
||||
IsLatest: true,
|
||||
})
|
||||
if len(out) >= count {
|
||||
return out
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Fatalf("could not find %d entries in shard %d for bucket %s after 4096 attempts", count, shardID, bucket)
|
||||
return nil
|
||||
}
|
||||
|
||||
func intToStr(i int) string {
|
||||
const hex = "0123456789abcdef"
|
||||
if i == 0 {
|
||||
return "0"
|
||||
}
|
||||
var out []byte
|
||||
for n := i; n > 0; n /= 16 {
|
||||
out = append([]byte{hex[n%16]}, out...)
|
||||
}
|
||||
return string(out)
|
||||
}
|
||||
|
||||
func tries(i, count int) bool { return i < count*256+256 }
|
||||
|
||||
func snapshotForBucketRule(t *testing.T, bucket string, days int) *engine.Snapshot {
|
||||
t.Helper()
|
||||
e := engine.New()
|
||||
e.Compile([]engine.CompileInput{
|
||||
{Bucket: bucket, Rules: []*s3lifecycle.Rule{
|
||||
{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days},
|
||||
}},
|
||||
}, engine.CompileOptions{})
|
||||
snap := e.Snapshot()
|
||||
for _, a := range snap.AllActions() {
|
||||
snap.MarkActive(a.Key)
|
||||
}
|
||||
return snap
|
||||
}
|
||||
|
||||
func TestWalkBuckets_DispatchesOnlyShardMatchingEntries(t *testing.T) {
|
||||
bucket := "b1"
|
||||
const targetShard = 0
|
||||
// Build entries: half in shard 0, half elsewhere. The walker's
|
||||
// per-shard filter must drop the non-matching ones.
|
||||
inShard := fixedShardEntries(t, bucket, targetShard, 3)
|
||||
otherShard := fixedShardEntries(t, bucket, (targetShard+1)%16, 3)
|
||||
all := append([]*bootstrap.Entry{}, inShard...)
|
||||
all = append(all, otherShard...)
|
||||
|
||||
snap := snapshotForBucketRule(t, bucket, 7) // 30-day-old objects expire
|
||||
d := &recordingDispatcher{}
|
||||
err := WalkBuckets(context.Background(), snap, targetShard, []string{bucket},
|
||||
bootstrap.EntryCallback(all), d)
|
||||
require.NoError(t, err)
|
||||
|
||||
gotPaths := make([]string, 0, len(d.calls))
|
||||
for _, c := range d.calls {
|
||||
gotPaths = append(gotPaths, c.path)
|
||||
assert.Equal(t, bucket, c.bucket)
|
||||
}
|
||||
sort.Strings(gotPaths)
|
||||
|
||||
wantPaths := make([]string, 0, len(inShard))
|
||||
for _, e := range inShard {
|
||||
wantPaths = append(wantPaths, e.Path)
|
||||
}
|
||||
sort.Strings(wantPaths)
|
||||
assert.Equal(t, wantPaths, gotPaths, "walker must dispatch exactly the in-shard entries")
|
||||
}
|
||||
|
||||
func TestWalkBuckets_NilGuards(t *testing.T) {
|
||||
require.Error(t, WalkBuckets(context.Background(), nil, 0, nil, bootstrap.EntryCallback(nil), &recordingDispatcher{}))
|
||||
require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), 0, nil, nil, &recordingDispatcher{}))
|
||||
require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), 0, nil, bootstrap.EntryCallback(nil), nil))
|
||||
}
|
||||
|
||||
func TestWalkBuckets_OneBucketErrorDoesNotStopOthers(t *testing.T) {
|
||||
// Two buckets, first ListFunc errors. WalkBuckets should still
|
||||
// process the second bucket but return the first bucket's error.
|
||||
listErr := errors.New("filer flake")
|
||||
list := func(_ context.Context, bucket, _ string, _ func(*bootstrap.Entry) error) error {
|
||||
if bucket == "bad" {
|
||||
return listErr
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
snap := snapshotForBucketRule(t, "good", 7)
|
||||
err := WalkBuckets(context.Background(), snap, 0, []string{"bad", "good"}, list, &recordingDispatcher{})
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, listErr)
|
||||
}
|
||||
|
||||
func TestWalkBuckets_HonorsContextCancellationBetweenBuckets(t *testing.T) {
|
||||
// Pre-cancel ctx before invocation. Even an empty bucket list
|
||||
// must surface ctx.Err early so a long walk in progress can
|
||||
// short-circuit.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
err := WalkBuckets(ctx, snapshotForBucketRule(t, "b", 7), 0, []string{"b"},
|
||||
bootstrap.EntryCallback(nil), &recordingDispatcher{})
|
||||
assert.ErrorIs(t, err, context.Canceled)
|
||||
}
|
||||
|
||||
func TestEntryShardID_MPUUsesDestKey(t *testing.T) {
|
||||
bucket := "bkt"
|
||||
// Force a (Path, DestKey) pair that maps to different shards
|
||||
// so we can prove the filter picks the right one.
|
||||
for i := 0; i < 4096; i++ {
|
||||
path := ".uploads/" + intToStr(i)
|
||||
destKey := "user/" + intToStr(i)
|
||||
if s3lifecycle.ShardID(bucket, path) != s3lifecycle.ShardID(bucket, destKey) {
|
||||
e := &bootstrap.Entry{Path: path, DestKey: destKey, IsMPUInit: true}
|
||||
assert.Equal(t, s3lifecycle.ShardID(bucket, destKey), entryShardID(bucket, e),
|
||||
"MPU init must use DestKey for shard, not the .uploads/<id> path")
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatal("could not find an MPU path/destkey pair with diverging shards in 4096 attempts")
|
||||
}
|
||||
@@ -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/<id>
|
||||
// directory itself only when ActionKind=ABORT_MPU, and the
|
||||
// server resolves the upload from (bucket, object_path) +
|
||||
// the init record's metadata.
|
||||
if entry.DestKey == "" {
|
||||
return fmt.Errorf("walker dispatch: MPU init entry with empty DestKey: %s", entry.Path)
|
||||
}
|
||||
objectPath = entry.DestKey
|
||||
}
|
||||
rh := action.Key.RuleHash
|
||||
req := &s3_lifecycle_pb.LifecycleDeleteRequest{
|
||||
Bucket: action.Bucket,
|
||||
ObjectPath: objectPath,
|
||||
VersionId: entry.VersionID,
|
||||
RuleHash: rh[:],
|
||||
ActionKind: toProtoActionKind(action.Key.ActionKind),
|
||||
// ExpectedIdentity intentionally nil; server bootstraps from
|
||||
// the live entry on this code path.
|
||||
}
|
||||
kindLabel := action.Key.ActionKind.String()
|
||||
resp, err := d.Client.LifecycleDelete(ctx, req)
|
||||
if err != nil {
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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"}))
|
||||
}
|
||||
@@ -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.
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user