configurable cli for channel max length

protocol/app/app.go

@@ -1958,7 +1958,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/streaming/grpc/grpc_streaming_manager.go

@@ -17,11 +17,6 @@ import (
 
 var _ types.GrpcStreamingManager = (*GrpcStreamingManagerImpl)(nil)
 
-const (
-	// TODO CHANGE THIS CONFIGURABLE
-	MAX_BUFF_LEN = 1000
-)
-
 // GrpcStreamingManagerImpl is an implementation for managing gRPC streaming subscriptions.
 type GrpcStreamingManagerImpl struct {
 	sync.Mutex
@@ -39,7 +34,8 @@ type GrpcStreamingManagerImpl struct {
 	streamUpdateCache map[uint32][]clobtypes.StreamUpdate
 	numUpdatesInCache uint32
 
-	maxUpdatesInCache uint32
+	maxUpdatesInCache          uint32
+	maxSubscriptionChannelSize uint32
 }
 
 // OrderbookSubscription represents a active subscription to the orderbook updates stream.
@@ -63,6 +59,7 @@ func NewGrpcStreamingManager(
 	logger log.Logger,
 	flushIntervalMs uint32,
 	maxUpdatesInCache uint32,
+	maxSubscriptionChannelSize uint32,
 ) *GrpcStreamingManagerImpl {
 	logger = logger.With(log.ModuleKey, "grpc-streaming")
 	grpcStreamingManager := &GrpcStreamingManagerImpl{
@@ -75,7 +72,8 @@ func NewGrpcStreamingManager(
 		streamUpdateCache: make(map[uint32][]clobtypes.StreamUpdate),
 		numUpdatesInCache: 0,
 
-		maxUpdatesInCache: maxUpdatesInCache,
+		maxUpdatesInCache:          maxUpdatesInCache,
+		maxSubscriptionChannelSize: maxSubscriptionChannelSize,
 	}
 
 	// Start the goroutine for pushing order updates through.
@@ -137,7 +135,7 @@ func (sm *GrpcStreamingManagerImpl) Subscribe(
 		subscriptionId: sm.nextSubscriptionId,
 		clobPairIds:    clobPairIds,
 		srv:            srv,
-		updatesChannel: make(chan []clobtypes.StreamUpdate, MAX_BUFF_LEN),
+		updatesChannel: make(chan []clobtypes.StreamUpdate, sm.maxSubscriptionChannelSize),
 	}
 
 	sm.logger.Info(