ACP-77: Implement RegisterSubnetValidatorTx

tests/e2e/p/permissionless_layer_one.go

@@ -4,33 +4,53 @@
 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/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"
 	platformvmsdk "github.com/ava-labs/avalanchego/vms/platformvm"
+	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
 )
 
 var _ = e2e.DescribePChain("[Permissionless L1]", func() {
@@ -135,6 +155,27 @@ var _ = e2e.DescribePChain("[Permissionless 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)
+		defer func() {
+			genesisPeerMessages.Close()
+			genesisPeer.StartClose()
+			require.NoError(genesisPeer.AwaitClosed(tc.DefaultContext()))
+		}()
+
 		address := []byte{}
 		tc.By("issuing a ConvertSubnetTx", func() {
 			_, err := pWallet.IssueConvertSubnetTx(
@@ -205,5 +246,207 @@ var _ = e2e.DescribePChain("[Permissionless L1]", func() {
 				})
 			})
 		})
+
+		advanceProposerVMPChainHeight := func() {
+			// We first must wait at least [RecentlyAcceptedWindowTTL] to ensure
+			// the next block will evict the prior block from the windower.
+			time.Sleep((5 * platformvmvalidators.RecentlyAcceptedWindowTTL) / 4)
+
+			// Now we must:
+			// 1. issue a block which should include the old P-chain height.
+			// 2. issue a block which should include the new P-chain height.
+			for range 2 {
+				_, err = pWallet.IssueBaseTx(nil, tc.WithDefaultContext())
+				require.NoError(err)
+			}
+			// Now that a block has been issued with the new P-chain height, the
+			// next block will use that height for warp message verification.
+		}
+		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")
+		registerSubnetValidatorMessage, err := warpmessage.NewRegisterSubnetValidator(
+			subnetID,
+			subnetRegisterNode.NodeID,
+			registerNodePoP.PublicKey,
+			uint64(time.Now().Add(5*time.Minute).Unix()),
+			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")
+			signers := set.NewBits()
+			signers.Add(0) // [signers] has weight from the genesis peer
+
+			var sigBytes [bls.SignatureLen]byte
+			copy(sigBytes[:], bls.SignatureToBytes(registerSubnetValidatorSignature))
+			registerSubnetValidator, err := warp.NewMessage(
+				unsignedRegisterSubnetValidator,
+				&warp.BitSetSignature{
+					Signers:   signers.Bytes(),
+					Signature: sigBytes,
+				},
+			)
+			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,
+					},
+				})
+			})
+		})
 	})
 })
+
+func wrapWarpSignatureRequest(
+	msg *warp.UnsignedMessage,
+	justification []byte,
+) (p2pmessage.OutboundMessage, error) {
+	p2pMessageFactory, err := p2pmessage.NewCreator(
+		logging.NoLog{},
+		prometheus.NewRegistry(),
+		constants.DefaultNetworkCompressionType,
+		10*time.Second,
+	)
+	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

@@ -132,16 +132,23 @@ func (m *txMetrics) TransferSubnetOwnershipTx(*txs.TransferSubnetOwnershipTx) er
 	return nil
 }
 
+func (m *txMetrics) BaseTx(*txs.BaseTx) error {
+	m.numTxs.With(prometheus.Labels{
+		txLabel: "base",
+	}).Inc()
+	return nil
+}
+
 func (m *txMetrics) ConvertSubnetTx(*txs.ConvertSubnetTx) error {
 	m.numTxs.With(prometheus.Labels{
 		txLabel: "convert_subnet",
 	}).Inc()
 	return nil
 }
 
-func (m *txMetrics) BaseTx(*txs.BaseTx) error {
+func (m *txMetrics) RegisterSubnetValidatorTx(*txs.RegisterSubnetValidatorTx) error {
 	m.numTxs.With(prometheus.Labels{
-		txLabel: "base",
+		txLabel: "register_subnet_validator",
 	}).Inc()
 	return nil
 }

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

@@ -66,15 +66,15 @@ func (*AtomicTxExecutor) TransformSubnetTx(*txs.TransformSubnetTx) error {
 	return ErrWrongTxType
 }
 
-func (*AtomicTxExecutor) TransferSubnetOwnershipTx(*txs.TransferSubnetOwnershipTx) error {
+func (*AtomicTxExecutor) AddPermissionlessValidatorTx(*txs.AddPermissionlessValidatorTx) error {
 	return ErrWrongTxType
 }
 
-func (*AtomicTxExecutor) AddPermissionlessValidatorTx(*txs.AddPermissionlessValidatorTx) error {
+func (*AtomicTxExecutor) AddPermissionlessDelegatorTx(*txs.AddPermissionlessDelegatorTx) error {
 	return ErrWrongTxType
 }
 
-func (*AtomicTxExecutor) AddPermissionlessDelegatorTx(*txs.AddPermissionlessDelegatorTx) error {
+func (*AtomicTxExecutor) TransferSubnetOwnershipTx(*txs.TransferSubnetOwnershipTx) error {
 	return ErrWrongTxType
 }
 
@@ -86,6 +86,10 @@ func (*AtomicTxExecutor) ConvertSubnetTx(*txs.ConvertSubnetTx) error {
 	return ErrWrongTxType
 }
 
+func (*AtomicTxExecutor) RegisterSubnetValidatorTx(*txs.RegisterSubnetValidatorTx) error {
+	return ErrWrongTxType
+}
+
 func (e *AtomicTxExecutor) ImportTx(tx *txs.ImportTx) error {
 	return e.atomicTx(tx)
 }

vms/platformvm/txs/executor/proposal_tx_executor.go

@@ -103,6 +103,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