From 0f6c6b0524291e7fbeb6357ef0e597ce5c6a1e16 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 7 May 2026 15:42:37 -0700 Subject: [PATCH] feat(s3/lifecycle): shard-aware meta-log reader (Phase 3 PR-B) (#9354) feat(s3/lifecycle): shard-aware meta-log reader - ShardCount=16; ShardID(bucket,key)=top-4-bits of sha256(bucket||/||key) - Reader subscribes via SubscribeMetadata starting at Cursor.MinTsNs(), filters events by shard, emits to caller-owned Events channel - Cursor: per-(shard, ActionKey) position with monotonic Advance, Freeze for blocked actions, MinTsNs for subscription resume - Persister interface with InMemoryPersister for tests; filer-backed impl lands with the worker integration --- weed/s3api/s3lifecycle/reader/cursor.go | 123 ++++++++++++ weed/s3api/s3lifecycle/reader/cursor_test.go | 109 ++++++++++ weed/s3api/s3lifecycle/reader/persister.go | 55 +++++ weed/s3api/s3lifecycle/reader/reader.go | 201 +++++++++++++++++++ weed/s3api/s3lifecycle/reader/reader_test.go | 167 +++++++++++++++ weed/s3api/s3lifecycle/shard.go | 20 ++ weed/s3api/s3lifecycle/shard_test.go | 46 +++++ 7 files changed, 721 insertions(+) create mode 100644 weed/s3api/s3lifecycle/reader/cursor.go create mode 100644 weed/s3api/s3lifecycle/reader/cursor_test.go create mode 100644 weed/s3api/s3lifecycle/reader/persister.go create mode 100644 weed/s3api/s3lifecycle/reader/reader.go create mode 100644 weed/s3api/s3lifecycle/reader/reader_test.go create mode 100644 weed/s3api/s3lifecycle/shard.go create mode 100644 weed/s3api/s3lifecycle/shard_test.go diff --git a/weed/s3api/s3lifecycle/reader/cursor.go b/weed/s3api/s3lifecycle/reader/cursor.go new file mode 100644 index 000000000..4b917c435 --- /dev/null +++ b/weed/s3api/s3lifecycle/reader/cursor.go @@ -0,0 +1,123 @@ +package reader + +import ( + "sync" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// Cursor tracks the meta-log position for each ActionKey within one shard. +// Position = tsNs of the last event the shard's reader+downstream considers +// "fully resolved" for this ActionKey. Subscription resumes at the minimum +// position across all keys. +// +// A frozen key is pinned at its current position (BLOCKED outcome): Advance +// is a no-op until Unfreeze. The meta-log retention bounds how long a freeze +// can last before the entry expires from the log; operator action via the +// blocker-resolve flow lifts the freeze. +type Cursor struct { + mu sync.RWMutex + state map[s3lifecycle.ActionKey]int64 + frozen map[s3lifecycle.ActionKey]struct{} +} + +func NewCursor() *Cursor { + return &Cursor{ + state: map[s3lifecycle.ActionKey]int64{}, + frozen: map[s3lifecycle.ActionKey]struct{}{}, + } +} + +// MinTsNs returns the smallest position across all keys; zero if empty. +// Used as the resume point for the meta-log subscription. +func (c *Cursor) MinTsNs() int64 { + c.mu.RLock() + defer c.mu.RUnlock() + if len(c.state) == 0 { + return 0 + } + var min int64 = -1 + for _, v := range c.state { + if min < 0 || v < min { + min = v + } + } + if min < 0 { + return 0 + } + return min +} + +// Get returns the current position for key, or 0 if unset. +func (c *Cursor) Get(key s3lifecycle.ActionKey) int64 { + c.mu.RLock() + defer c.mu.RUnlock() + return c.state[key] +} + +// Advance moves key's position forward to tsNs. No-op if key is frozen, if +// tsNs <= current, or if tsNs <= 0. +func (c *Cursor) Advance(key s3lifecycle.ActionKey, tsNs int64) { + if tsNs <= 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if _, isFrozen := c.frozen[key]; isFrozen { + return + } + if cur := c.state[key]; tsNs > cur { + c.state[key] = tsNs + } +} + +// Freeze pins key at its current position. If key has no recorded position +// yet, it is set to tsNs. +func (c *Cursor) Freeze(key s3lifecycle.ActionKey, tsNs int64) { + c.mu.Lock() + defer c.mu.Unlock() + if _, exists := c.state[key]; !exists { + c.state[key] = tsNs + } + c.frozen[key] = struct{}{} +} + +// Unfreeze releases a freeze. Subsequent Advance calls take effect. +func (c *Cursor) Unfreeze(key s3lifecycle.ActionKey) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.frozen, key) +} + +// IsFrozen reports whether key is currently pinned. +func (c *Cursor) IsFrozen(key s3lifecycle.ActionKey) bool { + c.mu.RLock() + defer c.mu.RUnlock() + _, ok := c.frozen[key] + return ok +} + +// Snapshot copies the cursor map for persistence. Frozen keys are not +// distinguished in the snapshot; callers persist freezes via the separate +// blocker-record store. +func (c *Cursor) Snapshot() map[s3lifecycle.ActionKey]int64 { + c.mu.RLock() + defer c.mu.RUnlock() + out := make(map[s3lifecycle.ActionKey]int64, len(c.state)) + for k, v := range c.state { + out[k] = v + } + return out +} + +// Restore replaces the cursor map. Freezes are not restored here; the +// caller re-applies them from blocker records on startup. +func (c *Cursor) Restore(state map[s3lifecycle.ActionKey]int64) { + c.mu.Lock() + defer c.mu.Unlock() + c.state = make(map[s3lifecycle.ActionKey]int64, len(state)) + for k, v := range state { + c.state[k] = v + } + c.frozen = map[s3lifecycle.ActionKey]struct{}{} +} diff --git a/weed/s3api/s3lifecycle/reader/cursor_test.go b/weed/s3api/s3lifecycle/reader/cursor_test.go new file mode 100644 index 000000000..111c121d2 --- /dev/null +++ b/weed/s3api/s3lifecycle/reader/cursor_test.go @@ -0,0 +1,109 @@ +package reader + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +func key(b string, kind s3lifecycle.ActionKind) s3lifecycle.ActionKey { + return s3lifecycle.ActionKey{Bucket: b, ActionKind: kind} +} + +func TestCursorMinTsNsEmpty(t *testing.T) { + c := NewCursor() + if got := c.MinTsNs(); got != 0 { + t.Fatalf("empty MinTsNs=%d, want 0", got) + } +} + +func TestCursorAdvanceMonotonic(t *testing.T) { + c := NewCursor() + k := key("b", s3lifecycle.ActionKindExpirationDays) + c.Advance(k, 100) + c.Advance(k, 50) // backward, ignored + c.Advance(k, 100) // equal, ignored + c.Advance(k, 200) + if got := c.Get(k); got != 200 { + t.Fatalf("Get=%d, want 200", got) + } +} + +func TestCursorAdvanceIgnoresZero(t *testing.T) { + c := NewCursor() + k := key("b", s3lifecycle.ActionKindExpirationDays) + c.Advance(k, 0) + c.Advance(k, -1) + if got := c.Get(k); got != 0 { + t.Fatalf("Get=%d, want 0", got) + } +} + +func TestCursorMinTsNsAcrossKeys(t *testing.T) { + c := NewCursor() + c.Advance(key("a", s3lifecycle.ActionKindExpirationDays), 100) + c.Advance(key("b", s3lifecycle.ActionKindExpirationDays), 50) + c.Advance(key("c", s3lifecycle.ActionKindAbortMPU), 200) + if got := c.MinTsNs(); got != 50 { + t.Fatalf("MinTsNs=%d, want 50", got) + } +} + +func TestCursorFreeze(t *testing.T) { + c := NewCursor() + k := key("b", s3lifecycle.ActionKindExpirationDays) + c.Advance(k, 100) + c.Freeze(k, 100) + c.Advance(k, 200) // frozen — ignored + if got := c.Get(k); got != 100 { + t.Fatalf("Get after freeze=%d, want 100", got) + } + if !c.IsFrozen(k) { + t.Fatal("IsFrozen=false, want true") + } + c.Unfreeze(k) + c.Advance(k, 200) + if got := c.Get(k); got != 200 { + t.Fatalf("Get after unfreeze=%d, want 200", got) + } +} + +func TestCursorFreezeOnUnsetKeySeedsPosition(t *testing.T) { + // Freezing a key with no recorded position seeds it at the freeze tsNs; + // subsequent MinTsNs reflects the freeze, so the subscription doesn't + // rewind past it. + c := NewCursor() + k := key("b", s3lifecycle.ActionKindExpirationDays) + c.Freeze(k, 500) + if got := c.Get(k); got != 500 { + t.Fatalf("Get after freeze-on-unset=%d, want 500", got) + } + if got := c.MinTsNs(); got != 500 { + t.Fatalf("MinTsNs=%d, want 500", got) + } +} + +func TestCursorSnapshotRestore(t *testing.T) { + c := NewCursor() + a := key("a", s3lifecycle.ActionKindExpirationDays) + b := key("b", s3lifecycle.ActionKindAbortMPU) + c.Advance(a, 100) + c.Advance(b, 200) + + snap := c.Snapshot() + if len(snap) != 2 || snap[a] != 100 || snap[b] != 200 { + t.Fatalf("Snapshot=%v", snap) + } + + c2 := NewCursor() + c2.Restore(snap) + if got := c2.Get(a); got != 100 { + t.Fatalf("restored Get(a)=%d, want 100", got) + } + if got := c2.Get(b); got != 200 { + t.Fatalf("restored Get(b)=%d, want 200", got) + } + if c2.IsFrozen(a) { + t.Fatal("Restore should not carry freezes") + } +} diff --git a/weed/s3api/s3lifecycle/reader/persister.go b/weed/s3api/s3lifecycle/reader/persister.go new file mode 100644 index 000000000..9c0366c91 --- /dev/null +++ b/weed/s3api/s3lifecycle/reader/persister.go @@ -0,0 +1,55 @@ +package reader + +import ( + "context" + "sync" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// Persister loads and stores per-shard cursor state. The contract: +// - Load returns an empty map (not an error) when no prior state exists. +// - Save replaces the prior state atomically; partial writes must not be +// observable on a later Load. +// - Concurrent Save calls for the same shardID are serialized by the +// caller (one Reader per shard at a time). +// +// The filer-backed implementation lives in the worker package alongside the +// rest of the persistence wiring; this package only ships the in-memory +// variant used by tests. +type Persister interface { + Load(ctx context.Context, shardID int) (map[s3lifecycle.ActionKey]int64, error) + Save(ctx context.Context, shardID int, state map[s3lifecycle.ActionKey]int64) error +} + +// InMemoryPersister is a Persister for tests. Save copies the input so +// later mutations by the caller don't leak into stored state. +type InMemoryPersister struct { + mu sync.Mutex + storage map[int]map[s3lifecycle.ActionKey]int64 +} + +func NewInMemoryPersister() *InMemoryPersister { + return &InMemoryPersister{storage: map[int]map[s3lifecycle.ActionKey]int64{}} +} + +func (p *InMemoryPersister) Load(ctx context.Context, shardID int) (map[s3lifecycle.ActionKey]int64, error) { + p.mu.Lock() + defer p.mu.Unlock() + out := map[s3lifecycle.ActionKey]int64{} + for k, v := range p.storage[shardID] { + out[k] = v + } + return out, nil +} + +func (p *InMemoryPersister) Save(ctx context.Context, shardID int, state map[s3lifecycle.ActionKey]int64) error { + p.mu.Lock() + defer p.mu.Unlock() + copied := make(map[s3lifecycle.ActionKey]int64, len(state)) + for k, v := range state { + copied[k] = v + } + p.storage[shardID] = copied + return nil +} diff --git a/weed/s3api/s3lifecycle/reader/reader.go b/weed/s3api/s3lifecycle/reader/reader.go new file mode 100644 index 000000000..5a21c9492 --- /dev/null +++ b/weed/s3api/s3lifecycle/reader/reader.go @@ -0,0 +1,201 @@ +package reader + +import ( + "context" + "errors" + "fmt" + "io" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// Event is one in-shard meta-log event delivered to the router. +type Event struct { + TsNs int64 + Bucket string + Key string + OldEntry *filer_pb.Entry + NewEntry *filer_pb.Entry + NewParent string +} + +// IsDelete reports whether this event removes an entry. +func (e *Event) IsDelete() bool { + return e.NewEntry == nil && e.OldEntry != nil +} + +// IsCreate reports whether this event creates an entry. +func (e *Event) IsCreate() bool { + return e.OldEntry == nil && e.NewEntry != nil +} + +// Reader subscribes to the filer meta-log for one shard. It owns a Cursor +// (per-ActionKey position within the shard) and emits in-shard Events to a +// channel; the downstream router consumes events and ack-advances the cursor +// for matched ActionKeys when their actions complete. +type Reader struct { + ShardID int // [0, s3lifecycle.ShardCount) + BucketsPath string // e.g. "/buckets" + Cursor *Cursor + Events chan<- *Event + + // EventBudget caps how many events Run processes before returning nil. + // Zero = unbounded; the run continues until ctx cancellation or stream + // error. Used by the worker scheduler to bound a single READ task. + EventBudget int +} + +// Run subscribes via SubscribeMetadata starting at Cursor.MinTsNs(), filters +// to this shard, and emits Events. Returns on ctx.Done(), io.EOF, or +// stream error. Caller is responsible for closing Events if it owns it. +func (r *Reader) Run(ctx context.Context, client filer_pb.SeaweedFilerClient, clientName string, clientID int32) error { + if r.ShardID < 0 || r.ShardID >= s3lifecycle.ShardCount { + return fmt.Errorf("reader: shard_id %d out of range", r.ShardID) + } + if r.Cursor == nil { + return errors.New("reader: nil Cursor") + } + if r.Events == nil { + return errors.New("reader: nil Events channel") + } + if r.BucketsPath == "" { + return errors.New("reader: empty BucketsPath") + } + + stream, err := client.SubscribeMetadata(ctx, &filer_pb.SubscribeMetadataRequest{ + ClientName: clientName, + PathPrefix: r.BucketsPath, + SinceNs: r.Cursor.MinTsNs(), + ClientId: clientID, + ClientSupportsBatching: true, + }) + if err != nil { + return fmt.Errorf("subscribe: %w", err) + } + + processed := 0 + for { + resp, recvErr := stream.Recv() + if recvErr == io.EOF { + return nil + } + if recvErr != nil { + return recvErr + } + + // First event in resp is the primary; resp.Events is the batched tail. + if err := r.dispatchOne(ctx, resp, &processed); err != nil { + return err + } + for _, ev := range resp.Events { + if err := r.dispatchOne(ctx, ev, &processed); err != nil { + return err + } + } + if r.EventBudget > 0 && processed >= r.EventBudget { + return nil + } + } +} + +func (r *Reader) dispatchOne(ctx context.Context, resp *filer_pb.SubscribeMetadataResponse, processed *int) error { + if resp == nil || resp.EventNotification == nil { + return nil + } + bucket, key, ok := r.extractBucketKey(resp) + if !ok { + return nil + } + if s3lifecycle.ShardID(bucket, key) != r.ShardID { + return nil + } + + ev := &Event{ + TsNs: resp.TsNs, + Bucket: bucket, + Key: key, + OldEntry: resp.EventNotification.OldEntry, + NewEntry: resp.EventNotification.NewEntry, + NewParent: resp.EventNotification.NewParentPath, + } + select { + case <-ctx.Done(): + return ctx.Err() + case r.Events <- ev: + *processed++ + return nil + } +} + +// extractBucketKey turns a meta-log event's path into (bucket, key) when the +// event lies under BucketsPath. Returns ok=false for events outside that +// subtree (cluster admin entries, system files, etc.) so the reader can skip +// them without engaging the routing index. +// +// The path is reconstructed as DirectoryPath/Name, where DirectoryPath comes +// from the entry context and Name from old_entry/new_entry. We prefer +// new_entry on creates/updates and old_entry on deletes; both carry the same +// Name on renames where new_parent_path differs. +func (r *Reader) extractBucketKey(resp *filer_pb.SubscribeMetadataResponse) (string, string, bool) { + notif := resp.EventNotification + dir := notif.NewParentPath + if dir == "" { + // On deletes, NewParentPath may be empty; the directory is encoded + // in resp.Directory. + dir = resp.Directory + } + var name string + switch { + case notif.NewEntry != nil: + name = notif.NewEntry.Name + case notif.OldEntry != nil: + name = notif.OldEntry.Name + default: + 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 += "/" + } + if !strings.HasPrefix(dir, prefix) { + 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 + } + if parentInBucket != "" { + return bucket, parentInBucket + "/" + name, true + } + return bucket, name, true +} + +// LogStartup is a small helper for callers that want a one-line readable +// description of where the reader is starting. +func (r *Reader) LogStartup() { + glog.V(1).Infof("lifecycle reader: shard=%d sinceNs=%d budget=%d", + r.ShardID, r.Cursor.MinTsNs(), r.EventBudget) +} diff --git a/weed/s3api/s3lifecycle/reader/reader_test.go b/weed/s3api/s3lifecycle/reader/reader_test.go new file mode 100644 index 000000000..9032aedf7 --- /dev/null +++ b/weed/s3api/s3lifecycle/reader/reader_test.go @@ -0,0 +1,167 @@ +package reader + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +func TestExtractBucketKeyCreate(t *testing.T) { + r := &Reader{BucketsPath: "/buckets"} + resp := &filer_pb.SubscribeMetadataResponse{ + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/buckets/mybucket/path/to", + NewEntry: &filer_pb.Entry{Name: "obj.txt"}, + }, + } + b, k, ok := r.extractBucketKey(resp) + if !ok || b != "mybucket" || k != "path/to/obj.txt" { + t.Fatalf("got (%q,%q,%v), want (mybucket,path/to/obj.txt,true)", b, k, ok) + } +} + +func TestExtractBucketKeyTopLevelObject(t *testing.T) { + r := &Reader{BucketsPath: "/buckets"} + resp := &filer_pb.SubscribeMetadataResponse{ + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/buckets/mybucket", + NewEntry: &filer_pb.Entry{Name: "a.txt"}, + }, + } + b, k, ok := r.extractBucketKey(resp) + if !ok || b != "mybucket" || k != "a.txt" { + t.Fatalf("got (%q,%q,%v), want (mybucket,a.txt,true)", b, k, ok) + } +} + +func TestExtractBucketKeyDeleteUsesOldEntry(t *testing.T) { + r := &Reader{BucketsPath: "/buckets"} + resp := &filer_pb.SubscribeMetadataResponse{ + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/buckets/b/dir", + OldEntry: &filer_pb.Entry{Name: "gone.txt"}, + }, + } + b, k, ok := r.extractBucketKey(resp) + if !ok || b != "b" || k != "dir/gone.txt" { + t.Fatalf("got (%q,%q,%v), want (b,dir/gone.txt,true)", b, k, ok) + } +} + +func TestExtractBucketKeyDeleteWithEmptyParent(t *testing.T) { + // Some delete events carry the directory in resp.Directory rather than + // EventNotification.NewParentPath. + r := &Reader{BucketsPath: "/buckets"} + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/b/dir", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "gone.txt"}, + }, + } + b, k, ok := r.extractBucketKey(resp) + if !ok || b != "b" || k != "dir/gone.txt" { + t.Fatalf("got (%q,%q,%v), want (b,dir/gone.txt,true)", b, k, ok) + } +} + +func TestExtractBucketKeyOutsideBucketsSkipped(t *testing.T) { + r := &Reader{BucketsPath: "/buckets"} + resp := &filer_pb.SubscribeMetadataResponse{ + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/etc/something", + NewEntry: &filer_pb.Entry{Name: "config"}, + }, + } + if _, _, ok := r.extractBucketKey(resp); ok { + t.Fatal("expected skip for non-bucket path") + } +} + +func TestExtractBucketKeyBucketCreateAtRoot(t *testing.T) { + // Creating a new bucket emits an event at /buckets/ with the bucket as + // the new entry's Name. + 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("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. + bucket, key := "mybucket", "thing.txt" + target := s3lifecycle.ShardID(bucket, key) + other := (target + 1) % s3lifecycle.ShardCount + + resp := &filer_pb.SubscribeMetadataResponse{ + TsNs: 12345, + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/buckets/" + bucket, + NewEntry: &filer_pb.Entry{Name: key}, + }, + } + + out := make(chan *Event, 1) + r := &Reader{ + ShardID: other, + BucketsPath: "/buckets", + Cursor: NewCursor(), + Events: out, + } + processed := 0 + if err := r.dispatchOne(context.Background(), resp, &processed); err != nil { + t.Fatalf("dispatchOne err: %v", err) + } + if processed != 0 || len(out) != 0 { + t.Fatalf("wrong-shard event should be skipped: processed=%d emitted=%d", processed, len(out)) + } + + r.ShardID = target + if err := r.dispatchOne(context.Background(), resp, &processed); err != nil { + t.Fatalf("dispatchOne err: %v", err) + } + if processed != 1 || len(out) != 1 { + t.Fatalf("matching-shard event should be emitted: processed=%d emitted=%d", processed, len(out)) + } + ev := <-out + if ev.Bucket != bucket || ev.Key != key || ev.TsNs != 12345 { + t.Fatalf("emitted event mismatch: %+v", ev) + } +} + +func TestDispatchOneRespectsCtxCancel(t *testing.T) { + // Buffered channel of size 0 (unbuffered) blocks the send; ctx cancel + // should unblock and return ctx.Err(). + bucket, key := "mybucket", "thing.txt" + target := s3lifecycle.ShardID(bucket, key) + resp := &filer_pb.SubscribeMetadataResponse{ + TsNs: 1, + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/buckets/" + bucket, + NewEntry: &filer_pb.Entry{Name: key}, + }, + } + out := make(chan *Event) + r := &Reader{ + ShardID: target, + BucketsPath: "/buckets", + Cursor: NewCursor(), + Events: out, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + processed := 0 + err := r.dispatchOne(ctx, resp, &processed) + if err != context.Canceled { + t.Fatalf("expected ctx.Canceled, got %v", err) + } +} diff --git a/weed/s3api/s3lifecycle/shard.go b/weed/s3api/s3lifecycle/shard.go new file mode 100644 index 000000000..502786459 --- /dev/null +++ b/weed/s3api/s3lifecycle/shard.go @@ -0,0 +1,20 @@ +package s3lifecycle + +import "crypto/sha256" + +// 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 + +// 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.Write([]byte(bucket)) + h.Write([]byte{'/'}) + h.Write([]byte(key)) + sum := h.Sum(nil) + return int(sum[0] >> 4) +} diff --git a/weed/s3api/s3lifecycle/shard_test.go b/weed/s3api/s3lifecycle/shard_test.go new file mode 100644 index 000000000..59b2eedf1 --- /dev/null +++ b/weed/s3api/s3lifecycle/shard_test.go @@ -0,0 +1,46 @@ +package s3lifecycle + +import "testing" + +func TestShardIDInRange(t *testing.T) { + cases := []struct{ bucket, key string }{ + {"", ""}, + {"a", "b"}, + {"my-bucket", "path/to/object.txt"}, + {"日本語", "キー"}, + {"bucket-with-dash", ""}, + } + for _, c := range cases { + got := ShardID(c.bucket, c.key) + if got < 0 || got >= ShardCount { + t.Fatalf("ShardID(%q,%q)=%d outside [0,%d)", c.bucket, c.key, got, ShardCount) + } + } +} + +func TestShardIDDeterministic(t *testing.T) { + a := ShardID("bucket", "key") + b := ShardID("bucket", "key") + if a != b { + t.Fatalf("non-deterministic: %d vs %d", a, b) + } +} + +func TestShardIDDistinctFromBucket(t *testing.T) { + // Same key, different buckets must be free to land on different shards. + // (Statistical: with 16 shards, picking 16 distinct buckets should hit + // at least 4 different shards.) + seen := map[int]struct{}{} + for i := 0; i < 16; i++ { + buckets := []string{ + "alpha", "bravo", "charlie", "delta", + "echo", "foxtrot", "golf", "hotel", + "india", "juliet", "kilo", "lima", + "mike", "november", "oscar", "papa", + } + seen[ShardID(buckets[i], "samekey")] = struct{}{} + } + if len(seen) < 4 { + t.Fatalf("expected >=4 distinct shards across 16 buckets, got %d", len(seen)) + } +}