Use dependent root to request duties (#15142)

* Use dependent root to request duties * Add actual roots to the event stream * Add feature flag * fix no go * fix test * add test * log duties changes on reorgs * send depdendent root on grpc response * fix wrong return status in tests * fix tests * gazelle * add unil wait for wg * parse slot twice * add slot deadline to update duties * fix conflict * add dependency * lint * Thanks James! * fix segfault * fix rpc tests --------- Co-authored-by: terence tsao <terence@prysmaticlabs.com>
2026-01-09 15:37:56 -05:00 · 2025-04-11 16:22:57 -05:00
parent 5a527a15c8
commit bab898d1d3
35 changed files with 3948 additions and 3311 deletions
--- a/validator/client/validator_test.go
+++ b/validator/client/validator_test.go
@@ -16,6 +16,7 @@ import (
 	"testing"
 	"time"

+	"github.com/OffchainLabs/prysm/v6/api/server/structs"
 	"github.com/OffchainLabs/prysm/v6/async/event"
 	"github.com/OffchainLabs/prysm/v6/cmd/validator/flags"
 	"github.com/OffchainLabs/prysm/v6/config/features"
@@ -2904,3 +2905,106 @@ func TestUpdateValidatorStatusCache(t *testing.T) {
 	// make sure the value is 0
 	assert.Equal(t, 0, len(v.pubkeyToStatus))
 }
+
+func TestValidator_CheckDependentRoots(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	ctx := context.Background()
+	client := validatormock.NewMockValidatorClient(ctrl)
+
+	v := &validator{
+		km:              newMockKeymanager(t, randKeypair(t)),
+		validatorClient: client,
+		duties: &ethpb.ValidatorDutiesContainer{
+			PrevDependentRoot: bytesutil.PadTo([]byte{0x01, 0x02, 0x03}, fieldparams.RootLength),
+			CurrDependentRoot: bytesutil.PadTo([]byte{0x04, 0x05, 0x06}, fieldparams.RootLength),
+			CurrentEpochDuties: []*ethpb.ValidatorDuty{
+				{
+					AttesterSlot:    params.BeaconConfig().SlotsPerEpoch,
+					ValidatorIndex:  200,
+					CommitteeIndex:  100,
+					CommitteeLength: 4,
+					PublicKey:       []byte("testPubKey_1"),
+					ProposerSlots:   []primitives.Slot{params.BeaconConfig().SlotsPerEpoch + 1},
+				},
+			},
+		},
+	}
+
+	t.Run("nil head event", func(t *testing.T) {
+		err := v.checkDependentRoots(ctx, nil)
+		require.ErrorContains(t, "received empty head event", err)
+	})
+
+	t.Run("invalid previous duty dependent root", func(t *testing.T) {
+		head := &structs.HeadEvent{
+			Slot:                      "0",
+			PreviousDutyDependentRoot: "invalid_hex",
+		}
+		err := v.checkDependentRoots(ctx, head)
+		require.ErrorContains(t, "failed to decode previous duty dependent root", err)
+	})
+
+	t.Run("invalid current duty dependent root", func(t *testing.T) {
+		head := &structs.HeadEvent{
+			Slot:                      "0",
+			PreviousDutyDependentRoot: "0x0102030000000000000000000000000000000000000000000000000000000000",
+			CurrentDutyDependentRoot:  "invalid_hex",
+		}
+		err := v.checkDependentRoots(ctx, head)
+		require.ErrorContains(t, "failed to decode current duty dependent root", err)
+	})
+
+	t.Run("update duties for previous root mismatch", func(t *testing.T) {
+		head := &structs.HeadEvent{
+			Slot:                      "1",
+			PreviousDutyDependentRoot: "0xe3f7a1b2c489d56f03a6b8d9c7e1fa2456bb09f3de42a67c8910fc3e7a5d4b12",
+			CurrentDutyDependentRoot:  "0xe3f7a1b2c489d56f03a6b8d9c7e1fa2456bb09f3de42a67c8910fc3e7a5d4b12",
+		}
+		client.EXPECT().SubscribeCommitteeSubnets(
+			gomock.Any(),
+			gomock.Any(),
+			gomock.Any(),
+		).DoAndReturn(func(_ context.Context, _ *ethpb.CommitteeSubnetsSubscribeRequest, _ []*ethpb.ValidatorDuty) (*emptypb.Empty, error) {
+			return nil, nil
+		}).AnyTimes()
+		client.EXPECT().Duties(gomock.Any(), gomock.Any()).Return(v.duties, nil)
+		err := v.checkDependentRoots(ctx, head)
+		require.NoError(t, err)
+	})
+
+	t.Run("update duties for current root mismatch", func(t *testing.T) {
+		head := &structs.HeadEvent{
+			Slot:                      "1",
+			PreviousDutyDependentRoot: "0x0102030000000000000000000000000000000000000000000000000000000000",
+			CurrentDutyDependentRoot:  "0xe3f7a1b2c489d56f03a6b8d9c7e1fa2456bb09f3de42a67c8910fc3e7a5d4b12",
+		}
+		client.EXPECT().Duties(gomock.Any(), gomock.Any()).Return(v.duties, nil)
+		var wg sync.WaitGroup
+		wg.Add(1)
+
+		client.EXPECT().SubscribeCommitteeSubnets(
+			gomock.Any(),
+			gomock.Any(),
+			gomock.Any(),
+		).DoAndReturn(func(_ context.Context, _ *ethpb.CommitteeSubnetsSubscribeRequest, _ []*ethpb.ValidatorDuty) (*emptypb.Empty, error) {
+			wg.Done()
+			return nil, nil
+		}).AnyTimes()
+		err := v.checkDependentRoots(ctx, head)
+		require.NoError(t, err)
+		util.WaitTimeout(&wg, 2*time.Second)
+	})
+	t.Run("no updates needed", func(t *testing.T) {
+		head := &structs.HeadEvent{
+			Slot:                      "0",
+			PreviousDutyDependentRoot: "0x0102030000000000000000000000000000000000000000000000000000000000",
+			CurrentDutyDependentRoot:  "0x0405060000000000000000000000000000000000000000000000000000000000",
+		}
+		curr, err := bytesutil.DecodeHexWithLength(head.CurrentDutyDependentRoot, fieldparams.RootLength)
+		require.NoError(t, err)
+		require.DeepEqual(t, curr, v.duties.CurrDependentRoot)
+		require.NoError(t, v.checkDependentRoots(ctx, head))
+	})
+}