feat(s3/lifecycle): sole-survivor delete-marker routing (Phase 5b/2) (#9381)

* feat(s3/lifecycle): sole-survivor delete-marker routing (Phase 5b/2)

Production stores delete markers under <object>.versions/<version-file>;
buildObjectInfo skips every version-folder event because it can't tell
current from noncurrent without sibling state. EXP_DM is the one rule
that can be routed from the file path alone: if the marker is the only
entry under .versions/<key>/ then it's necessarily the latest.

Add SiblingLister; gate the listing on (versioned, version-folder path,
delete-marker entry, EXP_DM rule active for the bucket, lister supplied)
so non-marker version writes pay nothing. The Match carries the LOGICAL
key in ObjectKey and the marker's version_id in VersionID so the
dispatcher can reach deleteSpecificObjectVersion(bucket, logical, vid);
without a version_id the dispatcher would BLOCK and freeze the cursor.

Server-side dispatch re-checks sole-survivor before deleting: another
PUT can land between schedule and dispatch, and identity-CAS only covers
the marker entry, not the directory shape. Lists .versions/<key>/ with
limit 2 and returns NOOP_RESOLVED if count != 1 or the surviving entry
is not the same version.

Fix isDeleteMarkerEntry to compare string(v) == "true" — production
writes []byte("true"), not {1}; every other reader of ExtDeleteMarkerKey
in this repo uses the string predicate.

filerSiblingLister.Count caps at limit 2 — callers only distinguish
"sole survivor" (1) from "more than one" (>=2).

Follows #9373.

* fix(s3/lifecycle): gate sibling listing on active event-driven EXP_DM

hasActionKind tested key presence only; an EXP_DM rule whose action was
inactive (BootstrapComplete=false) or scan-only would still trigger the
sibling-listing RPC even though no match could fire. Replace with a
helper that mirrors the per-key filter Route already applies before
emitting matches: kind matches, snap.Action returns non-nil, IsActive,
and Mode == ModeEventDriven.

Add a regression test that builds a versioned snapshot with the rule
but no PriorStates (action stays inactive) and asserts the lister is
never consulted.

* chore(s3/lifecycle): trim verbose comments

* fix(s3/lifecycle): null version, hard-delete trigger, latest pointer

Three correctness gaps in EXP_DM routing.

1. Null version. SiblingLister.Count saw only .versions/<key>/, but a
   pre-versioning bare-key object survives outside that folder when
   versioning is enabled later. Marker + null would look like count==1
   and EXP_DM would re-expose the old object. Replace Count with
   Survivors which also reports HasNullVersion; route- and dispatch-time
   suppress when set.

2. Hard-delete trigger. The router skipped events with NewEntry==nil so
   a noncurrent hard-delete that left the marker as sole survivor never
   fired EXP_DM. Allow version-folder hard-deletes through; the lister's
   LoneEntry carries the surviving marker's version_id and identity.

3. Latest pointer. checkSoleSurvivorMarker only checked count and
   filename. The worker can race createDeleteMarker between the file
   write and the .versions/ directory metadata update. Require
   ExtLatestVersionIdKey == versionId; missing pointer returns
   retry-later instead of deleting.

Adds a null-version exists check on the dispatch path too.

* fix(s3/lifecycle): normalize null-version lookup errors and detect dir markers

Two correctness bugs in the null-version check.

Route-side: filerSiblingLister called client.LookupDirectoryEntry
directly. SeaweedList wraps not-found via filer_pb.LookupEntry, which
normalizes the gRPC string-mapped not-found into ErrNotFound. The raw
client returns it as a generic error instead, so every absent
null-version (the common case) bubbled up as an error and the router
suppressed every otherwise-valid match. Use filer_pb.LookupEntry.

Both sides: explicit S3 directory-key markers (object names ending in /)
are stored as directory entries with Attributes.Mime set;
processExplicitDirectory in the listing path treats them as null
versions. The previous check was !IsDirectory only, which let the
marker delete proceed and re-expose the directory key. Add
IsDirectoryKeyObject() to both predicates. Also use
util.NewFullPath(...).DirAndName() for the parent/name split so a
trailing-slash key resolves to the same underlying entry path as the
listing code.

* fix(s3/lifecycle): EXP_DM ctx propagation, nil-entry guard, fast-path skip

Three small follow-ups on the EXP_DM dispatch path.

checkSoleSurvivorMarker now takes ctx instead of context.Background()
so worker shutdown / deadlines cancel the SeaweedList RPC instead of
stalling.

If SeaweedList fires the lone callback with entry==nil, firstName
stays empty and the marker-replaced check would short-circuit; that's
the one shape that bypasses the dispatch guard, so retry-later instead.

routeSoleSurvivorMarker now skips the Survivors RPC on regular
non-marker version creates — those always have Count >= 2, so the
listing was wasted load on every versioned write under an EXP_DM rule.
Hard-delete events (NewEntry==nil) and marker creates still flow
through. Added a regression test asserting the regular-create case
doesn't consult the lister.

Documented that logicalKeyFromVersionPath rejects bucket-root markers
intentionally.

weed/s3api/s3api_internal_lifecycle.go

@@ -120,6 +120,15 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 				return noopResolved("VERSION_IS_LATEST"), nil
 			}
 		}
+		// Re-check sole-survivor: a fresh PUT can land between schedule
+		// and dispatch. Identity-CAS upstream covers the marker bytes;
+		// this covers the directory shape.
+		if req.ActionKind == s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER {
+			outcome, err := s3a.checkSoleSurvivorMarker(ctx, req.Bucket, req.ObjectPath, req.VersionId)
+			if outcome != nil || err != nil {
+				return outcome, err
+			}
+		}
 		if err := s3a.deleteSpecificObjectVersion(req.Bucket, req.ObjectPath, req.VersionId); err != nil {
 			if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrObjectNotFound) {
 				return noopResolved("NOT_FOUND_AT_DELETE"), nil
@@ -175,6 +184,92 @@ func (s3a *S3ApiServer) lifecycleAbortMPU(ctx context.Context, req *s3_lifecycle
 	return done(), nil
 }
 
+// checkSoleSurvivorMarker returns nil to proceed with the delete, or a
+// terminal response when state has drifted: count != 1, the surviving
+// entry is a different version, the .versions/ directory's latest
+// pointer doesn't name versionId, or a bare null-version exists outside
+// .versions/. Pointer missing while a marker is present is treated as
+// retry-later — the create races with the directory metadata update.
+func (s3a *S3ApiServer) checkSoleSurvivorMarker(ctx context.Context, bucket, object, versionId string) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
+	bucketDir := s3a.bucketDir(bucket)
+	versionsDir := bucketDir + "/" + object + s3_constants.VersionsFolder
+	count := 0
+	var firstName string
+	err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return filer_pb.SeaweedList(ctx, client, versionsDir, "", func(entry *filer_pb.Entry, _ bool) error {
+			count++
+			if count == 1 && entry != nil {
+				firstName = entry.Name
+			}
+			return nil
+		}, "", false, 2)
+	})
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return noopResolved("NOT_FOUND"), nil
+		}
+		return retryLater("TRANSPORT_ERROR: sole-survivor list: " + err.Error()), nil
+	}
+	if count == 0 {
+		return noopResolved("NOT_FOUND"), nil
+	}
+	if count > 1 {
+		return noopResolved("NOT_SOLE_SURVIVOR"), nil
+	}
+	// SeaweedList delivered a single callback but with a nil entry; we
+	// can't compare names so retry rather than silently bypass the
+	// marker-replaced check.
+	if firstName == "" {
+		return retryLater("PENDING_SURVIVOR_ENTRY"), nil
+	}
+	if versionId != "" && firstName != s3a.getVersionFileName(versionId) {
+		return noopResolved("MARKER_REPLACED"), nil
+	}
+	// Latest-pointer check: createDeleteMarker writes the marker file
+	// and then updates the parent directory's Extended map. Reading
+	// before the second step lands would see count==1 but no pointer;
+	// retry-later rather than mistakenly delete.
+	parent, name := path.Split(versionsDir)
+	parent = strings.TrimRight(parent, "/")
+	if parent == "" {
+		parent = "/"
+	}
+	versionsEntry, err := s3a.getEntry(parent, name)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return noopResolved("NOT_FOUND"), nil
+		}
+		return retryLater("TRANSPORT_ERROR: latest-pointer lookup: " + err.Error()), nil
+	}
+	if versionsEntry == nil {
+		return noopResolved("NOT_FOUND"), nil
+	}
+	latest, hasPointer := versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey]
+	if !hasPointer || len(latest) == 0 {
+		return retryLater("PENDING_LATEST_POINTER"), nil
+	}
+	if string(latest) != versionId {
+		return noopResolved("MARKER_NOT_LATEST"), nil
+	}
+	// Null-version check: pre-versioning objects survive as the bare
+	// <bucket>/<key>. Both regular files and explicit S3 directory-key
+	// markers (object names ending in /) qualify; the listing path
+	// (s3api_object_versioning.go processExplicitDirectory) treats both
+	// as the null version. getEntry uses NewFullPath so a trailing slash
+	// in object splits the same as a regular key.
+	bareEntry, err := s3a.getEntry(bucketDir, object)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return nil, nil
+		}
+		return retryLater("TRANSPORT_ERROR: null-version lookup: " + err.Error()), nil
+	}
+	if bareEntry != nil && (!bareEntry.IsDirectory || bareEntry.IsDirectoryKeyObject()) {
+		return noopResolved("NULL_VERSION_PRESENT"), nil
+	}
+	return nil, nil
+}
+
 // isCurrentLatestVersion reports whether versionId is the version the
 // .versions/ directory currently points to. SeaweedFS records the latest
 // version on the parent directory's Extended map; without consulting it,

weed/s3api/s3lifecycle/dispatcher/pipeline.go

@@ -205,6 +205,8 @@ func (p *Pipeline) Run(ctx context.Context) error {
 		cancel() // wake the dispatcher goroutine to drain & exit
 	}()
 
+	lister := &filerSiblingLister{client: p.FilerClient, bucketsPath: p.BucketsPath}
+
 	// Router/dispatcher goroutine: pulls events, routes them to per-shard
 	// schedules, ticks every shard's dispatcher on the same cadence, and
 	// checkpoints every shard's cursor on the checkpoint cadence. One
@@ -257,7 +259,7 @@ func (p *Pipeline) Run(ctx context.Context) error {
 				// against the prior (empty) snapshot would silently drop
 				// every match. Engine.Snapshot is an atomic Load.
 				snap := p.Engine.Snapshot()
-				for _, m := range router.Route(snap, ev, time.Now()) {
+				for _, m := range router.Route(runCtx, snap, ev, time.Now(), lister) {
 					st.dispatch.Schedule.Add(m)
 				}
 			case <-dt.C:

weed/s3api/s3lifecycle/dispatcher/pipeline_test.go

@@ -48,7 +48,7 @@ func TestPipelineIntegrationInMemory(t *testing.T) {
 		},
 	}
 
-	matches := router.Route(snap, ev, now)
+	matches := router.Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match, got %v", matches)
 	}

weed/s3api/s3lifecycle/dispatcher/sibling_lister.go

@@ -0,0 +1,61 @@
+package dispatcher
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// filerSiblingLister inspects the .versions/<key>/ folder and the bare
+// null-version. Listing caps at 2 — callers only distinguish 1 vs >=2.
+type filerSiblingLister struct {
+	client      filer_pb.SeaweedFilerClient
+	bucketsPath string
+}
+
+func (l *filerSiblingLister) Survivors(ctx context.Context, bucket, objectKey string) (router.Survivors, error) {
+	bucketPath := strings.TrimSuffix(l.bucketsPath, "/") + "/" + bucket
+	versionsDir := bucketPath + "/" + objectKey + s3_constants.VersionsFolder
+
+	var s router.Survivors
+	err := filer_pb.SeaweedList(ctx, l.client, versionsDir, "", func(entry *filer_pb.Entry, _ bool) error {
+		s.Count++
+		if s.Count == 1 && entry != nil {
+			s.LoneEntry = entry
+		} else if s.Count > 1 {
+			s.LoneEntry = nil
+		}
+		return nil
+	}, "", false, 2)
+	if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
+		return router.Survivors{}, err
+	}
+
+	// NewFullPath strips a trailing slash from objectKey so directory-key
+	// objects (foo/) split the same as regular keys. LookupEntry
+	// normalizes the gRPC string-mapped not-found into ErrNotFound; a
+	// raw client.LookupDirectoryEntry would return that as a generic
+	// error and suppress every otherwise-valid match.
+	parent, name := util.NewFullPath(bucketPath, objectKey).DirAndName()
+	resp, err := filer_pb.LookupEntry(ctx, l.client, &filer_pb.LookupDirectoryEntryRequest{
+		Directory: parent,
+		Name:      name,
+	})
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			return s, nil
+		}
+		return router.Survivors{}, err
+	}
+	// Bare regular file or an explicit S3 directory-marker (an empty
+	// directory entry with Mime set) both count as the null version.
+	if resp.Entry != nil && (!resp.Entry.IsDirectory || resp.Entry.IsDirectoryKeyObject()) {
+		s.HasNullVersion = true
+	}
+	return s, nil
+}

weed/s3api/s3lifecycle/router/router.go

@@ -1,9 +1,11 @@
 package router
 
 import (
+	"context"
 	"strings"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
@@ -11,6 +13,22 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
 )
 
+// SiblingLister inspects the surviving versions of a versioned key.
+// nil receiver or an error means "unknown" — callers suppress.
+type SiblingLister interface {
+	Survivors(ctx context.Context, bucket, objectKey string) (Survivors, error)
+}
+
+// Survivors describes the state under .versions/<key>/ plus the bare
+// null-version that exists when versioning was turned on after the
+// object was first PUT (s3api_object_versioning.go treats <bucket>/<key>
+// as a regular file, the null version, in that case).
+type Survivors struct {
+	Count          int             // entries under .versions/<key>/, capped at 2
+	LoneEntry      *filer_pb.Entry // populated when Count == 1
+	HasNullVersion bool            // bare <bucket>/<key> exists as a regular file
+}
+
 // Match is one (event, action) pair where EvaluateAction fired. The
 // dispatcher runs `LifecycleDelete` at DueTime; identity-CAS in the RPC
 // guards against drift between schedule time and dispatch time.
@@ -42,26 +60,33 @@ type EntryIdentity struct {
 	ExtendedHash []byte
 }
 
-// Route returns the matches that fire for ev against snap. Only EVENT_DRIVEN
-// actions on this bucket are considered; actions in SCAN_AT_DATE or DISABLED
-// modes are out-of-band of the event stream. Inactive actions
-// (BootstrapComplete=false) are also skipped.
-func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match {
+// Route returns the matches that fire for ev against snap. Only active
+// event-driven actions are considered; SCAN_AT_DATE and DISABLED bypass
+// this path.
+func Route(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, now time.Time, lister SiblingLister) []Match {
 	if snap == nil || ev == nil {
 		return nil
 	}
-	// Hard deletes carry no schedule-relevant state: an Expiration would
-	// hit NOOP_RESOLVED at dispatch time anyway, ExpiredObjectDeleteMarker
-	// only fires on the latest-version delete-marker which is a Create
-	// from the server's perspective. Skip rather than burn a schedule slot.
-	if ev.NewEntry == nil {
-		return nil
-	}
 	keys := snap.BucketActionKeys(ev.Bucket)
 	if len(keys) == 0 {
 		return nil
 	}
-	info := buildObjectInfo(ev, snap.BucketVersioned(ev.Bucket))
+	versioned := snap.BucketVersioned(ev.Bucket)
+
+	// EXP_DM can fire on two version-folder events: the marker create
+	// (sole survivor immediately) and a noncurrent hard-delete that
+	// leaves only the marker behind. Both reach routeSoleSurvivorMarker.
+	if versioned && isVersionFolderPath(ev.Key) {
+		if !hasActiveEventDrivenAction(snap, keys, s3lifecycle.ActionKindExpiredDeleteMarker) {
+			return nil
+		}
+		return routeSoleSurvivorMarker(ctx, snap, ev, keys, lister)
+	}
+
+	if ev.NewEntry == nil {
+		return nil
+	}
+	info := buildObjectInfo(ev, versioned)
 	if info == nil {
 		return nil
 	}
@@ -112,42 +137,127 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []Match {
 	return matches
 }
 
-// buildObjectInfo derives an ObjectInfo from a meta-log event and the
-// bucket's versioning state. Returns nil when the event has no usable
-// shape (missing attributes, hard delete already handled upstream).
-//
-// On a versioned bucket the storage layout (<key>.versions/v_<id>) is
-// shared between the current latest and the noncurrent versions; the
-// latest pointer lives in the .versions/ directory's Extended map and
-// is updated separately. Without that pointer-transition signal here,
-// the router conservatively classifies every event as if it were the
-// current version (IsLatest=true) so it never deletes the live
-// latest; bootstrap walking + the server-side dispatch guard handle
-// noncurrent retention. NumVersions=0 keeps ExpiredObjectDeleteMarker
-// (which requires sole-survivor) suppressed.
-//
-// 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.
+// routeSoleSurvivorMarker emits an EXP_DM Match against the LOGICAL key
+// (so the dispatcher can call deleteSpecificObjectVersion) with the
+// marker's version_id. Handles two events: a marker create (the new
+// entry IS the marker) and a noncurrent hard-delete that leaves the
+// marker behind (the listing's lone entry IS the marker). The server
+// re-checks before deleting.
+func routeSoleSurvivorMarker(ctx context.Context, snap *engine.Snapshot, ev *reader.Event, keys []s3lifecycle.ActionKey, lister SiblingLister) []Match {
+	if lister == nil {
+		return nil
+	}
+	// Skip the listing RPC for events that can't possibly produce a
+	// sole-survivor marker: a regular non-marker version create/update
+	// always lands at Count >= 2.
+	if ev.NewEntry != nil && !isDeleteMarkerEntry(ev.NewEntry) {
+		return nil
+	}
+	logicalKey, ok := logicalKeyFromVersionPath(ev.Key)
+	if !ok {
+		return nil
+	}
+	s, err := lister.Survivors(ctx, ev.Bucket, logicalKey)
+	if err != nil {
+		glog.V(2).Infof("lifecycle router: survivors %s/%s: %v", ev.Bucket, logicalKey, err)
+		return nil
+	}
+	// Pre-versioning bare-key objects (the "null" version) live outside
+	// .versions/. Treating count==1 as sole-survivor while a null
+	// version exists would let lifecycle delete the marker and re-expose
+	// the old object.
+	if s.Count != 1 || s.HasNullVersion || s.LoneEntry == nil {
+		return nil
+	}
+	if !isDeleteMarkerEntry(s.LoneEntry) {
+		return nil
+	}
+	versionID := string(s.LoneEntry.Extended[s3_constants.ExtVersionIdKey])
+	if versionID == "" {
+		// Empty version_id would BLOCK at dispatch and freeze the cursor.
+		return nil
+	}
+	entry := s.LoneEntry
+	info := &s3lifecycle.ObjectInfo{
+		Key:            logicalKey,
+		ModTime:        time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)),
+		Size:           int64(entry.Attributes.FileSize),
+		IsLatest:       true,
+		IsDeleteMarker: true,
+		NumVersions:    1,
+	}
+	if tags := extractTags(entry.Extended); len(tags) > 0 {
+		info.Tags = tags
+	}
+	eventTime := time.Unix(0, ev.TsNs)
+	identity := buildIdentityFromEntry(entry)
+	var matches []Match
+	for _, key := range keys {
+		if key.ActionKind != s3lifecycle.ActionKindExpiredDeleteMarker {
+			continue
+		}
+		action := snap.Action(key)
+		if action == nil || !action.IsActive() || action.Mode != engine.ModeEventDriven {
+			continue
+		}
+		dueTime := info.ModTime.Add(action.Delay)
+		res := s3lifecycle.EvaluateAction(action.Rule, key.ActionKind, info, dueTime)
+		if res.Action == s3lifecycle.ActionNone {
+			continue
+		}
+		matches = append(matches, Match{
+			Key:       key,
+			Action:    action,
+			Result:    res,
+			EventTs:   eventTime,
+			DueTime:   dueTime,
+			Bucket:    ev.Bucket,
+			ObjectKey: logicalKey,
+			VersionID: versionID,
+			Identity:  identity,
+		})
+	}
+	return matches
+}
+
+// logicalKeyFromVersionPath extracts <logical> from <logical>.versions/<file>.
+// Returns false for a bucket-root marker (path = ".versions/<file>"), since
+// AWS S3 has no concept of an object at the bucket key itself.
+func logicalKeyFromVersionPath(versionPath string) (string, bool) {
+	lastSlash := strings.LastIndex(versionPath, "/")
+	if lastSlash <= 0 {
+		return "", false
+	}
+	parent := versionPath[:lastSlash]
+	if !strings.HasSuffix(parent, s3_constants.VersionsFolder) {
+		return "", false
+	}
+	logical := strings.TrimSuffix(parent, s3_constants.VersionsFolder)
+	if logical == "" {
+		return "", false
+	}
+	return logical, true
+}
+
+// buildObjectInfo derives an ObjectInfo from a meta-log event. Returns
+// nil for shapes the router can't classify safely: missing attributes,
+// non-MPU directories, version-folder files (those route through
+// routeSoleSurvivorMarker upstream when EXP_DM applies). On a versioned
+// bucket the latest pointer lives in the .versions/ directory's
+// Extended map; without it we leave NumVersions=0 so the bootstrap walk
+// drives noncurrent retention.
 func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo {
 	entry := ev.NewEntry
 	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,
 		}
 	}
-	// Directory entries that aren't MPU inits aren't lifecycle subjects:
-	// the .versions/ folder itself, prefix dirs, etc. — emit nothing.
 	if entry.IsDirectory {
 		return nil
 	}
@@ -159,22 +269,9 @@ func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo {
 		NumVersions: 1,
 	}
 	if versioned {
-		// On a versioned bucket the actual file path doesn't tell us
-		// whether the entry is the current latest or a noncurrent
-		// version — the latest pointer lives in the .versions/
-		// directory's Extended map and isn't part of this event. We
-		// also can't compute NumVersions / NoncurrentIndex here. Skip
-		// any version-folder file event for now; bootstrap walking
-		// drives noncurrent retention and current-version expiration
-		// for versioned buckets until pointer-transition routing
-		// lands. The bare-key path (null-version, pre-versioning
-		// objects) keeps the regular routing.
 		if isVersionFolderPath(ev.Key) {
 			return nil
 		}
-		// NumVersions=0 keeps ExpiredObjectDeleteMarker (sole-survivor
-		// gate) suppressed for the bare-key delete-marker case until
-		// sibling listing lands.
 		info.NumVersions = 0
 	}
 	if tags := extractTags(entry.Extended); len(tags) > 0 {
@@ -226,7 +323,10 @@ func mpuInitInfo(ev *reader.Event, entry *filer_pb.Entry) (destKey string, ok bo
 // 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 {
-	entry := ev.NewEntry
+	return buildIdentityFromEntry(ev.NewEntry)
+}
+
+func buildIdentityFromEntry(entry *filer_pb.Entry) *EntryIdentity {
 	if entry == nil {
 		return nil
 	}
@@ -266,10 +366,31 @@ func extractTags(ext map[string][]byte) map[string]string {
 	return out
 }
 
+// hasActiveEventDrivenAction gates I/O (e.g. sibling listing) on whether
+// a match could actually fire. Mirrors the per-key filter in Route so
+// disabled or scan-only actions don't pay the RPC.
+func hasActiveEventDrivenAction(snap *engine.Snapshot, keys []s3lifecycle.ActionKey, kind s3lifecycle.ActionKind) bool {
+	for _, k := range keys {
+		if k.ActionKind != kind {
+			continue
+		}
+		a := snap.Action(k)
+		if a == nil {
+			continue
+		}
+		if a.IsActive() && a.Mode == engine.ModeEventDriven {
+			return true
+		}
+	}
+	return false
+}
+
+// isDeleteMarkerEntry mirrors every read site for ExtDeleteMarkerKey:
+// production writes []byte("true").
 func isDeleteMarkerEntry(entry *filer_pb.Entry) bool {
 	if entry == nil || len(entry.Extended) == 0 {
 		return false
 	}
 	v, ok := entry.Extended[s3_constants.ExtDeleteMarkerKey]
-	return ok && len(v) == 1 && v[0] == 1
+	return ok && string(v) == "true"
 }

weed/s3api/s3lifecycle/router/router_test.go

@@ -1,6 +1,8 @@
 package router
 
 import (
+	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -45,7 +47,7 @@ func eventCreate(bucket, key string, modTimeS, size int64, ts int64) *reader.Eve
 }
 
 func TestRouteNoSnapshotNoMatches(t *testing.T) {
-	if got := Route(nil, eventCreate("bk", "k", 0, 1, 1), time.Now()); got != nil {
+	if got := Route(context.Background(), nil, eventCreate("bk", "k", 0, 1, 1), time.Now(), nil); got != nil {
 		t.Fatalf("nil snap should yield nil, got %v", got)
 	}
 }
@@ -54,7 +56,7 @@ func TestRouteMissingBucketNoMatches(t *testing.T) {
 	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
 	snap := compileWith(rule, activatedPrior(rule))
 	ev := eventCreate("other-bucket", "k", 0, 1, 1)
-	if got := Route(snap, ev, time.Now()); got != nil {
+	if got := Route(context.Background(), snap, ev, time.Now(), nil); got != nil {
 		t.Fatalf("foreign bucket should yield nil, got %v", got)
 	}
 }
@@ -68,7 +70,7 @@ func TestRouteInactiveSkipped(t *testing.T) {
 	old := now.Add(-48 * time.Hour) // past 1-day expiration
 	ev := eventCreate("bk", "k", old.Unix(), 1, old.UnixNano())
 
-	if got := Route(snap, ev, now); got != nil {
+	if got := Route(context.Background(), snap, ev, now, nil); got != nil {
 		t.Fatalf("inactive action should not match, got %v", got)
 	}
 }
@@ -81,7 +83,7 @@ func TestRouteExpirationDaysFires(t *testing.T) {
 	old := now.Add(-48 * time.Hour) // past 1-day expiration
 	ev := eventCreate("bk", "k", old.Unix(), 1, old.UnixNano())
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match, got %+v", matches)
 	}
@@ -107,7 +109,7 @@ func TestRouteFreshObjectSchedulesInFuture(t *testing.T) {
 	now := time.Now()
 	ev := eventCreate("bk", "k", now.Unix(), 1, now.UnixNano())
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match (scheduled), got %v", matches)
 	}
@@ -127,13 +129,13 @@ func TestRouteRespectsPrefixFilter(t *testing.T) {
 
 	// Out of prefix: no match.
 	ev := eventCreate("bk", "data/file", old.Unix(), 1, old.UnixNano())
-	if got := Route(snap, ev, now); got != nil {
+	if got := Route(context.Background(), snap, ev, now, nil); got != nil {
 		t.Fatalf("out-of-prefix should not match, got %v", got)
 	}
 
 	// In prefix: matches.
 	ev = eventCreate("bk", "logs/file", old.Unix(), 1, old.UnixNano())
-	if got := Route(snap, ev, now); len(got) != 1 {
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 1 {
 		t.Fatalf("in-prefix should match, got %v", got)
 	}
 }
@@ -153,7 +155,7 @@ func TestRouteSkipsHardDelete(t *testing.T) {
 			Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 1},
 		},
 	}
-	if got := Route(snap, ev, now); got != nil {
+	if got := Route(context.Background(), snap, ev, now, nil); got != nil {
 		t.Fatalf("hard delete should not route, got %v", got)
 	}
 }
@@ -169,7 +171,7 @@ func TestRouteSkipsMissingAttributes(t *testing.T) {
 		Key:      "k",
 		NewEntry: &filer_pb.Entry{Name: "k"}, // no Attributes
 	}
-	if got := Route(snap, ev, time.Now()); got != nil {
+	if got := Route(context.Background(), snap, ev, time.Now(), nil); got != nil {
 		t.Fatalf("missing-Attributes event should not route, got %v", got)
 	}
 }
@@ -183,7 +185,7 @@ func TestRouteIdentityCapturedForNewEntry(t *testing.T) {
 	ev := eventCreate("bk", "k", old.Unix(), 42, old.UnixNano())
 	ev.NewEntry.Chunks = []*filer_pb.FileChunk{{FileId: "1,abc"}}
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match, got %v", matches)
 	}
@@ -214,7 +216,7 @@ func TestRouteIdentityHashesExtended(t *testing.T) {
 		"Content-Type":    []byte("text/plain"),
 	}
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match, got %v", matches)
 	}
@@ -259,7 +261,7 @@ func TestRouteMPUInitFiresAbortAfterDelay(t *testing.T) {
 	init := now.AddDate(0, 0, -8)
 	ev := mpuInitEvent("bk", "u1", "logs/foo.txt", init.Unix(), init.UnixNano())
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match, got %v", matches)
 	}
@@ -285,7 +287,7 @@ func TestRouteMPUInitFilteredOutByPrefix(t *testing.T) {
 	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 {
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
 		t.Fatalf("expected 0 matches for foreign prefix, got %v", got)
 	}
 }
@@ -313,7 +315,7 @@ func TestRouteMPUInitMissingDestKeySkipped(t *testing.T) {
 		},
 	}
 
-	if got := Route(snap, ev, now); len(got) != 0 {
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
 		t.Fatalf("expected 0 matches for missing destKey, got %v", got)
 	}
 }
@@ -341,7 +343,7 @@ func TestRouteMPUPartEventSkipped(t *testing.T) {
 		},
 	}
 
-	if got := Route(snap, ev, now); len(got) != 0 {
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
 		t.Fatalf("expected 0 matches for part event, got %v", got)
 	}
 }
@@ -362,7 +364,7 @@ func TestRouteMPUInitDoesNotFireNoncurrent(t *testing.T) {
 	init := now.AddDate(0, 0, -30)
 	ev := mpuInitEvent("bk", "u1", "logs/foo.txt", init.Unix(), init.UnixNano())
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected exactly 1 match (ABORT_MPU only), got %v", matches)
 	}
@@ -390,7 +392,7 @@ func TestRouteRegularObjectUnderDualRuleSkipsAbortMPU(t *testing.T) {
 	old := now.AddDate(0, 0, -2) // past the 1d expiration
 	ev := eventCreate("bk", "obj", old.Unix(), 1, old.UnixNano())
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected exactly 1 match (EXPIRATION_DAYS only), got %v", matches)
 	}
@@ -427,7 +429,7 @@ func TestRouteVersionedNoncurrentEventDoesNotFireFromRouter(t *testing.T) {
 		s3_constants.ExtVersionIdKey: []byte("v1"),
 	}
 
-	if got := Route(snap, ev, now); len(got) != 0 {
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
 		t.Fatalf("router must not emit noncurrent matches yet, got %v", got)
 	}
 }
@@ -443,7 +445,7 @@ func TestRouteVersionedCurrentEventStaysLatest(t *testing.T) {
 	old := now.AddDate(0, 0, -2)
 	ev := eventCreate("bk", "logs/foo", old.Unix(), 1, old.UnixNano())
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match (EXPIRATION_DAYS), got %v", matches)
 	}
@@ -467,7 +469,7 @@ func TestRouteNonVersionedBucketIgnoresVersionsSuffix(t *testing.T) {
 	old := now.AddDate(0, 0, -2)
 	ev := eventCreate("bk", "logs/foo.versions/v1", old.Unix(), 1, old.UnixNano())
 
-	matches := Route(snap, ev, now)
+	matches := Route(context.Background(), snap, ev, now, nil)
 	if len(matches) != 1 {
 		t.Fatalf("expected 1 match, got %v", matches)
 	}
@@ -479,11 +481,49 @@ func TestRouteNonVersionedBucketIgnoresVersionsSuffix(t *testing.T) {
 	}
 }
 
-func TestRouteVersionedExpiredDeleteMarkerSuppressedWithoutSiblings(t *testing.T) {
-	// ExpiredObjectDeleteMarker requires NumVersions==1 — the marker is
-	// the sole-survivor. Without sibling listing the router can't
-	// confirm that, so the rule must NOT fire just because the latest
-	// is a delete marker. A future PR adds sibling listing.
+// markerEventBytes returns the production shape: a file event under
+// <key>.versions/v_<version-id>, with ExtDeleteMarkerKey="true" and
+// ExtVersionIdKey populated. Mirrors createDeleteMarker in
+// s3api_object_versioning.go.
+func markerEvent(bucket, logicalKey, versionID string, mtimeUnix, mtimeNs int64) *reader.Event {
+	versionPath := logicalKey + s3_constants.VersionsFolder + "/v_" + versionID
+	ev := eventCreate(bucket, versionPath, mtimeUnix, 0, mtimeNs)
+	ev.NewEntry.Extended = map[string][]byte{
+		s3_constants.ExtDeleteMarkerKey: []byte("true"),
+		s3_constants.ExtVersionIdKey:    []byte(versionID),
+	}
+	return ev
+}
+
+// recordingLister captures Survivors calls. Configure with the exact
+// state to return; calls list is appended on each invocation so tests
+// can assert whether the lister was consulted at all.
+type recordingLister struct {
+	calls    []string
+	survivors Survivors
+	err      error
+}
+
+func (r *recordingLister) Survivors(_ context.Context, bucket, key string) (Survivors, error) {
+	r.calls = append(r.calls, bucket+"/"+key)
+	return r.survivors, r.err
+}
+
+func markerLoneEntry(versionID string, mtimeUnix, mtimeNs int64) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name: "v_" + versionID,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:   mtimeUnix,
+			MtimeNs: int32(mtimeNs - mtimeUnix*int64(1e9)),
+		},
+		Extended: map[string][]byte{
+			s3_constants.ExtDeleteMarkerKey: []byte("true"),
+			s3_constants.ExtVersionIdKey:    []byte(versionID),
+		},
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerNilListerSuppresses(t *testing.T) {
 	rule := &s3lifecycle.Rule{
 		ID:                        "r",
 		Status:                    s3lifecycle.StatusEnabled,
@@ -493,13 +533,239 @@ func TestRouteVersionedExpiredDeleteMarkerSuppressedWithoutSiblings(t *testing.T
 
 	now := time.Now()
 	old := now.AddDate(0, 0, -1)
-	ev := eventCreate("bk", "logs/gone", old.Unix(), 0, old.UnixNano())
+	ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano())
+
+	if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
+		t.Fatalf("nil lister must suppress, got %v", got)
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerSoleSurvivorFires(t *testing.T) {
+	// Exactly one entry under .versions/<key>/ — the marker — and no
+	// bare null version: EXP_DM fires with the LOGICAL key + version_id.
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano())
+
+	lister := &recordingLister{survivors: Survivors{
+		Count:     1,
+		LoneEntry: markerLoneEntry("2026-05-09-abc", old.Unix(), old.UnixNano()),
+	}}
+	matches := Route(context.Background(), snap, ev, now, lister)
+	if len(matches) != 1 {
+		t.Fatalf("expected 1 match (ExpiredDeleteMarker), got %v", matches)
+	}
+	m := matches[0]
+	if m.Result.Action != s3lifecycle.ActionExpireDeleteMarker {
+		t.Fatalf("Action=%v, want ExpireDeleteMarker", m.Result.Action)
+	}
+	if m.ObjectKey != "logs/gone" {
+		t.Fatalf("ObjectKey=%q, want logical key logs/gone", m.ObjectKey)
+	}
+	if m.VersionID != "2026-05-09-abc" {
+		t.Fatalf("VersionID=%q, want 2026-05-09-abc", m.VersionID)
+	}
+	if len(lister.calls) != 1 || lister.calls[0] != "bk/logs/gone" {
+		t.Fatalf("lister calls=%v, want [bk/logs/gone]", lister.calls)
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerSiblingsRemainSuppressed(t *testing.T) {
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano())
+
+	lister := &recordingLister{survivors: Survivors{Count: 2}}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("siblings present, must not fire, got %v", got)
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerNullVersionSuppresses(t *testing.T) {
+	// A pre-versioning bare-key object (HasNullVersion=true) still survives
+	// outside .versions/. Firing EXP_DM would re-expose it; suppress.
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano())
+
+	lister := &recordingLister{survivors: Survivors{
+		Count:          1,
+		LoneEntry:      markerLoneEntry("2026-05-09-abc", old.Unix(), old.UnixNano()),
+		HasNullVersion: true,
+	}}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("null-version present, must not fire, got %v", got)
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerHardDeleteLeavesLoneMarkerFires(t *testing.T) {
+	// Sequence: object had v1 + DM, hard-delete of v1 leaves DM as the
+	// sole survivor. The hard-delete event has NewEntry=nil; the router
+	// must still consult the lister, see the lone DM, and emit a match
+	// using the LoneEntry's version_id and identity.
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	versionPath := "logs/gone" + s3_constants.VersionsFolder + "/v_2026-05-08-old"
+	ev := &reader.Event{
+		TsNs:     now.UnixNano(),
+		Bucket:   "bk",
+		Key:      versionPath,
+		OldEntry: &filer_pb.Entry{Name: "v_2026-05-08-old"},
+	}
+	lister := &recordingLister{survivors: Survivors{
+		Count:     1,
+		LoneEntry: markerLoneEntry("2026-05-09-abc", old.Unix(), old.UnixNano()),
+	}}
+	matches := Route(context.Background(), snap, ev, now, lister)
+	if len(matches) != 1 {
+		t.Fatalf("expected 1 match after hard-delete, got %v", matches)
+	}
+	if matches[0].VersionID != "2026-05-09-abc" {
+		t.Fatalf("VersionID=%q, want lone-entry's version 2026-05-09-abc", matches[0].VersionID)
+	}
+	if matches[0].ObjectKey != "logs/gone" {
+		t.Fatalf("ObjectKey=%q, want logs/gone", matches[0].ObjectKey)
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerHardDeleteLoneNonMarkerNoFire(t *testing.T) {
+	// After a hard-delete the surviving entry is a regular version, not
+	// a marker. Nothing to expire.
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	versionPath := "logs/gone" + s3_constants.VersionsFolder + "/v_2026-05-08-old"
+	ev := &reader.Event{
+		TsNs:     now.UnixNano(),
+		Bucket:   "bk",
+		Key:      versionPath,
+		OldEntry: &filer_pb.Entry{Name: "v_2026-05-08-old"},
+	}
+	regular := &filer_pb.Entry{
+		Name:       "v_v1",
+		Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix()},
+		Extended:   map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")},
+	}
+	lister := &recordingLister{survivors: Survivors{Count: 1, LoneEntry: regular}}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("non-marker survivor must not fire, got %v", got)
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerListerErrorSuppressed(t *testing.T) {
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano())
+
+	lister := &recordingLister{err: errors.New("filer down")}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("lister error must suppress, got %v", got)
+	}
+}
+
+func TestRouteVersionedExpiredDeleteMarkerInactiveActionSkipsListing(t *testing.T) {
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	// PriorStates omitted => BootstrapComplete=false => action stays inactive.
+	snap := compileWithVersioned(rule, nil)
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano())
+
+	lister := &recordingLister{survivors: Survivors{Count: 1}}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("inactive action must not produce a match, got %v", got)
+	}
+	if len(lister.calls) != 0 {
+		t.Fatalf("inactive action must not consult lister, calls=%v", lister.calls)
+	}
+}
+
+func TestRouteVersionedRegularVersionCreateSkipsListing(t *testing.T) {
+	// A non-marker version create under .versions/<key>/ can never be the
+	// sole survivor (Count >= 2 by definition), so the lister must NOT
+	// be consulted on every regular versioned PUT.
+	rule := &s3lifecycle.Rule{
+		ID:                        "r",
+		Status:                    s3lifecycle.StatusEnabled,
+		ExpiredObjectDeleteMarker: true,
+	}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	versionPath := "logs/keep" + s3_constants.VersionsFolder + "/v_v1"
+	ev := eventCreate("bk", versionPath, old.Unix(), 100, old.UnixNano())
 	ev.NewEntry.Extended = map[string][]byte{
-		s3_constants.ExtDeleteMarkerKey: {1},
+		s3_constants.ExtVersionIdKey: []byte("v1"),
 	}
 
-	if got := Route(snap, ev, now); len(got) != 0 {
-		t.Fatalf("ExpiredDeleteMarker without sibling count must not fire, got %v", got)
+	lister := &recordingLister{survivors: Survivors{Count: 1}}
+	if got := Route(context.Background(), snap, ev, now, lister); len(got) != 0 {
+		t.Fatalf("regular version create must not fire EXP_DM, got %v", got)
+	}
+	if len(lister.calls) != 0 {
+		t.Fatalf("lister consulted for regular version create: calls=%v", lister.calls)
+	}
+}
+
+func TestRouteVersionedDeleteMarkerNoExpDMRuleSkipsListing(t *testing.T) {
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWithVersioned(rule, activatedPrior(rule))
+
+	now := time.Now()
+	old := now.AddDate(0, 0, -1)
+	ev := markerEvent("bk", "logs/gone", "2026-05-09-abc", old.Unix(), old.UnixNano())
+
+	lister := &recordingLister{survivors: Survivors{Count: 1}}
+	Route(context.Background(), snap, ev, now, lister)
+	if len(lister.calls) != 0 {
+		t.Fatalf("lister consulted without EXP_DM rule: calls=%v", lister.calls)
 	}
 }
 
@@ -555,7 +821,7 @@ func TestRouteVersionedAllVersionFolderPathsSkipped(t *testing.T) {
 			if tc.isDir {
 				ev.NewEntry.IsDirectory = true
 			}
-			if got := Route(snap, ev, now); len(got) != 0 {
+			if got := Route(context.Background(), snap, ev, now, nil); len(got) != 0 {
 				t.Fatalf("version-folder event should be skipped, got %v", got)
 			}
 		})