CRE-1678: config option for maxConcurrency in wf syncer (#21324)

* cre-1678: config option for maxConcurrency in wf syncer

* cre-1678: test data fix

* cre-1678: minor

Author: mchain0

core/config/capabilities_config.go

@@ -32,6 +32,7 @@ type CapabilitiesWorkflowRegistry interface {
 	MaxConfigSize() utils.FileSize
 	RelayID() types.RelayID
 	SyncStrategy() string
+	MaxConcurrency() int
 	WorkflowStorage() WorkflowStorage
 	AdditionalSources() []AdditionalWorkflowSource
 }

core/config/docs/core.toml

@@ -519,6 +519,8 @@ MaxConfigSize = '50.00kb' # Default
 # SyncStrategy is the strategy that will be used to bring the node up to date with the latest Workflow Registry contract state.
 # Options are: event which watches for contract events or reconciliation which diffs workflow metadata state.
 SyncStrategy = 'event' # Default
+# MaxConcurrency controls the maximum number of concurrent event handlers in the workflow registry syncer.
+MaxConcurrency = 12 # Default
 
 [Capabilities.WorkflowRegistry.WorkflowStorage]
 # URL is the location for the workflow storage service to be communicated with.

core/config/toml/types.go

@@ -2264,6 +2264,7 @@ type WorkflowRegistry struct {
 	MaxEncryptedSecretsSize *utils.FileSize
 	MaxConfigSize           *utils.FileSize
 	SyncStrategy            *string
+	MaxConcurrency          *int
 	WorkflowStorage         WorkflowStorage
 	AdditionalSourcesConfig []AdditionalWorkflowSource `toml:"AdditionalSources"`
 }
@@ -2301,6 +2302,10 @@ func (r *WorkflowRegistry) setFrom(f *WorkflowRegistry) {
 		r.SyncStrategy = f.SyncStrategy
 	}
 
+	if f.MaxConcurrency != nil {
+		r.MaxConcurrency = f.MaxConcurrency
+	}
+
 	r.WorkflowStorage.setFrom(&f.WorkflowStorage)
 
 	if len(f.AdditionalSourcesConfig) > 0 {

core/services/chainlink/config_capabilities.go

@@ -229,6 +229,10 @@ func (c *capabilitiesWorkflowRegistry) SyncStrategy() string {
 	return *c.c.SyncStrategy
 }
 
+func (c *capabilitiesWorkflowRegistry) MaxConcurrency() int {
+	return *c.c.MaxConcurrency
+}
+
 func (c *capabilitiesWorkflowRegistry) WorkflowStorage() config.WorkflowStorage {
 	return &workflowStorage{
 		c: c.c.WorkflowStorage,

core/services/chainlink/config_test.go

@@ -531,6 +531,7 @@ func TestConfig_Marshal(t *testing.T) {
 			MaxEncryptedSecretsSize: ptr(utils.FileSize(26.4 * utils.KB)),
 			MaxConfigSize:           ptr(utils.FileSize(50 * utils.KB)),
 			SyncStrategy:            ptr("event"),
+			MaxConcurrency:          ptr(12),
 			WorkflowStorage: toml.WorkflowStorage{
 				ArtifactStorageHost: ptr(""),
 				URL:                 ptr(""),

core/services/chainlink/testdata/config-empty-effective.toml

@@ -326,6 +326,7 @@ MaxBinarySize = '20.00mb'
 MaxEncryptedSecretsSize = '26.40kb'
 MaxConfigSize = '50.00kb'
 SyncStrategy = 'event'
+MaxConcurrency = 12
 AdditionalSources = []
 
 [Capabilities.WorkflowRegistry.WorkflowStorage]

core/services/chainlink/testdata/config-full.toml

@@ -336,6 +336,7 @@ MaxBinarySize = '20.00mb'
 MaxEncryptedSecretsSize = '26.40kb'
 MaxConfigSize = '50.00kb'
 SyncStrategy = 'event'
+MaxConcurrency = 12
 
 [Capabilities.WorkflowRegistry.WorkflowStorage]
 ArtifactStorageHost = ''

core/services/chainlink/testdata/config-multi-chain-effective.toml

@@ -326,6 +326,7 @@ MaxBinarySize = '20.00mb'
 MaxEncryptedSecretsSize = '26.40kb'
 MaxConfigSize = '50.00kb'
 SyncStrategy = 'event'
+MaxConcurrency = 12
 AdditionalSources = []
 
 [Capabilities.WorkflowRegistry.WorkflowStorage]

core/services/cre/cre.go

@@ -936,6 +936,7 @@ func newWorkflowRegistrySyncerV2(
 	registryOpts := []syncerV2.Option{
 		syncerV2.WithAdditionalSources(addSourceConfigs),
 		syncerV2.WithShardOrchestratorClient(shardOrchestratorClient),
+		syncerV2.WithMaxConcurrency(capCfg.WorkflowRegistry().MaxConcurrency()),
 	}
 	if cfg.Sharding().ShardingEnabled() {
 		registryOpts = append(registryOpts,

core/services/workflows/syncer/v2/workflow_registry.go

@@ -34,7 +34,7 @@ var (
 	defaultTickInterval          = 12 * time.Second
 	defaultRetryInterval         = 12 * time.Second
 	defaultMaxRetryInterval      = 5 * time.Minute
-	defaultMaxConcurrency        = 50
+	defaultMaxConcurrency        = 12
 	WorkflowRegistryContractName = "WorkflowRegistry"
 
 	GetWorkflowsByDONMethodName                   = "getWorkflowListByDON"