From 0841ea033885715bf9a47c1907c1c9f82e03bf0f Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Fri, 28 Aug 2026 13:29:30 +0200 Subject: [PATCH 1/5] go fix, and modernize some code automated fixes from; `go fix -mod=readonly -all ./...` Signed-off-by: Sebastiaan van Stijn --- broadcast.go | 16 ++++------------ broadcast_test.go | 4 ++-- channel.go | 2 +- common_test.go | 4 ++-- event.go | 2 +- filter_test.go | 2 +- retry.go | 10 +++++----- retry_test.go | 7 ++----- 8 files changed, 18 insertions(+), 29 deletions(-) diff --git a/broadcast.go b/broadcast.go index c81f600..83d8693 100644 --- a/broadcast.go +++ b/broadcast.go @@ -2,6 +2,7 @@ package events import ( "fmt" + "slices" "sync" "github.com/sirupsen/logrus" @@ -131,19 +132,10 @@ func (b *Broadcaster) run() { case request := <-b.adds: // while we have to iterate for add/remove, common iteration for // send is faster against slice. - - var found bool - for _, sink := range b.sinks { - if request.sink == sink { - found = true - break - } - } - - if !found { + // b.sinks[request.sink] = struct{}{} + if !slices.Contains(b.sinks, request.sink) { b.sinks = append(b.sinks, request.sink) } - // b.sinks[request.sink] = struct{}{} request.response <- nil case request := <-b.removes: remove(request.sink) @@ -165,7 +157,7 @@ func (b *Broadcaster) String() string { // Serialize copy of this broadcaster without the sync.Once, to avoid // a data race. - b2 := map[string]interface{}{ + b2 := map[string]any{ "sinks": b.sinks, "events": b.events, "adds": b.adds, diff --git a/broadcast_test.go b/broadcast_test.go index 5108dd9..5afed3f 100644 --- a/broadcast_test.go +++ b/broadcast_test.go @@ -10,7 +10,7 @@ func TestBroadcaster(t *testing.T) { const nEvents = 1000 var sinks []Sink b := NewBroadcaster() - for i := 0; i < 10; i++ { + for i := range 10 { sinks = append(sinks, newTestSink(t, nEvents)) err := b.Add(sinks[i]) if err != nil { @@ -108,7 +108,7 @@ func benchmarkBroadcast(b *testing.B, nsinks int) { b.StopTimer() var sinks []Sink - for i := 0; i < nsinks; i++ { + for range nsinks { // counter.Inc(1) sinks = append(sinks, newTestSink(b, b.N)) // sinks = append(sinks, NewQueue(&testSink{t: b, expected: b.N})) diff --git a/channel.go b/channel.go index 802cf51..21b6ba9 100644 --- a/channel.go +++ b/channel.go @@ -53,7 +53,7 @@ func (ch *Channel) Close() error { func (ch *Channel) String() string { // Serialize a copy of the Channel that doesn't contain the sync.Once, // to avoid a data race. - ch2 := map[string]interface{}{ + ch2 := map[string]any{ "C": ch.C, "closed": ch.closed, } diff --git a/common_test.go b/common_test.go index 15e125e..72fd529 100644 --- a/common_test.go +++ b/common_test.go @@ -9,8 +9,8 @@ import ( ) type tOrB interface { - Fatalf(format string, args ...interface{}) - Logf(format string, args ...interface{}) + Fatalf(format string, args ...any) + Logf(format string, args ...any) } type testSink struct { diff --git a/event.go b/event.go index f0f1d9e..8cd997a 100644 --- a/event.go +++ b/event.go @@ -1,7 +1,7 @@ package events // Event marks items that can be sent as events. -type Event interface{} +type Event any // Sink accepts and sends events. type Sink interface { diff --git a/filter_test.go b/filter_test.go index ccb46c1..18ccd8a 100644 --- a/filter_test.go +++ b/filter_test.go @@ -10,7 +10,7 @@ func TestFilter(t *testing.T) { return ok && i%2 == 0 })) - for i := 0; i < nevents; i++ { + for i := range nevents { if err := filter.Write(i); err != nil { t.Fatalf("unexpected error writing event: %v", err) } diff --git a/retry.go b/retry.go index 9f75e23..bee26df 100644 --- a/retry.go +++ b/retry.go @@ -93,7 +93,7 @@ func (rs *RetryingSink) Close() error { func (rs *RetryingSink) String() string { // Serialize a copy of the RetryingSink without the sync.Once, to avoid // a data race. - rs2 := map[string]interface{}{ + rs2 := map[string]any{ "sink": rs.sink, "strategy": rs.strategy, "closed": rs.closed, @@ -201,7 +201,7 @@ type ExponentialBackoffConfig struct { // ExponentialBackoff implements random backoff with exponentially increasing // bounds as the number consecutive failures increase. type ExponentialBackoff struct { - failures uint64 // consecutive failure counter (needs to be 64-bit aligned) + failures atomic.Uint64 // consecutive failure counter (needs to be 64-bit aligned) config ExponentialBackoffConfig } @@ -215,17 +215,17 @@ func NewExponentialBackoff(config ExponentialBackoffConfig) *ExponentialBackoff // Proceed returns the next randomly bound exponential backoff time. func (b *ExponentialBackoff) Proceed(event Event) time.Duration { - return b.backoff(atomic.LoadUint64(&b.failures)) + return b.backoff(b.failures.Load()) } // Success resets the failures counter. func (b *ExponentialBackoff) Success(event Event) { - atomic.StoreUint64(&b.failures, 0) + b.failures.Store(0) } // Failure increments the failure counter. func (b *ExponentialBackoff) Failure(event Event, err error) bool { - atomic.AddUint64(&b.failures, 1) + b.failures.Add(1) return false } diff --git a/retry_test.go b/retry_test.go index 1da4618..57ef092 100644 --- a/retry_test.go +++ b/retry_test.go @@ -84,7 +84,7 @@ func TestExponentialBackoff(t *testing.T) { t.Errorf("no facilities for dropping events in ExponentialBackoff") } - for j := 0; j < 1000; j++ { + for range 1000 { // sample this several thousand times. backoff := strategy.Proceed(nil) if backoff > expected { @@ -92,10 +92,7 @@ func TestExponentialBackoff(t *testing.T) { } } - expected = strategy.config.Base + strategy.config.Factor*time.Duration(1< strategy.config.Max { - expected = strategy.config.Max - } + expected = min(strategy.config.Base+strategy.config.Factor*time.Duration(1< Date: Fri, 28 Aug 2026 13:36:27 +0200 Subject: [PATCH 2/5] rename/remove intermediate vars that shadowed Also move the backoff var closer to where used Signed-off-by: Sebastiaan van Stijn --- broadcast.go | 7 ++----- channel.go | 5 ++--- queue.go | 1 + retry.go | 23 +++++++++-------------- 4 files changed, 14 insertions(+), 22 deletions(-) diff --git a/broadcast.go b/broadcast.go index 83d8693..91dbf70 100644 --- a/broadcast.go +++ b/broadcast.go @@ -156,8 +156,7 @@ func (b *Broadcaster) run() { func (b *Broadcaster) String() string { // Serialize copy of this broadcaster without the sync.Once, to avoid // a data race. - - b2 := map[string]any{ + return fmt.Sprint(map[string]any{ "sinks": b.sinks, "events": b.events, "adds": b.adds, @@ -165,7 +164,5 @@ func (b *Broadcaster) String() string { "shutdown": b.shutdown, "closed": b.closed, - } - - return fmt.Sprint(b2) + }) } diff --git a/channel.go b/channel.go index 21b6ba9..0eb7287 100644 --- a/channel.go +++ b/channel.go @@ -53,9 +53,8 @@ func (ch *Channel) Close() error { func (ch *Channel) String() string { // Serialize a copy of the Channel that doesn't contain the sync.Once, // to avoid a data race. - ch2 := map[string]any{ + return fmt.Sprint(map[string]any{ "C": ch.C, "closed": ch.closed, - } - return fmt.Sprint(ch2) + }) } diff --git a/queue.go b/queue.go index 4bb770a..b286405 100644 --- a/queue.go +++ b/queue.go @@ -103,6 +103,7 @@ func (eq *Queue) next() Event { eq.cond.Wait() } + // Len is non-zero while holding eq.mu, so Front cannot be nil. front := eq.events.Front() block := front.Value.(Event) eq.events.Remove(front) diff --git a/retry.go b/retry.go index bee26df..c1ca031 100644 --- a/retry.go +++ b/retry.go @@ -26,13 +26,11 @@ type RetryingSink struct { // off on failure. Parameters threshold and backoff adjust the behavior of the // circuit breaker. func NewRetryingSink(sink Sink, strategy RetryStrategy) *RetryingSink { - rs := &RetryingSink{ + return &RetryingSink{ sink: sink, strategy: strategy, closed: make(chan struct{}), } - - return rs } // Write attempts to flush the events to the downstream sink until it succeeds @@ -66,14 +64,12 @@ retry: return err } - logger := logger.WithError(err) // shadow!! - if rs.strategy.Failure(event, err) { - logger.Errorf("retryingsink: dropped event") + logger.WithError(err).Error("retryingsink: dropped event") return nil } - logger.Errorf("retryingsink: error writing event, retrying") + logger.WithError(err).Error("retryingsink: error writing event, retrying") goto retry } @@ -242,15 +238,14 @@ func (b *ExponentialBackoff) backoff(failures uint64) time.Duration { factor = DefaultExponentialBackoffConfig.Factor } - backoff := b.config.Base + factor*time.Duration(1<<(failures-1)) - - max := b.config.Max - if max <= 0 { - max = DefaultExponentialBackoffConfig.Max + maxBackoff := b.config.Max + if maxBackoff <= 0 { + maxBackoff = DefaultExponentialBackoffConfig.Max } - if backoff > max || backoff < 0 { - backoff = max + backoff := b.config.Base + factor*time.Duration(1<<(failures-1)) + if backoff > maxBackoff || backoff < 0 { + backoff = maxBackoff } // Choose a uniformly distributed value from [0, backoff). From 056aa37b8e6b903084b00ebf1e8089ab18bcf4fd Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Fri, 28 Aug 2026 13:44:01 +0200 Subject: [PATCH 3/5] switch to math/rand/v2 Signed-off-by: Sebastiaan van Stijn --- common_test.go | 2 +- retry.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/common_test.go b/common_test.go index 72fd529..e27053e 100644 --- a/common_test.go +++ b/common_test.go @@ -2,7 +2,7 @@ package events import ( "fmt" - "math/rand" + "math/rand/v2" "sync" "testing" "time" diff --git a/retry.go b/retry.go index c1ca031..70fad9c 100644 --- a/retry.go +++ b/retry.go @@ -2,7 +2,7 @@ package events import ( "fmt" - "math/rand" + "math/rand/v2" "sync" "sync/atomic" "time" @@ -249,5 +249,5 @@ func (b *ExponentialBackoff) backoff(failures uint64) time.Duration { } // Choose a uniformly distributed value from [0, backoff). - return time.Duration(rand.Int63n(int64(backoff))) + return rand.N(backoff) } From 8a0b5e22dc26dbf31f8030553a47f777eeafe661 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Fri, 28 Aug 2026 14:24:08 +0200 Subject: [PATCH 4/5] avoid chainging logrus.WithField Signed-off-by: Sebastiaan van Stijn --- broadcast.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/broadcast.go b/broadcast.go index 91dbf70..2d1f8f1 100644 --- a/broadcast.go +++ b/broadcast.go @@ -125,8 +125,11 @@ func (b *Broadcaster) run() { remove(sink) continue } - logrus.WithField("event", event).WithField("events.sink", sink).WithError(err). - Errorf("broadcaster: dropping event") + logrus.WithFields(logrus.Fields{ + "error": err, + "event": event, + "events.sink": sink, + }).Error("broadcaster: dropping event") } } case request := <-b.adds: @@ -144,8 +147,10 @@ func (b *Broadcaster) run() { // close all the underlying sinks for _, sink := range b.sinks { if err := sink.Close(); err != nil && err != ErrSinkClosed { - logrus.WithField("events.sink", sink).WithError(err). - Errorf("broadcaster: closing sink failed") + logrus.WithFields(logrus.Fields{ + "error": err, + "events.sink": sink, + }).Error("broadcaster: closing sink failed") } } return From f1a1f5ca529937a10dccc0eed3bc4fce90c35815 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Fri, 28 Aug 2026 14:36:42 +0200 Subject: [PATCH 5/5] refactor: avoid redundant sink lookup when broadcasting Remove closed sinks directly by index while iterating over the broadcaster's sink slice. This avoids searching the slice again for every closed sink and prevents skipping entries when removing elements during iteration. Use slices.Index and slices.Delete for explicit sink removals, replacing the local remove helper and manual slice manipulation. Signed-off-by: Sebastiaan van Stijn --- broadcast.go | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/broadcast.go b/broadcast.go index 2d1f8f1..ef8dd3f 100644 --- a/broadcast.go +++ b/broadcast.go @@ -106,23 +106,15 @@ func (b *Broadcaster) Close() error { // Close is called, this goroutine will exit. func (b *Broadcaster) run() { defer close(b.closed) - remove := func(target Sink) { - for i, sink := range b.sinks { - if sink == target { - b.sinks = append(b.sinks[:i], b.sinks[i+1:]...) - break - } - } - } for { select { case event := <-b.events: - for _, sink := range b.sinks { + for i := 0; i < len(b.sinks); { + sink := b.sinks[i] if err := sink.Write(event); err != nil { if err == ErrSinkClosed { - // remove closed sinks - remove(sink) + b.sinks = slices.Delete(b.sinks, i, i+1) continue } logrus.WithFields(logrus.Fields{ @@ -131,6 +123,7 @@ func (b *Broadcaster) run() { "events.sink": sink, }).Error("broadcaster: dropping event") } + i++ } case request := <-b.adds: // while we have to iterate for add/remove, common iteration for @@ -141,7 +134,9 @@ func (b *Broadcaster) run() { } request.response <- nil case request := <-b.removes: - remove(request.sink) + if i := slices.Index(b.sinks, request.sink); i >= 0 { + b.sinks = slices.Delete(b.sinks, i, i+1) + } request.response <- nil case <-b.shutdown: // close all the underlying sinks