golangci-lint: enable containedctx (#511)

.golangci.yml

@@ -15,6 +15,7 @@ linters:
     - noctx
     - whitespace
     - depguard
+    - containedctx
 linters-settings:
   exhaustive:
     default-signifies-exhaustive: true

Makefile

@@ -15,7 +15,7 @@ install-protoc:
 
 .PHONY: mockery
 mockery: $(mockery) ## Install mockery.
-	go install github.com/vektra/mockery/v2@v2.38.0
+	go install github.com/vektra/mockery/v2@v2.43.0
 
 .PHONY: generate
 generate: mockery install-protoc

pkg/capabilities/consensus/ocr3/capability.go

@@ -216,8 +216,7 @@ func (o *capability) queueRequestForProcessing(
 	i *inputs) (<-chan capabilities.CapabilityResponse, error) {
 	callbackCh := make(chan capabilities.CapabilityResponse, o.callbackChannelBufferSize)
 	r := &request{
-		// TODO: set correct context
-		RequestCtx:          context.Background(),
+		StopCh:              make(chan struct{}),
 		CallbackCh:          callbackCh,
 		WorkflowExecutionID: metadata.WorkflowExecutionID,
 		WorkflowID:          metadata.WorkflowID,
@@ -263,7 +262,7 @@ func (o *capability) handleTransmitMsg(ctx context.Context, resp *outputs) {
 	select {
 	// This should only happen if the client has closed the upstream context.
 	// In this case, the request is cancelled and we shouldn't transmit.
-	case <-req.RequestCtx.Done():
+	case <-req.StopCh:
 	case req.CallbackCh <- resp.CapabilityResponse:
 		close(req.CallbackCh)
 	}

pkg/capabilities/consensus/ocr3/models.go

@@ -1,10 +1,10 @@
 package ocr3
 
 import (
-	"context"
 	"time"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/values"
 )
 
@@ -31,7 +31,7 @@ type request struct {
 	// CallbackCh is a channel to send a response back to the requester
 	// after the request has been processed or timed out.
 	CallbackCh chan capabilities.CapabilityResponse
-	RequestCtx context.Context
+	StopCh     services.StopChan
 
 	WorkflowExecutionID string
 	WorkflowID          string

pkg/capabilities/consensus/ocr3/transmitter_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	pbtypes "github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/ocr3/types"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services/servicetest"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/mocks"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
 	"github.com/smartcontractkit/chainlink-common/pkg/values"
@@ -38,8 +39,7 @@ func TestTransmitter(t *testing.T) {
 		lggr,
 		10,
 	)
-	cp.Start(ctx)
-	defer cp.Close()
+	servicetest.Run(t, cp)
 
 	payload, err := values.NewMap(map[string]any{"observations": []string{"something happened"}})
 	require.NoError(t, err)
@@ -111,8 +111,7 @@ func TestTransmitter_ShouldReportFalse(t *testing.T) {
 		lggr,
 		10,
 	)
-	cp.Start(ctx)
-	defer cp.Close()
+	servicetest.Run(t, cp)
 
 	payload, err := values.NewMap(map[string]any{"observations": []string{"something happened"}})
 	require.NoError(t, err)

pkg/capabilities/pb/capabilities_helpers_test.go

@@ -12,7 +12,7 @@ import (
 )
 
 const (
-	testWorkflowId  = "test-id-1"
+	testWorkflowID  = "test-id-1"
 	testConfigKey   = "test-key"
 	testConfigValue = "test-value"
 	testInputsKey   = "input-key"
@@ -23,7 +23,7 @@ const (
 func TestMarshalUnmarshalRequest(t *testing.T) {
 	req := capabilities.CapabilityRequest{
 		Metadata: capabilities.RequestMetadata{
-			WorkflowExecutionID: testWorkflowId,
+			WorkflowExecutionID: testWorkflowID,
 		},
 		Config: &values.Map{Underlying: map[string]values.Value{
 			testConfigKey: &values.String{Underlying: testConfigValue},

pkg/capabilities/triggers/mercury_remote_aggregator_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	eventId    = "ev_id_1"
+	eventID    = "ev_id_1"
 	timestamp  = 1000
 	rawReport1 = "abcd"
 	rawReport2 = "efgh"
@@ -70,17 +70,17 @@ func TestMercuryRemoteAggregator(t *testing.T) {
 		Signatures:           signatures,
 	}
 
-	node1Resp, err := wrapReports([]datastreams.FeedReport{feed1Old, feed2New}, eventId, 400, datastreams.SignersMetadata{})
+	node1Resp, err := wrapReports([]datastreams.FeedReport{feed1Old, feed2New}, eventID, 400, datastreams.SignersMetadata{})
 	require.NoError(t, err)
 	rawNode1Resp, err := pb.MarshalCapabilityResponse(node1Resp)
 	require.NoError(t, err)
-	node2Resp, err := wrapReports([]datastreams.FeedReport{feed1New, feed2Old}, eventId, 300, datastreams.SignersMetadata{})
+	node2Resp, err := wrapReports([]datastreams.FeedReport{feed1New, feed2Old}, eventID, 300, datastreams.SignersMetadata{})
 	require.NoError(t, err)
 	rawNode2Resp, err := pb.MarshalCapabilityResponse(node2Resp)
 	require.NoError(t, err)
 
 	// aggregator should return latest value for each feedID
-	aggResponse, err := agg.Aggregate(eventId, [][]byte{rawNode1Resp, rawNode2Resp})
+	aggResponse, err := agg.Aggregate(eventID, [][]byte{rawNode1Resp, rawNode2Resp})
 	require.NoError(t, err)
 	aggEvent := capabilities.TriggerEvent{}
 	require.NoError(t, aggResponse.Value.UnwrapTo(&aggEvent))

pkg/codec/encodings/binary/gen/ints_test.go.tmpl

@@ -187,7 +187,7 @@ func TestUint{{.}}(t *testing.T) {
 {{ end }}
 
 func TestBuilderInt(t *testing.T) {
-    {{ range . }}
+    {{ range . -}}
     t.Run("Wraps encoding and decoding for {{.}} bytes", func(t *testing.T) {
         codec, err := bi.Int({{div . 8}})
         require.NoError(t, err)
@@ -252,7 +252,7 @@ func TestBuilderInt(t *testing.T) {
 }
 
 func TestGetUintTypeCodecByByteSize(t *testing.T) {
-    {{ range . }}
+    {{ range . -}}
     t.Run("Wraps encoding and decoding for {{.}} bytes", func(t *testing.T) {
         codec, err := bi.Uint({{div . 8}})
         require.NoError(t, err)

pkg/codec/encodings/binary/int_gen_test.go

@@ -666,7 +666,6 @@ func TestUint64(t *testing.T) {
 }
 
 func TestBuilderInt(t *testing.T) {
-
 	t.Run("Wraps encoding and decoding for 8 bytes", func(t *testing.T) {
 		codec, err := bi.Int(1)
 		require.NoError(t, err)
@@ -697,7 +696,6 @@ func TestBuilderInt(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, size, 1)
 	})
-
 	t.Run("Wraps encoding and decoding for 16 bytes", func(t *testing.T) {
 		codec, err := bi.Int(2)
 		require.NoError(t, err)
@@ -728,7 +726,6 @@ func TestBuilderInt(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, size, 2)
 	})
-
 	t.Run("Wraps encoding and decoding for 32 bytes", func(t *testing.T) {
 		codec, err := bi.Int(4)
 		require.NoError(t, err)
@@ -759,7 +756,6 @@ func TestBuilderInt(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, size, 4)
 	})
-
 	t.Run("Wraps encoding and decoding for 64 bytes", func(t *testing.T) {
 		codec, err := bi.Int(8)
 		require.NoError(t, err)
@@ -823,7 +819,6 @@ func TestBuilderInt(t *testing.T) {
 }
 
 func TestGetUintTypeCodecByByteSize(t *testing.T) {
-
 	t.Run("Wraps encoding and decoding for 8 bytes", func(t *testing.T) {
 		codec, err := bi.Uint(1)
 		require.NoError(t, err)
@@ -854,7 +849,6 @@ func TestGetUintTypeCodecByByteSize(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, size, 1)
 	})
-
 	t.Run("Wraps encoding and decoding for 16 bytes", func(t *testing.T) {
 		codec, err := bi.Uint(2)
 		require.NoError(t, err)
@@ -885,7 +879,6 @@ func TestGetUintTypeCodecByByteSize(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, size, 2)
 	})
-
 	t.Run("Wraps encoding and decoding for 32 bytes", func(t *testing.T) {
 		codec, err := bi.Uint(4)
 		require.NoError(t, err)
@@ -916,7 +909,6 @@ func TestGetUintTypeCodecByByteSize(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, size, 4)
 	})
-
 	t.Run("Wraps encoding and decoding for 64 bytes", func(t *testing.T) {
 		codec, err := bi.Uint(8)
 		require.NoError(t, err)

pkg/codec/encodings/type_codec_test.go

@@ -16,8 +16,9 @@ import (
 	encodingtestutils "github.com/smartcontractkit/chainlink-common/pkg/codec/encodings/testutils"
 	"github.com/smartcontractkit/chainlink-common/pkg/loop/testutils"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
-	. "github.com/smartcontractkit/chainlink-common/pkg/types/interfacetests"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
+
+	. "github.com/smartcontractkit/chainlink-common/pkg/types/interfacetests" //nolint
 )
 
 // TestCodecFromTypeCodecs tests all the codecs in packages under pkg/codec/encodings

pkg/loop/internal/relayer/pluginprovider/chainreader/chain_reader_test.go

@@ -20,9 +20,10 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/loop/internal/relayer/pluginprovider/chainreader"
 	"github.com/smartcontractkit/chainlink-common/pkg/loop/internal/relayer/pluginprovider/chainreader/test"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
-	. "github.com/smartcontractkit/chainlink-common/pkg/types/interfacetests"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/query"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
+
+	. "github.com/smartcontractkit/chainlink-common/pkg/types/interfacetests" //nolint
 )
 
 func TestVersionedBytesFunctions(t *testing.T) {

pkg/loop/internal/relayer/pluginprovider/chainreader/helper_test.go

@@ -7,9 +7,10 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
-	. "github.com/smartcontractkit/chainlink-common/pkg/types/interfacetests"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/query"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives"
+
+	. "github.com/smartcontractkit/chainlink-common/pkg/types/interfacetests" //nolint
 )
 
 var errorTypes = []error{

pkg/monitoring/exporter_kafka_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 
@@ -17,7 +18,8 @@ func TestKafkaExporter(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond)
 		defer cancel()
 		log := newNullLogger()
-		producer := fakeProducer{make(chan producerMessage), ctx}
+		producer := fakeProducer{make(chan producerMessage), make(chan struct{})}
+		defer func() { assert.NoError(t, producer.Close()) }()
 		cfg := config.Config{}
 		cfg.Kafka.TransmissionTopic = "transmissions"
 		cfg.Kafka.ConfigSetSimplifiedTopic = "config-set-simplified"

pkg/monitoring/feed_monitor_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
@@ -60,7 +61,8 @@ func TestFeedMonitor(t *testing.T) {
 			poller2.Run(ctx)
 		})
 
-		producer := fakeProducer{make(chan producerMessage), ctx}
+		producer := fakeProducer{make(chan producerMessage), make(chan struct{})}
+		defer func() { assert.NoError(t, producer.Close()) }()
 
 		transmissionSchema := fakeSchema{transmissionCodec, SubjectFromTopic(cfg.Kafka.TransmissionTopic)}
 		configSetSimplifiedSchema := fakeSchema{configSetSimplifiedCodec, SubjectFromTopic(cfg.Kafka.ConfigSetSimplifiedTopic)}

pkg/monitoring/http.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils"
 )
 
@@ -17,13 +18,14 @@ type HTTPServer interface {
 	Run(ctx context.Context)
 }
 
-func NewHTTPServer(baseCtx context.Context, addr string, log Logger) HTTPServer {
+func NewHTTPServer(stopCh services.StopRChan, addr string, log Logger) HTTPServer {
 	mux := http.NewServeMux()
 	srv := &http.Server{
 		Addr:    addr,
 		Handler: mux,
-		BaseContext: func(_ net.Listener) context.Context {
-			return baseCtx
+		BaseContext: func(_ net.Listener) (ctx context.Context) {
+			ctx, _ = stopCh.NewCtx()
+			return
 		},
 		ReadHeaderTimeout: 60 * time.Second,
 	}

pkg/monitoring/manager_benchmark_test.go

@@ -5,6 +5,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/monitoring/config"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils"
 )
@@ -48,7 +50,8 @@ func BenchmarkManager(b *testing.B) {
 
 	transmissionSchema := fakeSchema{transmissionCodec, SubjectFromTopic(cfg.Kafka.TransmissionTopic)}
 
-	producer := fakeProducer{make(chan producerMessage), ctx}
+	producer := fakeProducer{make(chan producerMessage), make(chan struct{})}
+	defer func() { assert.NoError(b, producer.Close()) }()
 	factory := &fakeRandomDataSourceFactory{make(chan interface{})}
 
 	prometheusExporterFactory := NewPrometheusExporterFactory(