feat(runs): apply settings to the run spec at run creation

Signed-off-by: davidlin20dev <davidlin20.dev@gmail.com>

Author: davidlin20dev

runs/service/run_service.go

@@ -14,6 +14,7 @@ import (
 
 	"connectrpc.com/connect"
 	semver "github.com/Masterminds/semver/v3"
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/settings"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -40,6 +41,7 @@ import (
 // RunService implements the RunServiceHandler interface
 type RunService struct {
 	repo            interfaces.Repository
+	settingsRepo    interfaces.SettingsRepo
 	actionsClient   actionsconnect.ActionsServiceClient
 	projectClient   projectconnect.ProjectServiceClient
 	storagePrefix   string
@@ -143,6 +145,7 @@ func (s *RunService) WatchGroups(ctx context.Context, req *connect.Request[workf
 // NewRunService creates a new RunService instance
 func NewRunService(
 	repo interfaces.Repository,
+	settingsRepo interfaces.SettingsRepo,
 	actionsClient actionsconnect.ActionsServiceClient,
 	projectClient projectconnect.ProjectServiceClient,
 	storagePrefix string,
@@ -154,6 +157,7 @@ func NewRunService(
 ) *RunService {
 	return &RunService{
 		repo:            repo,
+		settingsRepo:    settingsRepo,
 		actionsClient:   actionsClient,
 		projectClient:   projectClient,
 		storagePrefix:   storagePrefix,
@@ -276,6 +280,19 @@ func (s *RunService) CreateRun(
 	}
 	request.RunSpec = runSpec
 
+	// Settings sit between an explicit request value and the static config defaults
+	// applied below. Org is empty when the caller passed a ProjectId rather than a
+	// RunId; the storage key encoder normalizes that to the default org.
+	resolved, err := resolveSettings(ctx, s.settingsRepo, &settings.SettingsKey{
+		Org:     runId.GetOrg(),
+		Domain:  runId.GetDomain(),
+		Project: runId.GetProject(),
+	})
+	if err != nil {
+		return nil, connect.NewError(connect.CodeInternal, err)
+	}
+	applyRunSettings(runSpec, resolved)
+
 	// Stamp the run start time, but only for SDKs that understand it (>= 2.3.6) — older task
 	// templates have no {{.runStartTime}} placeholder, so leaving it unset keeps the executor from
 	// substituting anything. The scheduler sets CreateRunRequest.run_start_time to a trigger's

runs/service/settings_apply.go

@@ -0,0 +1,28 @@
+package service
+
+import (
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/settings"
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/task"
+)
+
+// applyRunSettings fills fields the caller left unset with values resolved from
+// settings. An explicit value in the request always wins, so this only ever fills
+// gaps, and a setting that is INHERIT or UNSET contributes nothing.
+func applyRunSettings(spec *task.RunSpec, resolved *settings.Settings) {
+	if spec == nil {
+		return
+	}
+
+	if spec.GetQueue() == "" && resolved.GetRun().GetDefaultQueue().GetState() ==
+		settings.SettingState_SETTING_STATE_VALUE {
+		spec.Queue = resolved.GetRun().GetDefaultQueue().GetStringValue()
+	}
+
+	// The proto defines 0 as unset for this field, so there is no explicit request for
+	// "unlimited" to override. The settings validator bounds the value to 0 or
+	// [2, MaxUint32], so narrowing to uint32 cannot overflow.
+	if concurrency := resolved.GetRun().GetMaxActionConcurrency(); spec.GetMaxActionConcurrency() == 0 &&
+		concurrency.GetState() == settings.SettingState_SETTING_STATE_VALUE {
+		spec.MaxActionConcurrency = uint32(concurrency.GetIntValue())
+	}
+}

runs/service/settings_apply_test.go

@@ -0,0 +1 @@
+package service

runs/setup.go

@@ -97,6 +97,8 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 		return fmt.Errorf("runs: failed to create repository: %w", err)
 	}
 
+	settingsRepo := impl.NewSettingsRepo(sc.DB)
+
 	// In unified mode, intra-service calls go through the same mux.
 	actionsServiceCfg := cfg.ActionsService
 	if sc.BaseURL != "" {
@@ -129,7 +131,7 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 		return abortReconciler.Run(ctx)
 	})
 
-	runsSvc := service.NewRunService(repo, actionsClient, projectClient, cfg.StoragePrefix, sc.DataStore, abortReconciler, cfg.AuthMetadata.ExternalAuthServerBaseURL, cfg.TrustForwardedIdentityHeaders, cfg.IdentityHeaders)
+	runsSvc := service.NewRunService(repo, settingsRepo, actionsClient, projectClient, cfg.StoragePrefix, sc.DataStore, abortReconciler, cfg.AuthMetadata.ExternalAuthServerBaseURL, cfg.TrustForwardedIdentityHeaders, cfg.IdentityHeaders)
 	taskSvc := service.NewTaskService(repo, projectClient)
 
 	runsPath, runsHandler := workflowconnect.NewRunServiceHandler(runsSvc, connect.WithInterceptors(interceptors...))
@@ -178,7 +180,7 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 	sc.Mux.Handle(projectPath, projectHandler)
 	logger.Infof(ctx, "Mounted ProjectService at %s", projectPath)
 
-	settingsSvc := service.NewSettingsService(impl.NewSettingsRepo(sc.DB))
+	settingsSvc := service.NewSettingsService(settingsRepo)
 	settingsPath, settingsHandler := settingsconnect.NewSettingsServiceHandler(settingsSvc, connect.WithInterceptors(otelInterceptor))
 	sc.Mux.Handle(settingsPath, settingsHandler)
 	logger.Infof(ctx, "Mounted SettingsService at %s", settingsPath)

runs/test/api/setup_test.go

@@ -114,6 +114,9 @@ func TestMain(m *testing.M) {
 		exitCode = 1
 		return
 	}
+
+	settingsRepo := impl.NewSettingsRepo(testDB)
+
 	// Services validate project existence through the ProjectService client, so mount a
 	// real ProjectService on the same mux (mirrors production unified mode in setup.go).
 	endpointURL := fmt.Sprintf("http://localhost:%d", testPort)
@@ -123,7 +126,7 @@ func TestMain(m *testing.M) {
 
 	// Create RunService with a no-op actions client (points at test server; not used by watch tests)
 	actionsClient := actionsconnect.NewActionsServiceClient(http.DefaultClient, endpointURL)
-	runSvc := service.NewRunService(repo, actionsClient, projectClient, "", nil, nil, "", true, config.GetConfig().IdentityHeaders)
+	runSvc := service.NewRunService(repo, settingsRepo, actionsClient, projectClient, "", nil, nil, "", true, config.GetConfig().IdentityHeaders)
 
 	// Setup HTTP server
 	mux := http.NewServeMux()
@@ -139,7 +142,7 @@ func TestMain(m *testing.M) {
 	internalRunPath, internalRunHandler := workflowconnect.NewInternalRunServiceHandler(runSvc)
 	mux.Handle(internalRunPath, internalRunHandler)
 
-	settingsSvc := service.NewSettingsService(impl.NewSettingsRepo(testDB))
+	settingsSvc := service.NewSettingsService(settingsRepo)
 	settingsPath, settingsHandler := settingsconnect.NewSettingsServiceHandler(settingsSvc)
 	mux.Handle(settingsPath, settingsHandler)