diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index e4e38094f..6dd562f16 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/weed/s3api/s3api_internal_lifecycle.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "strings" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -118,9 +119,42 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle } func (s3a *S3ApiServer) lifecycleAbortMPU(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { - // TODO(phase-5): plumb to abortMultipartUpload (currently expects an - // *s3.AbortMultipartUploadInput; lifecycle has no HTTP request). - return retryLater("ABORT_MPU not yet wired"), nil + // req.ObjectPath is `.uploads/` (set by the router from the + // init directory's bucket-relative path); reject anything that isn't + // exactly that shape so a malformed event can't escalate to a wider rm. + const uploadsPrefix = s3_constants.MultipartUploadsFolder + "/" + if !strings.HasPrefix(req.ObjectPath, uploadsPrefix) { + return blocked("FATAL_EVENT_ERROR: ABORT_MPU object_path missing .uploads/ prefix"), nil + } + uploadID := req.ObjectPath[len(uploadsPrefix):] + // Reject "." and ".." explicitly: util.JoinPath in the filer cleans + // path components, so .uploads/.. would resolve to the bucket root. + if uploadID == "" || uploadID == "." || uploadID == ".." || strings.ContainsRune(uploadID, '/') { + return blocked("FATAL_EVENT_ERROR: ABORT_MPU object_path malformed: " + req.ObjectPath), nil + } + + uploadsFolder := s3a.genUploadsFolder(req.Bucket) + // Pre-check existence: filer.DeleteEntry suppresses ErrNotFound and + // returns success, so without this check an already-aborted upload + // would report DONE instead of the correct NOOP_RESOLVED. + exists, err := s3a.exists(uploadsFolder, uploadID, true) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return noopResolved("NOT_FOUND"), nil + } + return retryLater("TRANSPORT_ERROR: exists: " + err.Error()), nil + } + if !exists { + return noopResolved("NOT_FOUND"), nil + } + if err := s3a.rm(uploadsFolder, uploadID, true, true); err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return noopResolved("NOT_FOUND_AT_DELETE"), nil + } + glog.V(1).Infof("lifecycle abort_mpu %s/%s: %v", req.Bucket, req.ObjectPath, err) + return retryLater("TRANSPORT_ERROR: rm: " + err.Error()), nil + } + return done(), nil } // computeEntryIdentity captures (mtime, size, head fid, sorted-Extended hash): diff --git a/weed/s3api/s3api_internal_lifecycle_test.go b/weed/s3api/s3api_internal_lifecycle_test.go index 7d7ded78a..9d2d8b156 100644 --- a/weed/s3api/s3api_internal_lifecycle_test.go +++ b/weed/s3api/s3api_internal_lifecycle_test.go @@ -117,3 +117,33 @@ func TestLifecycleDelete_RejectsEmptyRequest(t *testing.T) { t.Fatalf("empty request should be BLOCKED, got %v", resp.Outcome) } } + +func TestLifecycleAbortMPU_RejectsTraversalUploadIDs(t *testing.T) { + // "." and ".." pass the no-slash check but resolve to the bucket + // root via util.JoinPath; they must be rejected before any rm call. + s := &S3ApiServer{} + cases := []string{ + "", + ".uploads", + ".uploads/", + ".uploads/.", + ".uploads/..", + ".uploads/u1/extra", + } + for _, path := range cases { + t.Run(path, func(t *testing.T) { + resp, err := s.LifecycleDelete(nil, &s3_lifecycle_pb.LifecycleDeleteRequest{ + Bucket: "bk", + ObjectPath: path, + ActionKind: s3_lifecycle_pb.ActionKind_ABORT_MPU, + }) + if err != nil { + t.Fatalf("unexpected gRPC error: %v", err) + } + if resp.Outcome != s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED { + t.Fatalf("path %q: outcome=%v reason=%q, want BLOCKED", + path, resp.Outcome, resp.Reason) + } + }) + } +} diff --git a/weed/s3api/s3lifecycle/bootstrap/walker.go b/weed/s3api/s3lifecycle/bootstrap/walker.go index 476fad261..6990a11cd 100644 --- a/weed/s3api/s3lifecycle/bootstrap/walker.go +++ b/weed/s3api/s3lifecycle/bootstrap/walker.go @@ -20,8 +20,14 @@ import ( // Entry is the routing-relevant slice of a filer entry. SuccessorModTime // and NoncurrentIndex are populated only on versioned-bucket walks; the // retention path bails out conservatively when they're zero / nil. +// +// MPU init directories at .uploads/ populate DestKey with the +// destination object key (from entry.Extended[ExtMultipartObjectKey]) so +// rule-prefix matching works on the user's intended path while Path +// stays as the upload directory the dispatcher must rm. type Entry struct { Path string + DestKey string ModTime time.Time Size int64 IsDirectory bool @@ -76,8 +82,9 @@ func Walk(ctx context.Context, snap *engine.Snapshot, bucket string, list ListFu if entry == nil || entry.Path == "" { return nil } - // Lifecycle never applies to directory entries. - if entry.IsDirectory { + // MPU init directories at .uploads/ are the one directory + // shape lifecycle cares about; everything else stays out. + if entry.IsDirectory && !entry.IsMPUInit { cp.LastScannedPath = entry.Path return nil } @@ -95,12 +102,23 @@ func Walk(ctx context.Context, snap *engine.Snapshot, bucket string, list ListFu } func walkEntry(ctx context.Context, snap *engine.Snapshot, bucket string, entry *Entry, dispatch Dispatcher, now time.Time, info *s3lifecycle.ObjectInfo) error { - keys := snap.MatchPath(bucket, entry.Path, nil) + // MPU init: rule-prefix matching uses the destination key, not the + // .uploads/ directory path. A bare directory with no DestKey is + // either a stray dir or an init mid-write before metadata landed — + // skip rather than guess. + matchKey := entry.Path + if entry.IsMPUInit { + if entry.DestKey == "" { + return nil + } + matchKey = entry.DestKey + } + keys := snap.MatchPath(bucket, matchKey, nil) if len(keys) == 0 { return nil } *info = s3lifecycle.ObjectInfo{ - Key: entry.Path, + Key: matchKey, ModTime: entry.ModTime, Size: entry.Size, IsLatest: entry.IsLatest, @@ -122,6 +140,17 @@ func walkEntry(ctx context.Context, snap *engine.Snapshot, bucket string, entry if action.Mode == engine.ModeScanAtDate || action.Mode == engine.ModeDisabled { continue } + // (kind, info) shape gate: ABORT_MPU only on MPU init records, + // every other kind only on regular objects/versions. Mismatched + // pairs would either dispatch a noncurrent action with empty + // version_id (server BLOCKs, cursor freezes) or dispatch + // ABORT_MPU against a regular object path. + if entry.IsMPUInit && key.ActionKind != s3lifecycle.ActionKindAbortMPU { + continue + } + if !entry.IsMPUInit && key.ActionKind == s3lifecycle.ActionKindAbortMPU { + continue + } res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, now) if res.Action == s3lifecycle.ActionNone { continue diff --git a/weed/s3api/s3lifecycle/bootstrap/walker_test.go b/weed/s3api/s3lifecycle/bootstrap/walker_test.go index 381d72264..58f73810f 100644 --- a/weed/s3api/s3lifecycle/bootstrap/walker_test.go +++ b/weed/s3api/s3lifecycle/bootstrap/walker_test.go @@ -3,6 +3,8 @@ package bootstrap import ( "context" "errors" + "reflect" + "sort" "testing" "time" @@ -107,8 +109,9 @@ func TestWalk_MultiActionRule_AllDueDispatched(t *testing.T) { {Path: "obj/a", IsLatest: true, ModTime: mod}, // Non-current version under NoncurrentDays. {Path: "obj/a/.versions/v1", IsLatest: false, ModTime: mod, SuccessorModTime: mod}, - // MPU init under AbortMPU. - {Path: ".uploads/u1/", IsMPUInit: true, ModTime: mod}, + // MPU init under AbortMPU. Real init is a directory; DestKey + // carries the eventual object key for prefix matching. + {Path: ".uploads/u1", IsDirectory: true, IsMPUInit: true, DestKey: "obj/a", ModTime: mod}, } rec := &recorder{} @@ -116,18 +119,24 @@ func TestWalk_MultiActionRule_AllDueDispatched(t *testing.T) { t.Fatalf("Walk: %v", err) } - gotKinds := map[s3lifecycle.ActionKind]bool{} - for _, c := range rec.calls { - gotKinds[c.kind] = true + // Exact-shape assertion: each entry dispatches exactly one action, + // and ABORT_MPU only fires on the .uploads/ entry. A weaker + // "kinds-as-set" check would have missed the (kind, info) gating + // regression where an MPU init also fired NONCURRENT_DAYS. + want := []dispatchCall{ + {kind: s3lifecycle.ActionKindAbortMPU, path: ".uploads/u1"}, + {kind: s3lifecycle.ActionKindExpirationDays, path: "obj/a"}, + {kind: s3lifecycle.ActionKindNoncurrentDays, path: "obj/a/.versions/v1"}, } - for _, want := range []s3lifecycle.ActionKind{ - s3lifecycle.ActionKindExpirationDays, - s3lifecycle.ActionKindNoncurrentDays, - s3lifecycle.ActionKindAbortMPU, - } { - if !gotKinds[want] { - t.Fatalf("expected dispatch for %v, got %v", want, rec.calls) + got := append([]dispatchCall(nil), rec.calls...) + sort.Slice(got, func(i, j int) bool { + if got[i].path != got[j].path { + return got[i].path < got[j].path } + return got[i].kind < got[j].kind + }) + if !reflect.DeepEqual(got, want) { + t.Fatalf("dispatch calls mismatch:\n got %+v\nwant %+v", got, want) } } @@ -310,3 +319,90 @@ func TestWalk_ResumeFromCheckpoint(t *testing.T) { t.Fatalf("checkpoint want c, got %q", cp.LastScannedPath) } } + +func TestWalk_MPUInitDirMatchesByDestKey(t *testing.T) { + // Existing in-flight MPUs predate the meta-log subscription, so they + // only get cleaned up via the bootstrap walk. The init record is a + // directory whose path is .uploads/; the rule's Filter.Prefix + // applies to the destination object key, not the upload directory. + rule := &s3lifecycle.Rule{ + ID: "r-mpu", + Status: s3lifecycle.StatusEnabled, + Prefix: "logs/", + AbortMPUDaysAfterInitiation: 7, + } + snap := compileEvDriven(t, "bk", rule) + mod := mustTime(t, "2024-01-01T00:00:00Z") + now := mod.AddDate(0, 0, 8) // past the 7d threshold + + entries := []*Entry{ + // Matches: dest key under logs/. + {Path: ".uploads/u-match", IsDirectory: true, IsMPUInit: true, DestKey: "logs/foo.txt", ModTime: mod}, + // Filtered out: dest key under data/. + {Path: ".uploads/u-skip", IsDirectory: true, IsMPUInit: true, DestKey: "data/foo.txt", ModTime: mod}, + // No DestKey: malformed init mid-write; skip rather than guess. + {Path: ".uploads/u-bare", IsDirectory: true, IsMPUInit: true, ModTime: mod}, + } + + rec := &recorder{} + if _, err := Walk(context.Background(), snap, "bk", EntryCallback(entries), rec, WalkOptions{Now: now}); err != nil { + t.Fatalf("Walk: %v", err) + } + if len(rec.calls) != 1 { + t.Fatalf("expected 1 dispatch (u-match only), got %v", rec.calls) + } + if rec.calls[0].path != ".uploads/u-match" { + t.Fatalf("dispatch path=%q, want .uploads/u-match (the rm target)", rec.calls[0].path) + } + if rec.calls[0].kind != s3lifecycle.ActionKindAbortMPU { + t.Fatalf("dispatch kind=%v, want AbortMPU", rec.calls[0].kind) + } +} + +func TestWalk_NonMPUDirectorySkipped(t *testing.T) { + // Non-MPU directories must still be skipped — the relaxed + // IsDirectory check is gated on IsMPUInit. + rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1} + snap := compileEvDriven(t, "bk", rule) + mod := mustTime(t, "2024-01-01T00:00:00Z") + now := mod.AddDate(0, 0, 100) + + rec := &recorder{} + if _, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{ + {Path: "a/", IsDirectory: true, IsLatest: true, ModTime: mod}, + }), rec, WalkOptions{Now: now}); err != nil { + t.Fatalf("Walk: %v", err) + } + if len(rec.calls) != 0 { + t.Fatalf("plain directory should not dispatch, got %v", rec.calls) + } +} + +func TestWalk_MPUInitDoesNotFireNoncurrent(t *testing.T) { + // Same rule covers both AbortMPU and NoncurrentVersionExpiration; the + // MPU init record must dispatch only the AbortMPU action. Without the + // engine guard, NONCURRENT_DAYS would fire (IsLatest=false) and the + // server would BLOCK on empty version_id, freezing the cursor. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + AbortMPUDaysAfterInitiation: 7, + NoncurrentVersionExpirationDays: 7, + } + snap := compileEvDriven(t, "bk", rule) + mod := mustTime(t, "2024-01-01T00:00:00Z") + now := mod.AddDate(0, 0, 30) + + rec := &recorder{} + if _, err := Walk(context.Background(), snap, "bk", EntryCallback([]*Entry{ + {Path: ".uploads/u1", IsDirectory: true, IsMPUInit: true, DestKey: "obj/a", ModTime: mod}, + }), rec, WalkOptions{Now: now}); err != nil { + t.Fatalf("Walk: %v", err) + } + if len(rec.calls) != 1 { + t.Fatalf("expected 1 dispatch (AbortMPU only), got %v", rec.calls) + } + if rec.calls[0].kind != s3lifecycle.ActionKindAbortMPU { + t.Fatalf("kind=%v, want AbortMPU", rec.calls[0].kind) + } +} diff --git a/weed/s3api/s3lifecycle/evaluate.go b/weed/s3api/s3lifecycle/evaluate.go index 6de9cf750..2375a96ec 100644 --- a/weed/s3api/s3lifecycle/evaluate.go +++ b/weed/s3api/s3lifecycle/evaluate.go @@ -17,6 +17,13 @@ func EvaluateAction(rule *Rule, kind ActionKind, info *ObjectInfo, now time.Time if !filterMatches(rule, info) { return none } + // MPU init records carry IsLatest=false (they are not yet versions); + // without this guard NoncurrentDays / NewerNoncurrent fire on them + // and the dispatcher BLOCKs because version_id is empty, freezing + // the cursor. Only ABORT_MPU is meaningful for an in-flight upload. + if info.IsMPUInit && kind != ActionKindAbortMPU { + return none + } switch kind { case ActionKindAbortMPU: diff --git a/weed/s3api/s3lifecycle/evaluate_test.go b/weed/s3api/s3lifecycle/evaluate_test.go index ea59b9dd6..99e1b08d9 100644 --- a/weed/s3api/s3lifecycle/evaluate_test.go +++ b/weed/s3api/s3lifecycle/evaluate_test.go @@ -305,3 +305,23 @@ func TestEvaluateAction_EmptyPrefixMatchesAll(t *testing.T) { t.Fatalf("empty prefix should match, got %v", got) } } + +func TestEvaluateAction_MPUInitDoesNotFireNoncurrent(t *testing.T) { + // IsLatest=false on an MPU init must not let NoncurrentDays / + // NewerNoncurrent fire. The dispatcher BLOCKs noncurrent kinds with + // empty version_id, which would freeze the cursor. + rule := &Rule{ + ID: "r", + Status: StatusEnabled, + NoncurrentVersionExpirationDays: 7, + NewerNoncurrentVersions: 3, + } + init := time.Now().AddDate(0, 0, -30) + info := &ObjectInfo{Key: "logs/foo.txt", IsMPUInit: true, ModTime: init} + + for _, kind := range []ActionKind{ActionKindNoncurrentDays, ActionKindNewerNoncurrent} { + if got := EvaluateAction(rule, kind, info, time.Now()); got.Action != ActionNone { + t.Fatalf("kind=%v on IsMPUInit must return None, got %+v", kind, got) + } + } +} diff --git a/weed/s3api/s3lifecycle/router/router.go b/weed/s3api/s3lifecycle/router/router.go index 9badf77e4..569d80b0b 100644 --- a/weed/s3api/s3lifecycle/router/router.go +++ b/weed/s3api/s3lifecycle/router/router.go @@ -71,6 +71,17 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match { if action.Mode != engine.ModeEventDriven { continue } + // (kind, info) shape gate: ABORT_MPU only fires on MPU init events, + // and other kinds never do. Without this an MPU init would be + // matched against NONCURRENT_DAYS (IsLatest=false reads as a + // non-current version) and the dispatcher would BLOCK on empty + // version_id. + if info.IsMPUInit && key.ActionKind != s3lifecycle.ActionKindAbortMPU { + continue + } + if !info.IsMPUInit && key.ActionKind == s3lifecycle.ActionKindAbortMPU { + continue + } // Schedule from ModTime, not the meta-log event time: a backdated // or out-of-band entry update has eventTime ≈ now but ModTime far // in the past, so eventTime+Delay would push the dispatch into the @@ -99,9 +110,13 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match { // buildObjectInfo derives a non-versioned ObjectInfo from a meta-log event. // Versioned-bucket semantics (IsLatest, NumVersions, NoncurrentIndex, // IsDeleteMarker for noncurrent versions) require listing siblings and land -// in Phase 5; for now any event is treated as IsLatest=true with the +// in Phase 5; for now any non-MPU event is treated as IsLatest=true with the // LifecycleDelete RPC's identity-CAS catching stale schedules. // +// MPU init directories at .uploads/ populate IsMPUInit and use the +// destination object key from the entry's Extended map for filter matching, +// so a rule with Filter.Prefix=foo/ matches an MPU uploading to foo/bar.txt. +// // Returns nil when Attributes are missing — without ModTime, EvaluateAction // would compute due against year-0001 and fire immediately. func buildObjectInfo(ev *reader.Event) *s3lifecycle.ObjectInfo { @@ -109,6 +124,17 @@ func buildObjectInfo(ev *reader.Event) *s3lifecycle.ObjectInfo { if entry == nil || entry.Attributes == nil { return nil } + if destKey, ok := mpuInitInfo(ev, entry); ok { + // MPU intermediate state has no tags, no versions, no delete-marker + // semantics. info.Key is the user's destination key so a rule's + // Filter.Prefix matches the eventual object; the dispatcher already + // carries .uploads/ in m.ObjectKey for ABORT_MPU. + return &s3lifecycle.ObjectInfo{ + Key: destKey, + ModTime: time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)), + IsMPUInit: true, + } + } info := &s3lifecycle.ObjectInfo{ Key: ev.Key, ModTime: time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)), @@ -125,6 +151,27 @@ func buildObjectInfo(ev *reader.Event) *s3lifecycle.ObjectInfo { return info } +// mpuInitInfo recognizes a multipart-upload init: a directory entry at +// `.uploads/` carrying the destination key in Extended. Sub-events +// for part uploads (deeper paths under the upload directory) are deliberately +// rejected — they ride a different mtime and would over-fire ABORT_MPU. +func mpuInitInfo(ev *reader.Event, entry *filer_pb.Entry) (destKey string, ok bool) { + uploadsPrefix := s3_constants.MultipartUploadsFolder + "/" + if !entry.IsDirectory || !strings.HasPrefix(ev.Key, uploadsPrefix) { + return "", false + } + rest := ev.Key[len(uploadsPrefix):] + if rest == "" || strings.ContainsRune(rest, '/') { + // `.uploads/` itself or `.uploads//...`; not the init. + return "", false + } + keyBytes, hasKey := entry.Extended[s3_constants.ExtMultipartObjectKey] + if !hasKey || len(keyBytes) == 0 { + return "", false + } + return string(keyBytes), true +} + // buildIdentity captures the entry's schedule-time fingerprint for the CAS // witness. Returns nil if the event has no entry to fingerprint (deletes). func buildIdentity(ev *reader.Event) *EntryIdentity { diff --git a/weed/s3api/s3lifecycle/router/router_test.go b/weed/s3api/s3lifecycle/router/router_test.go index 7d5e57551..0b243a7e3 100644 --- a/weed/s3api/s3lifecycle/router/router_test.go +++ b/weed/s3api/s3lifecycle/router/router_test.go @@ -5,6 +5,7 @@ import ( "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/engine" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" @@ -226,3 +227,174 @@ func TestRouteIdentityHashesExtended(t *testing.T) { t.Fatalf("ExtendedHash mismatch:\n got %x\nwant %x", id.ExtendedHash, want) } } + +func mpuInitEvent(bucket, uploadID, destKey string, initS, ts int64) *reader.Event { + return &reader.Event{ + TsNs: ts, + Bucket: bucket, + Key: ".uploads/" + uploadID, + NewEntry: &filer_pb.Entry{ + Name: uploadID, + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{Mtime: initS}, + Extended: map[string][]byte{ + s3_constants.ExtMultipartObjectKey: []byte(destKey), + }, + }, + } +} + +func TestRouteMPUInitFiresAbortAfterDelay(t *testing.T) { + // 7-day MPU abort. Init 8 days ago must fire; rule's prefix matches the + // destination key, not the .uploads/ path. + rule := &s3lifecycle.Rule{ + ID: "r-mpu", + Status: s3lifecycle.StatusEnabled, + Prefix: "logs/", + AbortMPUDaysAfterInitiation: 7, + } + snap := compileWith(rule, activatedPrior(rule)) + + now := time.Now() + init := now.AddDate(0, 0, -8) + ev := mpuInitEvent("bk", "u1", "logs/foo.txt", init.Unix(), init.UnixNano()) + + matches := Route(snap, ev, now) + if len(matches) != 1 { + t.Fatalf("expected 1 match, got %v", matches) + } + if got, want := matches[0].Key.ActionKind, s3lifecycle.ActionKindAbortMPU; got != want { + t.Fatalf("ActionKind=%v, want %v", got, want) + } + if got, want := matches[0].ObjectKey, ".uploads/u1"; got != want { + t.Fatalf("ObjectKey=%q, want %q (server needs the upload directory path)", got, want) + } +} + +func TestRouteMPUInitFilteredOutByPrefix(t *testing.T) { + // Same rule, MPU uploads to a different prefix → no match. + rule := &s3lifecycle.Rule{ + ID: "r-mpu", + Status: s3lifecycle.StatusEnabled, + Prefix: "logs/", + AbortMPUDaysAfterInitiation: 7, + } + snap := compileWith(rule, activatedPrior(rule)) + + now := time.Now() + init := now.AddDate(0, 0, -8) + ev := mpuInitEvent("bk", "u2", "data/foo.txt", init.Unix(), init.UnixNano()) + + if got := Route(snap, ev, now); len(got) != 0 { + t.Fatalf("expected 0 matches for foreign prefix, got %v", got) + } +} + +func TestRouteMPUInitMissingDestKeySkipped(t *testing.T) { + // .uploads/ directory without ExtMultipartObjectKey is malformed + // (mkdir wrote it before the metadata, or it's a stray dir). Skip. + rule := &s3lifecycle.Rule{ + ID: "r-mpu", + Status: s3lifecycle.StatusEnabled, + AbortMPUDaysAfterInitiation: 7, + } + snap := compileWith(rule, activatedPrior(rule)) + + now := time.Now() + init := now.AddDate(0, 0, -8) + ev := &reader.Event{ + TsNs: init.UnixNano(), + Bucket: "bk", + Key: ".uploads/u3", + NewEntry: &filer_pb.Entry{ + Name: "u3", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{Mtime: init.Unix()}, + }, + } + + if got := Route(snap, ev, now); len(got) != 0 { + t.Fatalf("expected 0 matches for missing destKey, got %v", got) + } +} + +func TestRouteMPUPartEventSkipped(t *testing.T) { + // Part-upload events at .uploads// ride a different mtime and + // must not over-fire ABORT_MPU. + rule := &s3lifecycle.Rule{ + ID: "r-mpu", + Status: s3lifecycle.StatusEnabled, + AbortMPUDaysAfterInitiation: 7, + } + snap := compileWith(rule, activatedPrior(rule)) + + now := time.Now() + init := now.AddDate(0, 0, -8) + ev := &reader.Event{ + TsNs: init.UnixNano(), + Bucket: "bk", + Key: ".uploads/u4/0001", + NewEntry: &filer_pb.Entry{ + Name: "0001", + Attributes: &filer_pb.FuseAttributes{Mtime: init.Unix()}, + Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("logs/foo.txt")}, + }, + } + + if got := Route(snap, ev, now); len(got) != 0 { + t.Fatalf("expected 0 matches for part event, got %v", got) + } +} + +func TestRouteMPUInitDoesNotFireNoncurrent(t *testing.T) { + // One rule with both ABORT_MPU and NONCURRENT_DAYS; an MPU init must + // only emit the ABORT_MPU match. Otherwise the dispatcher receives a + // NONCURRENT_DAYS action with version_id="" and freezes the cursor. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + AbortMPUDaysAfterInitiation: 7, + NoncurrentVersionExpirationDays: 7, + } + snap := compileWith(rule, activatedPrior(rule)) + + now := time.Now() + init := now.AddDate(0, 0, -30) + ev := mpuInitEvent("bk", "u1", "logs/foo.txt", init.Unix(), init.UnixNano()) + + matches := Route(snap, ev, now) + if len(matches) != 1 { + t.Fatalf("expected exactly 1 match (ABORT_MPU only), got %v", matches) + } + if got := matches[0].Key.ActionKind; got != s3lifecycle.ActionKindAbortMPU { + t.Fatalf("ActionKind=%v, want AbortMPU", got) + } +} + +func TestRouteRegularObjectUnderDualRuleSkipsAbortMPU(t *testing.T) { + // Converse of TestRouteMPUInitDoesNotFireNoncurrent: a regular + // current-version object under a rule that has both ExpirationDays + // and AbortIncompleteMultipartUpload must fire EXPIRATION_DAYS only. + // Without the gate the dispatcher would also receive an ABORT_MPU + // action targeting the object path, which would rm a regular object + // via the MPU code path. + rule := &s3lifecycle.Rule{ + ID: "r", + Status: s3lifecycle.StatusEnabled, + ExpirationDays: 1, + AbortMPUDaysAfterInitiation: 7, + } + snap := compileWith(rule, activatedPrior(rule)) + + now := time.Now() + old := now.AddDate(0, 0, -2) // past the 1d expiration + ev := eventCreate("bk", "obj", old.Unix(), 1, old.UnixNano()) + + matches := Route(snap, ev, now) + if len(matches) != 1 { + t.Fatalf("expected exactly 1 match (EXPIRATION_DAYS only), got %v", matches) + } + if got := matches[0].Key.ActionKind; got != s3lifecycle.ActionKindExpirationDays { + t.Fatalf("ActionKind=%v, want ExpirationDays", got) + } +}