Add etna-time configs

config/api_config.go

@@ -1,3 +1,6 @@
+// Copyright (C) 2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
 package config
 
 import (

relayer/config/config.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
+	"time"
 
 	basecfg "github.com/ava-labs/awm-relayer/config"
 	"github.com/ava-labs/awm-relayer/peers"
@@ -67,6 +68,9 @@ type Config struct {
 	DeciderURL             string                   `mapstructure:"decider-url" json:"decider-url"`
 	SignatureCacheSize     uint64                   `mapstructure:"signature-cache-size" json:"signature-cache-size"`
 
+	// mapstructure doesn't handle time.Time out of the box so handle it manually
+	EtnaTime time.Time `json:"etna-time"`
+
 	// convenience field to fetch a blockchain's subnet ID
 	blockchainIDToSubnetID map[ids.ID]ids.ID
 	overwrittenOptions     []string

relayer/config/keys.go

@@ -24,4 +24,5 @@ const (
 	ManualWarpMessagesKey     = "manual-warp-messages"
 	DBWriteIntervalSecondsKey = "db-write-interval-seconds"
 	SignatureCacheSizeKey     = "signature-cache-size"
+	EtnaTimeKey               = "etna-time"
 )

relayer/config/viper.go

@@ -84,6 +84,9 @@ func BuildConfig(v *viper.Viper) (Config, error) {
 		return cfg, fmt.Errorf("failed to unmarshal viper config: %w", err)
 	}
 
+	// Manually set EtnaTime field since it's not automatically parseable using mapstructure
+	cfg.EtnaTime = v.GetTime(EtnaTimeKey)
+
 	// Explicitly overwrite the configured account private key
 	// If account-private-key is set as a flag or environment variable,
 	// overwrite all destination subnet configurations to use that key

relayer/main/main.go

@@ -227,6 +227,7 @@ func main() {
 			prometheus.DefaultRegisterer,
 		),
 		messageCreator,
+		cfg.EtnaTime,
 	)
 	if err != nil {
 		logger.Fatal("Failed to create signature aggregator", zap.Error(err))

signature-aggregator/README.md

@@ -59,9 +59,9 @@ curl --location 'https://api.avax-test.network/ext/bc/C/rpc' \
    The topic of the message will be `0x56600c567728a800c0aa927500f831cb451df66a7af570eb4df4dfbf4674887d` which is the output of`cast keccak "SendWarpMessage(address,bytes32,bytes)"`
 4. Use the data field of the log message found in step 2 and send it to the locally running service via curl.
 ```bash
-curl --location 'http://localhost:8080/aggregate-signatures/by-raw-message' \
+curl --location 'http://localhost:8080/aggregate-signatures' \
 --header 'Content-Type: application/json' \
 --data '{
-    "data": "<hex encoded unsigned message bytes retrieved from the logs>",
+    "message": "<hex encoded unsigned message bytes retrieved from the logs>"
 }'
 ```

signature-aggregator/aggregator/aggregator.go

@@ -19,6 +19,7 @@ import (
 	"github.com/ava-labs/avalanchego/proto/pb/p2p"
 	"github.com/ava-labs/avalanchego/proto/pb/sdk"
 	"github.com/ava-labs/avalanchego/subnets"
+	"github.com/ava-labs/avalanchego/utils/constants"
 	"github.com/ava-labs/avalanchego/utils/crypto/bls"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/utils/set"
@@ -27,6 +28,8 @@ import (
 	"github.com/ava-labs/awm-relayer/signature-aggregator/aggregator/cache"
 	"github.com/ava-labs/awm-relayer/signature-aggregator/metrics"
 	"github.com/ava-labs/awm-relayer/utils"
+	corethMsg "github.com/ava-labs/coreth/plugin/evm/message"
+	msg "github.com/ava-labs/subnet-evm/plugin/evm/message"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 )
@@ -42,6 +45,9 @@ const (
 )
 
 var (
+	codec       = msg.Codec
+	corethCodec = corethMsg.Codec
+
 	// Errors
 	errNotEnoughSignatures     = errors.New("failed to collect a threshold of signatures")
 	errNotEnoughConnectedStake = errors.New("failed to connect to a threshold of stake")
@@ -57,6 +63,7 @@ type SignatureAggregator struct {
 	subnetsMapLock          sync.RWMutex
 	metrics                 *metrics.SignatureAggregatorMetrics
 	cache                   *cache.Cache
+	etnaTime                time.Time
 }
 
 func NewSignatureAggregator(
@@ -65,6 +72,7 @@ func NewSignatureAggregator(
 	signatureCacheSize uint64,
 	metrics *metrics.SignatureAggregatorMetrics,
 	messageCreator message.Creator,
+	etnaTime time.Time,
 ) (*SignatureAggregator, error) {
 	cache, err := cache.NewCache(signatureCacheSize, logger)
 	if err != nil {
@@ -81,6 +89,7 @@ func NewSignatureAggregator(
 		messageCreator:          messageCreator,
 		currentRequestID:        atomic.Uint32{},
 		cache:                   cache,
+		etnaTime:                etnaTime,
 	}
 	sa.currentRequestID.Store(rand.Uint32())
 	return &sa, nil
@@ -174,14 +183,7 @@ func (s *SignatureAggregator) CreateSignedMessage(
 		))
 	}
 
-	reqBytes := networkP2P.ProtocolPrefix(networkP2P.SignatureRequestHandlerID)
-	messageBytes, err := proto.Marshal(
-		&sdk.SignatureRequest{
-			Message:       unsignedMessage.Bytes(),
-			Justification: justification,
-		},
-	)
-	reqBytes = append(reqBytes, messageBytes...)
+	reqBytes, err := s.marshalRequest(unsignedMessage, justification, sourceSubnet)
 	if err != nil {
 		msg := "Failed to marshal request bytes"
 		s.logger.Error(
@@ -517,15 +519,13 @@ func (s *SignatureAggregator) isValidSignatureResponse(
 		return blsSignatureBuf{}, false
 	}
 
-	sigResponse := sdk.SignatureResponse{}
-	err := proto.Unmarshal(appResponse.AppBytes, &sigResponse)
+	signature, err := s.unmarshalResponse(appResponse.AppBytes)
 	if err != nil {
 		s.logger.Error(
 			"Error unmarshaling signature response",
 			zap.Error(err),
 		)
 	}
-	signature := sigResponse.Signature
 
 	// If the node returned an empty signature, then it has not yet seen the warp message. Retry later.
 	emptySignature := blsSignatureBuf{}
@@ -560,9 +560,8 @@ func (s *SignatureAggregator) isValidSignatureResponse(
 		)
 		return blsSignatureBuf{}, false
 	}
-	blsSig := blsSignatureBuf{}
-	copy(blsSig[:], signature[:])
-	return blsSig, true
+
+	return signature, true
 }
 
 // aggregateSignatures constructs a BLS aggregate signature from the collected validator signatures. Also
@@ -594,3 +593,58 @@ func (s *SignatureAggregator) aggregateSignatures(
 	}
 	return aggSig, vdrBitSet, nil
 }
+
+func (s *SignatureAggregator) marshalRequest(
+	unsignedMessage *avalancheWarp.UnsignedMessage,
+	justification []byte,
+	sourceSubnet ids.ID,
+) ([]byte, error) {
+	if !s.etnaTime.IsZero() && s.etnaTime.Before(time.Now()) {
+		// Post-Etna case
+		reqBytes := networkP2P.ProtocolPrefix(networkP2P.SignatureRequestHandlerID)
+		messageBytes, err := proto.Marshal(
+			&sdk.SignatureRequest{
+				Message:       unsignedMessage.Bytes(),
+				Justification: justification,
+			},
+		)
+		if err != nil {
+			return []byte{}, err
+		}
+		reqBytes = append(reqBytes, messageBytes...)
+		return reqBytes, nil
+	} else {
+		// Pre-Etna case
+		if sourceSubnet == constants.PrimaryNetworkID {
+			req := corethMsg.MessageSignatureRequest{
+				MessageID: unsignedMessage.ID(),
+			}
+			return corethMsg.RequestToBytes(corethCodec, req)
+		} else {
+			req := msg.MessageSignatureRequest{
+				MessageID: unsignedMessage.ID(),
+			}
+			return msg.RequestToBytes(codec, req)
+		}
+	}
+}
+
+func (s *SignatureAggregator) unmarshalResponse(responseBytes []byte) (blsSignatureBuf, error) {
+	if !s.etnaTime.IsZero() && s.etnaTime.Before(time.Now()) {
+		// Post-Etna case
+		var sigResponse sdk.SignatureResponse
+		err := proto.Unmarshal(responseBytes, &sigResponse)
+		if err != nil {
+			return blsSignatureBuf{}, err
+		}
+		return blsSignatureBuf(sigResponse.Signature), nil
+	} else {
+		// Pre-Etna case
+		var sigResponse msg.SignatureResponse
+		_, err := msg.Codec.Unmarshal(responseBytes, &sigResponse)
+		if err != nil {
+			return blsSignatureBuf{}, err
+		}
+		return sigResponse.Signature, nil
+	}
+}

signature-aggregator/config/config.go

@@ -5,6 +5,7 @@ package config
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/ava-labs/avalanchego/utils/logging"
 	basecfg "github.com/ava-labs/awm-relayer/config"
@@ -34,6 +35,9 @@ type Config struct {
 	APIPort            uint16             `mapstructure:"api-port" json:"api-port"`
 	MetricsPort        uint16             `mapstructure:"metrics-port" json:"metrics-port"`
 	SignatureCacheSize uint64             `mapstructure:"signature-cache-size" json:"signature-cache-size"`
+
+	// mapstructure doesn't support time.Time out of the box so handle it manually
+	EtnaTime time.Time `json:"etna-time"`
 }
 
 func DisplayUsageText() {

signature-aggregator/config/keys.go

@@ -16,4 +16,5 @@ const (
 	APIPortKey            = "api-port"
 	MetricsPortKey        = "metrics-port"
 	SignatureCacheSizeKey = "signature-cache-size"
+	EtnaTimeKey           = "etna-time"
 )

signature-aggregator/config/viper.go

@@ -76,5 +76,8 @@ func BuildConfig(v *viper.Viper) (Config, error) {
 		return cfg, fmt.Errorf("failed to unmarshal viper config: %w", err)
 	}
 
+	// mapstructure doesn't support time.Time out of the box so handle it manually
+	cfg.EtnaTime = v.GetTime(EtnaTimeKey)
+
 	return cfg, nil
 }

signature-aggregator/main/main.go

@@ -117,6 +117,7 @@ func main() {
 		cfg.SignatureCacheSize,
 		metricsInstance,
 		messageCreator,
+		cfg.EtnaTime,
 	)
 	if err != nil {
 		logger.Fatal("Failed to create signature aggregator", zap.Error(err))

tests/e2e_test.go

@@ -174,4 +174,7 @@ var _ = ginkgo.Describe("[AWM Relayer Integration Tests", func() {
 	ginkgo.It("Signature Aggregator", func() {
 		SignatureAggregatorAPI(localNetworkInstance)
 	})
+	ginkgo.It("Etna Upgrade", func() {
+		EtnaUpgrade(localNetworkInstance)
+	})
 })

tests/etna_upgrade.go

@@ -0,0 +1,109 @@
+// Copyright (C) 2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package tests
+
+import (
+	"context"
+	"time"
+
+	testUtils "github.com/ava-labs/awm-relayer/tests/utils"
+	"github.com/ava-labs/teleporter/tests/interfaces"
+	"github.com/ava-labs/teleporter/tests/utils"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/log"
+	. "github.com/onsi/gomega"
+)
+
+// This tests basic functionality of the relayer in the context
+// of Etna network upgrade using the following cases:
+// - Relaying from Subnet A to Subnet B using Pre-Etna config
+// - Relaying from Subnet B to Subnet A using Post-Etna config
+func EtnaUpgrade(network interfaces.LocalNetwork) {
+	subnetAInfo := network.GetPrimaryNetworkInfo()
+	subnetBInfo, _ := utils.GetTwoSubnets(network)
+	fundedAddress, fundedKey := network.GetFundedAccountInfo()
+	teleporterContractAddress := network.GetTeleporterContractAddress()
+	err := testUtils.ClearRelayerStorage()
+	Expect(err).Should(BeNil())
+	//
+	// Fund the relayer address on all subnets
+	//
+	ctx := context.Background()
+
+	log.Info("Funding relayer address on all subnets")
+	relayerKey, err := crypto.GenerateKey()
+	Expect(err).Should(BeNil())
+	testUtils.FundRelayers(ctx, []interfaces.SubnetTestInfo{subnetAInfo, subnetBInfo}, fundedKey, relayerKey)
+
+	//
+	// Set up relayer config
+	//
+	relayerConfig := testUtils.CreateDefaultRelayerConfig(
+		[]interfaces.SubnetTestInfo{subnetAInfo, subnetBInfo},
+		[]interfaces.SubnetTestInfo{subnetAInfo, subnetBInfo},
+		teleporterContractAddress,
+		fundedAddress,
+		relayerKey,
+	)
+	relayerConfig.EtnaTime = time.Now().AddDate(0, 0, 1)
+	// The config needs to be validated in order to be passed to database.GetConfigRelayerIDs
+	relayerConfig.Validate()
+
+	relayerConfigPath := testUtils.WriteRelayerConfig(relayerConfig, testUtils.DefaultRelayerCfgFname)
+
+	//
+	// Test Relaying from Subnet A to Subnet B
+	//
+	log.Info("Test Relaying from Subnet A to Subnet B")
+
+	log.Info("Starting the relayer")
+	relayerCleanup, readyChan := testUtils.RunRelayerExecutable(
+		ctx,
+		relayerConfigPath,
+		relayerConfig,
+	)
+	defer relayerCleanup()
+
+	// Wait for relayer to start up
+	startupCtx, startupCancel := context.WithTimeout(ctx, 15*time.Second)
+	defer startupCancel()
+	testUtils.WaitForChannelClose(startupCtx, readyChan)
+
+	log.Info("Sending transaction from Subnet A to Subnet B, Pre-Etna")
+	testUtils.RelayBasicMessage(
+		ctx,
+		subnetAInfo,
+		subnetBInfo,
+		teleporterContractAddress,
+		fundedKey,
+		fundedAddress,
+	)
+	// Shutdown the relayer and write a new config with EtnaTime set to yesterday
+	relayerCleanup()
+
+	relayerConfig.EtnaTime = time.Now().AddDate(0, 0, -1)
+	relayerConfigPath = testUtils.WriteRelayerConfig(relayerConfig, testUtils.DefaultRelayerCfgFname)
+
+	relayerCleanup, readyChan = testUtils.RunRelayerExecutable(
+		ctx,
+		relayerConfigPath,
+		relayerConfig,
+	)
+	defer relayerCleanup()
+
+	// Wait for relayer to start up
+	startupCtx, startupCancel = context.WithTimeout(ctx, 15*time.Second)
+	defer startupCancel()
+	testUtils.WaitForChannelClose(startupCtx, readyChan)
+
+	log.Info("Test Relaying from Subnet B to Subnet A - Post-Etna")
+	testUtils.RelayBasicMessage(
+		ctx,
+		subnetBInfo,
+		subnetAInfo,
+		teleporterContractAddress,
+		fundedKey,
+		fundedAddress,
+	)
+}