Merge pull request #16 from tellor-io/fix/goroutine-tracking

Fix/goroutine tracking and Gas Estimation

Author: brendaloya

cmd/main.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/joho/godotenv"
 	"github.com/rs/zerolog"
@@ -126,6 +127,7 @@ func init() {
 	rootCmd.Flags().Uint32("auto-unbonding-frequency", 0, "Enable automatic unbonding every N days (0 = disabled, 1 - 21 days = valid")
 	rootCmd.Flags().Uint32("auto-unbonding-amount", 0, "Amount of tokens in loya to unbond each unbonding transaction (0 = disabled)")
 	rootCmd.Flags().String("auto-unbonding-max-stake-percentage", "0.0", "Maximum percentage of stake to unbond each unbonding transaction (0 = disabled, 1.0 = 100%). If unbonding amount exceeds this percentage, we will skip the unbonding transaction until it exceeds this percentage again.")
+	rootCmd.Flags().Duration("refresh-gas-estimates-interval", 12*time.Hour, "Interval for resetting cached gas estimates and gas-adjustment levels (<=0 disables)")
 
 	// Marking required flags
 	if err := rootCmd.MarkFlagRequired(flags.FlagHome); err != nil {

pricefeed/client/client.go

@@ -137,6 +137,14 @@ func (c *Client) start(ctx context.Context,
 	exchangeIdToExchangeDetails map[types.ExchangeId]types.ExchangeQueryDetails,
 	subTaskRunner SubTaskRunner,
 ) (err error) {
+	// Release daemonStartup on any exit before the normal "startup complete" point (avoids Stop() blocking forever).
+	startupCompleted := false
+	defer func() {
+		if !startupCompleted {
+			c.daemonStartup.Done()
+		}
+	}()
+
 	// 1. Establish connections to gRPC servers.
 	queryConn, err := grpcClient.NewTcpConnection(ctx, grpcAddress)
 	if err != nil {
@@ -246,6 +254,7 @@ func (c *Client) start(ctx context.Context,
 	// Now that all persistent subtasks have been started and all tickers and stop channels are created,
 	// signal that the startup process is complete. This needs to be called before entering the
 	// price updater loop, which loops indefinitely until the daemon is stopped.
+	startupCompleted = true
 	c.daemonStartup.Done()
 
 	pricefeedClient := servertypes.NewPriceFeedServiceClient(daemonConn)

pricefeed/client/client_shutdown_test.go

@@ -0,0 +1,100 @@
+package client
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	daemonflags "github.com/tellor-io/layer-daemons/flags"
+	"github.com/tellor-io/layer-daemons/mocks"
+	"github.com/tellor-io/layer-daemons/pricefeed/client/types"
+	"github.com/tellor-io/layer-daemons/testutil/constants"
+	grpc_util "github.com/tellor-io/layer-daemons/testutil/grpc"
+
+	"cosmossdk.io/log"
+)
+
+// TestStop_CompletesWhenStartFailsBeforeDaemonStartupDone guards the contract that
+// daemonStartup must be released on every early return from start(); otherwise Stop()
+// blocks forever on daemonStartup.Wait().
+func TestStop_CompletesWhenStartFailsBeforeDaemonStartupDone(t *testing.T) {
+	t.Parallel()
+
+	tests := map[string]struct {
+		mockGrpcClient              *mocks.GrpcClient
+		exchangeIdToQueryConfig     map[types.ExchangeId]*types.ExchangeQueryConfig
+		exchangeIdToExchangeDetails map[types.ExchangeId]types.ExchangeQueryDetails
+		wantErrContains             string
+	}{
+		"tcp_connection_fails": {
+			mockGrpcClient: grpc_util.GenerateMockGrpcClientWithOptionalTcpConnectionErrors(
+				errors.New(connectionFailsErrorMsg),
+				nil,
+				false,
+			),
+			wantErrContains: connectionFailsErrorMsg,
+		},
+		"grpc_connection_fails": {
+			mockGrpcClient: grpc_util.GenerateMockGrpcClientWithOptionalGrpcConnectionErrors(
+				errors.New(connectionFailsErrorMsg),
+				nil,
+				false,
+			),
+			wantErrContains: connectionFailsErrorMsg,
+		},
+		"empty_exchange_config": {
+			mockGrpcClient: grpc_util.GenerateMockGrpcClientWithOptionalGrpcConnectionErrors(
+				nil,
+				nil,
+				true,
+			),
+			exchangeIdToQueryConfig:     map[types.ExchangeId]*types.ExchangeQueryConfig{},
+			exchangeIdToExchangeDetails: map[types.ExchangeId]types.ExchangeQueryDetails{},
+			wantErrContains:             "exchangeIds must not be empty",
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			cfg := tc.exchangeIdToQueryConfig
+			details := tc.exchangeIdToExchangeDetails
+			if cfg == nil {
+				cfg = constants.TestExchangeQueryConfigs
+			}
+			if details == nil {
+				details = constants.TestExchangeIdToExchangeQueryDetails
+			}
+
+			faketaskRunner := FakeSubTaskRunner{}
+
+			client := newClient(log.NewNopLogger())
+			err := client.start(
+				grpc_util.Ctx,
+				daemonflags.GetDefaultDaemonFlags(),
+				grpc_util.TcpEndpoint,
+				tc.mockGrpcClient,
+				[]types.MarketParam{},
+				cfg,
+				details,
+				&faketaskRunner,
+			)
+			require.Error(t, err)
+			require.Contains(t, err.Error(), tc.wantErrContains)
+
+			stopDone := make(chan struct{})
+			go func() {
+				client.Stop()
+				close(stopDone)
+			}()
+
+			select {
+			case <-stopDone:
+			case <-time.After(2 * time.Second):
+				t.Fatal("Stop() blocked — daemonStartup was likely not released on failed start()")
+			}
+		})
+	}
+}

reporter/client/client.go

@@ -75,6 +75,9 @@ type Client struct {
 	logger     log.Logger
 	txChan     chan TxChannelInfo
 	PriceGuard *PriceGuard
+	// Gas estimate refresh interval; <=0 disables periodic refresh.
+	refreshGasEstimatesInterval time.Duration
+	gasEstimator                *gasEstimateState
 
 	// Resources that need cleanup
 	grpcConn    *grpc.ClientConn
@@ -101,6 +104,16 @@ func NewClient(logger log.Logger, valGasMin string) *Client {
 		logger:    logger,
 		minGasFee: valGasMin,
 		txChan:    txChan,
+		gasEstimator: newGasEstimateState(map[string]gasBucketConfig{
+			bridgeGasBucketKey: {
+				levels:  []float64{1.75, 2.0},
+				baseIdx: 0,
+			},
+			defaultNonBridgeBucketConfigKey: {
+				levels:  []float64{1.25, 1.6, 2.0},
+				baseIdx: 0,
+			},
+		}),
 	}
 }
 
@@ -190,6 +203,7 @@ func (c *Client) Start(
 	autoUnbondingFrequency := viper.GetUint32("auto-unbonding-frequency")
 	autoUnbondingAmount := viper.GetUint32("auto-unbonding-amount")
 	autoUnbondingMaxStakePercentage := viper.GetString("auto-unbonding-max-stake-percentage")
+	c.refreshGasEstimatesInterval = viper.GetDuration("refresh-gas-estimates-interval")
 
 	if autoUnbondingFrequency > 0 {
 		if autoUnbondingAmount == 0 {
@@ -224,6 +238,11 @@ func (c *Client) Start(
 	} else {
 		c.logger.Info("Auto unbonding disabled")
 	}
+	if c.refreshGasEstimatesInterval > 0 {
+		c.logger.Info("Periodic gas estimate refresh enabled", "interval", c.refreshGasEstimatesInterval.String())
+	} else {
+		c.logger.Info("Periodic gas estimate refresh disabled")
+	}
 
 	c.cosmosCtx = c.cosmosCtx.WithChainID(chainId)
 	c.cosmosCtx = c.cosmosCtx.WithHomeDir(homeDir)
@@ -332,9 +351,31 @@ func StartReporterDaemonTaskLoop(
 	wg.Add(1)
 	go client.AutoUnbondStakePeriodically(ctx, wg)
 
+	if client.refreshGasEstimatesInterval > 0 {
+		wg.Add(1)
+		go client.RefreshGasEstimatesPeriodically(ctx, wg)
+	}
+
 	wg.Wait()
 }
 
+func (c *Client) RefreshGasEstimatesPeriodically(ctx context.Context, wg *sync.WaitGroup) {
+	defer wg.Done()
+	ticker := time.NewTicker(c.refreshGasEstimatesInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			c.logger.Debug("RefreshGasEstimatesPeriodically: context canceled, exiting")
+			return
+		case <-ticker.C:
+			c.logger.Info("Refreshing gas estimate buckets to base levels")
+			c.resetAllGasLevelsToBase()
+		}
+	}
+}
+
 func (c *Client) checkReporter(ctx context.Context) bool {
 	c.logger.Info("Checking if reporter is created", "address", c.accAddr.String())