Signature Request Handler (#459)

Co-authored-by: aaronbuchwald <aaron.buchwald56@gmail.com>
Co-authored-by: Darioush Jalali <darioush.jalali@avalabs.org>

handlers/handler.go

@@ -0,0 +1,61 @@
+// (c) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package handlers
+
+import (
+	"context"
+
+	"github.com/ava-labs/avalanchego/codec"
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/subnet-evm/handlers/stats"
+	warpHandlers "github.com/ava-labs/subnet-evm/handlers/warp"
+	"github.com/ava-labs/subnet-evm/metrics"
+	"github.com/ava-labs/subnet-evm/plugin/evm/message"
+	"github.com/ava-labs/subnet-evm/sync/handlers"
+	syncHandlers "github.com/ava-labs/subnet-evm/sync/handlers"
+	"github.com/ava-labs/subnet-evm/trie"
+)
+
+var _ message.RequestHandler = &networkHandler{}
+
+type networkHandler struct {
+	stateTrieLeafsRequestHandler *syncHandlers.LeafsRequestHandler
+	blockRequestHandler          *syncHandlers.BlockRequestHandler
+	codeRequestHandler           *syncHandlers.CodeRequestHandler
+	signatureRequestHandler      warpHandlers.SignatureRequestHandler
+}
+
+// NewNetworkHandler constructs the handler for serving network requests.
+func NewNetworkHandler(
+	provider handlers.SyncDataProvider,
+	evmTrieDB *trie.Database,
+	networkCodec codec.Manager,
+) message.RequestHandler {
+	handlerStats := stats.NewHandlerStats(metrics.Enabled)
+	return &networkHandler{
+		// State sync handlers
+		stateTrieLeafsRequestHandler: syncHandlers.NewLeafsRequestHandler(evmTrieDB, provider, networkCodec, handlerStats),
+		blockRequestHandler:          syncHandlers.NewBlockRequestHandler(provider, networkCodec, handlerStats),
+		codeRequestHandler:           syncHandlers.NewCodeRequestHandler(evmTrieDB.DiskDB(), networkCodec, handlerStats),
+
+		// TODO: initialize actual signature request handler when warp is ready
+		signatureRequestHandler: &warpHandlers.NoopSignatureRequestHandler{},
+	}
+}
+
+func (n networkHandler) HandleTrieLeafsRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, leafsRequest message.LeafsRequest) ([]byte, error) {
+	return n.stateTrieLeafsRequestHandler.OnLeafsRequest(ctx, nodeID, requestID, leafsRequest)
+}
+
+func (n networkHandler) HandleBlockRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, blockRequest message.BlockRequest) ([]byte, error) {
+	return n.blockRequestHandler.OnBlockRequest(ctx, nodeID, requestID, blockRequest)
+}
+
+func (n networkHandler) HandleCodeRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, codeRequest message.CodeRequest) ([]byte, error) {
+	return n.codeRequestHandler.OnCodeRequest(ctx, nodeID, requestID, codeRequest)
+}
+
+func (n networkHandler) HandleSignatureRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, signatureRequest message.SignatureRequest) ([]byte, error) {
+	return n.signatureRequestHandler.OnSignatureRequest(ctx, nodeID, requestID, signatureRequest)
+}

handlers/stats/mock_stats.go

@@ -0,0 +1,22 @@
+// (c) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package stats
+
+import (
+	warpStats "github.com/ava-labs/subnet-evm/handlers/warp/stats"
+	syncStats "github.com/ava-labs/subnet-evm/sync/handlers/stats"
+)
+
+var _ HandlerStats = &MockHandlerStats{}
+
+// MockHandlerStats is mock for capturing and asserting on handler metrics in test
+type MockHandlerStats struct {
+	syncStats.MockHandlerStats
+	warpStats.MockSignatureRequestHandlerStats
+}
+
+func (m *MockHandlerStats) Reset() {
+	m.MockHandlerStats.Reset()
+	m.MockSignatureRequestHandlerStats.Reset()
+}

handlers/stats/stats.go

@@ -0,0 +1,38 @@
+// (c) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package stats
+
+import (
+	warpStats "github.com/ava-labs/subnet-evm/handlers/warp/stats"
+	syncStats "github.com/ava-labs/subnet-evm/sync/handlers/stats"
+)
+
+var (
+	_ HandlerStats = &handlerStats{}
+	_ HandlerStats = &MockHandlerStats{}
+)
+
+// HandlerStats reports prometheus metrics for the network handlers
+type HandlerStats interface {
+	warpStats.SignatureRequestHandlerStats
+	syncStats.HandlerStats
+}
+
+type handlerStats struct {
+	// State sync handler metrics
+	syncStats.HandlerStats
+
+	// Warp handler metrics
+	warpStats.SignatureRequestHandlerStats
+}
+
+func NewHandlerStats(enabled bool) HandlerStats {
+	if !enabled {
+		return &MockHandlerStats{}
+	}
+	return &handlerStats{
+		HandlerStats:                 syncStats.NewHandlerStats(enabled),
+		SignatureRequestHandlerStats: warpStats.NewStats(enabled),
+	}
+}

handlers/warp/signature_request.go

@@ -0,0 +1,75 @@
+// (c) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package warp
+
+import (
+	"context"
+	"time"
+
+	"github.com/ava-labs/avalanchego/codec"
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/subnet-evm/handlers/warp/stats"
+	"github.com/ava-labs/subnet-evm/plugin/evm/message"
+	"github.com/ava-labs/subnet-evm/plugin/evm/warp"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+// SignatureRequestHandler is a peer.RequestHandler for message.SignatureRequest
+// serving requested BLS signature data
+type SignatureRequestHandler interface {
+	OnSignatureRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, signatureRequest message.SignatureRequest) ([]byte, error)
+}
+
+// signatureRequestHandler implements the SignatureRequestHandler interface
+type signatureRequestHandler struct {
+	backend warp.WarpBackend
+	codec   codec.Manager
+	stats   stats.SignatureRequestHandlerStats
+}
+
+func NewSignatureRequestHandler(backend warp.WarpBackend, codec codec.Manager, stats stats.SignatureRequestHandlerStats) SignatureRequestHandler {
+	return &signatureRequestHandler{
+		backend: backend,
+		codec:   codec,
+		stats:   stats,
+	}
+}
+
+// OnSignatureRequest handles message.SignatureRequest, and retrieves a warp signature for the requested message ID.
+// Never returns an error
+// Expects returned errors to be treated as FATAL
+// Returns empty response if signature is not found
+// Assumes ctx is active
+func (s *signatureRequestHandler) OnSignatureRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, signatureRequest message.SignatureRequest) ([]byte, error) {
+	startTime := time.Now()
+	s.stats.IncSignatureRequest()
+
+	// Always report signature request time
+	defer func() {
+		s.stats.UpdateSignatureRequestTime(time.Since(startTime))
+	}()
+
+	signature, err := s.backend.GetSignature(ctx, signatureRequest.MessageID)
+	if err != nil {
+		log.Debug("Unknown warp signature requested", "messageID", signatureRequest.MessageID)
+		s.stats.IncSignatureMiss()
+		return nil, nil
+	}
+
+	s.stats.IncSignatureHit()
+	response := message.SignatureResponse{Signature: signature}
+	responseBytes, err := s.codec.Marshal(message.Version, &response)
+	if err != nil {
+		log.Warn("could not marshal SignatureResponse, dropping request", "nodeID", nodeID, "requestID", requestID, "err", err)
+		return nil, nil
+	}
+
+	return responseBytes, nil
+}
+
+type NoopSignatureRequestHandler struct{}
+
+func (s *NoopSignatureRequestHandler) OnSignatureRequest(ctx context.Context, nodeID ids.NodeID, requestID uint32, signatureRequest message.SignatureRequest) ([]byte, error) {
+	return nil, nil
+}

handlers/warp/signature_request_test.go

@@ -0,0 +1,99 @@
+// (c) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package warp
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/ava-labs/avalanchego/database/memdb"
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/snow"
+	"github.com/ava-labs/avalanchego/utils/crypto/bls"
+	"github.com/ava-labs/avalanchego/utils/hashing"
+	"github.com/ava-labs/avalanchego/vms/platformvm/teleporter"
+	"github.com/ava-labs/subnet-evm/handlers/stats"
+	"github.com/ava-labs/subnet-evm/plugin/evm/message"
+	"github.com/ava-labs/subnet-evm/plugin/evm/warp"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSignatureHandler(t *testing.T) {
+	database := memdb.New()
+	snowCtx := snow.DefaultContextTest()
+	blsSecretKey, err := bls.NewSecretKey()
+	require.NoError(t, err)
+
+	snowCtx.TeleporterSigner = teleporter.NewSigner(blsSecretKey, snowCtx.ChainID)
+	warpBackend := warp.NewWarpBackend(snowCtx, database, 100)
+
+	msg, err := teleporter.NewUnsignedMessage(snowCtx.ChainID, snowCtx.CChainID, []byte("test"))
+	require.NoError(t, err)
+
+	messageID := hashing.ComputeHash256Array(msg.Bytes())
+	require.NoError(t, warpBackend.AddMessage(context.Background(), msg))
+	signature, err := warpBackend.GetSignature(context.Background(), messageID)
+	require.NoError(t, err)
+	unknownMessageID := ids.GenerateTestID()
+
+	mockHandlerStats := &stats.MockHandlerStats{}
+	signatureRequestHandler := NewSignatureRequestHandler(warpBackend, message.Codec, mockHandlerStats)
+
+	tests := map[string]struct {
+		setup       func() (request message.SignatureRequest, expectedResponse []byte)
+		verifyStats func(t *testing.T, stats *stats.MockHandlerStats)
+	}{
+		"normal": {
+			setup: func() (request message.SignatureRequest, expectedResponse []byte) {
+				return message.SignatureRequest{
+					MessageID: messageID,
+				}, signature[:]
+			},
+			verifyStats: func(t *testing.T, stats *stats.MockHandlerStats) {
+				require.EqualValues(t, 1, mockHandlerStats.SignatureRequestCount)
+				require.EqualValues(t, 1, mockHandlerStats.SignatureRequestHit)
+				require.EqualValues(t, 0, mockHandlerStats.SignatureRequestMiss)
+				require.Greater(t, mockHandlerStats.SignatureRequestDuration, time.Duration(0))
+			},
+		},
+		"unknown": {
+			setup: func() (request message.SignatureRequest, expectedResponse []byte) {
+				return message.SignatureRequest{
+					MessageID: unknownMessageID,
+				}, nil
+			},
+			verifyStats: func(t *testing.T, stats *stats.MockHandlerStats) {
+				require.EqualValues(t, 1, mockHandlerStats.SignatureRequestCount)
+				require.EqualValues(t, 1, mockHandlerStats.SignatureRequestMiss)
+				require.EqualValues(t, 0, mockHandlerStats.SignatureRequestHit)
+				require.Greater(t, mockHandlerStats.SignatureRequestDuration, time.Duration(0))
+			},
+		},
+	}
+
+	for name, test := range tests {
+		// Reset stats before each test
+		mockHandlerStats.Reset()
+
+		t.Run(name, func(t *testing.T) {
+			request, expectedResponse := test.setup()
+			responseBytes, err := signatureRequestHandler.OnSignatureRequest(context.Background(), ids.GenerateTestNodeID(), 1, request)
+			require.NoError(t, err)
+
+			// If the expected response is empty, assert that the handler returns an empty response and return early.
+			if len(expectedResponse) == 0 {
+				test.verifyStats(t, mockHandlerStats)
+				require.Len(t, responseBytes, 0, "expected response to be empty")
+				return
+			}
+			var response message.SignatureResponse
+			_, err = message.Codec.Unmarshal(responseBytes, &response)
+			require.NoError(t, err, "error unmarshalling SignatureResponse")
+
+			require.Equal(t, expectedResponse, response.Signature[:])
+			test.verifyStats(t, mockHandlerStats)
+		})
+	}
+}

handlers/warp/stats/stats.go

@@ -0,0 +1,95 @@
+// (c) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package stats
+
+import (
+	"sync"
+	"time"
+
+	"github.com/ava-labs/subnet-evm/metrics"
+)
+
+var (
+	_ SignatureRequestHandlerStats = (*handlerStats)(nil)
+	_ SignatureRequestHandlerStats = (*MockSignatureRequestHandlerStats)(nil)
+)
+
+type SignatureRequestHandlerStats interface {
+	IncSignatureRequest()
+	IncSignatureHit()
+	IncSignatureMiss()
+	UpdateSignatureRequestTime(duration time.Duration)
+}
+
+type handlerStats struct {
+	// SignatureRequestHandler metrics
+	signatureRequest        metrics.Counter
+	signatureHit            metrics.Counter
+	signatureMiss           metrics.Counter
+	signatureProcessingTime metrics.Timer
+}
+
+func NewStats(enabled bool) SignatureRequestHandlerStats {
+	if !enabled {
+		return &MockSignatureRequestHandlerStats{}
+	}
+
+	return &handlerStats{
+		signatureRequest:        metrics.GetOrRegisterCounter("signature_request_count", nil),
+		signatureHit:            metrics.GetOrRegisterCounter("signature_request_hit", nil),
+		signatureMiss:           metrics.GetOrRegisterCounter("signature_request_miss", nil),
+		signatureProcessingTime: metrics.GetOrRegisterTimer("signature_request_duration", nil),
+	}
+}
+
+func (h *handlerStats) IncSignatureRequest() { h.signatureRequest.Inc(1) }
+func (h *handlerStats) IncSignatureHit()     { h.signatureHit.Inc(1) }
+func (h *handlerStats) IncSignatureMiss()    { h.signatureMiss.Inc(1) }
+func (h *handlerStats) UpdateSignatureRequestTime(duration time.Duration) {
+	h.signatureProcessingTime.Update(duration)
+}
+
+// MockSignatureRequestHandlerStats is mock for capturing and asserting on handler metrics in test
+type MockSignatureRequestHandlerStats struct {
+	lock sync.Mutex
+
+	SignatureRequestCount,
+	SignatureRequestHit,
+	SignatureRequestMiss uint32
+	SignatureRequestDuration time.Duration
+}
+
+func (m *MockSignatureRequestHandlerStats) Reset() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	m.SignatureRequestCount = 0
+	m.SignatureRequestHit = 0
+	m.SignatureRequestMiss = 0
+	m.SignatureRequestDuration = 0
+}
+
+func (m *MockSignatureRequestHandlerStats) IncSignatureRequest() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.SignatureRequestCount++
+}
+
+func (m *MockSignatureRequestHandlerStats) IncSignatureHit() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.SignatureRequestHit++
+}
+
+func (m *MockSignatureRequestHandlerStats) IncSignatureMiss() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.SignatureRequestMiss++
+}
+
+func (m *MockSignatureRequestHandlerStats) UpdateSignatureRequestTime(duration time.Duration) {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.SignatureRequestDuration += duration
+}