Add regression test for PR 15369 (#15379)

* Update go.uber.org/mock to v0.5.2.

* Regression test for #15369

* Changelog fragment

* Use SetActiveTestCleanup

* Remove logrus log level change

changelog/pvl-regression-15369.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Added regression test for [PR 15369](https://github.com/OffchainLabs/prysm/pull/15369)

deps.bzl

@@ -4894,8 +4894,8 @@ def prysm_deps():
     go_repository(
         name = "org_uber_go_mock",
         importpath = "go.uber.org/mock",
-        sum = "h1:VcM4ZOtdbR4f6VXfiOpwpVJDL6lCReaZ6mw31wqh7KU=",
-        version = "v0.4.0",
+        sum = "h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko=",
+        version = "v0.5.2",
     )
     go_repository(
         name = "org_uber_go_multierr",

go.mod

@@ -87,7 +87,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.34.0
 	go.opentelemetry.io/otel/trace v1.35.0
 	go.uber.org/automaxprocs v1.5.2
-	go.uber.org/mock v0.4.0
+	go.uber.org/mock v0.5.2
 	golang.org/x/crypto v0.36.0
 	golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa
 	golang.org/x/sync v0.12.0

go.sum

@@ -1127,8 +1127,8 @@ go.uber.org/fx v1.22.2/go.mod h1:o/D9n+2mLP6v1EG+qsdT1O8wKopYAsqZasju97SDFCU=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-go.uber.org/mock v0.4.0 h1:VcM4ZOtdbR4f6VXfiOpwpVJDL6lCReaZ6mw31wqh7KU=
-go.uber.org/mock v0.4.0/go.mod h1:a6FSlNadKUHUa9IP5Vyt1zh4fC7uAwxMutEAscFbkZc=
+go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko=
+go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=

validator/client/runner_test.go

@@ -2,22 +2,36 @@ package client
 
 import (
 	"context"
+	"fmt"
 	"math/bits"
+	"math/rand"
+	"runtime/debug"
+	"strconv"
 	"testing"
 	"time"
 
 	"github.com/OffchainLabs/prysm/v6/api/client/beacon/health"
 	"github.com/OffchainLabs/prysm/v6/async/event"
+	"github.com/OffchainLabs/prysm/v6/cache/lru"
 	fieldparams "github.com/OffchainLabs/prysm/v6/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/config/proposer"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
+	"github.com/OffchainLabs/prysm/v6/encoding/bytesutil"
+	ethpb "github.com/OffchainLabs/prysm/v6/proto/prysm/v1alpha1"
 	"github.com/OffchainLabs/prysm/v6/testing/assert"
 	"github.com/OffchainLabs/prysm/v6/testing/require"
+	"github.com/OffchainLabs/prysm/v6/testing/util"
+	validatormock "github.com/OffchainLabs/prysm/v6/testing/validator-mock"
+	"github.com/OffchainLabs/prysm/v6/time/slots"
 	"github.com/OffchainLabs/prysm/v6/validator/client/iface"
 	"github.com/OffchainLabs/prysm/v6/validator/client/testutil"
+	testing2 "github.com/OffchainLabs/prysm/v6/validator/db/testing"
+	"github.com/OffchainLabs/prysm/v6/validator/keymanager/local"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/go-bitfield"
+	"github.com/sirupsen/logrus"
 	logTest "github.com/sirupsen/logrus/hooks/test"
 	"go.uber.org/mock/gomock"
 )
@@ -366,3 +380,231 @@ func TestUpdateProposerSettingsAt_EpochEndOk(t *testing.T) {
 	// can't test "Failed to update proposer settings" because of log.fatal
 	assert.LogsContain(t, hook, "Mock updated proposer settings")
 }
+
+type tlogger struct {
+	t testing.TB
+}
+
+func (t tlogger) Write(p []byte) (n int, err error) {
+	t.t.Log(fmt.Sprintf("%s", p))
+	return len(p), nil
+}
+
+func delay(t testing.TB) {
+	const timeout = 100 * time.Millisecond
+
+	select {
+	case <-t.Context().Done():
+		return
+	case <-time.After(timeout):
+		return
+	}
+}
+
+// assertValidContext, but only when the parent context is still valid. This is testing that mocked methods are called
+// and maintain a valid context while processing, except when the test is shutting down.
+func assertValidContext(t testing.TB, parent, ctx context.Context) {
+	if ctx.Err() != nil && parent.Err() == nil && t.Context().Err() == nil {
+		t.Logf("stack: %s", debug.Stack())
+		t.Fatalf("Context is no longer valid during a mocked RPC call: %v", ctx.Err())
+	}
+}
+
+func TestRunnerPushesProposerSettings_ValidContext(t *testing.T) {
+	logrus.SetOutput(tlogger{t})
+
+	cfg := params.BeaconConfig()
+	cfg.SecondsPerSlot = 1
+	params.SetActiveTestCleanup(t, cfg)
+
+	timedCtx, cancel := context.WithTimeout(t.Context(), 1*time.Minute)
+	defer cancel()
+
+	// This test is meant to ensure that PushProposerSettings is called successfully on a next slot event.
+	// This is a regresion test for PR 15369, however the same methodology of context checking is applied
+	// to many other methods as well.
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	// We want to test that mocked methods are called with a live context, but only while the timed context is valid.
+	liveCtx := gomock.Cond(func(ctx context.Context) bool { return ctx.Err() == nil || timedCtx.Err() != nil })
+	// Mocked client(s) setup.
+	vcm := validatormock.NewMockValidatorClient(ctrl)
+	vcm.EXPECT().WaitForChainStart(liveCtx, gomock.Any()).Return(&ethpb.ChainStartResponse{
+		GenesisTime: uint64(time.Now().Unix()) - params.BeaconConfig().SecondsPerSlot,
+	}, nil)
+	vcm.EXPECT().MultipleValidatorStatus(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.MultipleValidatorStatusRequest) (*ethpb.MultipleValidatorStatusResponse, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		res := &ethpb.MultipleValidatorStatusResponse{}
+		for i, pk := range req.PublicKeys {
+			res.PublicKeys = append(res.PublicKeys, pk)
+			res.Statuses = append(res.Statuses, &ethpb.ValidatorStatusResponse{Status: ethpb.ValidatorStatus_ACTIVE})
+			res.Indices = append(res.Indices, primitives.ValidatorIndex(i))
+		}
+		return res, nil
+	}).AnyTimes()
+	vcm.EXPECT().Duties(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.DutiesRequest) (*ethpb.ValidatorDutiesContainer, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+
+		s := slots.UnsafeEpochStart(req.Epoch)
+		res := &ethpb.ValidatorDutiesContainer{}
+		for i, pk := range req.PublicKeys {
+			var ps []primitives.Slot
+			if i < int(params.BeaconConfig().SlotsPerEpoch) {
+				ps = []primitives.Slot{s + primitives.Slot(i)}
+			}
+			res.CurrentEpochDuties = append(res.CurrentEpochDuties, &ethpb.ValidatorDuty{
+				CommitteeLength:  uint64(len(req.PublicKeys)),
+				CommitteeIndex:   0,
+				AttesterSlot:     s + primitives.Slot(i)%params.BeaconConfig().SlotsPerEpoch,
+				ProposerSlots:    ps,
+				PublicKey:        pk,
+				Status:           ethpb.ValidatorStatus_ACTIVE,
+				ValidatorIndex:   primitives.ValidatorIndex(i),
+				IsSyncCommittee:  i%5 == 0,
+				CommitteesAtSlot: 1,
+			})
+			res.NextEpochDuties = append(res.NextEpochDuties, &ethpb.ValidatorDuty{
+				CommitteeLength:  uint64(len(req.PublicKeys)),
+				CommitteeIndex:   0,
+				AttesterSlot:     s + primitives.Slot(i)%params.BeaconConfig().SlotsPerEpoch + params.BeaconConfig().SlotsPerEpoch,
+				ProposerSlots:    ps,
+				PublicKey:        pk,
+				Status:           ethpb.ValidatorStatus_ACTIVE,
+				ValidatorIndex:   primitives.ValidatorIndex(i),
+				IsSyncCommittee:  i%7 == 0,
+				CommitteesAtSlot: 1,
+			})
+		}
+		return res, nil
+	}).AnyTimes()
+	vcm.EXPECT().PrepareBeaconProposer(liveCtx, gomock.Any()).Return(nil, nil).AnyTimes().Do(func(ctx context.Context, _ any) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+	})
+	vcm.EXPECT().EventStreamIsRunning().Return(true).AnyTimes().Do(func() { delay(t) })
+	vcm.EXPECT().SubmitValidatorRegistrations(liveCtx, gomock.Any()).Do(func(ctx context.Context, _ any) {
+		defer assertValidContext(t, timedCtx, ctx) // This is the specific regression test assertion for PR 15369.
+		delay(t)
+	}).MinTimes(1)
+	// DomainData calls are really fast, no delay needed.
+	vcm.EXPECT().DomainData(liveCtx, gomock.Any()).Return(&ethpb.DomainResponse{SignatureDomain: make([]byte, 32)}, nil).AnyTimes()
+	vcm.EXPECT().SubscribeCommitteeSubnets(liveCtx, gomock.Any(), gomock.Any()).AnyTimes().Do(func(_, _, _ any) { delay(t) })
+	vcm.EXPECT().AttestationData(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.AttestationDataRequest) (*ethpb.AttestationData, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		r := rand.New(rand.NewSource(123))
+		root := bytesutil.PadTo([]byte("root_"+strconv.Itoa(r.Intn(100_000))), 32)
+		root2 := bytesutil.PadTo([]byte("root_"+strconv.Itoa(r.Intn(100_000))), 32)
+		ckpt := &ethpb.Checkpoint{Root: root2, Epoch: slots.ToEpoch(req.Slot)}
+		return &ethpb.AttestationData{
+			Slot:            req.Slot,
+			CommitteeIndex:  req.CommitteeIndex,
+			BeaconBlockRoot: root,
+			Target:          ckpt,
+			Source:          ckpt,
+		}, nil
+	}).AnyTimes()
+	vcm.EXPECT().ProposeAttestation(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.Attestation) (*ethpb.AttestResponse, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		return &ethpb.AttestResponse{AttestationDataRoot: make([]byte, fieldparams.RootLength)}, nil
+	}).AnyTimes()
+	vcm.EXPECT().SubmitAggregateSelectionProof(liveCtx, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.AggregateSelectionRequest, index primitives.ValidatorIndex, committeeLength uint64) (*ethpb.AggregateSelectionResponse, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		ckpt := &ethpb.Checkpoint{Root: make([]byte, fieldparams.RootLength)}
+		return &ethpb.AggregateSelectionResponse{
+			AggregateAndProof: &ethpb.AggregateAttestationAndProof{
+				AggregatorIndex: index,
+				Aggregate: &ethpb.Attestation{
+					Data:            &ethpb.AttestationData{Slot: req.Slot, BeaconBlockRoot: make([]byte, fieldparams.RootLength), Source: ckpt, Target: ckpt},
+					AggregationBits: bitfield.Bitlist{0b00011111},
+					Signature:       make([]byte, fieldparams.BLSSignatureLength),
+				},
+				SelectionProof: make([]byte, fieldparams.BLSSignatureLength),
+			},
+		}, nil
+	}).AnyTimes()
+	vcm.EXPECT().SubmitSignedAggregateSelectionProof(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.SignedAggregateSubmitRequest) (*ethpb.SignedAggregateSubmitResponse, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		return &ethpb.SignedAggregateSubmitResponse{AttestationDataRoot: make([]byte, fieldparams.RootLength)}, nil
+	}).AnyTimes()
+	vcm.EXPECT().BeaconBlock(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.BlockRequest) (*ethpb.GenericBeaconBlock, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		return &ethpb.GenericBeaconBlock{Block: &ethpb.GenericBeaconBlock_Electra{Electra: &ethpb.BeaconBlockContentsElectra{Block: util.HydrateBeaconBlockElectra(&ethpb.BeaconBlockElectra{})}}}, nil
+	}).AnyTimes()
+	vcm.EXPECT().ProposeBeaconBlock(liveCtx, gomock.Any()).AnyTimes().DoAndReturn(func(ctx context.Context, req *ethpb.GenericSignedBeaconBlock) (*ethpb.ProposeResponse, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		return &ethpb.ProposeResponse{BlockRoot: make([]byte, fieldparams.RootLength)}, nil
+	})
+	vcm.EXPECT().SyncSubcommitteeIndex(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.SyncSubcommitteeIndexRequest) (*ethpb.SyncSubcommitteeIndexResponse, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		//delay(t)
+		return &ethpb.SyncSubcommitteeIndexResponse{Indices: []primitives.CommitteeIndex{0}}, nil
+	}).AnyTimes()
+	vcm.EXPECT().SyncMessageBlockRoot(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, _ any) (*ethpb.SyncMessageBlockRootResponse, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		return &ethpb.SyncMessageBlockRootResponse{Root: make([]byte, fieldparams.RootLength)}, nil
+	}).AnyTimes()
+	vcm.EXPECT().SubmitSyncMessage(liveCtx, gomock.Any()).Do(func(ctx context.Context, _ any) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+	}).AnyTimes()
+	vcm.EXPECT().SyncCommitteeContribution(liveCtx, gomock.Any()).DoAndReturn(func(ctx context.Context, req *ethpb.SyncCommitteeContributionRequest) (*ethpb.SyncCommitteeContribution, error) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+		bits := bitfield.NewBitvector128()
+		bits.SetBitAt(0, true)
+		return &ethpb.SyncCommitteeContribution{Slot: req.Slot, BlockRoot: make([]byte, fieldparams.RootLength), SubcommitteeIndex: req.SubnetId, AggregationBits: bits, Signature: make([]byte, fieldparams.BLSSignatureLength)}, nil
+	}).AnyTimes()
+	vcm.EXPECT().SubmitSignedContributionAndProof(liveCtx, gomock.Any()).Do(func(ctx context.Context, _ any) {
+		defer assertValidContext(t, timedCtx, ctx)
+		delay(t)
+	}).AnyTimes()
+	hcm := health.NewMockHealthClient(ctrl)
+	hcm.EXPECT().IsHealthy(liveCtx).Return(true).AnyTimes().Do(func(_ any) { delay(t) })
+	ncm := validatormock.NewMockNodeClient(ctrl)
+	ncm.EXPECT().SyncStatus(liveCtx, gomock.Any()).Return(&ethpb.SyncStatus{Syncing: false}, nil)
+	ncm.EXPECT().HealthTracker().Return(health.NewTracker(hcm)).AnyTimes()
+	ccm := validatormock.NewMockChainClient(ctrl)
+	ccm.EXPECT().ChainHead(liveCtx, gomock.Any()).Return(&ethpb.ChainHead{}, nil).Do(func(_, _ any) { delay(t) })
+
+	// Setup the actual validator service.
+	v := &validator{
+		validatorClient: vcm,
+		nodeClient:      ncm,
+		chainClient:     ccm,
+		db:              testing2.SetupDB(t, t.TempDir(), [][fieldparams.BLSPubkeyLength]byte{}, false),
+		interopKeysConfig: &local.InteropKeymanagerConfig{
+			NumValidatorKeys: uint64(params.BeaconConfig().SlotsPerEpoch) * 4, // 4 Attesters per slot.
+		},
+		proposerSettings: &proposer.Settings{
+			ProposeConfig: make(map[[fieldparams.BLSPubkeyLength]byte]*proposer.Option),
+			DefaultConfig: &proposer.Option{
+				FeeRecipientConfig: &proposer.FeeRecipientConfig{
+					FeeRecipient: common.BytesToAddress([]byte{1}),
+				},
+				BuilderConfig: &proposer.BuilderConfig{
+					Enabled:  true,
+					GasLimit: 60_000_000,
+					Relays:   []string{"https://example.com"},
+				},
+				GraffitiConfig: &proposer.GraffitiConfig{
+					Graffiti: "foobar",
+				},
+			},
+		},
+		signedValidatorRegistrations:   make(map[[fieldparams.BLSPubkeyLength]byte]*ethpb.SignedValidatorRegistrationV1),
+		slotFeed:                       &event.Feed{},
+		aggregatedSlotCommitteeIDCache: lru.New(100),
+		submittedAtts:                  make(map[submittedAttKey]*submittedAtt),
+		submittedAggregates:            make(map[submittedAttKey]*submittedAtt),
+	}
+
+	run(timedCtx, v)
+}

validator/db/testing/setup_db.go

@@ -1,7 +1,6 @@
 package testing
 
 import (
-	"context"
 	"testing"
 
 	fieldparams "github.com/OffchainLabs/prysm/v6/config/fieldparams"
@@ -25,7 +24,7 @@ func SetupDB(t testing.TB, dataPath string, pubkeys [][fieldparams.BLSPubkeyLeng
 		db, err = filesystem.NewStore(dataPath, config)
 	} else {
 		config := &kv.Config{PubKeys: pubkeys}
-		db, err = kv.NewKVStore(context.Background(), dataPath, config)
+		db, err = kv.NewKVStore(t.Context(), dataPath, config)
 	}
 
 	if err != nil {