Merge

protocol/app/app.go

@@ -89,20 +89,21 @@ import (
 	"github.com/cosmos/ibc-go/modules/capability"
 	capabilitykeeper "github.com/cosmos/ibc-go/modules/capability/keeper"
 	capabilitytypes "github.com/cosmos/ibc-go/modules/capability/types"
-	"github.com/dydxprotocol/v4-chain/protocol/daemons/configs"
 	"github.com/gorilla/mux"
 	"github.com/rakyll/statik/fs"
 	"github.com/spf13/cast"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
 
 	// App
 	appconstants "github.com/dydxprotocol/v4-chain/protocol/app/constants"
 	"github.com/dydxprotocol/v4-chain/protocol/app/flags"
 	"github.com/dydxprotocol/v4-chain/protocol/app/middleware"
 	"github.com/dydxprotocol/v4-chain/protocol/app/prepare"
-	"github.com/dydxprotocol/v4-chain/protocol/app/prepare/prices"
 	"github.com/dydxprotocol/v4-chain/protocol/app/process"
 
+	priceUpdateGenerator "github.com/dydxprotocol/v4-chain/protocol/app/prepare/prices"
+	"github.com/dydxprotocol/v4-chain/protocol/app/vote_extensions"
 	"github.com/dydxprotocol/v4-chain/protocol/lib"
 	"github.com/dydxprotocol/v4-chain/protocol/lib/metrics"
 	timelib "github.com/dydxprotocol/v4-chain/protocol/lib/time"
@@ -113,6 +114,7 @@ import (
 
 	// Daemons
 	bridgeclient "github.com/dydxprotocol/v4-chain/protocol/daemons/bridge/client"
+	"github.com/dydxprotocol/v4-chain/protocol/daemons/configs"
 	daemonflags "github.com/dydxprotocol/v4-chain/protocol/daemons/flags"
 	liquidationclient "github.com/dydxprotocol/v4-chain/protocol/daemons/liquidation/client"
 	metricsclient "github.com/dydxprotocol/v4-chain/protocol/daemons/metrics/client"
@@ -206,9 +208,16 @@ import (
 	streamingtypes "github.com/dydxprotocol/v4-chain/protocol/streaming/grpc/types"
 
 	// Slinky
+	slinkyproposals "github.com/skip-mev/slinky/abci/proposals"
+	"github.com/skip-mev/slinky/abci/strategies/aggregator"
+	compression "github.com/skip-mev/slinky/abci/strategies/codec"
+	"github.com/skip-mev/slinky/abci/strategies/currencypair"
+	"github.com/skip-mev/slinky/abci/ve"
 	oracleconfig "github.com/skip-mev/slinky/oracle/config"
+	"github.com/skip-mev/slinky/pkg/math/voteweighted"
 	oracleclient "github.com/skip-mev/slinky/service/clients/oracle"
 	servicemetrics "github.com/skip-mev/slinky/service/metrics"
+	promserver "github.com/skip-mev/slinky/service/servers/prometheus"
 )
 
 var (
@@ -322,11 +331,13 @@ type App struct {
 	PriceFeedClient    *pricefeedclient.Client
 	LiquidationsClient *liquidationclient.Client
 	BridgeClient       *bridgeclient.Client
+	SlinkyClient       *slinkyclient.Client
 
 	DaemonHealthMonitor *daemonservertypes.HealthMonitor
 
 	// Slinky
-	SlinkyClient *slinkyclient.Client
+	oraclePrometheusServer *promserver.PrometheusServer
+	oracleMetrics          servicemetrics.Metrics
 }
 
 // assertAppPreconditions assert invariants required for an application to start.
@@ -1332,52 +1343,103 @@ func New(
 	app.SetPrecommiter(app.Precommitter)
 	app.SetPrepareCheckStater(app.PrepareCheckStater)
 
+	strategy := currencypair.NewDefaultCurrencyPairStrategy(app.PricesKeeper)
+	veCodec := compression.NewCompressionVoteExtensionCodec(
+		compression.NewDefaultVoteExtensionCodec(),
+		compression.NewZLibCompressor(),
+	)
+	extCommitCodec := compression.NewCompressionExtendedCommitCodec(
+		compression.NewDefaultExtendedCommitCodec(),
+		compression.NewZLibCompressor(),
+	)
+	priceUpdateGenerator := priceUpdateGenerator.NewSlinkyPriceUpdateGenerator(
+		aggregator.NewDefaultVoteAggregator(
+			logger,
+			voteweighted.MedianFromContext(
+				logger,
+				app.StakingKeeper,
+				voteweighted.DefaultPowerThreshold,
+			),
+			strategy,
+		),
+		extCommitCodec,
+		veCodec,
+		strategy,
+	)
+
+	var dydxPrepareProposalHandler sdk.PrepareProposalHandler
 	// PrepareProposal setup.
-	priceUpdateGenerator := prices.NewDefaultPriceUpdateGenerator(app.PricesKeeper)
 	if appFlags.NonValidatingFullNode {
-		app.SetPrepareProposal(prepare.FullNodePrepareProposalHandler())
+		dydxPrepareProposalHandler = prepare.FullNodePrepareProposalHandler()
 	} else {
-		app.SetPrepareProposal(
-			prepare.PrepareProposalHandler(
-				txConfig,
-				app.BridgeKeeper,
-				app.ClobKeeper,
-				app.PerpetualsKeeper,
-				priceUpdateGenerator,
-			),
+		// setup slinky prepare-proposal handler
+		dydxPrepareProposalHandler = prepare.PrepareProposalHandler(
+			txConfig,
+			app.BridgeKeeper,
+			app.ClobKeeper,
+			app.PerpetualsKeeper,
+			priceUpdateGenerator,
 		)
 	}
 
+	priceUpdateDecoder := process.NewSlinkyMarketPriceDecoder(
+		process.NewDefaultUpdateMarketPriceTxDecoder(app.PricesKeeper, app.txConfig.TxDecoder()),
+		priceUpdateGenerator,
+	)
+
+	// if the node is a NonValidatingFullNode, we don't need to run any of the oracle code
+	if !appFlags.NonValidatingFullNode {
+		if app.oracleMetrics == nil {
+			app.oracleMetrics = app.initOracleMetrics(appOpts)
+		}
+		app.initOracle(priceUpdateDecoder)
+	}
+
+	if app.oracleMetrics == nil {
+		app.oracleMetrics = servicemetrics.NewNopMetrics()
+	}
+
 	// ProcessProposal setup.
-	priceUpdateDecoder := process.NewDefaultUpdateMarketPriceTxDecoder(app.PricesKeeper, app.txConfig.TxDecoder())
+	var dydxProcessProposalHandler sdk.ProcessProposalHandler
 	if appFlags.NonValidatingFullNode {
 		// Note: If the command-line flag `--non-validating-full-node` is enabled, this node will use
 		// an implementation of `ProcessProposal` which always returns `abci.ResponseProcessProposal_ACCEPT`.
 		// Full-nodes do not participate in consensus, and therefore should not participate in voting / `ProcessProposal`.
-		app.SetProcessProposal(
-			process.FullNodeProcessProposalHandler(
-				txConfig,
-				app.BridgeKeeper,
-				app.ClobKeeper,
-				app.StakingKeeper,
-				app.PerpetualsKeeper,
-				priceUpdateDecoder,
-			),
+		dydxProcessProposalHandler = process.FullNodeProcessProposalHandler(
+			txConfig,
+			app.BridgeKeeper,
+			app.ClobKeeper,
+			app.StakingKeeper,
+			app.PerpetualsKeeper,
+			priceUpdateDecoder,
 		)
 	} else {
-		app.SetProcessProposal(
-			process.ProcessProposalHandler(
-				txConfig,
-				app.BridgeKeeper,
-				app.ClobKeeper,
-				app.StakingKeeper,
-				app.PerpetualsKeeper,
-				app.PricesKeeper,
-				priceUpdateDecoder,
-			),
+		dydxProcessProposalHandler = process.ProcessProposalHandler(
+			txConfig,
+			app.BridgeKeeper,
+			app.ClobKeeper,
+			app.StakingKeeper,
+			app.PerpetualsKeeper,
+			app.PricesKeeper,
+			priceUpdateDecoder,
 		)
 	}
 
+	proposalHandler := slinkyproposals.NewProposalHandler(
+		logger,
+		dydxPrepareProposalHandler,
+		dydxProcessProposalHandler,
+		ve.NewDefaultValidateVoteExtensionsFn(app.ChainID(), app.StakingKeeper),
+		veCodec,
+		extCommitCodec,
+		strategy,
+		app.oracleMetrics,
+		slinkyproposals.RetainOracleDataInWrappedProposalHandler(),
+	)
+
+	app.SetPrepareProposal(proposalHandler.PrepareProposalHandler())
+	app.SetProcessProposal(proposalHandler.ProcessProposalHandler())
+
 	// Note that panics from out of gas errors won't get logged, since the `OutOfGasMiddleware` is added in front of this,
 	// so error will get handled by that middleware and subsequent middlewares won't get executed.
 	// Also note that `AddRunTxRecoveryHandler` adds the handler in reverse order, meaning that handlers that appear
@@ -1426,8 +1488,7 @@ func New(
 	return app
 }
 
-func (app *App) initSlinkySidecarClient(appOpts servertypes.AppOptions) oracleclient.OracleClient {
-	// Slinky setup
+func (app *App) initOracleMetrics(appOpts servertypes.AppOptions) servicemetrics.Metrics {
 	cfg, err := oracleconfig.ReadConfigFromAppOpts(appOpts)
 	if err != nil {
 		panic(err)
@@ -1436,18 +1497,67 @@ func (app *App) initSlinkySidecarClient(appOpts servertypes.AppOptions) oraclecl
 	if err != nil {
 		panic(err)
 	}
+	// run prometheus metrics
+	if cfg.MetricsEnabled {
+		promLogger, err := zap.NewProduction()
+		if err != nil {
+			panic(err)
+		}
+		app.oraclePrometheusServer, err = promserver.NewPrometheusServer(cfg.PrometheusServerAddress, promLogger)
+		if err != nil {
+			panic(err)
+		}
+		// start the prometheus server
+		go app.oraclePrometheusServer.Start()
+	}
+	return oracleMetrics
+}
+
+func (app *App) initSlinkySidecarClient(appOpts servertypes.AppOptions) oracleclient.OracleClient {
 	// Create the oracle service.
+	if app.oracleMetrics == nil {
+		app.oracleMetrics = app.initOracleMetrics(appOpts)
+	}
+	cfg, err := oracleconfig.ReadConfigFromAppOpts(appOpts)
+	if err != nil {
+		panic(err)
+	}
 	slinkyClient, err := oracleclient.NewClientFromConfig(
 		cfg,
 		app.Logger().With("client", "oracle"),
-		oracleMetrics,
+		app.oracleMetrics,
 	)
 	if err != nil {
 		panic(err)
 	}
 	return slinkyClient
 }
 
+func (app *App) initOracle(pricesTxDecoder process.UpdateMarketPriceTxDecoder) {
+	// Vote Extension setup.
+	slinkyVoteExtensionsHandler := ve.NewVoteExtensionHandler(
+		app.Logger(),
+		vote_extensions.NewOracleClient(app.PricesKeeper),
+		time.Second,
+		currencypair.NewDefaultCurrencyPairStrategy(app.PricesKeeper),
+		compression.NewCompressionVoteExtensionCodec(
+			compression.NewDefaultVoteExtensionCodec(),
+			compression.NewZLibCompressor(),
+		),
+		app.PreBlocker,
+		app.oracleMetrics,
+	)
+
+	dydxExtendVoteHandler := vote_extensions.ExtendVoteHandler{
+		SlinkyExtendVoteHandler: slinkyVoteExtensionsHandler.ExtendVoteHandler(),
+		PricesTxDecoder:         pricesTxDecoder,
+		PricesKeeper:            app.PricesKeeper,
+	}
+
+	app.SetExtendVoteHandler(dydxExtendVoteHandler.ExtendVoteHandler())
+	app.SetVerifyVoteExtensionHandler(slinkyVoteExtensionsHandler.VerifyVoteExtensionHandler())
+}
+
 // RegisterDaemonWithHealthMonitor registers a daemon service with the update monitor, which will commence monitoring
 // the health of the daemon. If the daemon does not register, the method will panic.
 func (app *App) RegisterDaemonWithHealthMonitor(
@@ -1699,9 +1809,7 @@ func (app *App) buildAnteHandler(txConfig client.TxConfig) sdk.AnteHandler {
 				FeegrantKeeper:  app.FeeGrantKeeper,
 				SigGasConsumer:  ante.DefaultSigVerificationGasConsumer,
 			},
-			ClobKeeper:   app.ClobKeeper,
-			Codec:        app.appCodec,
-			AuthStoreKey: app.keys[authtypes.StoreKey],
+			ClobKeeper: app.ClobKeeper,
 		},
 	)
 	if err != nil {
@@ -1722,6 +1830,9 @@ func (app *App) setAnteHandler(txConfig client.TxConfig) {
 // Close invokes an ordered shutdown of routines.
 func (app *App) Close() error {
 	app.BaseApp.Close()
+	if app.oraclePrometheusServer != nil {
+		app.oraclePrometheusServer.Close()
+	}
 	return app.closeOnce()
 }
 

protocol/cmd/dydxprotocold/cmd/config.go

@@ -90,11 +90,6 @@ func initTendermintConfig() *tmcfg.Config {
 	// Expose the Tendermint RPC.
 	cfg.RPC.ListenAddress = "tcp://0.0.0.0:26657"
 	cfg.RPC.CORSAllowedOrigins = []string{"*"}
-	// goroutine profiling showed that we were using exactly 900 threads (the default) which was throttling
-	// the maximum amount of load that the process could take. As of the last load test, at max QPS we were
-	// seeing ~1700 threads being used.
-	cfg.RPC.MaxOpenConnections = 4000
-	cfg.RPC.GRPCMaxOpenConnections = 4000
 
 	// Mempool config.
 	// We specifically are using a number greater than max QPS (currently set at 5000) * ShortBlockWindow to prevent

protocol/daemons/slinky/client/market_pair_fetcher.go

@@ -8,7 +8,7 @@ import (
 	"cosmossdk.io/log"
 	"google.golang.org/grpc"
 
-	oracletypes "github.com/skip-mev/slinky/x/oracle/types"
+	oracletypes "github.com/skip-mev/slinky/pkg/types"
 
 	appflags "github.com/dydxprotocol/v4-chain/protocol/app/flags"
 	daemontypes "github.com/dydxprotocol/v4-chain/protocol/daemons/types"

protocol/daemons/slinky/client/market_pair_fetcher_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
-	slinkytypes "github.com/skip-mev/slinky/x/oracle/types"
+	slinkytypes "github.com/skip-mev/slinky/pkg/types"
 
 	"github.com/dydxprotocol/v4-chain/protocol/daemons/slinky/client"
 	"github.com/dydxprotocol/v4-chain/protocol/mocks"

protocol/daemons/slinky/client/price_fetcher.go

@@ -8,9 +8,9 @@ import (
 	"cosmossdk.io/log"
 	"google.golang.org/grpc"
 
+	oracletypes "github.com/skip-mev/slinky/pkg/types"
 	oracleclient "github.com/skip-mev/slinky/service/clients/oracle"
 	"github.com/skip-mev/slinky/service/servers/oracle/types"
-	oracletypes "github.com/skip-mev/slinky/x/oracle/types"
 
 	"github.com/dydxprotocol/v4-chain/protocol/daemons/pricefeed/api"
 	daemontypes "github.com/dydxprotocol/v4-chain/protocol/daemons/types"

protocol/lib/slinky/utils.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/skip-mev/slinky/x/oracle/types"
+	"github.com/skip-mev/slinky/pkg/types"
 )
 
 /*

protocol/lib/slinky/utils_test.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/skip-mev/slinky/x/oracle/types"
+	"github.com/skip-mev/slinky/pkg/types"
 	"github.com/stretchr/testify/require"
 
 	"github.com/dydxprotocol/v4-chain/protocol/lib/slinky"