fix(s3/lifecycle): address Phase 3 post-merge review (#9354 #9355 #9356) (#9357)

* fix(s3/lifecycle): reader handles bare /buckets parent and pre-normalizes prefix

extractBucketKey accepted /buckets/ but rejected /buckets (no trailing
slash); some delete events emit the bare form, so bucket-root events
were silently dropped. Pre-normalize BucketsPath once on Run instead
of recomputing per event.

* perf(s3/lifecycle): pool sha256 hashers in ShardID

ShardID runs on every meta-log event before the shard filter; a fresh
sha256.New per call produces measurable allocator pressure under load.
sync.Pool reuses hashers across calls.

* fix(s3/lifecycle): router skips hard deletes and missing-attribute events

A hard delete carries no schedule-relevant state — Expiration would hit
NOOP_RESOLVED at dispatch and ExpiredObjectDeleteMarker fires from a
Create on the latest version. Skip rather than burn a schedule slot.

Missing Attributes leaves ModTime at year 0001, which makes
ExpirationDays fire immediately at dispatch. Skip the event instead.

Drop the unused 'versioned' parameter from buildObjectInfo; the
dispatcher's identity-CAS handles version drift in Phase 5.

* fix(s3/lifecycle): EntryIdentity.MtimeNs holds true nanoseconds

Both computeEntryIdentity (server) and buildIdentity (router) wrote
entry.Attributes.Mtime (seconds) into a field named MtimeNs. The CAS
worked because both sides agreed, but the encoding contradicted the
field name and would break if either side later started using true
nanoseconds. Combine Mtime*1e9 + the FuseAttributes.MtimeNs nanosecond
component on both sides; the test was updated to match.

* fix(s3/lifecycle): dispatcher distinguishes ctx cancel from transport errors

A canceled or deadline-exceeded RPC is shutdown, not a transport
failure: re-queue the Match at its original DueTime with no retry-budget
burn so a quick restart can't escalate it to BLOCKED.

* fix(s3/lifecycle): reader fallback prefix normalization mirrors Run

The fallback path that builds prefix from r.BucketsPath when
bucketsPathSlash is empty (test-only entry into extractBucketKey) was
appending an unconditional '/', producing '//' if BucketsPath already
ended with one. Use the same normalization Run does.

* fix(s3/lifecycle): ObjectInfo.ModTime carries the nanosecond component

ModTime dropped FuseAttributes.MtimeNs, leaving ExpirationDays one
nanosecond off relative to EntryIdentity.MtimeNs. Pass both to
time.Unix so the precision matches the CAS witness.

weed/s3api/s3api_internal_lifecycle.go

@@ -135,7 +135,10 @@ func computeEntryIdentity(entry *filer_pb.Entry) *s3_lifecycle_pb.EntryIdentity
 	}
 	id := &s3_lifecycle_pb.EntryIdentity{}
 	if entry.Attributes != nil {
-		id.MtimeNs = entry.Attributes.Mtime
+		// FuseAttributes splits the timestamp across Mtime (seconds) and
+		// MtimeNs (nanosecond component); EntryIdentity.MtimeNs is the
+		// combined nanoseconds-since-epoch value.
+		id.MtimeNs = entry.Attributes.Mtime*int64(1e9) + int64(entry.Attributes.MtimeNs)
 		id.Size = int64(entry.Attributes.FileSize)
 	}
 	if len(entry.GetChunks()) > 0 {

weed/s3api/s3api_internal_lifecycle_test.go

@@ -10,15 +10,16 @@ import (
 
 func TestComputeEntryIdentity_BasicFields(t *testing.T) {
 	entry := &filer_pb.Entry{
-		Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, FileSize: 4096},
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, MtimeNs: 123, FileSize: 4096},
 		Chunks: []*filer_pb.FileChunk{
 			{FileId: "1,abc"},
 			{FileId: "1,def"},
 		},
 	}
 	id := computeEntryIdentity(entry)
-	if id.MtimeNs != 1700000000 {
+	want := int64(1700000000)*int64(1e9) + int64(123)
-		t.Fatalf("MtimeNs want 1700000000, got %d", id.MtimeNs)
+	if id.MtimeNs != want {
+		t.Fatalf("MtimeNs want %d, got %d", want, id.MtimeNs)
 	}
 	if id.Size != 4096 {
 		t.Fatalf("Size want 4096, got %d", id.Size)

weed/s3api/s3lifecycle/dispatcher/dispatcher.go

@@ -2,6 +2,7 @@ package dispatcher
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -121,6 +122,14 @@ func (d *Dispatcher) dispatchOne(ctx context.Context, m router.Match, now time.T
 	}
 	resp, err := d.Client.LifecycleDelete(ctx, req)
 	if err != nil {
+		// Context cancellation is shutdown, not a transport failure: put
+		// the Match back on the schedule untouched so the next worker
+		// run picks it up at its original DueTime, with no retry-budget
+		// burn.
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			d.Schedule.Add(m)
+			return
+		}
 		// Transport error: classify as RETRY_LATER. The remote handler
 		// already classifies its own filer-side errors; the only path
 		// that hits this branch is the RPC itself failing.

weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go

@@ -173,6 +173,33 @@ func TestDispatchBlockedFreezesCursor(t *testing.T) {
 	}
 }
 
+func TestDispatchContextCancelDoesNotBurnBudget(t *testing.T) {
+	// Worker shutdown causes the in-flight RPC to return context.Canceled.
+	// The Match should go back on the schedule untouched; no retry-budget
+	// burn means a quick restart can't escalate it to BLOCKED.
+	client := &fakeClient{
+		respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
+			return nil, context.Canceled
+		},
+	}
+	d, sched := newDispatcher(client)
+	d.RetryBudget = 1
+	t0 := time.Now()
+	m := mkMatch(t0, t0, "obj")
+	sched.Add(m)
+
+	d.Tick(context.Background(), t0)
+	if sched.Len() != 1 {
+		t.Fatalf("expected re-queue on ctx cancel, sched.Len=%d", sched.Len())
+	}
+	if d.Cursor.IsFrozen(m.Key) {
+		t.Fatal("ctx cancel must not freeze cursor")
+	}
+	if got := d.retries[keyOf(m)]; got != 0 {
+		t.Fatalf("ctx cancel must not burn retry budget, retries=%d", got)
+	}
+}
+
 func TestDispatchTransportErrorRetries(t *testing.T) {
 	// gRPC error: classified as RETRY_LATER. After the budget the cursor freezes.
 	client := &fakeClient{

weed/s3api/s3lifecycle/reader/reader.go

@@ -46,6 +46,11 @@ type Reader struct {
 	// Zero = unbounded; the run continues until ctx cancellation or stream
 	// error. Used by the worker scheduler to bound a single READ task.
 	EventBudget int
+
+	// bucketsPathSlash is BucketsPath with a guaranteed trailing slash,
+	// computed once on Run and reused per event to avoid recomputing the
+	// normalized prefix in extractBucketKey.
+	bucketsPathSlash string
 }
 
 // Run subscribes via SubscribeMetadata starting at Cursor.MinTsNs(), filters
@@ -64,6 +69,10 @@ func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, cl
 	if r.BucketsPath == "" {
 		return errors.New("reader: empty BucketsPath")
 	}
+	r.bucketsPathSlash = r.BucketsPath
+	if !strings.HasSuffix(r.bucketsPathSlash, "/") {
+		r.bucketsPathSlash += "/"
+	}
 
 	stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{
 		ClientName:             clientName,
@@ -157,35 +166,42 @@ func (r *Reader) extractBucketKey(resp *filer_pb.SubscribeMetadataResponse) (str
 		return "", "", false
 	}
 
-	// dir starts with BucketsPath when it's an in-bucket event. The bucket
+	// Pre-normalized prefix (BucketsPath with trailing slash) is computed
-	// is the first segment after BucketsPath; the key is the rest plus name.
+	// once in Run; bucket-root events arrive as either "/buckets" or
-	prefix := r.BucketsPath
+	// "/buckets/", so accept both. The fallback path mirrors Run's
-	if !strings.HasSuffix(prefix, "/") {
+	// normalization for tests that call extractBucketKey directly.
-		prefix += "/"
+	prefix := r.bucketsPathSlash
+	if prefix == "" {
+		prefix = r.BucketsPath
+		if !strings.HasSuffix(prefix, "/") {
+			prefix += "/"
+		}
 	}
-	if !strings.HasPrefix(dir, prefix) {
+	bare := strings.TrimSuffix(prefix, "/")
+	var rest string
+	switch {
+	case dir == bare || dir == prefix:
+		// Bucket create/delete at /buckets root: bucket name is the entry name.
+		if name == "" {
+			return "", "", false
+		}
+		return name, "", true
+	case strings.HasPrefix(dir, prefix):
+		rest = dir[len(prefix):]
+	default:
 		return "", "", false
 	}
-	rest := dir[len(prefix):]
 	// rest = "<bucket>" or "<bucket>/<sub>/<sub>..."
 	slash := strings.IndexByte(rest, '/')
 	var bucket, parentInBucket string
 	if slash < 0 {
 		bucket = rest
-		parentInBucket = ""
 	} else {
 		bucket = rest[:slash]
 		parentInBucket = rest[slash+1:]
 	}
 	if bucket == "" {
-		// rest was empty: event was at /buckets/ itself (bucket create/delete).
+		return "", "", false
-		// Bucket name is the entry name.
-		bucket = name
-		key := ""
-		if bucket == "" {
-			return "", "", false
-		}
-		return bucket, key, true
 	}
 	if parentInBucket != "" {
 		return bucket, parentInBucket + "/" + name, true

weed/s3api/s3lifecycle/reader/reader_test.go

@@ -95,6 +95,23 @@ func TestExtractBucketKeyBucketCreateAtRoot(t *testing.T) {
 	}
 }
 
+func TestExtractBucketKeyBucketCreateBareRoot(t *testing.T) {
+	// Some delete and create events emit the parent without a trailing
+	// slash. /buckets (no trailing slash) must also resolve to a bucket-
+	// root event with the entry name as the bucket.
+	r := &Reader{BucketsPath: "/buckets"}
+	resp := &filer_pb.SubscribeMetadataResponse{
+		EventNotification: &filer_pb.EventNotification{
+			NewParentPath: "/buckets",
+			NewEntry:      &filer_pb.Entry{Name: "newbucket"},
+		},
+	}
+	b, k, ok := r.extractBucketKey(resp)
+	if !ok || b != "newbucket" || k != "" {
+		t.Fatalf("bare /buckets parent: got (%q,%q,%v), want (newbucket,,true)", b, k, ok)
+	}
+}
+
 func TestDispatchOneFiltersByShard(t *testing.T) {
 	// Pick a bucket+key whose shard is known, set Reader to a different
 	// shard, expect skip; same shard, expect emit.

weed/s3api/s3lifecycle/router/router.go

@@ -44,12 +44,18 @@ func Route(snap *engine.Snapshot, ev *reader.Event, now time.Time) []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
 	}
-	versioned := snap.BucketVersioned(ev.Bucket)
+	info := buildObjectInfo(ev)
-	info := buildObjectInfo(ev, versioned)
 	if info == nil {
 		return nil
 	}
@@ -90,33 +96,29 @@ 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 an event on a versioned bucket is treated as
+// in Phase 5; for now any event is treated as IsLatest=true with the
-// IsLatest=true with the same caveat that the LifecycleDelete RPC's
+// LifecycleDelete RPC's identity-CAS catching stale schedules.
-// identity-CAS catches stale schedules.
+//
-func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo {
+// 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 {
 	entry := ev.NewEntry
-	if entry == nil {
+	if entry == nil || entry.Attributes == nil {
-		entry = ev.OldEntry
-	}
-	if entry == nil {
 		return nil
 	}
 	info := &s3lifecycle.ObjectInfo{
 		Key:         ev.Key,
+		ModTime:     time.Unix(entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)),
+		Size:        int64(entry.Attributes.FileSize),
 		IsLatest:    true,
 		NumVersions: 1,
 	}
-	if entry.Attributes != nil {
-		info.ModTime = time.Unix(entry.Attributes.Mtime, 0)
-		info.Size = int64(entry.Attributes.FileSize)
-	}
 	if tags := extractTags(entry.Extended); len(tags) > 0 {
 		info.Tags = tags
 	}
 	if isDeleteMarkerEntry(entry) {
 		info.IsDeleteMarker = true
 	}
-	_ = versioned
 	return info
 }
 
@@ -129,7 +131,10 @@ func buildIdentity(ev *reader.Event) *EntryIdentity {
 	}
 	id := &EntryIdentity{}
 	if entry.Attributes != nil {
-		id.MtimeNs = entry.Attributes.Mtime
+		// Mirror the server-side computeEntryIdentity encoding: Mtime
+		// (seconds) and MtimeNs (nanosecond component) combine into
+		// EntryIdentity.MtimeNs as nanoseconds-since-epoch.
+		id.MtimeNs = entry.Attributes.Mtime*int64(1e9) + int64(entry.Attributes.MtimeNs)
 		id.Size = int64(entry.Attributes.FileSize)
 	}
 	if len(entry.GetChunks()) > 0 {

weed/s3api/s3lifecycle/router/router_test.go

@@ -137,6 +137,42 @@ func TestRouteRespectsPrefixFilter(t *testing.T) {
 	}
 }
 
+func TestRouteSkipsHardDelete(t *testing.T) {
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWith(rule, activatedPrior(rule))
+	now := time.Now()
+	old := now.Add(-48 * time.Hour)
+	// Hard delete: NewEntry is nil; OldEntry holds the gone object.
+	ev := &reader.Event{
+		TsNs:   old.UnixNano(),
+		Bucket: "bk",
+		Key:    "gone.txt",
+		OldEntry: &filer_pb.Entry{
+			Name: "gone.txt",
+			Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 1},
+		},
+	}
+	if got := Route(snap, ev, now); got != nil {
+		t.Fatalf("hard delete should not route, got %v", got)
+	}
+}
+
+func TestRouteSkipsMissingAttributes(t *testing.T) {
+	// Without Attributes there's no ModTime, and EvaluateAction would
+	// compute due against year-0001 and fire immediately. Skip the event.
+	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
+	snap := compileWith(rule, activatedPrior(rule))
+	ev := &reader.Event{
+		TsNs:     time.Now().UnixNano(),
+		Bucket:   "bk",
+		Key:      "k",
+		NewEntry: &filer_pb.Entry{Name: "k"}, // no Attributes
+	}
+	if got := Route(snap, ev, time.Now()); got != nil {
+		t.Fatalf("missing-Attributes event should not route, got %v", got)
+	}
+}
+
 func TestRouteIdentityCapturedForNewEntry(t *testing.T) {
 	rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
 	snap := compileWith(rule, activatedPrior(rule))
@@ -154,4 +190,8 @@ func TestRouteIdentityCapturedForNewEntry(t *testing.T) {
 	if id == nil || id.Size != 42 || id.HeadFid != "1,abc" {
 		t.Fatalf("identity capture: %+v", id)
 	}
+	wantNs := old.Unix()*int64(1e9) + 0
+	if id.MtimeNs != wantNs {
+		t.Fatalf("MtimeNs=%d, want %d (Mtime*1e9)", id.MtimeNs, wantNs)
+	}
 }

weed/s3api/s3lifecycle/shard.go

@@ -1,20 +1,33 @@
 package s3lifecycle
 
-import "crypto/sha256"
+import (
+	"crypto/sha256"
+	"hash"
+	"sync"
+)
 
 // ShardCount partitions the (bucket, key) keyspace for the per-shard
 // lifecycle reader. Workers receive one READ task per owned shard;
 // each shard's cursor advances independently.
 const ShardCount = 16
 
+// shardHashPool reuses sha256 hashers — ShardID is in the per-event hot
+// path and a fresh allocation per call shows up under load.
+var shardHashPool = sync.Pool{
+	New: func() interface{} { return sha256.New() },
+}
+
 // ShardID maps (bucket, key) to a shard in [0, ShardCount). Stable across
 // restarts and across processes — identical (bucket, key) always lands on
 // the same shard. Implementation: top 4 bits of sha256(bucket || "/" || key).
 func ShardID(bucket, key string) int {
-	h := sha256.New()
+	h := shardHashPool.Get().(hash.Hash)
+	h.Reset()
 	h.Write([]byte(bucket))
 	h.Write([]byte{'/'})
 	h.Write([]byte(key))
-	sum := h.Sum(nil)
+	var buf [sha256.Size]byte
+	sum := h.Sum(buf[:0])
+	shardHashPool.Put(h)
 	return int(sum[0] >> 4)
 }