From b7ffcabddf91d35430b0b4a4daea362e01c313c6 Mon Sep 17 00:00:00 2001 From: Harshavardhana Date: Tue, 1 Sep 2026 23:41:13 -0700 Subject: [PATCH 1/3] feat(dedup): add a bounded de-duplicating channel Go channels cannot de-duplicate: there is no hook on send and no way to inspect a buffer. Chan wraps one with a pending-key map so a value whose key is already queued is dropped instead of enqueued twice, coalescing repeated notifications about the same entity into a single unit of work. A sender acquires a buffer slot before reserving its key, so a send that is cancelled while the channel is full cannot drop a duplicate and leave nothing queued in its place. --- sync/dedup/dedup.go | 195 ++++++++++++++++++++++ sync/dedup/dedup_test.go | 337 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 532 insertions(+) create mode 100644 sync/dedup/dedup.go create mode 100644 sync/dedup/dedup_test.go diff --git a/sync/dedup/dedup.go b/sync/dedup/dedup.go new file mode 100644 index 00000000..e16cc694 --- /dev/null +++ b/sync/dedup/dedup.go @@ -0,0 +1,195 @@ +// 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), nil +} + +// 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), nil +} + +// 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. Already queued values remain available to Recv. +// Close may be called more than once. +func (c *Chan[K, T]) Close() { + c.once.Do(func() { close(c.closed) }) +} + +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. +func (c *Chan[K, T]) enqueue(k K, v T) bool { + c.mu.Lock() + defer c.mu.Unlock() + if _, ok := c.pending[k]; ok { + c.slots <- struct{}{} + return false + } + c.pending[k] = struct{}{} + c.items <- entry[K, T]{key: k, val: v} + return true +} + +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..8b049491 --- /dev/null +++ b/sync/dedup/dedup_test.go @@ -0,0 +1,337 @@ +// 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 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) + } +} From 371b0a0eedfd6c722c6108de8d1d86897ad01281 Mon Sep 17 00:00:00 2001 From: Harshavardhana Date: Wed, 2 Sep 2026 00:07:12 -0700 Subject: [PATCH 2/3] build: bump golang.org/x/crypto to v0.55.0 govulncheck flags GO-2026-6303 in golang.org/x/crypto v0.52.0, reached from sftp.Server.handleConnection via ssh.NewServerConn. --- go.mod | 14 +++++++------- go.sum | 32 ++++++++++++++++---------------- 2 files changed, 23 insertions(+), 23 deletions(-) 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= From 7a36b03be2b1d03ba2ea4759908c0adaa95250a6 Mon Sep 17 00:00:00 2001 From: Harshavardhana Date: Wed, 2 Sep 2026 00:15:20 -0700 Subject: [PATCH 3/3] fix(dedup): serialize enqueue with Close A sender that had already acquired a slot could enqueue after Close, by which point a receiver may have observed closure over an empty buffer and given up, leaving the value unreachable. Close now takes the same mutex as the enqueue path, so an enqueue either lands before closure is observable or returns its slot and reports ErrClosed. --- sync/dedup/dedup.go | 30 ++++++++++++++------- sync/dedup/dedup_test.go | 56 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 9 deletions(-) diff --git a/sync/dedup/dedup.go b/sync/dedup/dedup.go index e16cc694..78ae32e2 100644 --- a/sync/dedup/dedup.go +++ b/sync/dedup/dedup.go @@ -99,7 +99,7 @@ func (c *Chan[K, T]) Send(ctx context.Context, v T) (bool, error) { case <-ctx.Done(): return false, ctx.Err() } - return c.enqueue(k, v), nil + return c.enqueue(k, v) } // TrySend queues v without blocking. It reports whether v was queued; a false @@ -120,7 +120,7 @@ func (c *Chan[K, T]) TrySend(v T) (bool, error) { default: return false, ErrFull } - return c.enqueue(k, v), nil + return c.enqueue(k, v) } // Recv returns the next queued value, blocking until one arrives. It reports @@ -160,10 +160,15 @@ func (c *Chan[K, T]) Len() int { return len(c.items) } -// Close stops further sends. Already queued values remain available to Recv. -// Close may be called more than once. +// 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() { close(c.closed) }) + c.once.Do(func() { + c.mu.Lock() + close(c.closed) + c.mu.Unlock() + }) } func (c *Chan[K, T]) queued(k K) bool { @@ -174,17 +179,24 @@ func (c *Chan[K, T]) queued(k K) bool { } // enqueue is called holding a slot, so neither the send to items nor the -// return of the slot can block. -func (c *Chan[K, T]) enqueue(k K, v T) bool { +// 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 + return false, nil } c.pending[k] = struct{}{} c.items <- entry[K, T]{key: k, val: v} - return true + return true, nil } func (c *Chan[K, T]) release(e entry[K, T]) { diff --git a/sync/dedup/dedup_test.go b/sync/dedup/dedup_test.go index 8b049491..301cb423 100644 --- a/sync/dedup/dedup_test.go +++ b/sync/dedup/dedup_test.go @@ -271,6 +271,62 @@ func TestConcurrentDuplicateSends(t *testing.T) { } } +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