bump slinky & forwarding module dependencies

app/app.go

@@ -164,6 +164,7 @@ import (
 	// slinky oracle dependencies
 	oraclepreblock "github.com/skip-mev/slinky/abci/preblock/oracle"
 	oracleproposals "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"
@@ -582,6 +583,7 @@ func NewInitiaApp(
 		runtime.NewKVStoreService(keys[forwardingtypes.StoreKey]),
 		runtime.NewTransientStoreService(tkeys[forwardingtypes.TransientStoreKey]),
 		appheaderinfo.NewHeaderInfoService(),
+		runtime.ProvideEventService(),
 		authorityAddr,
 		app.AccountKeeper,
 		app.BankKeeper,
@@ -1150,12 +1152,17 @@ func NewInitiaApp(
 	app.SetPrepareProposal(oracleProposalHandler.PrepareProposalHandler())
 	app.SetProcessProposal(oracleProposalHandler.ProcessProposalHandler())
 
+	// Create the aggregation function that will be used to aggregate oracle data
+	// from each validator.
+	aggregatorFn := voteweighted.MedianFromContext(
+		app.Logger(),
+		stakingkeeper.NewCompatibilityKeeper(app.StakingKeeper),
+		voteweighted.DefaultPowerThreshold,
+	)
+
 	app.oraclePreBlockHandler = oraclepreblock.NewOraclePreBlockHandler(
 		app.Logger(),
-		voteweighted.MedianFromContext(
-			app.Logger(),
-			stakingkeeper.NewCompatibilityKeeper(app.StakingKeeper),
-			voteweighted.DefaultPowerThreshold),
+		aggregatorFn,
 		app.OracleKeeper,
 		serviceMetrics,
 		currencypair.NewHashCurrencyPairStrategy(app.OracleKeeper),
@@ -1169,18 +1176,42 @@ func NewInitiaApp(
 		),
 	)
 
+	// see app.PreBlocker
+	// app.SetPreBlocker(oraclePreBlockHandler.WrappedPreBlocker(app.ModuleManager))
+
+	// Create the vote extensions handler that will be used to extend and verify
+	// vote extensions (i.e. oracle data).
+	cps := currencypair.NewDeltaCurrencyPairStrategy(app.OracleKeeper)
+	veCodec := compression.NewCompressionVoteExtensionCodec(
+		compression.NewDefaultVoteExtensionCodec(),
+		compression.NewZLibCompressor(),
+	)
+	extCommitCodec := compression.NewCompressionExtendedCommitCodec(
+		compression.NewDefaultExtendedCommitCodec(),
+		compression.NewZStdCompressor(),
+	)
+
 	// Create the vote extensions handler that will be used to extend and verify
 	// vote extensions (i.e. oracle data).
 	voteExtensionsHandler := ve.NewVoteExtensionHandler(
 		app.Logger(),
 		app.OracleClient,
 		time.Second,
-		currencypair.NewHashCurrencyPairStrategy(app.OracleKeeper),
-		compression.NewCompressionVoteExtensionCodec(
-			compression.NewDefaultVoteExtensionCodec(),
-			compression.NewZLibCompressor(),
+		cps,
+		veCodec,
+		aggregator.NewOraclePriceApplier(
+			aggregator.NewDefaultVoteAggregator(
+				app.Logger(),
+				aggregatorFn,
+				// we need a separate price strategy here, so that we can optimistically apply the latest prices
+				// and extend our vote based on these prices
+				currencypair.NewDeltaCurrencyPairStrategy(app.OracleKeeper),
+			),
+			app.OracleKeeper,
+			veCodec,
+			extCommitCodec,
+			app.Logger(),
 		),
-		app.oraclePreBlockHandler.PreBlocker(),
 		serviceMetrics,
 	)
 	app.SetExtendVoteHandler(voteExtensionsHandler.ExtendVoteHandler())
@@ -1272,13 +1303,7 @@ func (app *InitiaApp) Name() string { return app.BaseApp.Name() }
 
 // PreBlocker application updates every pre block
 func (app *InitiaApp) PreBlocker(ctx sdk.Context, req *abci.RequestFinalizeBlock) (*sdk.ResponsePreBlock, error) {
-	res, err := app.ModuleManager.PreBlock(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	_, err = app.oraclePreBlockHandler.PreBlocker()(ctx, req)
-	return res, err
+	return app.oraclePreBlockHandler.WrappedPreBlocker(app.ModuleManager)(ctx, req)
 }
 
 // BeginBlocker application updates every begin block

app/oracle/client.go

@@ -10,11 +10,13 @@
 	"cosmossdk.io/log"
 
 	"github.com/skip-mev/slinky/oracle/config"
+	slinkygrpc "github.com/skip-mev/slinky/pkg/grpc"
 	oracleclient "github.com/skip-mev/slinky/service/clients/oracle"
 	"github.com/skip-mev/slinky/service/metrics"
 	oracleservertypes "github.com/skip-mev/slinky/service/servers/oracle/types"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 
 	l2slinky "github.com/initia-labs/OPinit/x/opchild/l2slinky"
@@ -172,10 +174,6 @@
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 
-	if c.blockingDial {
-		opts = append(opts, grpc.WithBlock())
-	}
-
 	// dial the client, but defer to context closure, if necessary
 	var (
 		conn *grpc.ClientConn
@@ -184,7 +182,17 @@
 	)
 	go func() {
 		defer close(done)
-		conn, err = grpc.DialContext(ctx, c.addr, opts...)
+		conn, err = slinkygrpc.NewClient(c.addr, opts...)
+
+		// attempt to connect + wait for change in connection state
+		if c.blockingDial {
+			// connect
+			conn.Connect()
+
+			if err == nil {
+				conn.WaitForStateChange(ctx, connectivity.Ready)
+			}
+		}
 	}()
 
 	// wait for either the context to close or the dial to complete
@@ -198,10 +206,61 @@
 		return fmt.Errorf("failed to dial oracle gRPC server: %w", err)
 	}
 
+	c.mutex.Lock()
 	c.client = oracleservertypes.NewOracleClient(conn)
 	c.conn = conn
+	c.mutex.Unlock()
 
 	c.logger.Info("oracle client started")
 
 	return nil
 }
+
+func (c *GRPCClient) MarketMap(
+	ctx context.Context,
+	req *oracleservertypes.QueryMarketMapRequest,
+	_ ...grpc.CallOption,
+) (res *oracleservertypes.QueryMarketMapResponse, err error) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	start := time.Now()
+	defer func() {
+		// Observe the duration of the call as well as the error.
+		c.metrics.ObserveOracleResponseLatency(time.Since(start))
+		c.metrics.AddOracleResponse(metrics.StatusFromError(err))
+	}()
+
+	// set deadline on the context
+	ctx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("oracle client not started")
+	}
+
+	return c.client.MarketMap(ctx, req, grpc.WaitForReady(true))
+}
+
+// Version returns the version of the oracle service.
+func (c *GRPCClient) Version(ctx context.Context, req *oracleservertypes.QueryVersionRequest, _ ...grpc.CallOption) (res *oracleservertypes.QueryVersionResponse, err error) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	start := time.Now()
+	defer func() {
+		// Observe the duration of the call as well as the error.
+		c.metrics.ObserveOracleResponseLatency(time.Since(start))
+		c.metrics.AddOracleResponse(metrics.StatusFromError(err))
+	}()
+
+	// set deadline on the context
+	ctx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("oracle client not started")
+	}
+
+	return c.client.Version(ctx, req, grpc.WaitForReady(true))
+}