core/services/workflows/syncerlimiter: expand keyedOwnerSettings to check default settings.Getter (#21687)

Author: jmank88

core/services/workflows/syncerlimiter/limiter.go

@@ -27,11 +27,18 @@ type Config struct {
 }
 
 type keyedOwnerSettings struct {
-	key  string
-	vals map[string]string
+	getter settings.Getter
+	key    string
+	vals   map[string]string
 }
 
 func (k keyedOwnerSettings) GetScoped(ctx context.Context, scope settings.Scope, key string) (value string, err error) {
+	if k.getter != nil {
+		value, err = k.getter.GetScoped(ctx, scope, key)
+	}
+	if value != "" {
+		return
+	}
 	if k.key != key || scope != settings.ScopeOwner {
 		return "", nil
 	}
@@ -50,7 +57,7 @@ func NewWorkflowLimits(lggr logger.Logger, cfg Config, lf limits.Factory) (limit
 	for k, v := range cfg.PerOwnerOverrides {
 		perOwner[k] = strconv.Itoa(int(v))
 	}
-	lf.Settings = keyedOwnerSettings{key: ownerLimit.Key, vals: perOwner}
+	lf.Settings = keyedOwnerSettings{getter: lf.Settings, key: ownerLimit.Key, vals: perOwner}
 	owner, err := limits.MakeResourcePoolLimiter(lf, ownerLimit)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create owner resource limiter: %w", err)

core/services/workflows/v2/engine_test.go

@@ -98,10 +98,17 @@ func TestEngine_Init(t *testing.T) {
 
 func TestEngine_Start_RateLimited(t *testing.T) {
 	t.Parallel()
+	getter, err := settings.NewTOMLGetter([]byte(`
+[global]
+WorkflowExecutionConcurrencyLimit = "2"
+[global.PerOwner]
+WorkflowExecutionConcurrencyLimit = "1"
+`))
+	require.NoError(t, err)
 	sLimiter, err := syncerlimiter.NewWorkflowLimits(logger.Test(t), syncerlimiter.Config{
-		Global:   2,
-		PerOwner: 1,
-	}, limits.Factory{})
+		Global:   0,
+		PerOwner: 0,
+	}, limits.Factory{Settings: getter})
 	require.NoError(t, err)
 
 	module := modulemocks.NewModuleV2(t)