feat(s3/lifecycle): wire AbortIncompleteMultipartUpload (Phase 5a) (#9368)

* feat(s3/lifecycle/router): emit ABORT_MPU events for .uploads/<id> init dirs Detect a meta-log event at exactly .uploads/<upload_id> (a directory) and build the ObjectInfo from its destination key (entry.Extended[key]) so a rule with Filter.Prefix=foo/ matches an MPU uploading to foo/bar. Sub-events under .uploads/<id>/<part> ride a different mtime and would over-fire the ABORT_MPU schedule, so they're rejected explicitly. m.ObjectKey stays as ev.Key (.uploads/<upload_id>) — the dispatcher needs the upload directory path, not the destination key, to actually remove the in-flight upload. * feat(s3api): wire LifecycleDelete ABORT_MPU to remove the upload dir Replaces the retryLater stub. Validates the .uploads/<upload_id> shape of req.ObjectPath (so a malformed event can't escalate to a wider rm), then deletes the upload directory under <bucket>/.uploads/<id>. Maps NotFound to NOOP_RESOLVED, transport errors to RETRY_LATER, success to DONE. * refactor(s3api): drop redundant exists check before lifecycle ABORT_MPU rm s3a.rm already does a NotFound-returning lookup, so the pre-check just adds a round-trip. Map filer_pb.ErrNotFound to NOOP_RESOLVED on rm, keep transport errors as RETRY_LATER. * refactor(s3/lifecycle/router): use s3_constants for MPU paths + Extended key Drop the hardcoded ".uploads/" and "key" string literals; the symbols already exist as s3_constants.MultipartUploadsFolder and ExtMultipartObjectKey, and the server side reaches them through the same constants. Keeping the test helpers tied to those names also makes the negative-result tests meaningful — they'd otherwise still pass if the lookup constant drifted. * fix(s3api): close lifecycle ABORT_MPU traversal + NOT_FOUND gaps Two issues with the recent ABORT_MPU plumbing: - "." and ".." passed the no-slash check but resolve to the bucket root via util.JoinPath, so .uploads/.. could rm the wrong directory. - filer.DeleteEntry suppresses ErrNotFound and returns success, so the rm path can't distinguish missing from deleted; the previous version reported DONE for an already-aborted upload instead of NOOP_RESOLVED. Reject the two reserved names explicitly and restore the existence pre-check so the outcome map stays correct. Add a table-test covering the rejected paths. * fix(s3/lifecycle/bootstrap): walk MPU init dirs by destination key A real MPU init record is a directory under .uploads/<id> created by mkdir; the bootstrap walker was skipping every directory entry, so an MPU that existed before the meta-log subscription was never aborted. Even with the skip relaxed, MatchPath used the .uploads/<id> path, so a rule with Filter.Prefix=logs/ would never fire on an MPU uploading to logs/foo.txt. Add Entry.DestKey, let IsMPUInit directories through, and use DestKey for both MatchPath and ObjectInfo.Key. A bare init directory with no DestKey means metadata hasn't landed yet — skip rather than guess. * fix(s3/lifecycle): gate (kind, info) shape so MPU init only fires ABORT_MPU An MPU init record carries IsMPUInit=true and IsLatest=false. Without gating, the router and bootstrap walker matched it against every active ActionKey for the bucket, so NONCURRENT_DAYS / NEWER_NONCURRENT fired (IsLatest=false reads as a noncurrent version). The dispatcher would then BLOCK on empty version_id and freeze the cursor. Add a shape gate at both call sites: - IsMPUInit + non-ABORT_MPU kind → continue - regular object + ABORT_MPU kind → continue Plus a defense-in-depth check at the top of EvaluateAction so future callers can't reintroduce the bug. Tests cover all three layers. * test(s3/lifecycle): tighten dual-action coverage at the call sites - Walk multi-action: replace the kinds-as-set check with an exact-shape DeepEqual on (path, kind) tuples. The set check would have missed an MPU init wrongly firing NONCURRENT_DAYS — exactly the regression the (kind, info) gate fixes. - Router: add a converse case for the dual ExpirationDays + AbortIncompleteMultipartUpload rule. A regular current-version object must fire only EXPIRATION_DAYS; without the gate the dispatcher would also receive ABORT_MPU and rm the object via the MPU code path.
2026-08-18 21:26:56 +00:00 · 2026-05-08 12:12:42 -07:00
parent 8b87ceb0d1
commit 89aab30821
8 changed files with 455 additions and 20 deletions
@@ -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/<upload_id>` (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):
@@ -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)
+			}
+		})
+	}
+}
@@ -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/<id> 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/<id> 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/<id> 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
@@ -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/<id> 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/<id>; 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)
+	}
+}
@@ -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:
@@ -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)
+		}
+	}
+}
@@ -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/<upload_id> 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/<upload_id> 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/<upload_id>` 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/<id>/<part>...`; 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 {
@@ -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/<id> 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/<id>/<part> 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)
+	}
+}