[OTE-456] FNS x OE: stage FinalizeBlock events and emit in Precommit

indexer/packages/v4-protos/src/codegen/gogoproto/bundle.ts

@@ -1,3 +1,3 @@
-import * as _127 from "./gogo";
-export const gogoproto = { ..._127
+import * as _128 from "./gogo";
+export const gogoproto = { ..._128
 };

indexer/packages/v4-protos/src/codegen/google/bundle.ts

@@ -1,16 +1,16 @@
-import * as _128 from "./api/annotations";
-import * as _129 from "./api/http";
-import * as _130 from "./protobuf/descriptor";
-import * as _131 from "./protobuf/duration";
-import * as _132 from "./protobuf/timestamp";
-import * as _133 from "./protobuf/any";
+import * as _129 from "./api/annotations";
+import * as _130 from "./api/http";
+import * as _131 from "./protobuf/descriptor";
+import * as _132 from "./protobuf/duration";
+import * as _133 from "./protobuf/timestamp";
+import * as _134 from "./protobuf/any";
 export namespace google {
-  export const api = { ..._128,
-    ..._129
+  export const api = { ..._129,
+    ..._130
   };
-  export const protobuf = { ..._130,
-    ..._131,
+  export const protobuf = { ..._131,
     ..._132,
-    ..._133
+    ..._133,
+    ..._134
   };
 }

proto/dydxprotocol/clob/streaming.proto

@@ -0,0 +1,16 @@
+syntax = "proto3";
+package dydxprotocol.clob;
+
+import "dydxprotocol/subaccounts/streaming.proto";
+import "dydxprotocol/clob/query.proto";
+
+option go_package = "github.com/dydxprotocol/v4-chain/protocol/x/clob/types";
+
+// StagedFinalizeBlockEvent is an event staged during `FinalizeBlock`.
+message StagedFinalizeBlockEvent {
+  // Contains one of StreamOrderbookFill, StreamSubaccountUpdate.
+  oneof event {
+    StreamOrderbookFill order_fill = 1;
+    dydxprotocol.subaccounts.StreamSubaccountUpdate subaccount_update = 2;
+  }
+}

protocol/app/app.go

@@ -469,6 +469,7 @@ func New(
 		statsmoduletypes.TransientStoreKey,
 		rewardsmoduletypes.TransientStoreKey,
 		indexer_manager.TransientStoreKey,
+		streaming.StreamingManagerTransientStoreKey,
 		perpetualsmoduletypes.TransientStoreKey,
 	)
 	memKeys := storetypes.NewMemoryStoreKeys(capabilitytypes.MemStoreKey, clobmoduletypes.MemStoreKey)
@@ -764,6 +765,7 @@ func New(
 		appFlags,
 		appCodec,
 		logger,
+		tkeys[streaming.StreamingManagerTransientStoreKey],
 	)
 
 	timeProvider := &timelib.TimeProviderImpl{}
@@ -2059,6 +2061,7 @@ func getFullNodeStreamingManagerFromOptions(
 	appFlags flags.Flags,
 	cdc codec.Codec,
 	logger log.Logger,
+	streamingManagerTransientStoreKey storetypes.StoreKey,
 ) (manager streamingtypes.FullNodeStreamingManager, wsServer *ws.WebsocketServer) {
 	logger = logger.With(log.ModuleKey, "full-node-streaming")
 	if appFlags.GrpcStreamingEnabled {
@@ -2072,6 +2075,7 @@ func getFullNodeStreamingManagerFromOptions(
 			appFlags.GrpcStreamingMaxBatchSize,
 			appFlags.GrpcStreamingMaxChannelBufferSize,
 			appFlags.FullNodeStreamingSnapshotInterval,
+			streamingManagerTransientStoreKey,
 		)
 
 		// Start websocket server.

protocol/lib/metrics/constants.go

@@ -202,6 +202,7 @@ const (
 	UpdateType                                              = "update_type"
 	ValidateMatches                                         = "validate_matches"
 	ValidateOrder                                           = "validate_order"
+	StreamBatchUpdatesAfterFinalizeBlock                    = "stream_batch_updates_after_finalize_block"
 
 	// MemCLOB.
 	AddedToOrderBook                     = "added_to_orderbook"

protocol/lib/metrics/metric_keys.go

@@ -70,19 +70,22 @@ const (
 	GateWithdrawalsIfNegativeTncSubaccountSeenLatency = "gate_withdrawals_if_negative_tnc_subaccount_seen_latency"
 
 	// Full node grpc
-	FullNodeGrpc                              = "full_node_grpc"
-	GrpcSendOrderbookUpdatesLatency           = "grpc_send_orderbook_updates_latency"
-	GrpcSendOrderbookSnapshotLatency          = "grpc_send_orderbook_snapshot_latency"
-	GrpcSendSubaccountSnapshotLatency         = "grpc_send_subaccount_snapshot_latency"
-	GrpcSendOrderbookFillsLatency             = "grpc_send_orderbook_fills_latency"
-	GrpcSendFinalizedSubaccountUpdatesLatency = "grpc_send_finalized_subaccount_updates_latency"
-	GrpcAddUpdateToBufferCount                = "grpc_add_update_to_buffer_count"
-	GrpcAddToSubscriptionChannelCount         = "grpc_add_to_subscription_channel_count"
-	GrpcSendResponseToSubscriberCount         = "grpc_send_response_to_subscriber_count"
-	GrpcStreamSubscriberCount                 = "grpc_stream_subscriber_count"
-	GrpcStreamNumUpdatesBuffered              = "grpc_stream_num_updates_buffered"
-	GrpcFlushUpdatesLatency                   = "grpc_flush_updates_latency"
-	GrpcSubscriptionChannelLength             = "grpc_subscription_channel_length"
+	FullNodeGrpc                                  = "full_node_grpc"
+	GrpcSendOrderbookUpdatesLatency               = "grpc_send_orderbook_updates_latency"
+	GrpcSendOrderbookSnapshotLatency              = "grpc_send_orderbook_snapshot_latency"
+	GrpcSendSubaccountSnapshotLatency             = "grpc_send_subaccount_snapshot_latency"
+	GrpcSendOrderbookFillsLatency                 = "grpc_send_orderbook_fills_latency"
+	GrpcSendFinalizedSubaccountUpdatesLatency     = "grpc_send_finalized_subaccount_updates_latency"
+	GrpcAddUpdateToBufferCount                    = "grpc_add_update_to_buffer_count"
+	GrpcAddToSubscriptionChannelCount             = "grpc_add_to_subscription_channel_count"
+	GrpcSendResponseToSubscriberCount             = "grpc_send_response_to_subscriber_count"
+	GrpcStreamSubscriberCount                     = "grpc_stream_subscriber_count"
+	GrpcStreamNumUpdatesBuffered                  = "grpc_stream_num_updates_buffered"
+	GrpcFlushUpdatesLatency                       = "grpc_flush_updates_latency"
+	GrpcSubscriptionChannelLength                 = "grpc_subscription_channel_length"
+	GrpcStagedAllFinalizeBlockUpdatesCount        = "grpc_staged_all_finalize_block_updates_count"
+	GrpcStagedFillFinalizeBlockUpdatesCount       = "grpc_staged_finalize_block_fill_updates_count"
+	GrpcStagedSubaccountFinalizeBlockUpdatesCount = "grpc_staged_finalize_block_subaccount_updates_count"
 
 	EndBlocker    = "end_blocker"
 	EndBlockerLag = "end_blocker_lag"

protocol/streaming/constants.go

@@ -0,0 +1,13 @@
+package streaming
+
+// Constants for FullNodeStreamingManager.
+const (
+	// Transient store key for storing staged events.
+	StreamingManagerTransientStoreKey = "tmp_streaming"
+
+	// Key for storing the count of staged events.
+	StagedEventsCountKey = "EvtCnt"
+
+	// Key prefix for staged events.
+	StagedEventsKeyPrefix = "Evt:"
+)

protocol/streaming/full_node_streaming_manager.go

@@ -1,15 +1,20 @@
 package streaming
 
 import (
+	"encoding/binary"
 	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/dydxprotocol/v4-chain/protocol/lib"
 	satypes "github.com/dydxprotocol/v4-chain/protocol/x/subaccounts/types"
 
 	"cosmossdk.io/log"
+	"cosmossdk.io/store/prefix"
+	storetypes "cosmossdk.io/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	ante_types "github.com/dydxprotocol/v4-chain/protocol/app/ante/types"
 	"github.com/dydxprotocol/v4-chain/protocol/lib/metrics"
 	"github.com/dydxprotocol/v4-chain/protocol/streaming/types"
 	streaming_util "github.com/dydxprotocol/v4-chain/protocol/streaming/util"
@@ -50,6 +55,9 @@ type FullNodeStreamingManagerImpl struct {
 	// Block interval in which snapshot info should be sent out in.
 	// Defaults to 0, which means only one snapshot will be sent out.
 	snapshotBlockInterval uint32
+
+	// stores the staged FinalizeBlock events for full node streaming.
+	streamingManagerTransientStoreKey storetypes.StoreKey
 }
 
 // OrderbookSubscription represents a active subscription to the orderbook updates stream.
@@ -86,6 +94,7 @@ func NewFullNodeStreamingManager(
 	maxUpdatesInCache uint32,
 	maxSubscriptionChannelSize uint32,
 	snapshotBlockInterval uint32,
+	streamingManagerTransientStoreKey storetypes.StoreKey,
 ) *FullNodeStreamingManagerImpl {
 	fullNodeStreamingManager := &FullNodeStreamingManagerImpl{
 		logger:                 logger,
@@ -102,6 +111,8 @@ func NewFullNodeStreamingManager(
 		maxUpdatesInCache:          maxUpdatesInCache,
 		maxSubscriptionChannelSize: maxSubscriptionChannelSize,
 		snapshotBlockInterval:      snapshotBlockInterval,
+
+		streamingManagerTransientStoreKey: streamingManagerTransientStoreKey,
 	}
 
 	// Start the goroutine for pushing order updates through.
@@ -367,6 +378,88 @@ func (sm *FullNodeStreamingManagerImpl) sendStreamUpdates(
 	}
 }
 
+func getStagedEventsCount(store storetypes.KVStore) uint32 {
+	countsBytes := store.Get([]byte(StagedEventsCountKey))
+	if countsBytes == nil {
+		return 0
+	}
+	return binary.BigEndian.Uint32(countsBytes)
+}
+
+// Stage a subaccount update event in transient store, during `FinalizeBlock`.
+func (sm *FullNodeStreamingManagerImpl) StageFinalizeBlockSubaccountUpdate(
+	ctx sdk.Context,
+	subaccountUpdate satypes.StreamSubaccountUpdate,
+) {
+	stagedEvent := clobtypes.StagedFinalizeBlockEvent{
+		Event: &clobtypes.StagedFinalizeBlockEvent_SubaccountUpdate{
+			SubaccountUpdate: &subaccountUpdate,
+		},
+	}
+	sm.stageFinalizeBlockEvent(
+		ctx,
+		clobtypes.Amino.MustMarshal(stagedEvent),
+	)
+}
+
+// Stage a fill event in transient store, during `FinalizeBlock`.
+// Since `FinalizeBlock` code block can be called more than once with optimistc
+// execution (once optimistcally and optionally once on the canonical block),
+// we need to stage the events in transient store and later emit them
+// during `Precommit`.
+func (sm *FullNodeStreamingManagerImpl) StageFinalizeBlockFill(
+	ctx sdk.Context,
+	fill clobtypes.StreamOrderbookFill,
+) {
+	stagedEvent := clobtypes.StagedFinalizeBlockEvent{
+		Event: &clobtypes.StagedFinalizeBlockEvent_OrderFill{
+			OrderFill: &fill,
+		},
+	}
+	sm.stageFinalizeBlockEvent(
+		ctx,
+		clobtypes.Amino.MustMarshal(stagedEvent),
+	)
+}
+
+func getStagedFinalizeBlockEvents(store storetypes.KVStore) []clobtypes.StagedFinalizeBlockEvent {
+	count := getStagedEventsCount(store)
+	events := make([]clobtypes.StagedFinalizeBlockEvent, count)
+	store = prefix.NewStore(store, []byte(StagedEventsKeyPrefix))
+	for i := uint32(0); i < count; i++ {
+		var event clobtypes.StagedFinalizeBlockEvent
+		bytes := store.Get(lib.Uint32ToKey(i))
+		clobtypes.Amino.MustUnmarshal(bytes, &event)
+		events[i] = event
+	}
+	return events
+}
+
+// Retrieve all events staged during `FinalizeBlock`.
+func (sm *FullNodeStreamingManagerImpl) GetStagedFinalizeBlockEvents(
+	ctx sdk.Context,
+) []clobtypes.StagedFinalizeBlockEvent {
+	noGasCtx := ctx.WithGasMeter(ante_types.NewFreeInfiniteGasMeter())
+	store := noGasCtx.TransientStore(sm.streamingManagerTransientStoreKey)
+	return getStagedFinalizeBlockEvents(store)
+}
+
+func (sm *FullNodeStreamingManagerImpl) stageFinalizeBlockEvent(
+	ctx sdk.Context,
+	eventBytes []byte,
+) {
+	noGasCtx := ctx.WithGasMeter(ante_types.NewFreeInfiniteGasMeter())
+	store := noGasCtx.TransientStore(sm.streamingManagerTransientStoreKey)
+
+	// Increment events count.
+	count := getStagedEventsCount(store)
+	store.Set([]byte(StagedEventsCountKey), lib.Uint32ToKey(count+1))
+
+	// Store events keyed by index.
+	store = prefix.NewStore(store, []byte(StagedEventsKeyPrefix))
+	store.Set(lib.Uint32ToKey(count), eventBytes)
+}
+
 // SendCombinedSnapshot sends messages to a particular subscriber without buffering.
 // Note this method requires the lock and assumes that the lock has already been
 // acquired by the caller.
@@ -703,6 +796,80 @@ func (sm *FullNodeStreamingManagerImpl) GetSubaccountSnapshotsForInitStreams(
 	return ret
 }
 
+// Grpc Streaming logic after consensus agrees on a block.
+// - Stream all events staged during `FinalizeBlock`.
+// - Stream orderbook updates to sync fills in local ops queue.
+func (sm *FullNodeStreamingManagerImpl) StreamBatchUpdatesAfterFinalizeBlock(
+	ctx sdk.Context,
+	orderBookUpdatesToSyncLocalOpsQueue *clobtypes.OffchainUpdates,
+	perpetualIdToClobPairId map[uint32][]clobtypes.ClobPairId,
+) {
+	// Flush all pending updates, since we want the onchain updates to arrive in a batch.
+	sm.FlushStreamUpdates()
+
+	finalizedFills, finalizedSubaccountUpdates := sm.getStagedEventsFromFinalizeBlock(ctx)
+
+	// TODO(CT-1190): Stream below in a single batch.
+	// Send orderbook updates to sync optimistic orderbook onchain state after FinalizeBlock.
+	sm.SendOrderbookUpdates(
+		orderBookUpdatesToSyncLocalOpsQueue,
+		uint32(ctx.BlockHeight()),
+		ctx.ExecMode(),
+	)
+
+	// Send finalized fills from FinalizeBlock.
+	sm.SendOrderbookFillUpdates(
+		finalizedFills,
+		uint32(ctx.BlockHeight()),
+		ctx.ExecMode(),
+		perpetualIdToClobPairId,
+	)
+
+	// Send finalized subaccount updates from FinalizeBlock.
+	sm.SendFinalizedSubaccountUpdates(
+		finalizedSubaccountUpdates,
+		uint32(ctx.BlockHeight()),
+		ctx.ExecMode(),
+	)
+}
+
+// getStagedEventsFromFinalizeBlock returns staged events from `FinalizeBlock`.
+// It should be called after the consensus agrees on a block (e.g. Precommitter).
+func (sm *FullNodeStreamingManagerImpl) getStagedEventsFromFinalizeBlock(
+	ctx sdk.Context,
+) (
+	finalizedFills []clobtypes.StreamOrderbookFill,
+	finalizedSubaccountUpdates []satypes.StreamSubaccountUpdate,
+) {
+	// Get onchain stream events stored in transient store.
+	stagedEvents := sm.GetStagedFinalizeBlockEvents(ctx)
+
+	metrics.SetGauge(
+		metrics.GrpcStagedAllFinalizeBlockUpdatesCount,
+		float32(len(stagedEvents)),
+	)
+
+	for _, stagedEvent := range stagedEvents {
+		switch event := stagedEvent.Event.(type) {
+		case *clobtypes.StagedFinalizeBlockEvent_OrderFill:
+			finalizedFills = append(finalizedFills, *event.OrderFill)
+		case *clobtypes.StagedFinalizeBlockEvent_SubaccountUpdate:
+			finalizedSubaccountUpdates = append(finalizedSubaccountUpdates, *event.SubaccountUpdate)
+		}
+	}
+
+	metrics.SetGauge(
+		metrics.GrpcStagedSubaccountFinalizeBlockUpdatesCount,
+		float32(len(finalizedSubaccountUpdates)),
+	)
+	metrics.SetGauge(
+		metrics.GrpcStagedFillFinalizeBlockUpdatesCount,
+		float32(len(finalizedFills)),
+	)
+
+	return finalizedFills, finalizedSubaccountUpdates
+}
+
 func (sm *FullNodeStreamingManagerImpl) InitializeNewStreams(
 	getOrderbookSnapshot func(clobPairId clobtypes.ClobPairId) *clobtypes.OffchainUpdates,
 	subaccountSnapshots map[satypes.SubaccountId]*satypes.StreamSubaccountUpdate,

protocol/streaming/noop_streaming_manager.go

@@ -78,3 +78,28 @@ func (sm *NoopGrpcStreamingManager) InitializeNewStreams(
 
 func (sm *NoopGrpcStreamingManager) Stop() {
 }
+
+func (sm *NoopGrpcStreamingManager) StageFinalizeBlockFill(
+	ctx sdk.Context,
+	fill clobtypes.StreamOrderbookFill,
+) {
+}
+
+func (sm *NoopGrpcStreamingManager) GetStagedFinalizeBlockEvents(
+	ctx sdk.Context,
+) []clobtypes.StagedFinalizeBlockEvent {
+	return nil
+}
+
+func (sm *NoopGrpcStreamingManager) StageFinalizeBlockSubaccountUpdate(
+	ctx sdk.Context,
+	subaccountUpdate satypes.StreamSubaccountUpdate,
+) {
+}
+
+func (sm *NoopGrpcStreamingManager) StreamBatchUpdatesAfterFinalizeBlock(
+	ctx sdk.Context,
+	orderBookUpdatesToSyncLocalOpsQueue *clobtypes.OffchainUpdates,
+	perpetualIdToClobPairId map[uint32][]clobtypes.ClobPairId,
+) {
+}