ACP-77: Implement RegisterSubnetValidatorTx (#3420)

tests/e2e/p/l1.go

@@ -4,33 +4,58 @@
 package p
 
 import (
+	"context"
+	"errors"
 	"math"
+	"slices"
 	"time"
 
 	"github.com/onsi/ginkgo/v2"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/ava-labs/avalanchego/api/info"
 	"github.com/ava-labs/avalanchego/config"
 	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/peer"
+	"github.com/ava-labs/avalanchego/proto/pb/sdk"
+	"github.com/ava-labs/avalanchego/snow/networking/router"
 	"github.com/ava-labs/avalanchego/tests/fixture/e2e"
 	"github.com/ava-labs/avalanchego/tests/fixture/tmpnet"
+	"github.com/ava-labs/avalanchego/utils"
+	"github.com/ava-labs/avalanchego/utils/buffer"
 	"github.com/ava-labs/avalanchego/utils/constants"
 	"github.com/ava-labs/avalanchego/utils/crypto/bls"
 	"github.com/ava-labs/avalanchego/utils/crypto/secp256k1"
+	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/ava-labs/avalanchego/utils/set"
 	"github.com/ava-labs/avalanchego/utils/units"
 	"github.com/ava-labs/avalanchego/vms/example/xsvm/genesis"
 	"github.com/ava-labs/avalanchego/vms/platformvm"
 	"github.com/ava-labs/avalanchego/vms/platformvm/txs"
+	"github.com/ava-labs/avalanchego/vms/platformvm/warp"
+	"github.com/ava-labs/avalanchego/vms/platformvm/warp/payload"
 	"github.com/ava-labs/avalanchego/vms/secp256k1fx"
 
+	p2pmessage "github.com/ava-labs/avalanchego/message"
+	p2psdk "github.com/ava-labs/avalanchego/network/p2p"
+	p2ppb "github.com/ava-labs/avalanchego/proto/pb/p2p"
 	snowvalidators "github.com/ava-labs/avalanchego/snow/validators"
+	platformvmvalidators "github.com/ava-labs/avalanchego/vms/platformvm/validators"
 	warpmessage "github.com/ava-labs/avalanchego/vms/platformvm/warp/message"
 )
 
 const (
-	genesisWeight  = units.Schmeckle
-	genesisBalance = units.Avax
+	genesisWeight   = units.Schmeckle
+	genesisBalance  = units.Avax
+	registerWeight  = genesisWeight / 10
+	registerBalance = 0
+
+	// Validator registration attempts expire 5 minutes after they are created
+	expiryDelay = 5 * time.Minute
+	// P2P message requests timeout after 10 seconds
+	p2pTimeout = 10 * time.Second
 )
 
 var _ = e2e.DescribePChain("[L1]", func() {
@@ -164,6 +189,22 @@ var _ = e2e.DescribePChain("[L1]", func() {
 		genesisNodePK, err := bls.PublicKeyFromCompressedBytes(genesisNodePoP.PublicKey[:])
 		require.NoError(err)
 
+		tc.By("connecting to the genesis validator")
+		var (
+			networkID           = env.GetNetwork().GetNetworkID()
+			genesisPeerMessages = buffer.NewUnboundedBlockingDeque[p2pmessage.InboundMessage](1)
+		)
+		genesisPeer, err := peer.StartTestPeer(
+			tc.DefaultContext(),
+			subnetGenesisNode.StakingAddress,
+			networkID,
+			router.InboundHandlerFunc(func(_ context.Context, m p2pmessage.InboundMessage) {
+				tc.Outf("received %s %s from %s\n", m.Op(), m.Message(), m.NodeID())
+				genesisPeerMessages.PushRight(m)
+			}),
+		)
+		require.NoError(err)
+
 		address := []byte{}
 		tc.By("issuing a ConvertSubnetTx", func() {
 			_, err := pWallet.IssueConvertSubnetTx(
@@ -227,6 +268,200 @@ var _ = e2e.DescribePChain("[L1]", func() {
 			})
 		})
 
+		advanceProposerVMPChainHeight := func() {
+			// We must wait at least [RecentlyAcceptedWindowTTL] to ensure the
+			// next block will reference the last accepted P-chain height.
+			time.Sleep((5 * platformvmvalidators.RecentlyAcceptedWindowTTL) / 4)
+		}
+		tc.By("advancing the proposervm P-chain height", advanceProposerVMPChainHeight)
+
+		tc.By("creating the validator to register")
+		subnetRegisterNode := e2e.AddEphemeralNode(tc, env.GetNetwork(), tmpnet.FlagsMap{
+			config.TrackSubnetsKey: subnetID.String(),
+		})
+
+		registerNodePoP, err := subnetRegisterNode.GetProofOfPossession()
+		require.NoError(err)
+
+		tc.By("ensuring the subnet nodes are healthy", func() {
+			e2e.WaitForHealthy(tc, subnetGenesisNode)
+			e2e.WaitForHealthy(tc, subnetRegisterNode)
+		})
+
+		tc.By("creating the RegisterSubnetValidatorMessage")
+		expiry := uint64(time.Now().Add(expiryDelay).Unix()) // This message will expire in 5 minutes
+		registerSubnetValidatorMessage, err := warpmessage.NewRegisterSubnetValidator(
+			subnetID,
+			subnetRegisterNode.NodeID,
+			registerNodePoP.PublicKey,
+			expiry,
+			warpmessage.PChainOwner{},
+			warpmessage.PChainOwner{},
+			registerWeight,
+		)
+		require.NoError(err)
+
+		tc.By("registering the validator", func() {
+			tc.By("creating the unsigned warp message")
+			unsignedRegisterSubnetValidator := must[*warp.UnsignedMessage](tc)(warp.NewUnsignedMessage(
+				networkID,
+				chainID,
+				must[*payload.AddressedCall](tc)(payload.NewAddressedCall(
+					address,
+					registerSubnetValidatorMessage.Bytes(),
+				)).Bytes(),
+			))
+
+			tc.By("sending the request to sign the warp message", func() {
+				registerSubnetValidatorRequest, err := wrapWarpSignatureRequest(
+					unsignedRegisterSubnetValidator,
+					nil,
+				)
+				require.NoError(err)
+
+				require.True(genesisPeer.Send(tc.DefaultContext(), registerSubnetValidatorRequest))
+			})
+
+			tc.By("getting the signature response")
+			registerSubnetValidatorSignature, ok, err := findMessage(genesisPeerMessages, unwrapWarpSignature)
+			require.NoError(err)
+			require.True(ok)
+
+			tc.By("creating the signed warp message to register the validator")
+			registerSubnetValidator, err := warp.NewMessage(
+				unsignedRegisterSubnetValidator,
+				&warp.BitSetSignature{
+					Signers: set.NewBits(0).Bytes(), // [signers] has weight from the genesis peer
+					Signature: ([bls.SignatureLen]byte)(
+						bls.SignatureToBytes(registerSubnetValidatorSignature),
+					),
+				},
+			)
+			require.NoError(err)
+
+			tc.By("issuing a RegisterSubnetValidatorTx", func() {
+				_, err := pWallet.IssueRegisterSubnetValidatorTx(
+					registerBalance,
+					registerNodePoP.ProofOfPossession,
+					registerSubnetValidator.Bytes(),
+				)
+				require.NoError(err)
+			})
+		})
+
+		tc.By("verifying the validator was registered", func() {
+			tc.By("verifying the validator set was updated", func() {
+				verifyValidatorSet(map[ids.NodeID]*snowvalidators.GetValidatorOutput{
+					subnetGenesisNode.NodeID: {
+						NodeID:    subnetGenesisNode.NodeID,
+						PublicKey: genesisNodePK,
+						Weight:    genesisWeight,
+					},
+					ids.EmptyNodeID: { // The validator is not active
+						NodeID: ids.EmptyNodeID,
+						Weight: registerWeight,
+					},
+				})
+			})
+		})
+
+		genesisPeerMessages.Close()
+		genesisPeer.StartClose()
+		require.NoError(genesisPeer.AwaitClosed(tc.DefaultContext()))
+
 		_ = e2e.CheckBootstrapIsPossible(tc, env.GetNetwork())
 	})
 })
+
+func wrapWarpSignatureRequest(
+	msg *warp.UnsignedMessage,
+	justification []byte,
+) (p2pmessage.OutboundMessage, error) {
+	p2pMessageFactory, err := p2pmessage.NewCreator(
+		logging.NoLog{},
+		prometheus.NewRegistry(),
+		constants.DefaultNetworkCompressionType,
+		p2pTimeout,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	request := sdk.SignatureRequest{
+		Message:       msg.Bytes(),
+		Justification: justification,
+	}
+	requestBytes, err := proto.Marshal(&request)
+	if err != nil {
+		return nil, err
+	}
+
+	return p2pMessageFactory.AppRequest(
+		msg.SourceChainID,
+		0,
+		time.Hour,
+		p2psdk.PrefixMessage(
+			p2psdk.ProtocolPrefix(p2psdk.SignatureRequestHandlerID),
+			requestBytes,
+		),
+	)
+}
+
+func findMessage[T any](
+	q buffer.BlockingDeque[p2pmessage.InboundMessage],
+	parser func(p2pmessage.InboundMessage) (T, bool, error),
+) (T, bool, error) {
+	var messagesToReprocess []p2pmessage.InboundMessage
+	defer func() {
+		slices.Reverse(messagesToReprocess)
+		for _, msg := range messagesToReprocess {
+			q.PushLeft(msg)
+		}
+	}()
+
+	for {
+		msg, ok := q.PopLeft()
+		if !ok {
+			return utils.Zero[T](), false, nil
+		}
+
+		parsed, ok, err := parser(msg)
+		if err != nil {
+			return utils.Zero[T](), false, err
+		}
+		if ok {
+			return parsed, true, nil
+		}
+
+		messagesToReprocess = append(messagesToReprocess, msg)
+	}
+}
+
+// unwrapWarpSignature assumes the only type of AppResponses that will be
+// received are ACP-118 compliant responses.
+func unwrapWarpSignature(msg p2pmessage.InboundMessage) (*bls.Signature, bool, error) {
+	var appResponse *p2ppb.AppResponse
+	switch msg := msg.Message().(type) {
+	case *p2ppb.AppResponse:
+		appResponse = msg
+	case *p2ppb.AppError:
+		return nil, false, errors.New(msg.ErrorMessage)
+	default:
+		return nil, false, nil
+	}
+
+	var response sdk.SignatureResponse
+	if err := proto.Unmarshal(appResponse.AppBytes, &response); err != nil {
+		return nil, false, err
+	}
+
+	warpSignature, err := bls.SignatureFromBytes(response.Signature)
+	return warpSignature, true, err
+}
+
+func must[T any](t require.TestingT) func(T, error) T {
+	return func(val T, err error) T {
+		require.NoError(t, err)
+		return val
+	}
+}

vms/platformvm/metrics/tx_metrics.go

@@ -145,3 +145,10 @@ func (m *txMetrics) ConvertSubnetTx(*txs.ConvertSubnetTx) error {
 	}).Inc()
 	return nil
 }
+
+func (m *txMetrics) RegisterSubnetValidatorTx(*txs.RegisterSubnetValidatorTx) error {
+	m.numTxs.With(prometheus.Labels{
+		txLabel: "register_subnet_validator",
+	}).Inc()
+	return nil
+}

vms/platformvm/signer/proof_of_possession.go

@@ -14,7 +14,7 @@ import (
 var (
 	_ Signer = (*ProofOfPossession)(nil)
 
-	errInvalidProofOfPossession = errors.New("invalid proof of possession")
+	ErrInvalidProofOfPossession = errors.New("invalid proof of possession")
 )
 
 type ProofOfPossession struct {
@@ -52,7 +52,7 @@ func (p *ProofOfPossession) Verify() error {
 		return err
 	}
 	if !bls.VerifyProofOfPossession(publicKey, signature, p.PublicKey[:]) {
-		return errInvalidProofOfPossession
+		return ErrInvalidProofOfPossession
 	}
 
 	p.publicKey = publicKey

vms/platformvm/signer/proof_of_possession_test.go

@@ -35,7 +35,7 @@ func TestProofOfPossession(t *testing.T) {
 	require.NoError(err)
 	newBLSPOP.ProofOfPossession = blsPOP.ProofOfPossession
 	err = newBLSPOP.Verify()
-	require.ErrorIs(err, errInvalidProofOfPossession)
+	require.ErrorIs(err, ErrInvalidProofOfPossession)
 }
 
 func TestNewProofOfPossessionDeterministic(t *testing.T) {

vms/platformvm/txs/codec.go

@@ -123,5 +123,6 @@ func RegisterDurangoTypes(targetCodec linearcodec.Codec) error {
 func RegisterEtnaTypes(targetCodec linearcodec.Codec) error {
 	return errors.Join(
 		targetCodec.RegisterType(&ConvertSubnetTx{}),
+		targetCodec.RegisterType(&RegisterSubnetValidatorTx{}),
 	)
 }

vms/platformvm/txs/executor/atomic_tx_executor.go

@@ -112,6 +112,10 @@ func (*atomicTxExecutor) ConvertSubnetTx(*txs.ConvertSubnetTx) error {
 	return ErrWrongTxType
 }
 
+func (*atomicTxExecutor) RegisterSubnetValidatorTx(*txs.RegisterSubnetValidatorTx) error {
+	return ErrWrongTxType
+}
+
 func (e *atomicTxExecutor) ImportTx(*txs.ImportTx) error {
 	return e.atomicTx()
 }

vms/platformvm/txs/executor/proposal_tx_executor.go

@@ -131,6 +131,10 @@ func (*proposalTxExecutor) ConvertSubnetTx(*txs.ConvertSubnetTx) error {
 	return ErrWrongTxType
 }
 
+func (*proposalTxExecutor) RegisterSubnetValidatorTx(*txs.RegisterSubnetValidatorTx) error {
+	return ErrWrongTxType
+}
+
 func (e *proposalTxExecutor) AddValidatorTx(tx *txs.AddValidatorTx) error {
 	// AddValidatorTx is a proposal transaction until the Banff fork
 	// activation. Following the activation, AddValidatorTxs must be issued into