sync client

protocol/app/app.go

@@ -1419,6 +1419,7 @@ func New(
 
 	if app.GrpcStreamingManager.Enabled() {
 		app.GrpcStreamingTestClient = streamingclient.NewGrpcClient(appFlags, app.Logger())
+		app.GrpcStreamingManager.SubscribeTestClient(app.GrpcStreamingTestClient)
 	}
 
 	// Report out app version and git commit. This will be run when validators restart.

protocol/streaming/grpc/client/client.go

@@ -1,13 +1,11 @@
 package client
 
 import (
-	"context"
 	"sync"
 
 	"cosmossdk.io/log"
 
 	appflags "github.com/dydxprotocol/v4-chain/protocol/app/flags"
-	daemontypes "github.com/dydxprotocol/v4-chain/protocol/daemons/types"
 	v1 "github.com/dydxprotocol/v4-chain/protocol/indexer/protocol/v1"
 	v1types "github.com/dydxprotocol/v4-chain/protocol/indexer/protocol/v1/types"
 	clobtypes "github.com/dydxprotocol/v4-chain/protocol/x/clob/types"
@@ -40,44 +38,44 @@ func NewGrpcClient(appflags appflags.Flags, logger log.Logger) *GrpcClient {
 	}
 
 	// Subscribe to grpc orderbook updates.
-	go func() {
-		grpcClient := daemontypes.GrpcClientImpl{}
-
-		// Make a connection to the Cosmos gRPC query services.
-		queryConn, err := grpcClient.NewTcpConnection(context.Background(), appflags.GrpcAddress)
-		if err != nil {
-			logger.Error("Failed to establish gRPC connection to Cosmos gRPC query services", "error", err)
-			return
-		}
-		defer func() {
-			if err := grpcClient.CloseConnection(queryConn); err != nil {
-				logger.Error("Failed to close gRPC connection", "error", err)
-			}
-		}()
-
-		clobQueryClient := clobtypes.NewQueryClient(queryConn)
-		updateClient, err := clobQueryClient.StreamOrderbookUpdates(
-			context.Background(),
-			&clobtypes.StreamOrderbookUpdatesRequest{
-				ClobPairId: []uint32{0, 1},
-			},
-		)
-		if err != nil {
-			logger.Error("Failed to stream orderbook updates", "error", err)
-			return
-		}
-
-		for {
-			update, err := updateClient.Recv()
-			if err != nil {
-				logger.Error("Failed to receive orderbook update", "error", err)
-				return
-			}
-
-			logger.Info("Received orderbook update", "update", update)
-			client.Update(update)
-		}
-	}()
+	// go func() {
+	// 	grpcClient := daemontypes.GrpcClientImpl{}
+
+	// 	// Make a connection to the Cosmos gRPC query services.
+	// 	queryConn, err := grpcClient.NewTcpConnection(context.Background(), appflags.GrpcAddress)
+	// 	if err != nil {
+	// 		logger.Error("Failed to establish gRPC connection to Cosmos gRPC query services", "error", err)
+	// 		return
+	// 	}
+	// 	defer func() {
+	// 		if err := grpcClient.CloseConnection(queryConn); err != nil {
+	// 			logger.Error("Failed to close gRPC connection", "error", err)
+	// 		}
+	// 	}()
+
+	// 	clobQueryClient := clobtypes.NewQueryClient(queryConn)
+	// 	updateClient, err := clobQueryClient.StreamOrderbookUpdates(
+	// 		context.Background(),
+	// 		&clobtypes.StreamOrderbookUpdatesRequest{
+	// 			ClobPairId: []uint32{0, 1},
+	// 		},
+	// 	)
+	// 	if err != nil {
+	// 		logger.Error("Failed to stream orderbook updates", "error", err)
+	// 		return
+	// 	}
+
+	// 	for {
+	// 		update, err := updateClient.Recv()
+	// 		if err != nil {
+	// 			logger.Error("Failed to receive orderbook update", "error", err)
+	// 			return
+	// 		}
+
+	// 		logger.Info("Received orderbook update", "update", update)
+	// 		client.Update(update)
+	// 	}
+	// }()
 	return client
 }
 

protocol/streaming/grpc/grpc_streaming_manager.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cosmos/gogoproto/proto"
 	ocutypes "github.com/dydxprotocol/v4-chain/protocol/indexer/off_chain_updates/types"
 	"github.com/dydxprotocol/v4-chain/protocol/lib/metrics"
+	"github.com/dydxprotocol/v4-chain/protocol/streaming/grpc/client"
 	"github.com/dydxprotocol/v4-chain/protocol/streaming/grpc/types"
 	clobtypes "github.com/dydxprotocol/v4-chain/protocol/x/clob/types"
 )
@@ -52,6 +53,9 @@ type OrderbookSubscription struct {
 
 	// Channel to buffer writes before the stream
 	updatesChannel chan []clobtypes.StreamUpdate
+
+	// Testing
+	client *client.GrpcClient
 }
 
 func NewGrpcStreamingManager(
@@ -185,7 +189,20 @@ func (sm *GrpcStreamingManagerImpl) Subscribe(
 	return err
 }
 
-// removeSubscription removes a subscription from the grpc streaming manager.
+func (sm *GrpcStreamingManagerImpl) SubscribeTestClient(client *client.GrpcClient) {
+	subscription := &OrderbookSubscription{
+		clobPairIds: []uint32{0, 1},
+		client:      client,
+	}
+
+	sm.Lock()
+	defer sm.Unlock()
+
+	sm.orderbookSubscriptions[sm.nextSubscriptionId] = subscription
+	sm.nextSubscriptionId++
+}
+
+// removeSubscription removes a subscription from the streaming manager.
 // The streaming manager's lock should already be acquired before calling this.
 func (sm *GrpcStreamingManagerImpl) removeSubscription(
 	subscriptionIdToRemove uint32,
@@ -253,17 +270,11 @@ func (sm *GrpcStreamingManagerImpl) SendSnapshot(
 			metrics.GrpcAddToSubscriptionChannelCount,
 			1,
 		)
-		select {
-		case subscription.updatesChannel <- streamUpdates:
-		default:
-			sm.logger.Error(
-				fmt.Sprintf(
-					"GRPC Streaming subscription id %+v channel full capacity. Dropping subscription connection.",
-					subscriptionId,
-				),
-			)
-			removeSubscription = true
-		}
+		subscription.client.Update(
+			&clobtypes.StreamOrderbookUpdatesResponse{
+				Updates: streamUpdates,
+			},
+		)
 	}
 
 	// Clean up subscriptions that have been closed.
@@ -317,7 +328,25 @@ func (sm *GrpcStreamingManagerImpl) SendOrderbookUpdates(
 		}
 	}
 
-	sm.AddUpdatesToCache(updatesByClobPairId, uint32(len(updates)))
+	sm.Lock()
+	defer sm.Unlock()
+
+	for _, subscription := range sm.orderbookSubscriptions {
+		updatesToSend := make([]clobtypes.StreamUpdate, 0)
+		for _, clobPairId := range subscription.clobPairIds {
+			if updates, ok := updatesByClobPairId[clobPairId]; ok {
+				updatesToSend = append(updatesToSend, updates...)
+			}
+		}
+
+		if len(updatesToSend) > 0 {
+			subscription.client.Update(
+				&clobtypes.StreamOrderbookUpdatesResponse{
+					Updates: updatesToSend,
+				},
+			)
+		}
+	}
 }
 
 // SendOrderbookFillUpdates groups fills by their clob pair ids and
@@ -354,7 +383,25 @@ func (sm *GrpcStreamingManagerImpl) SendOrderbookFillUpdates(
 		updatesByClobPairId[clobPairId] = append(updatesByClobPairId[clobPairId], streamUpdate)
 	}
 
-	sm.AddUpdatesToCache(updatesByClobPairId, uint32(len(orderbookFills)))
+	sm.Lock()
+	defer sm.Unlock()
+
+	for _, subscription := range sm.orderbookSubscriptions {
+		updatesToSend := make([]clobtypes.StreamUpdate, 0)
+		for _, clobPairId := range subscription.clobPairIds {
+			if updates, ok := updatesByClobPairId[clobPairId]; ok {
+				updatesToSend = append(updatesToSend, updates...)
+			}
+		}
+
+		if len(updatesToSend) > 0 {
+			subscription.client.Update(
+				&clobtypes.StreamOrderbookUpdatesResponse{
+					Updates: updatesToSend,
+				},
+			)
+		}
+	}
 }
 
 func (sm *GrpcStreamingManagerImpl) AddUpdatesToCache(

protocol/streaming/grpc/noop_streaming_manager.go

@@ -2,6 +2,7 @@ package grpc
 
 import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/dydxprotocol/v4-chain/protocol/streaming/grpc/client"
 	"github.com/dydxprotocol/v4-chain/protocol/streaming/grpc/types"
 	clobtypes "github.com/dydxprotocol/v4-chain/protocol/x/clob/types"
 )
@@ -42,6 +43,9 @@ func (sm *NoopGrpcStreamingManager) SendOrderbookUpdates(
 ) {
 }
 
+func (sm *NoopGrpcStreamingManager) SubscribeTestClient(client *client.GrpcClient) {
+}
+
 func (sm *NoopGrpcStreamingManager) SendOrderbookFillUpdates(
 	ctx sdk.Context,
 	orderbookFills []clobtypes.StreamOrderbookFill,

protocol/streaming/grpc/types/manager.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/dydxprotocol/v4-chain/protocol/streaming/grpc/client"
 	clobtypes "github.com/dydxprotocol/v4-chain/protocol/x/clob/types"
 )
 
@@ -31,6 +32,9 @@ type GrpcStreamingManager interface {
 		blockHeight uint32,
 		execMode sdk.ExecMode,
 	)
+	SubscribeTestClient(
+		client *client.GrpcClient,
+	)
 	SendOrderbookFillUpdates(
 		ctx sdk.Context,
 		orderbookFills []clobtypes.StreamOrderbookFill,

protocol/x/clob/abci.go

@@ -143,6 +143,12 @@ func PrepareCheckState(
 		log.BlockHeight, ctx.BlockHeight()+1,
 	)
 
+	// Prune any rate limiting information that is no longer relevant.
+	keeper.PruneRateLimits(ctx)
+
+	// Initialize new streams with orderbook snapshots, if any.
+	keeper.InitializeNewGrpcStreams(ctx)
+
 	// Get the events generated from processing the matches in the latest block.
 	processProposerMatchesEvents := keeper.GetProcessProposerMatchesEvents(ctx)
 	if ctx.BlockHeight() != int64(processProposerMatchesEvents.BlockHeight) {
@@ -254,9 +260,6 @@ func PrepareCheckState(
 		types.GetInternalOperationsQueueTextString(newLocalValidatorOperationsQueue),
 	)
 
-	// Initialize new GRPC streams with orderbook snapshots, if any.
-	keeper.InitializeNewGrpcStreams(ctx)
-
 	// Set per-orderbook gauges.
 	keeper.MemClob.SetMemclobGauges(ctx)
 }