configurable buffer

Author: grutt

cmd/hatchet-engine/engine/run.go

@@ -356,6 +356,7 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 			dispatcher.WithCache(cacheInstance),
 			dispatcher.WithPayloadSizeThreshold(sc.Runtime.GRPCMaxMsgSize),
 			dispatcher.WithDefaultMaxWorkerBacklogSize(int64(sc.Runtime.GRPCWorkerStreamMaxBacklogSize)),
+			dispatcher.WithWorkflowRunBufferSize(sc.Runtime.WorkflowRunBufferSize),
 		)
 
 		if err != nil {
@@ -722,6 +723,7 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
 			dispatcher.WithCache(cacheInstance),
 			dispatcher.WithPayloadSizeThreshold(sc.Runtime.GRPCMaxMsgSize),
 			dispatcher.WithDefaultMaxWorkerBacklogSize(int64(sc.Runtime.GRPCWorkerStreamMaxBacklogSize)),
+			dispatcher.WithWorkflowRunBufferSize(sc.Runtime.WorkflowRunBufferSize),
 		)
 
 		if err != nil {

internal/services/dispatcher/dispatcher.go

@@ -43,6 +43,7 @@ type DispatcherImpl struct {
 	cache                       cache.Cacheable
 	payloadSizeThreshold        int
 	defaultMaxWorkerBacklogSize int64
+	workflowRunBufferSize       int
 
 	dispatcherId string
 	workers      *workers
@@ -122,6 +123,7 @@ type DispatcherOpts struct {
 	cache                       cache.Cacheable
 	payloadSizeThreshold        int
 	defaultMaxWorkerBacklogSize int64
+	workflowRunBufferSize       int
 }
 
 func defaultDispatcherOpts() *DispatcherOpts {
@@ -135,6 +137,7 @@ func defaultDispatcherOpts() *DispatcherOpts {
 		alerter:                     alerter,
 		payloadSizeThreshold:        3 * 1024 * 1024,
 		defaultMaxWorkerBacklogSize: 20,
+		workflowRunBufferSize:       1000,
 	}
 }
 
@@ -192,6 +195,12 @@ func WithDefaultMaxWorkerBacklogSize(size int64) DispatcherOpt {
 	}
 }
 
+func WithWorkflowRunBufferSize(size int) DispatcherOpt {
+	return func(opts *DispatcherOpts) {
+		opts.workflowRunBufferSize = size
+	}
+}
+
 func New(fs ...DispatcherOpt) (*DispatcherImpl, error) {
 	opts := defaultDispatcherOpts()
 
@@ -240,6 +249,7 @@ func New(fs ...DispatcherOpt) (*DispatcherImpl, error) {
 		cache:                       opts.cache,
 		payloadSizeThreshold:        opts.payloadSizeThreshold,
 		defaultMaxWorkerBacklogSize: opts.defaultMaxWorkerBacklogSize,
+		workflowRunBufferSize:       opts.workflowRunBufferSize,
 	}, nil
 }
 

internal/services/dispatcher/server_v1.go

@@ -366,7 +366,7 @@ func (s *DispatcherImpl) subscribeToWorkflowRunsV1(server contracts.Dispatcher_S
 		iterCtx, iterSpan := telemetry.NewSpan(ctx, "subscribe_to_workflow_runs_v1.iter")
 		defer iterSpan.End()
 
-		bufferSize := 1000
+		bufferSize := s.workflowRunBufferSize
 
 		if len(workflowRunIds) > bufferSize {
 			ringMu.Lock()

pkg/config/server/server.go

@@ -284,6 +284,9 @@ type ConfigFileRuntime struct {
 
 	// TaskOperationLimits controls the limits for various task operations
 	TaskOperationLimits TaskOperationLimitsConfigFile `mapstructure:"taskOperationLimits" json:"taskOperationLimits,omitempty"`
+
+	// WorkflowRunBufferSize is the buffer size for workflow run event batching in the dispatcher
+	WorkflowRunBufferSize int `mapstructure:"workflowRunBufferSize" json:"workflowRunBufferSize,omitempty" default:"1000"`
 }
 
 type InternalClientTLSConfigFile struct {
@@ -851,6 +854,9 @@ func BindAllEnv(v *viper.Viper) {
 	_ = v.BindEnv("taskOperationLimits.retryQueueLimit", "SERVER_TASK_OPERATION_LIMITS_RETRY_QUEUE_LIMIT")
 	_ = v.BindEnv("taskOperationLimits.durableSleepLimit", "SERVER_TASK_OPERATION_LIMITS_DURABLE_SLEEP_LIMIT")
 
+	// dispatcher options
+	_ = v.BindEnv("runtime.workflowRunBufferSize", "SERVER_WORKFLOW_RUN_BUFFER_SIZE")
+
 	// payload store options
 	_ = v.BindEnv("payloadStore.enablePayloadDualWrites", "SERVER_PAYLOAD_STORE_ENABLE_PAYLOAD_DUAL_WRITES")
 	_ = v.BindEnv("payloadStore.enableTaskEventPayloadDualWrites", "SERVER_PAYLOAD_STORE_ENABLE_TASK_EVENT_PAYLOAD_DUAL_WRITES")