Use in process Relayer for race detection (#996)

* clone maps before passing to pathprocessor

* Add local relayer implementation to help catch race conditions in CI

* use existing relayerfactory

* Isolate mutex to totalFees

* handle feedback

* fix fatalf format

.github/workflows/ibctest.yml

@@ -27,7 +27,7 @@ jobs:
             ${{ runner.os }}-go-
 
       - name: ibctest
-        run: make ibctest-docker-events
+        run: make ibctest-events
   legacy:
     runs-on: self-hosted
     steps:
@@ -48,7 +48,7 @@ jobs:
             ${{ runner.os }}-go-
 
       - name: ibctest
-        run: make ibctest-docker-legacy
+        run: make ibctest-legacy
   multiple-paths:
     runs-on: self-hosted
     steps:
@@ -69,4 +69,4 @@ jobs:
             ${{ runner.os }}-go-
 
       - name: ibctest
-        run: make ibctest-docker-multiple
+        run: make ibctest-multiple

Makefile

@@ -78,7 +78,13 @@ ibctest-docker-events:
 ibctest-docker-legacy:
 	cd ibctest && go test -race -v -run TestRelayerDockerLegacyProcessor .
 
-ibctest-docker-multiple:
+ibctest-events:
+	cd ibctest && go test -race -v -run TestRelayerEventProcessor .
+
+ibctest-legacy:
+	cd ibctest && go test -race -v -run TestRelayerLegacyProcessor .
+
+ibctest-multiple:
 	cd ibctest && go test -race -v -run TestRelayerMultiplePathsSingleProcess .
 
 coverage:

cmd/config.go

@@ -153,7 +153,7 @@ $ %s cfg i`, appName, defaultHome, appName)),
 				memo, _ := cmd.Flags().GetString(flagMemo)
 
 				// And write the default config to that location...
-				if _, err = f.Write(defaultConfig(memo)); err != nil {
+				if _, err = f.Write(defaultConfigYAML(memo)); err != nil {
 					return err
 				}
 
@@ -432,12 +432,16 @@ func (c Config) MustYAML() []byte {
 	return out
 }
 
-func defaultConfig(memo string) []byte {
-	return Config{
+func defaultConfigYAML(memo string) []byte {
+	return DefaultConfig(memo).MustYAML()
+}
+
+func DefaultConfig(memo string) *Config {
+	return &Config{
 		Global: newDefaultGlobalConfig(memo),
-		Chains: relayer.Chains{},
-		Paths:  relayer.Paths{},
-	}.MustYAML()
+		Chains: make(relayer.Chains),
+		Paths:  make(relayer.Paths),
+	}
 }
 
 // GlobalConfig describes any global relayer settings

cmd/start.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/cosmos/relayer/v2/internal/relaydebug"
 	"github.com/cosmos/relayer/v2/relayer"
+	"github.com/cosmos/relayer/v2/relayer/chains/cosmos"
 	"github.com/cosmos/relayer/v2/relayer/processor"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
@@ -96,6 +97,11 @@ $ %s start demo-path2 --max-tx-size 10`, appName, appName, appName)),
 				log.Info("Debug server listening", zap.String("addr", debugAddr))
 				relaydebug.StartDebugServer(cmd.Context(), log, ln)
 				prometheusMetrics = processor.NewPrometheusMetrics()
+				for _, chain := range chains {
+					if ccp, ok := chain.ChainProvider.(*cosmos.CosmosProvider); ok {
+						ccp.SetMetrics(prometheusMetrics)
+					}
+				}
 			}
 
 			processorType, err := cmd.Flags().GetString(flagProcessor)

ibctest/ibc_test.go

@@ -41,12 +41,11 @@ func TestRelayerInProcess(t *testing.T) {
 	ibctestConformance(t, relayeribctest.RelayerFactory{})
 }
 
-// TestRelayerDocker runs the ibctest conformance tests against
+// TestRelayerDockerEventProcessor runs the ibctest conformance tests against
 // the current state of this relayer implementation built in docker.
 // Relayer runs using the event processor.
 func TestRelayerDockerEventProcessor(t *testing.T) {
 	t.Parallel()
-	relayeribctest.BuildRelayerImage(t)
 
 	rf := ibctest.NewBuiltinRelayerFactory(
 		ibc.CosmosRly,
@@ -59,7 +58,7 @@ func TestRelayerDockerEventProcessor(t *testing.T) {
 	ibctestConformance(t, rf)
 }
 
-// TestRelayerDocker runs the ibctest conformance tests against
+// TestRelayerDockerLegacyProcessor runs the ibctest conformance tests against
 // the current state of this relayer implementation built in docker.
 // Relayer runs using the legacy processor.
 func TestRelayerDockerLegacyProcessor(t *testing.T) {
@@ -76,3 +75,26 @@ func TestRelayerDockerLegacyProcessor(t *testing.T) {
 
 	ibctestConformance(t, rf)
 }
+
+// TestRelayerEventProcessor runs the ibctest conformance tests against
+// the local relayer code. This is helpful for detecting race conditions.
+// Relayer runs using the event processor.
+func TestRelayerEventProcessor(t *testing.T) {
+	t.Parallel()
+
+	ibctestConformance(t, relayeribctest.NewRelayerFactory(relayeribctest.RelayerConfig{
+		Processor:           relayer.ProcessorEvents,
+		InitialBlockHistory: 100,
+	}))
+}
+
+// TestRelayerLegacyProcessor runs the ibctest conformance tests against
+// the local relayer code. This is helpful for detecting race conditions.
+// Relayer runs using the legacy processor.
+func TestRelayerLegacyProcessor(t *testing.T) {
+	t.Parallel()
+
+	ibctestConformance(t, relayeribctest.NewRelayerFactory(relayeribctest.RelayerConfig{
+		Processor: relayer.ProcessorLegacy,
+	}))
+}

ibctest/relay_many_test.go

@@ -21,15 +21,7 @@ import (
 // from the same process using the go relayer. A single
 // CosmosChainProcessor (gaia) will feed data to two PathProcessors (gaia-osmosis and gaia-juno).
 func TestRelayerMultiplePathsSingleProcess(t *testing.T) {
-	relayeribctest.BuildRelayerImage(t)
-
-	client, network := ibctest.DockerSetup(t)
-	r := ibctest.NewBuiltinRelayerFactory(
-		ibc.CosmosRly,
-		zaptest.NewLogger(t),
-		ibctestrelayer.CustomDockerImage(relayeribctest.RelayerImageName, "latest", "100:1000"),
-		ibctestrelayer.ImagePull(false),
-	).Build(t, client, network)
+	r := relayeribctest.NewRelayer(t, relayeribctest.RelayerConfig{})
 
 	rep := testreporter.NewNopReporter()
 	eRep := rep.RelayerExecReporter(t)
@@ -84,6 +76,8 @@ func TestRelayerMultiplePathsSingleProcess(t *testing.T) {
 			Path:    pathGaiaJuno,
 		})
 
+	client, network := ibctest.DockerSetup(t)
+
 	require.NoError(t, ic.Build(ctx, eRep, ibctest.InterchainBuildOptions{
 		TestName:          t.Name(),
 		Client:            client,

ibctest/relayer.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
 	"strings"
 	"testing"
 
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
 	"github.com/cosmos/relayer/v2/cmd"
 	"github.com/cosmos/relayer/v2/internal/relayertest"
+	"github.com/cosmos/relayer/v2/relayer"
 	"github.com/cosmos/relayer/v2/relayer/chains/cosmos"
 	"github.com/strangelove-ventures/ibctest/v3/ibc"
 	"go.uber.org/zap"
@@ -19,13 +21,33 @@ import (
 type Relayer struct {
 	t *testing.T
 
-	home string
+	config RelayerConfig
+	home   string
 
 	// Set during StartRelayer.
 	errCh  chan error
 	cancel context.CancelFunc
 }
 
+// Build returns a relayer interface
+func NewRelayer(
+	t *testing.T,
+	config RelayerConfig,
+) ibc.Relayer {
+	r := &Relayer{
+		t:      t,
+		home:   t.TempDir(),
+		config: config,
+	}
+
+	res := r.sys().Run(zaptest.NewLogger(t), "config", "init", "--memo", config.Memo)
+	if res.Err != nil {
+		t.Fatalf("failed to rly config init: %v", res.Err)
+	}
+
+	return r
+}
+
 func (r *Relayer) sys() *relayertest.System {
 	return &relayertest.System{HomeDir: r.home}
 }
@@ -47,10 +69,11 @@ func (r *Relayer) AddChainConfiguration(ctx context.Context, _ ibc.RelayerExecRe
 			KeyringBackend: keyring.BackendTest,
 			GasAdjustment:  chainConfig.GasAdjustment,
 			GasPrices:      chainConfig.GasPrices,
-			Debug:          true,
-			Timeout:        "10s",
-			OutputFormat:   "json",
-			SignModeStr:    "direct",
+			// MinGasAmount: chainConfig.MinGasAmount, // TODO
+			Debug:        true,
+			Timeout:      "10s",
+			OutputFormat: "json",
+			SignModeStr:  "direct",
 		},
 	})
 
@@ -197,12 +220,22 @@ func (r *Relayer) StartRelayer(ctx context.Context, _ ibc.RelayerExecReporter, p
 	r.errCh = make(chan error, 1)
 	ctx, r.cancel = context.WithCancel(ctx)
 
-	args := append([]string{"--processor=events"}, pathNames...)
+	if r.config.Processor == "" {
+		r.config.Processor = relayer.ProcessorEvents
+	}
+	args := append([]string{
+		"--processor", r.config.Processor,
+		"--block-history", strconv.FormatUint(r.config.InitialBlockHistory, 10),
+	}, pathNames...)
+
 	go r.start(ctx, args...)
 	return nil
 }
 
 func (r *Relayer) StopRelayer(ctx context.Context, _ ibc.RelayerExecReporter) error {
+	if r.cancel == nil {
+		return nil
+	}
 	r.cancel()
 	err := <-r.errCh
 

ibctest/relayer_factory.go

@@ -1,36 +1,39 @@
 package ibctest
 
 import (
-	"fmt"
 	"testing"
 
 	"github.com/docker/docker/client"
 	"github.com/strangelove-ventures/ibctest/v3/ibc"
 	"github.com/strangelove-ventures/ibctest/v3/label"
 	ibctestrelayer "github.com/strangelove-ventures/ibctest/v3/relayer"
-	"go.uber.org/zap/zaptest"
 )
 
 // RelayerFactory implements the ibctest RelayerFactory interface.
-type RelayerFactory struct{}
+type RelayerFactory struct {
+	config RelayerConfig
+}
+
+// LocalRelayerConfig defines parameters for customizing a LocalRelayer.
+type RelayerConfig struct {
+	Processor           string
+	Memo                string
+	InitialBlockHistory uint64
+}
+
+func NewRelayerFactory(config RelayerConfig) RelayerFactory {
+	return RelayerFactory{
+		config: config,
+	}
+}
 
 // Build returns a relayer interface
-func (RelayerFactory) Build(
+func (rf RelayerFactory) Build(
 	t *testing.T,
 	_ *client.Client,
 	networkID string,
 ) ibc.Relayer {
-	r := &Relayer{
-		t:    t,
-		home: t.TempDir(),
-	}
-
-	res := r.sys().Run(zaptest.NewLogger(t), "config", "init")
-	if res.Err != nil {
-		panic(fmt.Errorf("failed to rly config init: %w", res.Err))
-	}
-
-	return r
+	return NewRelayer(t, rf.config)
 }
 
 func (RelayerFactory) Capabilities() map[ibctestrelayer.Capability]bool {

relayer/chains/cosmos/cosmos_chain_processor.go

@@ -55,9 +55,6 @@ type CosmosChainProcessor struct {
 }
 
 func NewCosmosChainProcessor(log *zap.Logger, provider *CosmosProvider, metrics *processor.PrometheusMetrics) *CosmosChainProcessor {
-
-	provider.SetMetrics(metrics)
-
 	return &CosmosChainProcessor{
 		log:                  log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", provider.ChainId())),
 		chainProvider:        provider,

relayer/chains/cosmos/provider.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
-        "sync"
+	"sync"
 	"time"
 
 	"github.com/Stride-Labs/stride/x/interchainquery"
@@ -107,8 +107,10 @@ type CosmosProvider struct {
 	txMu           sync.Mutex
 
 	// metrics to monitor the provider
-	TotalFees sdk.Coins
-	metrics   *processor.PrometheusMetrics
+	TotalFees   sdk.Coins
+	totalFeesMu sync.Mutex
+
+	metrics *processor.PrometheusMetrics
 }
 
 type CosmosIBCHeader struct {

relayer/chains/cosmos/tx.go

@@ -1155,7 +1155,9 @@ func (cc *CosmosProvider) UpdateFeesSpent(chain, key string, fees sdk.Coins) {
 		return
 	}
 
+	cc.totalFeesMu.Lock()
 	cc.TotalFees = cc.TotalFees.Add(fees...)
+	cc.totalFeesMu.Unlock()
 
 	for _, fee := range cc.TotalFees {
 		// Convert to a big float to get a float64 for metrics