Listen to initialization message instead of sleeping

relayer/main/main.go

@@ -275,6 +275,7 @@ func main() {
 			)
 		})
 	}
+	logger.Info("Initialization complete")
 	err = errGroup.Wait()
 	logger.Error("Relayer exiting.", zap.Error(err))
 }

signature-aggregator/main/main.go

@@ -123,6 +123,7 @@ func main() {
 		signatureAggregator,
 	)
 
+	logger.Info("Initialization complete")
 	err = http.ListenAndServe(fmt.Sprintf(":%d", cfg.APIPort), nil)
 	if errors.Is(err, http.ErrServerClosed) {
 		logger.Info("server closed")

tests/allowed_addresses.go

@@ -172,12 +172,17 @@ func AllowedAddresses(network interfaces.LocalNetwork) {
 
 	// Test Relayer 1
 	log.Info("Testing Relayer 1: All sources -> All destinations")
-	relayerCleanup := testUtils.RunRelayerExecutable(ctx, relayerConfigPath1)
+	relayerCleanup, readyChan := testUtils.RunRelayerExecutable(ctx, relayerConfigPath1)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
-	log.Info("Waiting for the relayers to start up")
-	time.Sleep(10 * time.Second)
+	// Wait for relayer to start up
+	log.Info("Waiting for the relayer to start up")
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	// Allowed by Relayer 1
 	testUtils.RelayBasicMessage(
@@ -196,12 +201,17 @@ func AllowedAddresses(network interfaces.LocalNetwork) {
 
 	// Test Relayer 2
 	log.Info("Testing Relayer 2: Specific source -> All destinations")
-	relayerCleanup = testUtils.RunRelayerExecutable(ctx, relayerConfigPath2)
+	relayerCleanup, readyChan = testUtils.RunRelayerExecutable(ctx, relayerConfigPath2)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
-	log.Info("Waiting for the relayers to start up")
-	time.Sleep(10 * time.Second)
+	// Wait for relayer to start up
+	log.Info("Waiting for the relayer to start up")
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	// Disallowed by Relayer 2
 	_, _, id := testUtils.SendBasicTeleporterMessage(
@@ -236,12 +246,17 @@ func AllowedAddresses(network interfaces.LocalNetwork) {
 
 	// Test Relayer 3
 	log.Info("Testing Relayer 3: All sources -> Specific destination")
-	relayerCleanup = testUtils.RunRelayerExecutable(ctx, relayerConfigPath3)
+	relayerCleanup, readyChan = testUtils.RunRelayerExecutable(ctx, relayerConfigPath3)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
-	log.Info("Waiting for the relayers to start up")
-	time.Sleep(10 * time.Second)
+	// Wait for relayer to start up
+	log.Info("Waiting for the relayer to start up")
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	// Disallowed by Relayer 3
 	_, _, id = testUtils.SendBasicTeleporterMessage(
@@ -276,12 +291,17 @@ func AllowedAddresses(network interfaces.LocalNetwork) {
 
 	// Test Relayer 4
 	log.Info("Testing Relayer 4: Specific source -> Specific destination")
-	relayerCleanup = testUtils.RunRelayerExecutable(ctx, relayerConfigPath4)
+	relayerCleanup, readyChan = testUtils.RunRelayerExecutable(ctx, relayerConfigPath4)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
-	log.Info("Waiting for the relayers to start up")
-	time.Sleep(10 * time.Second)
+	// Wait for relayer to start up
+	log.Info("Waiting for the relayer to start up")
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	// Disallowed by Relayer 4
 	_, _, id = testUtils.SendBasicTeleporterMessage(

tests/basic_relay.go

@@ -63,12 +63,16 @@ func BasicRelay(network interfaces.LocalNetwork) {
 	log.Info("Test Relaying from Subnet A to Subnet B")
 
 	log.Info("Starting the relayer")
-	relayerCleanup := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
+	relayerCleanup, readyChan := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
+	// Wait for relayer to start up
 	log.Info("Waiting for the relayer to start up")
-	time.Sleep(15 * time.Second)
+	select {
+	case <-readyChan:
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	log.Info("Sending transaction from Subnet A to Subnet B")
 	testUtils.RelayBasicMessage(
@@ -143,9 +147,18 @@ func BasicRelay(network interfaces.LocalNetwork) {
 
 	// Run the relayer
 	log.Info("Creating new relayer instance to test already delivered message")
-	relayerCleanup = testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
+	relayerCleanup, readyChan = testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
 	defer relayerCleanup()
 
+	// Wait for relayer to start up
+	log.Info("Waiting for the relayer to start up")
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
+
 	// We should not receive a new block on subnet B, since the relayer should have
 	// seen the Teleporter message was already delivered.
 	log.Info("Waiting for 10s to ensure no new block confirmations on destination chain")

tests/batch_relay.go

@@ -67,12 +67,17 @@ func BatchRelay(network interfaces.LocalNetwork) {
 	relayerConfigPath := testUtils.WriteRelayerConfig(relayerConfig, testUtils.DefaultRelayerCfgFname)
 
 	log.Info("Starting the relayer")
-	relayerCleanup := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
+	relayerCleanup, readyChan := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
+	// Wait for relayer to start up
 	log.Info("Waiting for the relayer to start up")
-	time.Sleep(15 * time.Second)
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	//
 	// Send a batch message from subnet A -> B

tests/e2e_test.go

@@ -40,6 +40,7 @@ func TestE2E(t *testing.T) {
 		if r := recover(); r != nil {
 			fmt.Sprintf("Panic caught: %v", r)
 			cleanup()
+			os.Exit(1)
 		}
 	}()
 	// Handle SIGINT and SIGTERM signals as well.
@@ -49,6 +50,7 @@ func TestE2E(t *testing.T) {
 		sig := <-signalChan
 		fmt.Printf("Caught signal %s: Shutting down...\n", sig)
 		cleanup()
+		os.Exit(1)
 	}()
 
 	if os.Getenv("RUN_E2E") == "" {
@@ -117,12 +119,12 @@ var _ = ginkgo.BeforeSuite(func() {
 
 	decider = exec.CommandContext(ctx, "./tests/cmd/decider/decider")
 	decider.Start()
-	go func() { // panic if the decider exits abnormally
+	go func() {
 		err := decider.Wait()
 		// Context cancellation is the only expected way for the
-		// process to exit, otherwise panic
+		// otherwise log an error but don't panic to allow for easier cleanup
 		if !errors.Is(ctx.Err(), context.Canceled) {
-			panic(fmt.Errorf("decider exited abnormally: %w", err))
+			log.Error("Decider exited abnormally: ", "error", err)
 		}
 	}()
 	log.Info("Started decider service")
@@ -138,14 +140,13 @@ var _ = ginkgo.BeforeSuite(func() {
 })
 
 func cleanup() {
-	fmt.Printf("Cleaning up\n")
-	if localNetworkInstance != nil {
-		fmt.Printf("Started cleaning up\n")
-		localNetworkInstance.TearDownNetwork()
-		fmt.Printf("Finished cleaning up\n")
-	}
 	if decider != nil {
 		cancelDecider()
+		decider = nil
+	}
+	if localNetworkInstance != nil {
+		localNetworkInstance.TearDownNetwork()
+		localNetworkInstance = nil
 	}
 }
 

tests/manual_message.go

@@ -109,12 +109,16 @@ func ManualMessage(network interfaces.LocalNetwork) {
 	relayerConfigPath := testUtils.WriteRelayerConfig(relayerConfig, testUtils.DefaultRelayerCfgFname)
 
 	log.Info("Starting the relayer")
-	relayerCleanup := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
+	relayerCleanup, readyChan := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
+	// Wait for relayer to startup.
 	log.Info("Waiting for the relayer to start up")
-	time.Sleep(15 * time.Second)
+	select {
+	case <-readyChan:
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	reqBody := api.ManualWarpMessageRequest{
 		UnsignedMessageBytes: unsignedMessage.Bytes(),

tests/relay_message_api.go

@@ -66,12 +66,17 @@ func RelayMessageAPI(network interfaces.LocalNetwork) {
 	relayerConfigPath := testUtils.WriteRelayerConfig(relayerConfig, testUtils.DefaultRelayerCfgFname)
 
 	log.Info("Starting the relayer")
-	relayerCleanup := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
+	relayerCleanup, readyChan := testUtils.RunRelayerExecutable(ctx, relayerConfigPath)
 	defer relayerCleanup()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
+	// Wait for relayer to start up
 	log.Info("Waiting for the relayer to start up")
-	time.Sleep(15 * time.Second)
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Relayer did not start up in time")
+	}
 
 	reqBody := api.RelayMessageRequest{
 		BlockchainID: subnetAInfo.BlockchainID.String(),

tests/shared_db.go

@@ -2,7 +2,7 @@ package tests
 
 import (
 	"context"
-	"time"
+	"sync"
 
 	testUtils "github.com/ava-labs/awm-relayer/tests/utils"
 	"github.com/ava-labs/teleporter/tests/interfaces"
@@ -68,14 +68,18 @@ func SharedDatabaseAccess(network interfaces.LocalNetwork) {
 	log.Info("Test Relaying from Subnet A to Subnet B")
 
 	log.Info("Starting the relayers")
-	relayerCleanupA := testUtils.RunRelayerExecutable(ctx, relayerConfigPathA)
+	relayerCleanupA, readyChanA := testUtils.RunRelayerExecutable(ctx, relayerConfigPathA)
 	defer relayerCleanupA()
-	relayerCleanupB := testUtils.RunRelayerExecutable(ctx, relayerConfigPathB)
+	relayerCleanupB, readyChanB := testUtils.RunRelayerExecutable(ctx, relayerConfigPathB)
 	defer relayerCleanupB()
 
-	// Sleep for some time to make sure relayer has started up and subscribed.
+	// Wait for the relayers to start up
 	log.Info("Waiting for the relayers to start up")
-	time.Sleep(15 * time.Second)
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go testUtils.WaitFunc(&wg, readyChanA)
+	go testUtils.WaitFunc(&wg, readyChanB)
+	wg.Wait()
 
 	log.Info("Sending transaction from Subnet A to Subnet B")
 	testUtils.RelayBasicMessage(

tests/signature_aggregator_api.go

@@ -52,12 +52,17 @@ func SignatureAggregatorAPI(network interfaces.LocalNetwork) {
 		testUtils.DefaultSignatureAggregatorCfgFname,
 	)
 	log.Info("Starting the signature aggregator", "configPath", signatureAggregatorConfigPath)
-	signatureAggregatorCancel := testUtils.RunSignatureAggregatorExecutable(ctx, signatureAggregatorConfigPath)
+	signatureAggregatorCancel, readyChan := testUtils.RunSignatureAggregatorExecutable(ctx, signatureAggregatorConfigPath)
 	defer signatureAggregatorCancel()
 
-	// Sleep for some time to make sure signature aggregator has started up and subscribed.
-	log.Info("Waiting for the signature aggregator to start up")
-	time.Sleep(5 * time.Second)
+	// Wait for signature-aggregator to start up
+	log.Info("Waiting for the relayer to start up")
+	select {
+	case <-readyChan:
+		close(readyChan)
+	case <-time.After(15 * time.Second):
+		Expect(false).To(BeTrue(), "Signature Aggregator did not start up in time")
+	}
 
 	// End setup step
 	// Begin Test Case 1