// Package runtimeproduction is the explicit independently serving Runtime API // composition root. It imports no L1 App or legacy backend distribution. package runtimeproduction import ( "context" "crypto/tls" "database/sql" "errors" "fmt" navigationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationnavigationv1" "os" "sort" "strings" "time" featureauth "github.com/nexuscrm/auth" backendmiddleware "github.com/nexuscrm/backend/internal/host/adapters/apihost/middleware" "github.com/nexuscrm/backend/internal/managedruntime" "github.com/nexuscrm/backend/internal/runtimeapi" "github.com/nexuscrm/backend/internal/runtimeconfig" "github.com/nexuscrm/backend/internal/runtimeidentity" "github.com/nexuscrm/backend/internal/runtimepermissionstate" "github.com/nexuscrm/backend/internal/runtimeprocess" database "github.com/nexuscrm/db" kernelerrors "github.com/nexuscrm/kernel/pkg/errors" "github.com/nexuscrm/metadata/catalogsnapshot" "github.com/nexuscrm/persistence" "github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc" catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1" applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1" approvalnotificationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalnotificationflowv1" approvalsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalv1" auditsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/auditlogv1" bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1" commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1" permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1" filessemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/filesv1" flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1" layoutauthoringv2semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv2" listviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/listviewv1" mailboxauthorizationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/mailboxauthorizationv1" metadataadminsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/metadataadminreadv1" outboundintentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1" pagelayoutsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/pagelayoutresolutionv1" permissionauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionauthoringv1" querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1" accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1" attachmentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordattachmentsv1" defaultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recorddefaultsv1" mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1" recordviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordviewv1" recyclemutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinmutationv1" recyclesemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinv1" relatedsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1" reportsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/reportv1" searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1" sessionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/sessionv1" "github.com/pingcap-inc/nexuscrm-controlplane/catalogconsumer" "github.com/pingcap-inc/nexuscrm-controlplane/catalogread" "github.com/prometheus/client_golang/prometheus" "google.golang.org/grpc/credentials" ) // runtimeCapabilityDefinition is the one Runtime-owned declaration of a // capability that a production Runtime image can serve. The image's // host-compatibility artifact is checked against this set, so deployment // compatibility cannot drift from production assembly. type runtimeCapabilityDefinition struct { id string requiresOutcomeRetention bool selectionModes runtimeCapabilitySelectionMode } type runtimeCapabilitySelectionMode uint8 const runtimeCapabilityServingSelection runtimeCapabilitySelectionMode = 0 const ( runtimeCapabilityCatalogConsumerSelection runtimeCapabilitySelectionMode = 1 << iota runtimeCapabilityWorkloadSelection ) func (definition runtimeCapabilityDefinition) supportsSelection( mode runtimeCapabilitySelectionMode, ) bool { if mode == runtimeCapabilityServingSelection { return runtimeapi.UsesServingAdmission(definition.id) } return definition.selectionModes&mode != 0 } var productionRuntimeCapabilityDefinitions = []runtimeCapabilityDefinition{ {id: accesssemantic.CapabilityName}, {id: querysemantic.CapabilityName}, {id: relatedsemantic.CapabilityName}, {id: defaultsemantic.CapabilityName}, {id: recyclesemantic.CapabilityName}, {id: recyclemutationsemantic.CapabilityName}, {id: recordviewsemantic.CapabilityName}, {id: attachmentsemantic.CapabilityName}, {id: mutationsemantic.CapabilityName, requiresOutcomeRetention: true}, {id: permissionauthoringsemantic.CapabilityName, requiresOutcomeRetention: true}, {id: catalogsemantic.AppMetadataCapabilityName, selectionModes: runtimeCapabilityCatalogConsumerSelection}, {id: catalogsemantic.AppMetadataV2CapabilityName, selectionModes: runtimeCapabilityCatalogConsumerSelection}, {id: catalogsemantic.AppMetadataV3CapabilityName, selectionModes: runtimeCapabilityCatalogConsumerSelection}, {id: catalogsemantic.IntegrationBindingCapabilityName, selectionModes: runtimeCapabilityCatalogConsumerSelection}, {id: searchsemantic.CapabilityName}, {id: reportsemantic.CapabilityName}, {id: listviewsemantic.CapabilityName}, {id: filessemantic.CapabilityName}, {id: bulkingestsemantic.CapabilityName}, {id: commandsemantic.CapabilityName, selectionModes: runtimeCapabilityCatalogConsumerSelection}, {id: outboundintentsemantic.CapabilityName, selectionModes: runtimeCapabilityCatalogConsumerSelection}, {id: applicationresultsemantic.CapabilityName, selectionModes: runtimeCapabilityWorkloadSelection}, {id: mailboxauthorizationsemantic.CapabilityName, selectionModes: runtimeCapabilityWorkloadSelection}, {id: metadataadminsemantic.CapabilityName, selectionModes: runtimeCapabilityWorkloadSelection}, {id: layoutauthoringv2semantic.CapabilityName, selectionModes: runtimeCapabilityWorkloadSelection}, {id: navigationsemantic.CapabilityName, selectionModes: runtimeCapabilityWorkloadSelection}, {id: pagelayoutsemantic.CapabilityName, selectionModes: runtimeCapabilityCatalogConsumerSelection}, {id: approvalnotificationsemantic.CapabilityName, selectionModes: runtimeCapabilityWorkloadSelection}, {id: flowsemantic.CapabilityName}, {id: sessionsemantic.CapabilityName, selectionModes: runtimeCapabilityWorkloadSelection}, {id: approvalsemantic.ExecuteCapabilityName}, {id: approvalsemantic.ReadCapabilityName}, {id: auditsemantic.CapabilityName}, {id: permissionsemantic.CapabilityName}, } // productionRuntimeFeatureCapabilities are deploy-time composition facts, not // independently selectable Runtime services. Keep them out of // productionRuntimeCapabilityDefinitions so Catalog or operator config cannot // mistake a complete-host feature for a second serving path. var productionRuntimeFeatureCapabilities = []string{ outboundintentsemantic.MailAttachmentDeliveryCapabilityName, // This is a host-binary compatibility fact, not another serving route. // App releases require it together with the v1 authoring service so an // older Runtime cannot bind commands with pre-object-group semantics. permissionauthoringsemantic.RecordTypeObjectGroupsCapabilityName, permissionsemantic.FieldCreateCapabilityName, } func productionRuntimeCapabilityDefinition(id string) (runtimeCapabilityDefinition, bool) { for _, definition := range productionRuntimeCapabilityDefinitions { if definition.id == id { return definition, true } } return runtimeCapabilityDefinition{}, false } func productionRuntimeCapabilities() []string { capabilities := make( []string, 0, len(productionRuntimeCapabilityDefinitions)+len(productionRuntimeFeatureCapabilities), ) for _, definition := range productionRuntimeCapabilityDefinitions { capabilities = append(capabilities, definition.id) } capabilities = append(capabilities, productionRuntimeFeatureCapabilities...) sort.Strings(capabilities) return capabilities } func newFederatedIDTokenVerifiers( providers []runtimeconfig.FederatedOIDCProvider, ) (map[string]runtimeidentity.FederatedIDTokenVerifier, error) { verifiers := make(map[string]runtimeidentity.FederatedIDTokenVerifier, len(providers)) for _, provider := range providers { verifier, err := featureauth.NewOIDCIDTokenVerifier( featureauth.OIDCProviderConfig{ Audience: provider.Audience, TrustedIssuers: provider.TrustedIssuers, JWKSURL: provider.JWKSURL, }, ) if err != nil { return nil, fmt.Errorf( "runtime production: federated identity verifier %q: %w", provider.ProviderID, err, ) } verifiers[provider.ProviderID] = func(ctx context.Context, raw string) (runtimeidentity.FederatedIdentity, error) { identity, verifyErr := verifier.Verify(ctx, raw) return runtimeidentity.FederatedIdentity{LoginIdentifier: identity.Email}, verifyErr } } return verifiers, nil } func federatedSessionScopes( providers []runtimeconfig.FederatedOIDCProvider, ) map[string]string { scopes := make(map[string]string, len(providers)) for _, provider := range providers { scopes[provider.ProviderID] = provider.SessionScope } return scopes } // One serving AppInstallation may use only the Connected Apps declared by its // active AppRelease. The activated operation closes the binding over the exact // installation revision instead of trusting another installation that happens // to share the same database or source pack. const independentIdentityConnectedAppVisibilityQuery = `SELECT c.ComponentKind, c.BusinessKey FROM _System_CatalogAppInstallation AS i JOIN _System_CatalogAppOperation AS o ON o.AppInstallationId = i.AppInstallationId AND o.OrgId = i.OrgId AND o.AppId = i.AppId AND o.TargetReleaseDigest = i.ActiveReleaseDigest AND o.ExpectedCatalogRevision = i.CatalogRevision JOIN _System_CatalogAppReleaseComponent AS c ON c.AppId = i.AppId AND c.AppReleaseDigest = i.ActiveReleaseDigest WHERE i.AppInstallationId = ? AND i.State = 'ACTIVE' AND o.State = 'ACTIVE' AND o.Step = 'ACTIVATED' AND c.ComponentKind IN ('metadata.pack.connected-app.v1', 'metadata.pack.oauth-custom-scope.v1') ORDER BY BINARY c.ComponentKind, BINARY c.BusinessKey` const independentIdentityEffectiveOAuthVisibilityQuery = `SELECT c.ComponentKind, c.BusinessKey FROM _System_CatalogAppInstallation AS i JOIN _System_CatalogAppOperation AS o ON o.AppInstallationId = i.AppInstallationId AND o.OrgId = i.OrgId AND o.AppId = i.AppId AND o.TargetReleaseDigest = i.ActiveReleaseDigest AND o.ExpectedCatalogRevision = i.CatalogRevision JOIN _System_CatalogEffectiveMetadataComponent AS c ON c.OperationId = o.OperationId AND c.OrgId = i.OrgId AND c.AppInstallationId = i.AppInstallationId WHERE i.AppInstallationId = ? AND i.State = 'ACTIVE' AND o.State = 'ACTIVE' AND o.Step = 'ACTIVATED' AND c.ComponentKind IN ( 'metadata.effective.oauth-authority.v1', 'metadata.effective.connected-app.v1', 'metadata.effective.oauth-custom-scope.v1' ) ORDER BY BINARY c.ComponentKind, BINARY c.BusinessKey` type independentIdentityConnectedAppVisibility struct { connectedApps map[string]struct{} publicScopes map[string]struct{} } func newIndependentIdentitySystemRepository( connection *database.TiDBConnection, visibility independentIdentityConnectedAppVisibility, ) (*persistence.SystemRepository, error) { if connection == nil || connection.DB() == nil { return nil, errors.New("runtime production: identity database authority is required") } repository := persistence.NewSystemRepository(connection.DB()) if err := repository.SetTransactionManager(persistence.NewTransactionManager(connection)); err != nil { return nil, fmt.Errorf("runtime production: identity transaction authority: %w", err) } repository.SetConnectedAppRuntimeVisibilityPolicy(visibility) return repository, nil } func (p independentIdentityConnectedAppVisibility) AllowsConnectedApp(_ string, connectedAppID string) bool { _, allowed := p.connectedApps[strings.TrimSpace(connectedAppID)] return allowed } func (p independentIdentityConnectedAppVisibility) AllowsPublicOAuthScope(scope string) bool { _, allowed := p.publicScopes[strings.ToLower(strings.TrimSpace(scope))] return allowed } // AppRelease is the immutable ownership boundary in the split deployment. A // legacy SourcePackName remains audit provenance, but cannot override or widen // the exact active AppRelease's scope components. func (p independentIdentityConnectedAppVisibility) AllowsPublicOAuthScopeForPack(scope, packName string) bool { return strings.TrimSpace(packName) != "" && p.AllowsPublicOAuthScope(scope) } func loadIndependentIdentityConnectedAppVisibility( ctx context.Context, db *sql.DB, appInstallationID string, ) (independentIdentityConnectedAppVisibility, error) { if db == nil || strings.TrimSpace(appInstallationID) != appInstallationID || appInstallationID == "" { return independentIdentityConnectedAppVisibility{}, errors.New("runtime production: Connected App installation authority is required") } effective, published, err := loadIndependentIdentityEffectiveOAuthVisibility( ctx, db, appInstallationID, ) if err != nil { return independentIdentityConnectedAppVisibility{}, err } if published { return effective, nil } return loadIndependentIdentityReleaseOAuthVisibility(ctx, db, appInstallationID) } func loadIndependentIdentityEffectiveOAuthVisibility( ctx context.Context, db *sql.DB, appInstallationID string, ) (independentIdentityConnectedAppVisibility, bool, error) { rows, err := db.QueryContext( ctx, independentIdentityEffectiveOAuthVisibilityQuery, appInstallationID, ) if err != nil { return independentIdentityConnectedAppVisibility{}, false, fmt.Errorf("runtime production: read effective OAuth installation authority: %w", err) } defer rows.Close() visibility := independentIdentityConnectedAppVisibility{connectedApps: map[string]struct{}{}, publicScopes: map[string]struct{}{}} marker := false entries := 0 for rows.Next() { var componentKind, businessKey string if err := rows.Scan(&componentKind, &businessKey); err != nil { return independentIdentityConnectedAppVisibility{}, false, fmt.Errorf("runtime production: scan effective OAuth installation authority: %w", err) } businessKey = strings.TrimSpace(businessKey) if businessKey == "" { return independentIdentityConnectedAppVisibility{}, false, errors.New("runtime production: effective OAuth installation authority has an empty business key") } var target map[string]struct{} switch componentKind { case catalogsnapshot.EffectiveOAuthAuthorityMarkerComponentKind: if businessKey != catalogsnapshot.EffectiveOAuthAuthorityMarkerBusinessKey || marker { return independentIdentityConnectedAppVisibility{}, false, errors.New("runtime production: effective OAuth installation authority has an invalid completion marker") } marker = true continue case catalogsnapshot.EffectiveConnectedAppComponentKind: target = visibility.connectedApps case catalogsnapshot.EffectiveOAuthCustomScopeComponentKind: businessKey = strings.ToLower(businessKey) target = visibility.publicScopes default: return independentIdentityConnectedAppVisibility{}, false, errors.New("runtime production: effective OAuth installation authority contains an unexpected component") } if _, duplicate := target[businessKey]; duplicate { return independentIdentityConnectedAppVisibility{}, false, errors.New("runtime production: effective OAuth installation authority is ambiguous") } target[businessKey] = struct{}{} entries++ } if err := rows.Err(); err != nil { return independentIdentityConnectedAppVisibility{}, false, fmt.Errorf("runtime production: read effective OAuth installation authority: %w", err) } if !marker && entries > 0 { return independentIdentityConnectedAppVisibility{}, false, errors.New("runtime production: effective OAuth installation authority is incomplete") } return visibility, marker, nil } func loadIndependentIdentityReleaseOAuthVisibility( ctx context.Context, db *sql.DB, appInstallationID string, ) (independentIdentityConnectedAppVisibility, error) { rows, err := db.QueryContext(ctx, independentIdentityConnectedAppVisibilityQuery, appInstallationID) if err != nil { return independentIdentityConnectedAppVisibility{}, fmt.Errorf("runtime production: read Connected App installation authority: %w", err) } defer rows.Close() visibility := independentIdentityConnectedAppVisibility{connectedApps: map[string]struct{}{}, publicScopes: map[string]struct{}{}} for rows.Next() { var componentKind, businessKey string if err := rows.Scan(&componentKind, &businessKey); err != nil { return independentIdentityConnectedAppVisibility{}, fmt.Errorf("runtime production: scan Connected App installation authority: %w", err) } businessKey = strings.TrimSpace(businessKey) if businessKey == "" { return independentIdentityConnectedAppVisibility{}, errors.New("runtime production: identity installation authority has an empty business key") } var target map[string]struct{} switch componentKind { case "metadata.pack.connected-app.v1": target = visibility.connectedApps case "metadata.pack.oauth-custom-scope.v1": businessKey = strings.ToLower(businessKey) target = visibility.publicScopes default: return independentIdentityConnectedAppVisibility{}, errors.New("runtime production: identity installation authority contains an unexpected component") } if _, duplicate := target[businessKey]; duplicate { return independentIdentityConnectedAppVisibility{}, errors.New("runtime production: identity installation authority is ambiguous") } target[businessKey] = struct{}{} } if err := rows.Err(); err != nil { return independentIdentityConnectedAppVisibility{}, fmt.Errorf("runtime production: read Connected App installation authority: %w", err) } return visibility, nil } func newIndependentIdentityAuthService( ctx context.Context, db *sql.DB, connectedAppVisibility independentIdentityConnectedAppVisibility, ) (*featureauth.AuthService, error) { if err := persistence.VerifySessionBearerSchemaReadiness(ctx, db); err != nil { return nil, fmt.Errorf("runtime production: %w", err) } permissionState, err := persistence.NewPermissionStateRepository(db) if err != nil { return nil, fmt.Errorf("runtime production: permission state readiness: %w", err) } if _, err := permissionState.CheckReady(ctx); err != nil { return nil, fmt.Errorf("runtime production: permission state readiness: %w", err) } sessionRepository := persistence.NewSessionRepository(db) sessionRepository.SetConnectedAppRuntimeVisibilityPolicy(connectedAppVisibility) authService := featureauth.NewAuthService( persistence.NewUserRepository(db), sessionRepository, persistence.NewPermissionRepository(db), ) // This process builds its own auth service, so the HTTP host's wiring does not reach it. The // Runtime admissions it performs are exactly the unscoped credentials an operator would // otherwise conclude are gone — a zero here would be read as migration readiness. authService.SetCredentialScopeObserver(func(observation featureauth.CredentialScopeObservation) { backendmiddleware.RecordCredentialScopeAdmission(observation.Class, observation.Scope) }) return authService, nil } func Assemble(ctx context.Context, version string, config runtimeconfig.API) (*runtimeprocess.Process, *database.TiDBConnection, error) { identityAttemptCap, err := time.ParseDuration(config.Serving.IdentityAttemptCap) if err != nil || identityAttemptCap <= 0 { return nil, nil, errors.New("runtime production: identity attempt cap is invalid") } capabilityLeases := make(map[string]time.Duration, len(config.Serving.Capabilities)) allowedCapabilities := make([]string, 0, len(config.Serving.Capabilities)) outcomeRetentions := make(map[string]time.Duration) for _, capability := range config.Serving.Capabilities { if capability.AppCapabilityID == capability.RuntimeCapabilityID { return nil, nil, errors.New("runtime production: App handler and Runtime data capabilities must be distinct") } definition, supported := productionRuntimeCapabilityDefinition(capability.RuntimeCapabilityID) if !supported { return nil, nil, fmt.Errorf("runtime production: unsupported Runtime capability %q", capability.RuntimeCapabilityID) } if !definition.supportsSelection(runtimeCapabilityServingSelection) { return nil, nil, fmt.Errorf( "runtime production: capability %q cannot select a Catalog serving projection", capability.RuntimeCapabilityID, ) } if !definition.requiresOutcomeRetention { if capability.OutcomeRetention != "" { return nil, nil, fmt.Errorf("runtime production: capability %q cannot own durable outcomes", capability.RuntimeCapabilityID) } } else { outcomeRetention, parseErr := time.ParseDuration(capability.OutcomeRetention) if parseErr != nil || outcomeRetention <= 0 { return nil, nil, fmt.Errorf( "runtime production: outcome retention for capability %q is invalid", capability.RuntimeCapabilityID, ) } outcomeRetentions[capability.RuntimeCapabilityID] = outcomeRetention } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf("runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New("runtime production: Runtime capability is selected more than once") } capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append(allowedCapabilities, capability.RuntimeCapabilityID) } if err := runtimeapi.ValidateServingCapabilitySet(allowedCapabilities); err != nil { return nil, nil, fmt.Errorf("runtime production: invalid Catalog serving capability set: %w", err) } recordMutationRetention := outcomeRetentions[mutationsemantic.CapabilityName] permissionAuthoringRetention := outcomeRetentions[permissionauthoringsemantic.CapabilityName] var ( sessionWorkloadID string applicationResultConfig *runtimeconfig.WorkloadCapability mailboxAuthorizationConfig *runtimeconfig.WorkloadCapability metadataAdminReadConfig *runtimeconfig.WorkloadCapability layoutAuthoringV2Config *runtimeconfig.WorkloadCapability applicationNavigationConfig *runtimeconfig.WorkloadCapability approvalNotificationFlowReadConfig *runtimeconfig.WorkloadCapability ) for _, capability := range config.Serving.WorkloadCapabilities { definition, supported := productionRuntimeCapabilityDefinition(capability.RuntimeCapabilityID) if !supported || !definition.supportsSelection(runtimeCapabilityWorkloadSelection) { return nil, nil, fmt.Errorf( "runtime production: unsupported workload capability %q", capability.RuntimeCapabilityID, ) } switch capability.RuntimeCapabilityID { case sessionsemantic.CapabilityName: if capability.RequiredCustomPermission != "" || capability.RequiredSystemPermission != "" { return nil, nil, fmt.Errorf( "runtime production: capability %q cannot declare a permission requirement", capability.RuntimeCapabilityID, ) } if capability.AdmissionLease != "" { return nil, nil, fmt.Errorf( "runtime production: capability %q uses the identity attempt cap and cannot declare a Platform admission lease", capability.RuntimeCapabilityID, ) } if sessionWorkloadID != "" { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } sessionWorkloadID = capability.AuthorizedWorkloadID case applicationresultsemantic.CapabilityName: if capability.RequiredCustomPermission != "" || capability.RequiredSystemPermission != "" { return nil, nil, fmt.Errorf( "runtime production: capability %q cannot declare a permission requirement", capability.RuntimeCapabilityID, ) } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf( "runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID, ) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } selected := capability applicationResultConfig = &selected capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append( allowedCapabilities, capability.RuntimeCapabilityID, ) case mailboxauthorizationsemantic.CapabilityName: if (capability.RequiredCustomPermission == "") == (capability.RequiredSystemPermission == "") { return nil, nil, errors.New( "runtime production: mailbox authorization requires exactly one system or custom permission", ) } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf( "runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID, ) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } selected := capability mailboxAuthorizationConfig = &selected capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append( allowedCapabilities, capability.RuntimeCapabilityID, ) case metadataadminsemantic.CapabilityName: if capability.RequiredCustomPermission != "" || capability.RequiredSystemPermission != "" { return nil, nil, fmt.Errorf( "runtime production: capability %q cannot declare a permission requirement", capability.RuntimeCapabilityID, ) } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf( "runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID, ) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } selected := capability metadataAdminReadConfig = &selected capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append( allowedCapabilities, capability.RuntimeCapabilityID, ) case layoutauthoringv2semantic.CapabilityName: if capability.RequiredCustomPermission != "" || capability.RequiredSystemPermission != "" { return nil, nil, fmt.Errorf( "runtime production: capability %q cannot declare a permission requirement", capability.RuntimeCapabilityID, ) } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf( "runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID, ) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } selected := capability layoutAuthoringV2Config = &selected capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append( allowedCapabilities, capability.RuntimeCapabilityID, ) case navigationsemantic.CapabilityName: if capability.RequiredCustomPermission != "" || capability.RequiredSystemPermission != "" || !navigationsemantic.ValidIdentifier(capability.ApplicationOwnerID) { return nil, nil, fmt.Errorf( "runtime production: capability %q cannot declare a permission requirement", capability.RuntimeCapabilityID, ) } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf( "runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID, ) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } selected := capability applicationNavigationConfig = &selected capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append( allowedCapabilities, capability.RuntimeCapabilityID, ) case approvalnotificationsemantic.CapabilityName: if capability.RequiredCustomPermission != "" || capability.RequiredSystemPermission != "" { return nil, nil, fmt.Errorf( "runtime production: capability %q cannot declare a permission requirement", capability.RuntimeCapabilityID, ) } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf( "runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID, ) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } selected := capability approvalNotificationFlowReadConfig = &selected capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append( allowedCapabilities, capability.RuntimeCapabilityID, ) default: return nil, nil, fmt.Errorf( "runtime production: unsupported workload capability %q", capability.RuntimeCapabilityID, ) } } if sessionWorkloadID == "" && len(config.Serving.FederatedOIDCProviders) != 0 { return nil, nil, errors.New( "runtime production: federated OIDC trust requires the session workload capability", ) } var servingConsumers []runtimeapi.ServingConsumerSelection for _, capability := range config.Serving.CatalogConsumerCapabilities { definition, supported := productionRuntimeCapabilityDefinition( capability.RuntimeCapabilityID, ) if !supported || definition.requiresOutcomeRetention || !definition.supportsSelection(runtimeCapabilityCatalogConsumerSelection) { return nil, nil, fmt.Errorf( "runtime production: unsupported Catalog consumer capability %q", capability.RuntimeCapabilityID, ) } switch capability.RuntimeCapabilityID { case catalogsemantic.AppMetadataCapabilityName, catalogsemantic.AppMetadataV2CapabilityName, catalogsemantic.AppMetadataV3CapabilityName, catalogsemantic.IntegrationBindingCapabilityName: if capability.CatalogSnapshotDigest != "" { return nil, nil, fmt.Errorf( "runtime production: Catalog disclosure capability %q cannot select a Runtime snapshot", capability.RuntimeCapabilityID, ) } case commandsemantic.CapabilityName: if capability.CatalogSnapshotDigest == "" { return nil, nil, errors.New( "runtime production: command consumer requires an exact Catalog snapshot", ) } case outboundintentsemantic.CapabilityName: if capability.CatalogSnapshotDigest == "" { return nil, nil, errors.New( "runtime production: outbound intent consumer requires an exact Catalog snapshot", ) } case pagelayoutsemantic.CapabilityName: if capability.CatalogSnapshotDigest == "" { return nil, nil, errors.New( "runtime production: page-layout consumer requires an exact Catalog snapshot", ) } default: return nil, nil, fmt.Errorf( "runtime production: capability %q is not a Catalog consumer capability", capability.RuntimeCapabilityID, ) } if capability.CatalogSnapshotDigest != "" { servingConsumers = append(servingConsumers, runtimeapi.ServingConsumerSelection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, WorkloadID: capability.AuthorizedWorkloadID, CapabilityID: capability.RuntimeCapabilityID, DeploymentSnapshotDigest: catalogsnapshot.Digest(capability.CatalogSnapshotDigest), }) } lease, parseErr := time.ParseDuration(capability.AdmissionLease) if parseErr != nil || lease <= 0 { return nil, nil, fmt.Errorf( "runtime production: admission lease for %q is invalid", capability.RuntimeCapabilityID, ) } if _, duplicate := capabilityLeases[capability.RuntimeCapabilityID]; duplicate { return nil, nil, errors.New( "runtime production: Runtime capability is selected more than once", ) } capabilityLeases[capability.RuntimeCapabilityID] = lease allowedCapabilities = append( allowedCapabilities, capability.RuntimeCapabilityID, ) } // ExecuteFlow resolves and authorizes its exact closure through the // AppInstallation-scoped AppMetadataView V3 or N-1 V2. Do not advertise a // Flow endpoint whose request-time admission can never authorize that // mandatory read. V1 remains independently servable for legacy Catalog // consumers, but it is not sufficient authority for ExecuteFlow. if capabilityLeases[flowsemantic.CapabilityName] > 0 && capabilityLeases[catalogsemantic.AppMetadataV3CapabilityName] <= 0 && capabilityLeases[catalogsemantic.AppMetadataV2CapabilityName] <= 0 { return nil, nil, fmt.Errorf( "runtime production: capability %q requires capability %q or %q", flowsemantic.CapabilityName, catalogsemantic.AppMetadataV3CapabilityName, catalogsemantic.AppMetadataV2CapabilityName, ) } // The serving document is canonically ordered by its L1 App capability. // Delegation owns a different canonical set: sorted L0 Runtime capability // IDs. Do not make verifier construction depend on the L1 ordering domain. sort.Strings(allowedCapabilities) var ( workloadAuthority runtimegrpc.Authority bindingSource runtimeapi.RuntimeBindingSource ) if config.Serving.Schema == runtimeconfig.ServingConfigSchemaV4 { workloadAuthority, bindingSource, err = runtimeidentity.LoadBearerWorkloadAuthorities( config.WorkloadIdentityTokenFile, config.Serving.AuthorizedWorkloadIDs, ) } else { var rawToken []byte rawToken, err = os.ReadFile(config.WorkloadIdentityTokenFile) if err == nil { token := strings.TrimSpace(string(rawToken)) if token == "" || strings.ContainsAny(token, " \t\r\n") { err = errors.New( "runtime production: workload credential is invalid", ) } else { workloadAuthority, bindingSource, err = runtimeidentity.NewBearerWorkloadAuthority( config.Serving.AuthorizedWorkloadID, token, ) } } else { err = fmt.Errorf( "runtime production: read workload credential: %w", err, ) } } if err != nil { return nil, nil, err } keySnapshot, err := runtimeidentity.LoadSigningKeySnapshot(config.SigningKeyFile, time.Now()) if err != nil { return nil, nil, err } verifier, err := keySnapshot.Verifier(time.Now(), allowedCapabilities) if err != nil { return nil, nil, err } transportCredentials, err := LoadGRPCServerTransportCredentials( config.TLSCertificateFile, config.TLSPrivateKeyFile, ) if err != nil { return nil, nil, err } connection, err := database.NewConnectionContext(ctx, database.DefaultPoolOptions()) if err != nil { return nil, nil, err } var securityRenewalDB *sql.DB fail := func(err error) (*runtimeprocess.Process, *database.TiDBConnection, error) { if securityRenewalDB != nil { _ = securityRenewalDB.Close() } _ = connection.Close() return nil, nil, err } db := connection.DB() // Lease renewal is a tiny control-plane write whose fail-closed deadline // must not queue behind HeavyWork on the business pool. The independent pool // uses the exact same immutable driver/DSN authority, but owns one bounded // physical session and no business query capacity. securityRenewalDB, err = connection.OpenIndependentDB(database.PoolOptions{ MaxOpenConns: 1, MaxIdleConns: 1, ConnMaxLifetime: 5 * time.Minute, ConnMaxIdleTime: 3 * time.Minute, }) if err != nil { return fail(fmt.Errorf("runtime production: open security renewal database ownership: %w", err)) } if err := securityRenewalDB.PingContext(ctx); err != nil { return fail(fmt.Errorf("runtime production: probe security renewal database ownership: %w", err)) } connectedAppVisibility, err := loadIndependentIdentityConnectedAppVisibility( ctx, db, config.Serving.AppInstallationID, ) if err != nil { return fail(err) } authService, err := newIndependentIdentityAuthService(ctx, db, connectedAppVisibility) if err != nil { return fail(err) } userRepository := persistence.NewUserRepository(db) systemRepository, err := newIndependentIdentitySystemRepository(connection, connectedAppVisibility) if err != nil { return fail(err) } sessionRepository := persistence.NewSessionRepository(db) sessionRepository.SetConnectedAppRuntimeVisibilityPolicy(connectedAppVisibility) oauthService := featureauth.NewOAuthService( userRepository, systemRepository, sessionRepository, authService, persistence.NewOAuthRepository(db), ) oauthService.SetPublicScopePolicyProvider(connectedAppVisibility) permissionStateRole := permissionStateRoleForServing( recordMutationRetention > 0, capabilityLeases[recyclemutationsemantic.CapabilityName] > 0, capabilityLeases[outboundintentsemantic.CapabilityName] > 0, permissionAuthoringRetention > 0, ) permissionStateGate, err := runtimepermissionstate.NewGate( ctx, db, runtimepermissionstate.Config{ SourceRevision: version, ConfigurationSchema: config.Serving.Schema, PermissionStateProtocol: config.Serving.PermissionStateProtocol, Role: permissionStateRole, }, ) if err != nil { return fail(fmt.Errorf("runtime production: permission state capability: %w", err)) } securityRepository := persistence.NewRuntimeSecurityRepository(db) securityRenewalRepository := persistence.NewRuntimeSecurityRepository(securityRenewalDB) var applicationResultAdmission *runtimeapi.ApplicationResultAdmission if applicationResultConfig != nil { applicationResultAdmission, err = runtimeapi.NewApplicationResultAdmission( runtimeapi.ApplicationResultSelection{ AppInstallationID: config.Serving.AppInstallationID, WorkloadID: applicationResultConfig.AuthorizedWorkloadID, }, func( ctx context.Context, appInstallationID string, callerWorkloadID string, ) (persistence.RuntimeSecurityIssuanceState, error) { return securityRepository.ReadIssuanceState( ctx, config.Serving.ReplicaID, appInstallationID, callerWorkloadID, ) }, ) if err != nil { return fail(err) } } var applicationNavigationAdmission *runtimeapi.ApplicationNavigationAdmission if applicationNavigationConfig != nil { applicationNavigationAdmission, err = runtimeapi.NewApplicationNavigationAdmission(runtimeapi.ApplicationNavigationSelection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, WorkloadID: applicationNavigationConfig.AuthorizedWorkloadID, FrontendAppID: applicationNavigationConfig.ApplicationOwnerID, }) if err != nil { return fail(err) } } var metadataAdminReadAdmission *runtimeapi.MetadataAdminReadAdmission if metadataAdminReadConfig != nil { metadataAdminReadAdmission, err = runtimeapi.NewMetadataAdminReadAdmission( runtimeapi.MetadataAdminReadSelection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, WorkloadID: metadataAdminReadConfig.AuthorizedWorkloadID, }, ) if err != nil { return fail(err) } } var layoutAuthoringV2Admission *runtimeapi.LayoutAuthoringV2Admission if layoutAuthoringV2Config != nil { layoutAuthoringV2Admission, err = runtimeapi.NewLayoutAuthoringV2Admission( runtimeapi.LayoutAuthoringV2Selection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, WorkloadID: layoutAuthoringV2Config.AuthorizedWorkloadID, }, ) if err != nil { return fail(err) } } var approvalNotificationFlowReadAdmission *runtimeapi.ApprovalNotificationFlowReadAdmission if approvalNotificationFlowReadConfig != nil { approvalNotificationFlowReadAdmission, err = runtimeapi.NewApprovalNotificationFlowReadAdmission( runtimeapi.ApprovalNotificationFlowReadSelection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, WorkloadID: approvalNotificationFlowReadConfig.AuthorizedWorkloadID, }, ) if err != nil { return fail(err) } } var servingState *runtimeapi.ServingState userAPICallers := runtimeidentity.UserAPICallerCapabilities(func(org, installation, caller string, epoch uint64) ([]string, bool) { if servingState == nil { return nil, false } return servingState.UserAPICallerCapabilities(org, installation, caller, epoch) }) securityState, err := runtimeidentity.NewRenewingSecurityState( ctx, securityRenewalRepository, securityRepository, config.Serving.OrgID, config.Serving.ReplicaID, time.Now, userAPICallers, ) if err != nil { return fail(err) } orgID, _, securityEpoch, ok := securityState.PreparedIdentity() if !ok || orgID != config.Serving.OrgID { return fail(errors.New("runtime production: prepared security identity does not match serving selection")) } sessionCapabilities, err := runtimeidentity.NewSessionCapabilityPolicy(config.Serving.SessionScopeCapabilities) if err != nil { return fail(err) } issuanceSecurity := runtimeidentity.IssuanceSecuritySourceFrom( securityRepository, config.Serving.ReplicaID, time.Now, userAPICallers, ) issuer, err := runtimeidentity.NewIssuer(runtimeidentity.IssuerConfig{ Bindings: bindingSource, PublicSessions: runtimeidentity.PublicSessionAuthorityFrom(authService), SessionCapabilities: sessionCapabilities, Security: issuanceSecurity, SigningKeys: runtimeidentity.SnapshotSigningKeyProvider(keySnapshot), Now: time.Now, }) if err != nil { return fail(err) } var sessionLifecycle *runtimeidentity.SessionLifecycle if sessionWorkloadID != "" { federatedVerifiers, verifierErr := newFederatedIDTokenVerifiers( config.Serving.FederatedOIDCProviders, ) if verifierErr != nil { return fail(verifierErr) } sessionLifecycle, err = runtimeidentity.NewSessionLifecycle( runtimeidentity.SessionLifecycleConfig{ Auth: authService, OAuth: oauthService, Security: issuanceSecurity, DocumentPolicy: sessionDocumentPolicyReader(db), AppInstallationID: config.Serving.AppInstallationID, AuthorizedWorkloadID: sessionWorkloadID, FederatedVerifiers: federatedVerifiers, FederatedScopes: federatedSessionScopes(config.Serving.FederatedOIDCProviders), FederatedPrincipals: func(ctx context.Context, loginIdentifier string) (string, error) { user, findErr := userRepository.FindUserByEmailWithPassword(ctx, loginIdentifier) if findErr != nil { return "", findErr } if user == nil || user.User == nil { return "", kernelerrors.NewNotFoundError("User", loginIdentifier) } return user.ID, nil }, }, ) if err != nil { return fail(err) } } var authorizationLifecycle *runtimeidentity.AuthorizationLifecycle if sessionWorkloadID != "" { authorizationLifecycle, err = runtimeidentity.NewAuthorizationLifecycle(runtimeidentity.AuthorizationLifecycleConfig{ Auth: authService, OAuth: oauthService, Security: issuanceSecurity, AppInstallationID: config.Serving.AppInstallationID, AuthorizedWorkloadID: sessionWorkloadID, }) if err != nil { return fail(err) } } securityAdmission, err := runtimeapi.NewSecurityAdmission(runtimeapi.SecurityAdmissionConfig{ Verifier: verifier, Bindings: bindingSource, Sessions: runtimeidentity.SessionAuthority(authService), State: securityState, }) if err != nil { return fail(err) } catalogReader, err := catalogread.New(db) if err != nil { return fail(err) } catalogSource, err := runtimeapi.CatalogServingSource(catalogReader) if err != nil { return fail(err) } store, err := catalogsnapshot.NewDirectoryStore(config.Serving.CatalogSnapshotStoreRoot) if err != nil { return fail(err) } selections := make([]runtimeapi.ServingSelection, 0, len(config.Serving.Capabilities)) for _, capability := range config.Serving.Capabilities { workloadID := capability.AuthorizedWorkloadID if workloadID == "" { workloadID = config.Serving.AuthorizedWorkloadID } selections = append(selections, runtimeapi.ServingSelection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, ExpectedWorkloadID: workloadID, AppCapabilityID: capability.AppCapabilityID, RuntimeCapabilityID: capability.RuntimeCapabilityID, ExpectedServiceReleaseDigest: catalogsnapshot.Digest(config.Serving.ExpectedServiceReleaseDigest), ExpectedOwnershipEpoch: capability.ExpectedOwnershipEpoch, ExpectedCompatibilityReceiptDigest: catalogsnapshot.Digest(capability.ExpectedCompatibilityReceiptDigest), LocalCompatibility: catalogsnapshot.Compatibility{ SchemaGeneration: catalogsnapshot.SchemaGeneration(config.Serving.SchemaGeneration), EngineGeneration: catalogsnapshot.EngineGeneration(config.Serving.EngineGeneration), }, SecurityEpoch: securityEpoch, }) } var catalogConsumerAdmission *runtimeapi.CatalogConsumerAdmission if len(config.Serving.CatalogConsumerCapabilities) != 0 { consumerSelections := make( []runtimeapi.CatalogConsumerSelection, 0, len(config.Serving.CatalogConsumerCapabilities), ) for _, capability := range config.Serving.CatalogConsumerCapabilities { if capability.RuntimeCapabilityID == commandsemantic.CapabilityName || capability.RuntimeCapabilityID == outboundintentsemantic.CapabilityName || capability.RuntimeCapabilityID == pagelayoutsemantic.CapabilityName { continue } consumerSelections = append( consumerSelections, runtimeapi.CatalogConsumerSelection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, WorkloadID: capability.AuthorizedWorkloadID, CapabilityID: capability.RuntimeCapabilityID, }, ) } if len(consumerSelections) != 0 { catalogConsumerAdmission, err = runtimeapi.NewCatalogConsumerAdmission(consumerSelections) if err != nil { return fail(err) } } } var compositionResourceClasses []runtimegrpc.ResourceClass if capabilityLeases[outboundintentsemantic.CapabilityName] > 0 { // Outbound attachment delivery is an additive method of the existing // short-unary capability, not a separately selectable capability. The // production composition must therefore declare its stream pool directly. compositionResourceClasses = append(compositionResourceClasses, runtimegrpc.ResourceClassFileStream) } platformLeases, platformResourceLimits, err := productionPlatformAdmission( capabilityLeases, config.AdmissionLimit, compositionResourceClasses..., ) if err != nil { return fail(err) } var catalogConsumerViews runtimeapi.CatalogConsumerViewEngine if platformLeases[catalogsemantic.AppMetadataCapabilityName] > 0 || platformLeases[catalogsemantic.AppMetadataV2CapabilityName] > 0 || platformLeases[catalogsemantic.AppMetadataV3CapabilityName] > 0 || platformLeases[catalogsemantic.IntegrationBindingCapabilityName] > 0 || capabilityLeases[commandsemantic.CapabilityName] > 0 || capabilityLeases[outboundintentsemantic.CapabilityName] > 0 || mailboxAuthorizationConfig != nil || platformLeases[flowsemantic.CapabilityName] > 0 { reader, readerErr := catalogconsumer.New(db, store) if readerErr != nil { return fail(readerErr) } catalogConsumerViews, readerErr = newCatalogConsumerAdapter(reader) if readerErr != nil { return fail(readerErr) } } var mailboxAuthorizationProcessConfig *runtimeapi.MailboxAuthorizationProcessConfig if mailboxAuthorizationConfig != nil { mailboxAuthorizationAdmission, admissionErr := runtimeapi.NewMailboxAuthorizationAdmission( ctx, runtimeapi.MailboxAuthorizationSelection{ OrgID: config.Serving.OrgID, AppInstallationID: config.Serving.AppInstallationID, WorkloadID: mailboxAuthorizationConfig. AuthorizedWorkloadID, }, catalogConsumerViews, ) if admissionErr != nil { return fail(admissionErr) } mailboxAuthorizationProcessConfig = &runtimeapi.MailboxAuthorizationProcessConfig{ Admission: mailboxAuthorizationAdmission, RequiredPermission: runtimeapi.MailboxPermissionRequirement{ CustomPermission: mailboxAuthorizationConfig.RequiredCustomPermission, SystemPermission: mailboxAuthorizationConfig.RequiredSystemPermission, }, } } servingState, err = runtimeapi.NewServingState(runtimeapi.ServingStateConfig{ UserAPICallers: config.Serving.UserAPICallers, Store: store, Catalog: catalogSource, AppMetadata: catalogConsumerViews, Consumers: servingConsumers, ConsumerViews: catalogConsumerViews, AssignmentPolicyEligibility: func( prepareCtx context.Context, view *catalogsnapshot.RuntimeMetadataView, ) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error) { return managedruntime.NewIndependentAssignmentPolicyEligibilityAmbientStager(prepareCtx, db, view) }, }) if err != nil { return fail(err) } servingAdmission, err := runtimeapi.NewServingAdmission(servingState) if err != nil { return fail(err) } commandConsumerAdmission := servingState.CommandConsumerAdmission() pageLayoutConsumerAdmission := servingState.PageLayoutConsumerAdmission() outboundIntentAdmission := servingState.OutboundIntentAdmission() // NewRenewingSecurityState acquires the first lease before the production // engines and Catalog-backed admissions are assembled. A cold assembly may // consume most of that lease, especially against a remote TiDB authority. // Refresh at the last serving boundary so readiness never publishes a // nearly-expired startup lease. Reconcile the same epoch into every serving // selection before the gRPC process can accept work. if err := securityState.Refresh(ctx); err != nil { return fail(fmt.Errorf("runtime production: refresh pre-serving security authority: %w", err)) } _, _, refreshedSecurityEpoch, ready := securityState.PreparedIdentity() if !ready { return fail(errors.New("runtime production: refreshed security authority is not ready")) } if mailboxAuthorizationProcessConfig != nil { if err := mailboxAuthorizationProcessConfig.Admission.ReconcileSecurityEpoch( refreshedSecurityEpoch, ); err != nil { return fail(err) } } for i := range selections { selections[i].SecurityEpoch = refreshedSecurityEpoch } if err := servingState.ReconcileAll(ctx, selections); err != nil { return fail(fmt.Errorf("runtime production: reconcile refreshed security authority: %w", err)) } servingControllerMetrics := newServingControllerMetrics() process, err := runtimeapi.AssembleProductionWithConnection(ctx, runtimeapi.ProcessConfig{ Version: version, AdminAddress: config.AdminAddress, GRPCAddress: config.GRPCAddress, ProfilingEnabled: config.ProfilingEnabled, SQLDiagnosticsEnabled: config.SQLDiagnosticsEnabled, AdmissionLimit: config.AdmissionLimit, WorkloadAuthority: workloadAuthority, SecurityAdmission: securityAdmission, ServingAdmission: servingAdmission, CatalogConsumerAdmission: catalogConsumerAdmission, CommandConsumerAdmission: commandConsumerAdmission, PageLayoutConsumerAdmission: pageLayoutConsumerAdmission, OutboundIntentAdmission: outboundIntentAdmission, ApplicationResultAdmission: applicationResultAdmission, MailboxAuthorization: mailboxAuthorizationProcessConfig, MetadataAdminReadAdmission: metadataAdminReadAdmission, LayoutAuthoringV2Admission: layoutAuthoringV2Admission, ApplicationNavigationAdmission: applicationNavigationAdmission, ApprovalNotificationFlowReadAdmission: approvalNotificationFlowReadAdmission, IdentityExchange: issuer.Exchange, SessionLifecycle: sessionLifecycle, AuthorizationLifecycle: authorizationLifecycle, FederatedLoginTransactions: authorizationLifecycle, IdentityAttemptCap: identityAttemptCap, AdditionalMetricsCollectors: []prometheus.Collector{servingControllerMetrics}, RecordAccessMaxAdmissionLease: capabilityLeases[accesssemantic.CapabilityName], QueryMaxAdmissionLease: capabilityLeases[querysemantic.CapabilityName], RelatedCountMaxAdmissionLease: capabilityLeases[relatedsemantic.CapabilityName], CreateDefaultPreviewMaxAdmissionLease: capabilityLeases[defaultsemantic.CapabilityName], RecycleBinMaxAdmissionLease: capabilityLeases[recyclesemantic.CapabilityName], RecycleBinMutationMaxAdmissionLease: capabilityLeases[recyclemutationsemantic.CapabilityName], RecordViewMaxAdmissionLease: capabilityLeases[recordviewsemantic.CapabilityName], AttachmentSourceMaxAdmissionLease: capabilityLeases[attachmentsemantic.CapabilityName], ServingEnabled: true, RecordMutationAttemptCap: capabilityLeases[mutationsemantic.CapabilityName], RecordMutationOutcomeRetention: recordMutationRetention, PermissionAuthoringOutcomeRetention: permissionAuthoringRetention, PlatformCapabilityMaxAdmissionLeases: platformLeases, PlatformResourceClassAdmissionLimits: platformResourceLimits, CatalogConsumerViews: catalogConsumerViews, TransportCredentials: transportCredentials, PermissionStateCapabilities: permissionStateGate.Capabilities(), PermissionStateReady: permissionStateGate.Ready, ServingController: func(controllerCtx context.Context) error { defer securityRenewalDB.Close() return runServingController( controllerCtx, runtimeidentity.SecurityLeaseRenewalTarget, runtimeidentity.SecurityLeaseRenewalAttemptCap, refreshedSecurityEpoch, securityState.Refresh, func() (uint64, bool) { _, _, epoch, ready := securityState.PreparedIdentity(); return epoch, ready }, func(ctx context.Context, epoch uint64) error { if mailboxAuthorizationProcessConfig != nil { if err := mailboxAuthorizationProcessConfig.Admission.ReconcileSecurityEpoch(epoch); err != nil { return err } } for i := range selections { selections[i].SecurityEpoch = epoch } return servingState.ReconcileAll(ctx, selections) }, servingControllerMetrics, ) }, }, connection) if err != nil { return fail(err) } return process, connection, nil } func permissionStateRoleForServing( recordMutationEnabled bool, recycleBinMutationEnabled bool, outboundIntentEnabled bool, permissionAuthoringEnabled bool, ) runtimepermissionstate.Role { if recordMutationEnabled || recycleBinMutationEnabled || outboundIntentEnabled || permissionAuthoringEnabled { return runtimepermissionstate.RoleReaderWriter } return runtimepermissionstate.RoleReader } // LoadGRPCServerTransportCredentials is the single L0 owner of the private // Runtime endpoint's TLS policy. Serving and metadata-independent bootstrap // use the same TLS 1.3 floor and certificate loading behavior. func LoadGRPCServerTransportCredentials(certificateFile, privateKeyFile string) (credentials.TransportCredentials, error) { certificate, err := tls.LoadX509KeyPair(certificateFile, privateKeyFile) if err != nil { return nil, fmt.Errorf("runtime production: load gRPC server identity: %w", err) } return credentials.NewTLS(&tls.Config{ MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate}, }), nil } func productionPlatformAdmission( capabilityLeases map[string]time.Duration, admissionLimit int, compositionResourceClasses ...runtimegrpc.ResourceClass, ) (map[string]time.Duration, map[runtimegrpc.ResourceClass]int, error) { platformLeases := make(map[string]time.Duration) for capability, lease := range capabilityLeases { if _, ok := runtimegrpc.PlatformCapabilityResourceClass(capability); ok { platformLeases[capability] = lease } } requiredResourceClasses := make(map[runtimegrpc.ResourceClass]struct{}) for capability := range platformLeases { resourceClass, _ := runtimegrpc.PlatformCapabilityResourceClass(capability) if resourceClass != runtimegrpc.ResourceClassShortUnary { requiredResourceClasses[resourceClass] = struct{}{} } } for _, resourceClass := range compositionResourceClasses { switch resourceClass { case runtimegrpc.ResourceClassHeavyWork, runtimegrpc.ResourceClassFileStream, runtimegrpc.ResourceClassCatalogWatch: requiredResourceClasses[resourceClass] = struct{}{} default: return nil, nil, fmt.Errorf( "runtime production: composition resource class %d is not isolatable", resourceClass, ) } } if len(requiredResourceClasses) == 0 { return platformLeases, nil, nil } // File streams and Catalog watches keep small isolated pools, while HeavyWork // receives the remaining process capacity. Reserving one global slot outside // every isolated pool ensures a saturated expensive class cannot starve the // short-unary control and selected-root paths. remaining := admissionLimit - 1 platformResourceLimits := make(map[runtimegrpc.ResourceClass]int) for _, resourceClass := range []runtimegrpc.ResourceClass{ runtimegrpc.ResourceClassFileStream, runtimegrpc.ResourceClassCatalogWatch, } { if _, required := requiredResourceClasses[resourceClass]; required { platformResourceLimits[resourceClass] = 1 remaining-- } } if _, required := requiredResourceClasses[runtimegrpc.ResourceClassHeavyWork]; required { if remaining < 1 { return nil, nil, fmt.Errorf( "runtime production: admission limit %d cannot isolate enabled Platform resource classes and reserve short-unary capacity", admissionLimit, ) } platformResourceLimits[runtimegrpc.ResourceClassHeavyWork] = remaining remaining = 0 } if remaining < 0 { return nil, nil, fmt.Errorf( "runtime production: admission limit %d cannot isolate enabled Platform resource classes and reserve short-unary capacity", admissionLimit, ) } return platformLeases, platformResourceLimits, nil }