Add static analysis

.github/workflows/ci.yml

@@ -21,6 +21,25 @@ on:
   pull_request:
 
 jobs:
+  lint:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version: "oldstable" 
+          cache: false
+      - name: Create go.mod
+        run: |
+          ln -s vendor.mod go.mod
+          ln -s vendor.sum go.sum
+      - uses: golangci/golangci-lint-action@v6
+        with:
+          version: v1.60.1
+          skip-cache: true
+
   test:
     runs-on: ubuntu-24.04
     timeout-minutes: 5

README.md

@@ -1,6 +1,8 @@
 # Docker Events Package
 
 [![GoDoc](https://godoc.org/github.com/docker/go-events?status.svg)](https://godoc.org/github.com/docker/go-events)
+[![ci](https://github.com/docker/go-events/actions/workflows/ci.yml/badge.svg)](https://github.com/docker/go-events/actions/workflows/ci.yml)
+[![Go Report Card](https://goreportcard.com/badge/github.com/docker/go-events)](https://goreportcard.com/report/github.com/docker/go-events)
 
 The Docker `events` package implements a composable event distribution package
 for Go.

broadcast_test.go

@@ -1,6 +1,7 @@
 package events
 
 import (
+	"fmt"
 	"sync"
 	"testing"
 )
@@ -11,25 +12,46 @@ func TestBroadcaster(t *testing.T) {
 	b := NewBroadcaster()
 	for i := 0; i < 10; i++ {
 		sinks = append(sinks, newTestSink(t, nEvents))
-		b.Add(sinks[i])
-		b.Add(sinks[i]) // noop
+		err := b.Add(sinks[i])
+		if err != nil {
+			t.Errorf("expected nil error, got %v", err)
+		}
+
+		err = b.Add(sinks[i]) // noop
+		if err != nil {
+			t.Errorf("expected nil error, got %v", err)
+		}
 	}
 
-	var wg sync.WaitGroup
+	var (
+		wg       sync.WaitGroup
+		asyncErr error
+		once     sync.Once
+	)
 	for i := 1; i <= nEvents; i++ {
 		wg.Add(1)
 		go func(event Event) {
+			defer wg.Done()
+
 			if err := b.Write(event); err != nil {
-				t.Fatalf("error writing event %v: %v", event, err)
+				once.Do(func() {
+					asyncErr = fmt.Errorf("error writing event(%v): %v", event, err)
+				})
 			}
-			wg.Done()
-		}("event")
+		}(fmt.Sprintf("event-%d", i))
 	}
 
 	wg.Wait() // Wait until writes complete
 
+	if asyncErr != nil {
+		t.Fatalf("expected nil error, got %v", asyncErr)
+	}
+
 	for i := range sinks {
-		b.Remove(sinks[i])
+		err := b.Remove(sinks[i])
+		if err != nil {
+			t.Errorf("expected nil error, got %v", err)
+		}
 	}
 
 	// sending one more should trigger test failure if they weren't removed.
@@ -39,7 +61,10 @@ func TestBroadcaster(t *testing.T) {
 
 	// add them back to test closing.
 	for i := range sinks {
-		b.Add(sinks[i])
+		err := b.Add(sinks[i])
+		if err != nil {
+			t.Errorf("expected nil error, got %v", err)
+		}
 	}
 
 	checkClose(t, b)

channel_test.go

@@ -9,28 +9,55 @@ import (
 func TestChannel(t *testing.T) {
 	const nevents = 100
 
+	errCh := make(chan error)
 	sink := NewChannel(0)
 
 	go func() {
-		var wg sync.WaitGroup
+		var (
+			wg       sync.WaitGroup
+			asyncErr error
+			once     sync.Once
+		)
 		for i := 1; i <= nevents; i++ {
-			event := "event-" + fmt.Sprint(i)
 			wg.Add(1)
 			go func(event Event) {
 				defer wg.Done()
+
 				if err := sink.Write(event); err != nil {
-					t.Fatalf("error writing event: %v", err)
+					once.Do(func() {
+						asyncErr = fmt.Errorf("error writing event(%v): %v", event, err)
+					})
 				}
-			}(event)
+			}(fmt.Sprintf("event-%d", i))
 		}
+
 		wg.Wait()
+
+		if asyncErr != nil {
+			errCh <- asyncErr
+			return
+		}
+
 		sink.Close()
 
 		// now send another bunch of events and ensure we stay closed
 		for i := 1; i <= nevents; i++ {
-			if err := sink.Write(i); err != ErrSinkClosed {
-				t.Fatalf("unexpected error: %v != %v", err, ErrSinkClosed)
-			}
+			wg.Add(1)
+			go func(event Event) {
+				defer wg.Done()
+
+				if err := sink.Write(event); err != ErrSinkClosed {
+					once.Do(func() {
+						asyncErr = fmt.Errorf("expected %v, got %v", ErrSinkClosed, err)
+					})
+				}
+			}(fmt.Sprintf("event-%d", i))
+		}
+
+		wg.Wait()
+
+		if asyncErr != nil {
+			errCh <- asyncErr
 		}
 	}()
 
@@ -40,11 +67,15 @@ loop:
 		select {
 		case <-sink.C:
 			received++
+		case err := <-errCh:
+			t.Fatal(err)
 		case <-sink.Done():
 			break loop
 		}
 	}
 
+	close(errCh)
+
 	sink.Close()
 	_, ok := <-sink.Done() // test will timeout if this hangs
 	if ok {

common_test.go

@@ -112,6 +112,6 @@ func benchmarkSink(b *testing.B, sink Sink) {
 	defer sink.Close()
 	var event = "myevent"
 	for i := 0; i < b.N; i++ {
-		sink.Write(event)
+		_ = sink.Write(event)
 	}
 }

queue_test.go

@@ -19,18 +19,30 @@ func TestQueue(t *testing.T) {
 		})
 	time.Sleep(10 * time.Millisecond) // let's queue settle to wait conidition.
 
-	var wg sync.WaitGroup
+	var (
+		wg       sync.WaitGroup
+		asyncErr error
+		once     sync.Once
+	)
 	for i := 1; i <= nevents; i++ {
 		wg.Add(1)
 		go func(event Event) {
+			defer wg.Done()
+
 			if err := eq.Write(event); err != nil {
-				t.Fatalf("error writing event: %v", err)
+				once.Do(func() {
+					asyncErr = fmt.Errorf("error writing event(%v): %v", event, err)
+				})
 			}
-			wg.Done()
-		}("event-" + fmt.Sprint(i))
+		}(fmt.Sprintf("event-%d", i))
 	}
 
 	wg.Wait()
+
+	if asyncErr != nil {
+		t.Fatalf("expected nil error, got %v", asyncErr)
+	}
+
 	checkClose(t, eq)
 
 	ts.mu.Lock()

retry.go

@@ -151,7 +151,7 @@ func (b *Breaker) Proceed(event Event) time.Duration {
 		return 0
 	}
 
-	return b.last.Add(b.backoff).Sub(time.Now())
+	return time.Until(b.last.Add(b.backoff))
 }
 
 // Success resets the breaker.

retry_test.go

@@ -32,9 +32,13 @@ func testRetryingSink(t *testing.T, strategy RetryStrategy) {
 
 	s := NewRetryingSink(flaky, strategy)
 
-	var wg sync.WaitGroup
+	var (
+		wg       sync.WaitGroup
+		asyncErr error
+		once     sync.Once
+	)
 	for i := 1; i <= nevents; i++ {
-		event := "myevent-" + fmt.Sprint(i)
+		wg.Add(1)
 
 		// Above 50, set the failure rate lower
 		if i > 50 {
@@ -43,16 +47,23 @@ func testRetryingSink(t *testing.T, strategy RetryStrategy) {
 			flaky.mu.Unlock()
 		}
 
-		wg.Add(1)
 		go func(event Event) {
 			defer wg.Done()
+
 			if err := s.Write(event); err != nil {
-				t.Fatalf("error writing event: %v", err)
+				once.Do(func() {
+					asyncErr = fmt.Errorf("error writing event(%v): %v", event, err)
+				})
 			}
-		}(event)
+		}(fmt.Sprintf("event-%d", i))
 	}
 
 	wg.Wait()
+
+	if asyncErr != nil {
+		t.Fatalf("expected nil error, got %v", asyncErr)
+	}
+
 	checkClose(t, s)
 
 	ts.mu.Lock()