diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index 87d94882a..13e48a9cc 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/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 { diff --git a/weed/s3api/s3api_internal_lifecycle_test.go b/weed/s3api/s3api_internal_lifecycle_test.go index 0b5a3b53a..c7975f716 100644 --- a/weed/s3api/s3api_internal_lifecycle_test.go +++ b/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 { - t.Fatalf("MtimeNs want 1700000000, got %d", id.MtimeNs) + want := int64(1700000000)*int64(1e9) + int64(123) + 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) diff --git a/weed/s3api/s3lifecycle/dispatcher/dispatcher.go b/weed/s3api/s3lifecycle/dispatcher/dispatcher.go index bc3589e40..2e7015d86 100644 --- a/weed/s3api/s3lifecycle/dispatcher/dispatcher.go +++ b/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. diff --git a/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go b/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go index 8c812e871..90de082cf 100644 --- a/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go +++ b/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{ diff --git a/weed/s3api/s3lifecycle/reader/reader.go b/weed/s3api/s3lifecycle/reader/reader.go index 5a21c9492..1e0ab1f3e 100644 --- a/weed/s3api/s3lifecycle/reader/reader.go +++ b/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 - // is the first segment after BucketsPath; the key is the rest plus name. - prefix := r.BucketsPath - if !strings.HasSuffix(prefix, "/") { - prefix += "/" + // Pre-normalized prefix (BucketsPath with trailing slash) is computed + // once in Run; bucket-root events arrive as either "/buckets" or + // "/buckets/", so accept both. The fallback path mirrors Run's + // normalization for tests that call extractBucketKey directly. + 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 = "" or "//..." 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). - // Bucket name is the entry name. - bucket = name - key := "" - if bucket == "" { - return "", "", false - } - return bucket, key, true + return "", "", false } if parentInBucket != "" { return bucket, parentInBucket + "/" + name, true diff --git a/weed/s3api/s3lifecycle/reader/reader_test.go b/weed/s3api/s3lifecycle/reader/reader_test.go index 9032aedf7..1dccc7493 100644 --- a/weed/s3api/s3lifecycle/reader/reader_test.go +++ b/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. diff --git a/weed/s3api/s3lifecycle/router/router.go b/weed/s3api/s3lifecycle/router/router.go index 227cb8cbf..9f50fe402 100644 --- a/weed/s3api/s3lifecycle/router/router.go +++ b/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, versioned) + info := buildObjectInfo(ev) 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 -// IsLatest=true with the same caveat that the LifecycleDelete RPC's -// identity-CAS catches stale schedules. -func buildObjectInfo(ev *reader.Event, versioned bool) *s3lifecycle.ObjectInfo { +// in Phase 5; for now any event is treated as IsLatest=true with the +// LifecycleDelete RPC's identity-CAS catching stale schedules. +// +// 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 { - entry = ev.OldEntry - } - if entry == nil { + if entry == nil || entry.Attributes == 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 { diff --git a/weed/s3api/s3lifecycle/router/router_test.go b/weed/s3api/s3lifecycle/router/router_test.go index 11db7cba5..51038ce28 100644 --- a/weed/s3api/s3lifecycle/router/router_test.go +++ b/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) + } } diff --git a/weed/s3api/s3lifecycle/shard.go b/weed/s3api/s3lifecycle/shard.go index 502786459..080be9cdf 100644 --- a/weed/s3api/s3lifecycle/shard.go +++ b/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) }