diff --git a/go.mod b/go.mod
index c4dd9c9e..154522dd 100644
--- a/go.mod
+++ b/go.mod
@@ -26,9 +26,9 @@ require (
github.com/zeebo/xxh3 v1.1.0
go.etcd.io/etcd/client/v3 v3.6.12
go.yaml.in/yaml/v3 v3.0.4
- golang.org/x/crypto v0.52.0
+ golang.org/x/crypto v0.55.0
golang.org/x/oauth2 v0.36.0
- golang.org/x/sys v0.45.0
+ golang.org/x/sys v0.47.0
)
require (
@@ -229,9 +229,9 @@ require (
go.augendre.info/arangolint v0.4.0 // indirect
go.augendre.info/fatcontext v0.9.0 // indirect
golang.org/x/exp/typeparams v0.0.0-20260209203927-2842357ff358 // indirect
- golang.org/x/mod v0.35.0 // indirect
- golang.org/x/sync v0.20.0 // indirect
- golang.org/x/tools v0.44.0 // indirect
+ golang.org/x/mod v0.38.0 // indirect
+ golang.org/x/sync v0.22.0 // indirect
+ golang.org/x/tools v0.48.0 // indirect
gopkg.in/ini.v1 v1.67.2 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
@@ -264,8 +264,8 @@ require (
go.etcd.io/etcd/client/pkg/v3 v3.6.12 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.28.0 // indirect
- golang.org/x/net v0.55.0 // indirect
- golang.org/x/text v0.37.0 // indirect
+ golang.org/x/net v0.57.0 // indirect
+ golang.org/x/text v0.41.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
google.golang.org/grpc v1.82.1 // indirect
diff --git a/go.sum b/go.sum
index 0c5a8c87..46c4790b 100644
--- a/go.sum
+++ b/go.sum
@@ -596,8 +596,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
-golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
+golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M=
+golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis=
golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o=
golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8=
golang.org/x/exp/typeparams v0.0.0-20220428152302-39d4317da171/go.mod h1:AbB0pIl9nAr9wVwH+Z2ZpaocVmF5I4GyWCDIsVjR0bk=
@@ -610,8 +610,8 @@ golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3/go.mod h1:3p9vT2HGsQu2K1YbXdKPJLVgG5VJdoTa1poYQBtP1AY=
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
-golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM=
-golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU=
+golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk=
+golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -621,8 +621,8 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
-golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
+golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
+golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -632,8 +632,8 @@ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJ
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
-golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.0.0-20180926160741-c2ed4eda69e7/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -651,21 +651,21 @@ golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
-golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
-golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4=
-golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk=
+golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0=
+golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
-golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8=
+golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20200329025819-fd4102a86c65/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
@@ -677,8 +677,8 @@ golang.org/x/tools v0.1.1-0.20210302220138-2ac05c832e1a/go.mod h1:9bzcO0MWcOuT0t
golang.org/x/tools v0.1.10/go.mod h1:Uh6Zz+xoGYZom868N8YTex3t7RhtHDBrE8Gzo9bV56E=
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
-golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c=
-golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI=
+golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE=
+golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk=
golang.org/x/tools/go/expect v0.1.1-deprecated h1:jpBZDwmgPhXsKZC6WhL20P4b/wmnpsEAGHaNy0n/rJM=
golang.org/x/tools/go/expect v0.1.1-deprecated/go.mod h1:eihoPOH+FgIqa3FpoTwguz/bVUSGBlGQU67vpBeOrBY=
golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated h1:1h2MnaIAIXISqTFKdENegdpAgUXz6NrPEsbIeWaBRvM=
diff --git a/sync/dedup/dedup.go b/sync/dedup/dedup.go
new file mode 100644
index 00000000..78ae32e2
--- /dev/null
+++ b/sync/dedup/dedup.go
@@ -0,0 +1,207 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+// Package dedup provides a bounded, de-duplicating channel. A value whose key
+// is already queued is dropped instead of being enqueued a second time, which
+// makes it suitable for coalescing repeated notifications about the same
+// entity into a single unit of work.
+package dedup
+
+import (
+ "context"
+ "errors"
+ "sync"
+)
+
+// ErrClosed is returned by send operations on a closed Chan.
+var ErrClosed = errors.New("dedup: channel is closed")
+
+// ErrFull is returned by TrySend when the channel is at capacity.
+var ErrFull = errors.New("dedup: channel is full")
+
+// Chan is a buffered channel of T that holds at most one queued value per
+// key K. Keys are derived from values and stay reserved until the value is
+// received, so a duplicate is only dropped while its predecessor is still
+// waiting to be consumed.
+//
+// Chan is safe for concurrent use. It is not a Go channel and cannot be used
+// in a select statement; the context aware Send and Recv serve that purpose.
+type Chan[K comparable, T any] struct {
+ key func(T) K
+ items chan entry[K, T]
+ slots chan struct{}
+ closed chan struct{}
+ once sync.Once
+
+ mu sync.Mutex
+ pending map[K]struct{}
+}
+
+type entry[K comparable, T any] struct {
+ key K
+ val T
+}
+
+// NewChan returns a Chan buffering up to size values, keyed by key. key must
+// return the same result every time it is called for a given value.
+func NewChan[K comparable, T any](size int, key func(T) K) (*Chan[K, T], error) {
+ if size <= 0 {
+ return nil, errors.New("dedup: size must be > 0")
+ }
+ if key == nil {
+ return nil, errors.New("dedup: key must not be nil")
+ }
+ c := &Chan[K, T]{
+ key: key,
+ items: make(chan entry[K, T], size),
+ slots: make(chan struct{}, size),
+ closed: make(chan struct{}),
+ pending: make(map[K]struct{}, size),
+ }
+ for i := 0; i < size; i++ {
+ c.slots <- struct{}{}
+ }
+ return c, nil
+}
+
+// Send queues v, blocking while the channel is full. It reports whether v was
+// queued; a false return with a nil error means an equal key was already
+// queued. It returns ErrClosed once the channel is closed and the context
+// error if ctx ends first.
+func (c *Chan[K, T]) Send(ctx context.Context, v T) (bool, error) {
+ select {
+ case <-c.closed:
+ return false, ErrClosed
+ default:
+ }
+ k := c.key(v)
+ if c.queued(k) {
+ return false, nil
+ }
+ select {
+ case <-c.slots:
+ case <-c.closed:
+ return false, ErrClosed
+ case <-ctx.Done():
+ return false, ctx.Err()
+ }
+ return c.enqueue(k, v)
+}
+
+// TrySend queues v without blocking. It reports whether v was queued; a false
+// return with a nil error means an equal key was already queued. It returns
+// ErrFull if the channel is at capacity and ErrClosed once it is closed.
+func (c *Chan[K, T]) TrySend(v T) (bool, error) {
+ select {
+ case <-c.closed:
+ return false, ErrClosed
+ default:
+ }
+ k := c.key(v)
+ if c.queued(k) {
+ return false, nil
+ }
+ select {
+ case <-c.slots:
+ default:
+ return false, ErrFull
+ }
+ return c.enqueue(k, v)
+}
+
+// Recv returns the next queued value, blocking until one arrives. It reports
+// false once ctx ends, or once the channel is closed and drained. Receiving a
+// value frees its key, so an equal value may be queued again.
+func (c *Chan[K, T]) Recv(ctx context.Context) (T, bool) {
+ if v, ok := c.TryRecv(); ok {
+ return v, true
+ }
+ select {
+ case e := <-c.items:
+ c.release(e)
+ return e.val, true
+ case <-c.closed:
+ return c.TryRecv()
+ case <-ctx.Done():
+ var zero T
+ return zero, false
+ }
+}
+
+// TryRecv returns the next queued value without blocking, reporting false if
+// none is queued.
+func (c *Chan[K, T]) TryRecv() (T, bool) {
+ select {
+ case e := <-c.items:
+ c.release(e)
+ return e.val, true
+ default:
+ var zero T
+ return zero, false
+ }
+}
+
+// Len returns the number of queued values.
+func (c *Chan[K, T]) Len() int {
+ return len(c.items)
+}
+
+// Close stops further sends. No value is queued once Close returns, and every
+// value a successful send queued stays available to Recv. Close may be called
+// more than once.
+func (c *Chan[K, T]) Close() {
+ c.once.Do(func() {
+ c.mu.Lock()
+ close(c.closed)
+ c.mu.Unlock()
+ })
+}
+
+func (c *Chan[K, T]) queued(k K) bool {
+ c.mu.Lock()
+ _, ok := c.pending[k]
+ c.mu.Unlock()
+ return ok
+}
+
+// enqueue is called holding a slot, so neither the send to items nor the
+// return of the slot can block. Close takes the same mutex, so an enqueue
+// either completes before the channel is closed or gives up.
+func (c *Chan[K, T]) enqueue(k K, v T) (bool, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ select {
+ case <-c.closed:
+ c.slots <- struct{}{}
+ return false, ErrClosed
+ default:
+ }
+ if _, ok := c.pending[k]; ok {
+ c.slots <- struct{}{}
+ return false, nil
+ }
+ c.pending[k] = struct{}{}
+ c.items <- entry[K, T]{key: k, val: v}
+ return true, nil
+}
+
+func (c *Chan[K, T]) release(e entry[K, T]) {
+ c.mu.Lock()
+ delete(c.pending, e.key)
+ c.mu.Unlock()
+ c.slots <- struct{}{}
+}
diff --git a/sync/dedup/dedup_test.go b/sync/dedup/dedup_test.go
new file mode 100644
index 00000000..301cb423
--- /dev/null
+++ b/sync/dedup/dedup_test.go
@@ -0,0 +1,393 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package dedup
+
+import (
+ "context"
+ "errors"
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+type event struct {
+ bucket string
+ seq int
+}
+
+func eventKey(e event) string { return e.bucket }
+
+func newTestChan(t *testing.T, size int) *Chan[string, event] {
+ t.Helper()
+ c, err := NewChan(size, eventKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return c
+}
+
+func TestNewChanValidation(t *testing.T) {
+ if _, err := NewChan(0, eventKey); err == nil {
+ t.Fatal("expected an error for size 0")
+ }
+ if _, err := NewChan[string, event](4, nil); err == nil {
+ t.Fatal("expected an error for a nil key function")
+ }
+}
+
+func TestDuplicateDroppedWhileQueued(t *testing.T) {
+ ctx := context.Background()
+ c := newTestChan(t, 4)
+
+ if sent, err := c.Send(ctx, event{"a", 1}); !sent || err != nil {
+ t.Fatalf("first send: sent=%v err=%v", sent, err)
+ }
+ if sent, err := c.Send(ctx, event{"a", 2}); sent || err != nil {
+ t.Fatalf("duplicate send: sent=%v err=%v", sent, err)
+ }
+ if sent, err := c.TrySend(event{"a", 3}); sent || err != nil {
+ t.Fatalf("duplicate try send: sent=%v err=%v", sent, err)
+ }
+ if got := c.Len(); got != 1 {
+ t.Fatalf("expected 1 queued value, got %d", got)
+ }
+
+ v, ok := c.Recv(ctx)
+ if !ok || v.seq != 1 {
+ t.Fatalf("expected the first value, got %+v ok=%v", v, ok)
+ }
+ if sent, err := c.Send(ctx, event{"a", 4}); !sent || err != nil {
+ t.Fatalf("send after receive: sent=%v err=%v", sent, err)
+ }
+}
+
+func TestDistinctKeysQueueIndependently(t *testing.T) {
+ ctx := context.Background()
+ c := newTestChan(t, 3)
+
+ for _, bucket := range []string{"a", "b", "c"} {
+ if sent, err := c.Send(ctx, event{bucket, 0}); !sent || err != nil {
+ t.Fatalf("send %q: sent=%v err=%v", bucket, sent, err)
+ }
+ }
+ if got := c.Len(); got != 3 {
+ t.Fatalf("expected 3 queued values, got %d", got)
+ }
+ for _, want := range []string{"a", "b", "c"} {
+ v, ok := c.TryRecv()
+ if !ok || v.bucket != want {
+ t.Fatalf("expected %q, got %+v ok=%v", want, v, ok)
+ }
+ }
+ if _, ok := c.TryRecv(); ok {
+ t.Fatal("expected an empty channel")
+ }
+}
+
+func TestTrySendFull(t *testing.T) {
+ c := newTestChan(t, 1)
+
+ if sent, err := c.TrySend(event{"a", 0}); !sent || err != nil {
+ t.Fatalf("send: sent=%v err=%v", sent, err)
+ }
+ sent, err := c.TrySend(event{"b", 0})
+ if sent || !errors.Is(err, ErrFull) {
+ t.Fatalf("expected ErrFull, got sent=%v err=%v", sent, err)
+ }
+}
+
+func TestSendBlocksUntilCapacity(t *testing.T) {
+ ctx := context.Background()
+ c := newTestChan(t, 1)
+
+ if _, err := c.Send(ctx, event{"a", 0}); err != nil {
+ t.Fatal(err)
+ }
+
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ if sent, err := c.Send(ctx, event{"b", 0}); !sent || err != nil {
+ t.Errorf("blocked send: sent=%v err=%v", sent, err)
+ }
+ }()
+
+ select {
+ case <-done:
+ t.Fatal("send returned while the channel was full")
+ case <-time.After(50 * time.Millisecond):
+ }
+
+ if _, ok := c.Recv(ctx); !ok {
+ t.Fatal("expected a value")
+ }
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("send did not return after capacity was freed")
+ }
+}
+
+func TestSendContextCancel(t *testing.T) {
+ c := newTestChan(t, 1)
+ if _, err := c.Send(context.Background(), event{"a", 0}); err != nil {
+ t.Fatal(err)
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+ defer cancel()
+ sent, err := c.Send(ctx, event{"b", 0})
+ if sent || !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("expected a deadline error, got sent=%v err=%v", sent, err)
+ }
+ if sent, err := c.TrySend(event{"b", 1}); sent || !errors.Is(err, ErrFull) {
+ t.Fatalf("canceled send must not reserve a key: sent=%v err=%v", sent, err)
+ }
+}
+
+func TestRecvContextCancel(t *testing.T) {
+ c := newTestChan(t, 1)
+ ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+ defer cancel()
+
+ if _, ok := c.Recv(ctx); ok {
+ t.Fatal("expected no value from an empty channel")
+ }
+}
+
+func TestRecvPrefersQueuedValueOverCanceledContext(t *testing.T) {
+ c := newTestChan(t, 1)
+ if _, err := c.Send(context.Background(), event{"a", 0}); err != nil {
+ t.Fatal(err)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ if _, ok := c.Recv(ctx); !ok {
+ t.Fatal("expected the queued value")
+ }
+}
+
+func TestClose(t *testing.T) {
+ ctx := context.Background()
+ c := newTestChan(t, 2)
+
+ if _, err := c.Send(ctx, event{"a", 0}); err != nil {
+ t.Fatal(err)
+ }
+ c.Close()
+ c.Close()
+
+ if sent, err := c.Send(ctx, event{"b", 0}); sent || !errors.Is(err, ErrClosed) {
+ t.Fatalf("send after close: sent=%v err=%v", sent, err)
+ }
+ if sent, err := c.TrySend(event{"b", 0}); sent || !errors.Is(err, ErrClosed) {
+ t.Fatalf("try send after close: sent=%v err=%v", sent, err)
+ }
+ if sent, err := c.Send(ctx, event{"a", 1}); sent || !errors.Is(err, ErrClosed) {
+ t.Fatalf("duplicate send after close: sent=%v err=%v", sent, err)
+ }
+ if v, ok := c.Recv(ctx); !ok || v.bucket != "a" {
+ t.Fatalf("expected the queued value to drain, got %+v ok=%v", v, ok)
+ }
+ if _, ok := c.Recv(ctx); ok {
+ t.Fatal("expected no value from a closed and drained channel")
+ }
+}
+
+func TestCloseUnblocksSend(t *testing.T) {
+ ctx := context.Background()
+ c := newTestChan(t, 1)
+ if _, err := c.Send(ctx, event{"a", 0}); err != nil {
+ t.Fatal(err)
+ }
+
+ done := make(chan error, 1)
+ go func() {
+ _, err := c.Send(ctx, event{"b", 0})
+ done <- err
+ }()
+
+ time.Sleep(50 * time.Millisecond)
+ c.Close()
+
+ select {
+ case err := <-done:
+ if !errors.Is(err, ErrClosed) {
+ t.Fatalf("expected ErrClosed, got %v", err)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("close did not unblock the pending send")
+ }
+}
+
+func TestConcurrentDuplicateSends(t *testing.T) {
+ ctx := context.Background()
+ c := newTestChan(t, 8)
+
+ var wins atomic.Int64
+ var wg sync.WaitGroup
+ for i := 0; i < 32; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ sent, err := c.Send(ctx, event{"a", i})
+ if err != nil {
+ t.Errorf("send: %v", err)
+ return
+ }
+ if sent {
+ wins.Add(1)
+ }
+ }(i)
+ }
+ wg.Wait()
+
+ if wins.Load() != 1 {
+ t.Fatalf("expected exactly one send to win, got %d", wins.Load())
+ }
+ if got := c.Len(); got != 1 {
+ t.Fatalf("expected 1 queued value, got %d", got)
+ }
+ if got := len(c.slots); got != 7 {
+ t.Fatalf("expected 7 free slots, got %d", got)
+ }
+}
+
+func TestCloseDoesNotStrandValues(t *testing.T) {
+ const (
+ runs = 200
+ senders = 4
+ )
+ ctx := context.Background()
+
+ for run := 0; run < runs; run++ {
+ c := newTestChan(t, 2)
+
+ var queued, received atomic.Int64
+ var wg sync.WaitGroup
+
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for {
+ if _, ok := c.Recv(ctx); !ok {
+ return
+ }
+ received.Add(1)
+ }
+ }()
+
+ var sent sync.WaitGroup
+ sent.Add(senders)
+ for i := 0; i < senders; i++ {
+ go func(i int) {
+ defer sent.Done()
+ ok, err := c.Send(ctx, event{strconv.Itoa(i), i})
+ if err != nil && !errors.Is(err, ErrClosed) {
+ t.Errorf("send: %v", err)
+ return
+ }
+ if ok {
+ queued.Add(1)
+ }
+ }(i)
+ }
+
+ c.Close()
+ sent.Wait()
+ wg.Wait()
+
+ if got, ok := c.TryRecv(); ok {
+ t.Fatalf("run %d: value %+v was queued after close", run, got)
+ }
+ if queued.Load() != received.Load() {
+ t.Fatalf("run %d: queued %d values but received %d", run, queued.Load(), received.Load())
+ }
+ if got := len(c.slots); got != 2 {
+ t.Fatalf("run %d: expected all slots returned, got %d", run, got)
+ }
+ }
+}
+
+func TestConcurrentSendRecvLosesNothing(t *testing.T) {
+ const (
+ producers = 8
+ perRun = 500
+ keys = 16
+ consumers = 4
+ )
+ ctx := context.Background()
+ c := newTestChan(t, 4)
+
+ var queued, received atomic.Int64
+
+ var consumed sync.WaitGroup
+ consumed.Add(consumers)
+ for i := 0; i < consumers; i++ {
+ go func() {
+ defer consumed.Done()
+ for {
+ if _, ok := c.Recv(ctx); !ok {
+ return
+ }
+ received.Add(1)
+ }
+ }()
+ }
+
+ var produced sync.WaitGroup
+ produced.Add(producers)
+ for p := 0; p < producers; p++ {
+ go func(p int) {
+ defer produced.Done()
+ for i := 0; i < perRun; i++ {
+ sent, err := c.Send(ctx, event{strconv.Itoa((p*perRun + i) % keys), i})
+ if err != nil {
+ t.Errorf("send: %v", err)
+ return
+ }
+ if sent {
+ queued.Add(1)
+ }
+ }
+ }(p)
+ }
+
+ produced.Wait()
+ c.Close()
+ consumed.Wait()
+
+ if queued.Load() != received.Load() {
+ t.Fatalf("queued %d values but received %d", queued.Load(), received.Load())
+ }
+ if got := c.Len(); got != 0 {
+ t.Fatalf("expected an empty channel, got %d", got)
+ }
+ c.mu.Lock()
+ pending := len(c.pending)
+ c.mu.Unlock()
+ if pending != 0 {
+ t.Fatalf("expected no reserved keys, got %d", pending)
+ }
+ if got := len(c.slots); got != 4 {
+ t.Fatalf("expected all slots returned, got %d", got)
+ }
+}