GRPC Streaming add channel buffer per subscription and metrics (#1687)

protocol/app/app.go

@@ -1936,7 +1936,8 @@ func getGrpcStreamingManagerFromOptions(
 		return streaming.NewGrpcStreamingManager(
 			logger,
 			appFlags.GrpcStreamingFlushIntervalMs,
-			appFlags.GrpcStreamingMaxBufferSize,
+			appFlags.GrpcStreamingMaxBatchSize,
+			appFlags.GrpcStreamingMaxChannelBufferSize,
 		)
 	}
 	return streaming.NewNoopGrpcStreamingManager()

protocol/app/flags/flags.go

@@ -21,9 +21,10 @@ type Flags struct {
 	GrpcEnable  bool
 
 	// Grpc Streaming
-	GrpcStreamingEnabled         bool
-	GrpcStreamingFlushIntervalMs uint32
-	GrpcStreamingMaxBufferSize   uint32
+	GrpcStreamingEnabled              bool
+	GrpcStreamingFlushIntervalMs      uint32
+	GrpcStreamingMaxBatchSize         uint32
+	GrpcStreamingMaxChannelBufferSize uint32
 
 	VEOracleEnabled bool // Slinky Vote Extensions
 }
@@ -40,9 +41,10 @@ const (
 	GrpcEnable  = "grpc.enable"
 
 	// Grpc Streaming
-	GrpcStreamingEnabled         = "grpc-streaming-enabled"
-	GrpcStreamingFlushIntervalMs = "grpc-streaming-flush-interval-ms"
-	GrpcStreamingMaxBufferSize   = "grpc-streaming-max-buffer-size"
+	GrpcStreamingEnabled              = "grpc-streaming-enabled"
+	GrpcStreamingFlushIntervalMs      = "grpc-streaming-flush-interval-ms"
+	GrpcStreamingMaxBatchSize         = "grpc-streaming-max-batch-size"
+	GrpcStreamingMaxChannelBufferSize = "grpc-streaming-max-channel-buffer-size"
 
 	// Slinky VEs enabled
 	VEOracleEnabled = "slinky-vote-extension-oracle-enabled"
@@ -55,9 +57,10 @@ const (
 	DefaultNonValidatingFullNode = false
 	DefaultDdErrorTrackingFormat = false
 
-	DefaultGrpcStreamingEnabled         = false
-	DefaultGrpcStreamingFlushIntervalMs = 50
-	DefaultGrpcStreamingMaxBufferSize   = 10000
+	DefaultGrpcStreamingEnabled              = false
+	DefaultGrpcStreamingFlushIntervalMs      = 50
+	DefaultGrpcStreamingMaxBatchSize         = 10000
+	DefaultGrpcStreamingMaxChannelBufferSize = 10000
 
 	DefaultVEOracleEnabled = true
 )
@@ -99,9 +102,14 @@ func AddFlagsToCmd(cmd *cobra.Command) {
 		"Flush interval (in ms) for grpc streaming",
 	)
 	cmd.Flags().Uint32(
-		GrpcStreamingMaxBufferSize,
-		DefaultGrpcStreamingMaxBufferSize,
-		"Maximum buffer size before grpc streaming cancels all connections",
+		GrpcStreamingMaxBatchSize,
+		DefaultGrpcStreamingMaxBatchSize,
+		"Maximum batch size before grpc streaming cancels all subscriptions",
+	)
+	cmd.Flags().Uint32(
+		GrpcStreamingMaxChannelBufferSize,
+		DefaultGrpcStreamingMaxChannelBufferSize,
+		"Maximum per-subscription channel size before grpc streaming cancels a singular subscription",
 	)
 	cmd.Flags().Bool(
 		VEOracleEnabled,
@@ -122,12 +130,15 @@ func (f *Flags) Validate() error {
 		if !f.GrpcEnable {
 			return fmt.Errorf("grpc.enable must be set to true - grpc streaming requires gRPC server")
 		}
-		if f.GrpcStreamingMaxBufferSize == 0 {
-			return fmt.Errorf("grpc streaming buffer size must be positive number")
+		if f.GrpcStreamingMaxBatchSize == 0 {
+			return fmt.Errorf("grpc streaming batch size must be positive number")
 		}
 		if f.GrpcStreamingFlushIntervalMs == 0 {
 			return fmt.Errorf("grpc streaming flush interval must be positive number")
 		}
+		if f.GrpcStreamingMaxChannelBufferSize == 0 {
+			return fmt.Errorf("grpc streaming channel size must be positive number")
+		}
 	}
 	return nil
 }
@@ -148,9 +159,10 @@ func GetFlagValuesFromOptions(
 		GrpcAddress: config.DefaultGRPCAddress,
 		GrpcEnable:  true,
 
-		GrpcStreamingEnabled:         DefaultGrpcStreamingEnabled,
-		GrpcStreamingFlushIntervalMs: DefaultGrpcStreamingFlushIntervalMs,
-		GrpcStreamingMaxBufferSize:   DefaultGrpcStreamingMaxBufferSize,
+		GrpcStreamingEnabled:              DefaultGrpcStreamingEnabled,
+		GrpcStreamingFlushIntervalMs:      DefaultGrpcStreamingFlushIntervalMs,
+		GrpcStreamingMaxBatchSize:         DefaultGrpcStreamingMaxBatchSize,
+		GrpcStreamingMaxChannelBufferSize: DefaultGrpcStreamingMaxChannelBufferSize,
 
 		VEOracleEnabled: true,
 	}
@@ -204,9 +216,15 @@ func GetFlagValuesFromOptions(
 		}
 	}
 
-	if option := appOpts.Get(GrpcStreamingMaxBufferSize); option != nil {
+	if option := appOpts.Get(GrpcStreamingMaxBatchSize); option != nil {
+		if v, err := cast.ToUint32E(option); err == nil {
+			result.GrpcStreamingMaxBatchSize = v
+		}
+	}
+
+	if option := appOpts.Get(GrpcStreamingMaxChannelBufferSize); option != nil {
 		if v, err := cast.ToUint32E(option); err == nil {
-			result.GrpcStreamingMaxBufferSize = v
+			result.GrpcStreamingMaxChannelBufferSize = v
 		}
 	}
 

protocol/app/flags/flags_test.go

@@ -35,8 +35,11 @@ func TestAddFlagsToCommand(t *testing.T) {
 		fmt.Sprintf("Has %s flag", flags.GrpcStreamingFlushIntervalMs): {
 			flagName: flags.GrpcStreamingFlushIntervalMs,
 		},
-		fmt.Sprintf("Has %s flag", flags.GrpcStreamingMaxBufferSize): {
-			flagName: flags.GrpcStreamingMaxBufferSize,
+		fmt.Sprintf("Has %s flag", flags.GrpcStreamingMaxBatchSize): {
+			flagName: flags.GrpcStreamingMaxBatchSize,
+		},
+		fmt.Sprintf("Has %s flag", flags.GrpcStreamingMaxChannelBufferSize): {
+			flagName: flags.GrpcStreamingMaxChannelBufferSize,
 		},
 	}
 
@@ -69,11 +72,12 @@ func TestValidate(t *testing.T) {
 		},
 		"success - gRPC streaming enabled for validating nodes": {
 			flags: flags.Flags{
-				NonValidatingFullNode:        false,
-				GrpcEnable:                   true,
-				GrpcStreamingEnabled:         true,
-				GrpcStreamingFlushIntervalMs: 100,
-				GrpcStreamingMaxBufferSize:   10000,
+				NonValidatingFullNode:             false,
+				GrpcEnable:                        true,
+				GrpcStreamingEnabled:              true,
+				GrpcStreamingFlushIntervalMs:      100,
+				GrpcStreamingMaxBatchSize:         10000,
+				GrpcStreamingMaxChannelBufferSize: 10000,
 			},
 		},
 		"failure - gRPC disabled": {
@@ -90,26 +94,37 @@ func TestValidate(t *testing.T) {
 			},
 			expectedErr: fmt.Errorf("grpc.enable must be set to true - grpc streaming requires gRPC server"),
 		},
-		"failure - gRPC streaming enabled with zero buffer size": {
+		"failure - gRPC streaming enabled with zero batch size": {
 			flags: flags.Flags{
 				NonValidatingFullNode:        true,
 				GrpcEnable:                   true,
 				GrpcStreamingEnabled:         true,
 				GrpcStreamingFlushIntervalMs: 100,
-				GrpcStreamingMaxBufferSize:   0,
+				GrpcStreamingMaxBatchSize:    0,
 			},
-			expectedErr: fmt.Errorf("grpc streaming buffer size must be positive number"),
+			expectedErr: fmt.Errorf("grpc streaming batch size must be positive number"),
 		},
 		"failure - gRPC streaming enabled with zero flush interval ms": {
 			flags: flags.Flags{
 				NonValidatingFullNode:        true,
 				GrpcEnable:                   true,
 				GrpcStreamingEnabled:         true,
 				GrpcStreamingFlushIntervalMs: 0,
-				GrpcStreamingMaxBufferSize:   10000,
+				GrpcStreamingMaxBatchSize:    10000,
 			},
 			expectedErr: fmt.Errorf("grpc streaming flush interval must be positive number"),
 		},
+		"failure - gRPC streaming enabled with zero channel size ms": {
+			flags: flags.Flags{
+				NonValidatingFullNode:             true,
+				GrpcEnable:                        true,
+				GrpcStreamingEnabled:              true,
+				GrpcStreamingFlushIntervalMs:      100,
+				GrpcStreamingMaxBatchSize:         10000,
+				GrpcStreamingMaxChannelBufferSize: 0,
+			},
+			expectedErr: fmt.Errorf("grpc streaming channel size must be positive number"),
+		},
 	}
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {
@@ -129,44 +144,48 @@ func TestGetFlagValuesFromOptions(t *testing.T) {
 		optsMap map[string]any
 
 		// Expectations.
-		expectedNonValidatingFullNodeFlag bool
-		expectedDdAgentHost               string
-		expectedDdTraceAgentPort          uint16
-		expectedGrpcAddress               string
-		expectedGrpcEnable                bool
-		expectedGrpcStreamingEnable       bool
-		expectedGrpcStreamingFlushMs      uint32
-		expectedGrpcStreamingBufferSize   uint32
+		expectedNonValidatingFullNodeFlag         bool
+		expectedDdAgentHost                       string
+		expectedDdTraceAgentPort                  uint16
+		expectedGrpcAddress                       string
+		expectedGrpcEnable                        bool
+		expectedGrpcStreamingEnable               bool
+		expectedGrpcStreamingFlushMs              uint32
+		expectedGrpcStreamingBatchSize            uint32
+		expectedGrpcStreamingMaxChannelBufferSize uint32
 	}{
 		"Sets to default if unset": {
-			expectedNonValidatingFullNodeFlag: false,
-			expectedDdAgentHost:               "",
-			expectedDdTraceAgentPort:          8126,
-			expectedGrpcAddress:               "localhost:9090",
-			expectedGrpcEnable:                true,
-			expectedGrpcStreamingEnable:       false,
-			expectedGrpcStreamingFlushMs:      50,
-			expectedGrpcStreamingBufferSize:   10000,
+			expectedNonValidatingFullNodeFlag:         false,
+			expectedDdAgentHost:                       "",
+			expectedDdTraceAgentPort:                  8126,
+			expectedGrpcAddress:                       "localhost:9090",
+			expectedGrpcEnable:                        true,
+			expectedGrpcStreamingEnable:               false,
+			expectedGrpcStreamingFlushMs:              50,
+			expectedGrpcStreamingBatchSize:            10000,
+			expectedGrpcStreamingMaxChannelBufferSize: 10000,
 		},
 		"Sets values from options": {
 			optsMap: map[string]any{
-				flags.NonValidatingFullNodeFlag:    true,
-				flags.DdAgentHost:                  "agentHostTest",
-				flags.DdTraceAgentPort:             uint16(777),
-				flags.GrpcEnable:                   false,
-				flags.GrpcAddress:                  "localhost:9091",
-				flags.GrpcStreamingEnabled:         "true",
-				flags.GrpcStreamingFlushIntervalMs: uint32(408),
-				flags.GrpcStreamingMaxBufferSize:   uint32(650),
+				flags.NonValidatingFullNodeFlag:         true,
+				flags.DdAgentHost:                       "agentHostTest",
+				flags.DdTraceAgentPort:                  uint16(777),
+				flags.GrpcEnable:                        false,
+				flags.GrpcAddress:                       "localhost:9091",
+				flags.GrpcStreamingEnabled:              "true",
+				flags.GrpcStreamingFlushIntervalMs:      uint32(408),
+				flags.GrpcStreamingMaxBatchSize:         uint32(650),
+				flags.GrpcStreamingMaxChannelBufferSize: uint32(972),
 			},
-			expectedNonValidatingFullNodeFlag: true,
-			expectedDdAgentHost:               "agentHostTest",
-			expectedDdTraceAgentPort:          777,
-			expectedGrpcEnable:                false,
-			expectedGrpcAddress:               "localhost:9091",
-			expectedGrpcStreamingEnable:       true,
-			expectedGrpcStreamingFlushMs:      408,
-			expectedGrpcStreamingBufferSize:   650,
+			expectedNonValidatingFullNodeFlag:         true,
+			expectedDdAgentHost:                       "agentHostTest",
+			expectedDdTraceAgentPort:                  777,
+			expectedGrpcEnable:                        false,
+			expectedGrpcAddress:                       "localhost:9091",
+			expectedGrpcStreamingEnable:               true,
+			expectedGrpcStreamingFlushMs:              408,
+			expectedGrpcStreamingBatchSize:            650,
+			expectedGrpcStreamingMaxChannelBufferSize: 972,
 		},
 	}
 
@@ -216,8 +235,13 @@ func TestGetFlagValuesFromOptions(t *testing.T) {
 			)
 			require.Equal(
 				t,
-				tc.expectedGrpcStreamingBufferSize,
-				flags.GrpcStreamingMaxBufferSize,
+				tc.expectedGrpcStreamingBatchSize,
+				flags.GrpcStreamingMaxBatchSize,
+			)
+			require.Equal(
+				t,
+				tc.expectedGrpcStreamingMaxChannelBufferSize,
+				flags.GrpcStreamingMaxChannelBufferSize,
 			)
 		})
 	}

protocol/lib/metrics/metric_keys.go

@@ -69,11 +69,13 @@ const (
 	GrpcSendOrderbookUpdatesLatency   = "grpc_send_orderbook_updates_latency"
 	GrpcSendOrderbookSnapshotLatency  = "grpc_send_orderbook_snapshot_latency"
 	GrpcSendOrderbookFillsLatency     = "grpc_send_orderbook_fills_latency"
-	GrpcEmitProtocolUpdateCount       = "grpc_emit_protocol_update_count"
+	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"
 
 	EndBlocker    = "end_blocker"
 	EndBlockerLag = "end_blocker_lag"