From 5d25e123e27eb8d0d6403b5b38fd4cc121037126 Mon Sep 17 00:00:00 2001 From: JivusAyrus Date: Fri, 12 Jun 2026 13:28:20 +0530 Subject: [PATCH 1/7] feat: parallelize composition pipeline and upload pipeline for batch publishing of subgraphs --- .../subgraph/publishFederatedSubgraphs.ts | 2 +- .../src/core/services/CompositionService.ts | 554 ++++++++++++++++++ 2 files changed, 555 insertions(+), 1 deletion(-) diff --git a/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts b/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts index 225c92d349..9dcbc697da 100644 --- a/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts +++ b/controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts @@ -263,7 +263,7 @@ export function publishFederatedSubgraphs( ); const { compositionErrors, compositionWarnings, deploymentErrors } = - await compositionService.recomposeAndDeployAffected({ + await compositionService.recomposeAndDeployAffectedBatch({ actorId: authContext.userId, affectedFederatedGraphs, affectedFeatureFlags, diff --git a/controlplane/src/core/services/CompositionService.ts b/controlplane/src/core/services/CompositionService.ts index faaf0af821..634d99376f 100644 --- a/controlplane/src/core/services/CompositionService.ts +++ b/controlplane/src/core/services/CompositionService.ts @@ -7,6 +7,7 @@ import { and, eq, inArray } from 'drizzle-orm'; import { PostgresJsDatabase } from 'drizzle-orm/postgres-js'; import { FastifyBaseLogger } from 'fastify'; import { parse } from 'graphql'; +import pLimit from 'p-limit'; import { CompositionOptions, ROUTER_COMPATIBILITY_VERSION_ONE, @@ -49,6 +50,15 @@ import { FeatureFlagRepository, SubgraphsToCompose } from './../repositories/Fea import { GraphCompositionRepository } from './../repositories/GraphCompositionRepository.js'; import { SubgraphRepository } from './../repositories/SubgraphRepository.js'; +/** + * Window size for the batch publish pipeline ({@link CompositionService.recomposeAndDeployAffectedBatch}): the number + * of federated graphs / feature flags composed in parallel before being persisted and uploaded, and the max number of + * concurrent uploads / mapper rebuilds. Bounding the window caps how many composition artifacts are held in memory at + * once. Kept below the DB connection pool size (max 10, see `plugins/database.ts`) so concurrent batch publishing + * leaves connections for the rest of the control plane. Set to 1 to effectively restore sequential behavior. + */ +const COMPOSITION_DEPLOY_CONCURRENCY = 5; + @traced export class CompositionService { constructor( @@ -389,6 +399,550 @@ export class CompositionService { return result; } + async recomposeAndDeployAffectedBatch({ + actorId, + affectedFederatedGraphs, + affectedFeatureFlags, + isFeatureSubgraph, + }: { + actorId: string; + affectedFederatedGraphs: FederatedGraphDTO[]; + affectedFeatureFlags: FeatureFlagDTO[]; + isFeatureSubgraph: boolean; + }): Promise { + const orgFeatures = await this.getOrganizationFeatures(); + if (!orgFeatures.splitConfigLoading) { + return await this.legacyComposeAndDeploy({ + actorId, + federatedGraphs: affectedFederatedGraphs, + compositionOptions: { + disableResolvabilityValidation: this.disableResolvabilityValidation, + ignoreExternalKeys: orgFeatures.ignoreExternalKeys, + }, + }); + } + + const result: ComposeAndDeployResult = { + deploymentErrors: [], + compositionErrors: [], + compositionWarnings: [], + }; + + const compositionOptions: CompositionOptions = { + disableResolvabilityValidation: this.disableResolvabilityValidation, + ignoreExternalKeys: orgFeatures.ignoreExternalKeys, + }; + const limit = pLimit(COMPOSITION_DEPLOY_CONCURRENCY); + const composer = new Composer( + this.logger, + this.db, + new FederatedGraphRepository(this.logger, this.db, this.organizationId), + new SubgraphRepository(this.logger, this.db, this.organizationId), + new ContractRepository(this.logger, this.db, this.organizationId), + new GraphCompositionRepository(this.logger, this.db), + this.chClient, + this.webhookProxyUrl, + ); + const touchedGraphIds = new Set(); + + // Process in windows of COMPOSITION_DEPLOY_CONCURRENCY so only one window's composition artifacts are held in + // memory at a time: compose the window in parallel, persist it to the DB sequentially, upload it in parallel, then + // move on (releasing the artifacts). Base graphs are processed before feature flags so each graph's base + // composedSchemaVersionId exists before a feature flag references it. + const baseGraphs = isFeatureSubgraph ? [] : affectedFederatedGraphs; + for (let i = 0; i < baseGraphs.length; i += COMPOSITION_DEPLOY_CONCURRENCY) { + const window = baseGraphs.slice(i, i + COMPOSITION_DEPLOY_CONCURRENCY); + const composed = await Promise.all( + window.map((graph) => limit(() => this.composeAffectedBaseGraph(graph, compositionOptions))), + ); + await this.persistAndUploadBatch({ + actorId, + items: composed, + isFeatureFlagComposition: false, + result, + composer, + limit, + touchedGraphIds, + }); + } + + for (let i = 0; i < affectedFeatureFlags.length; i += COMPOSITION_DEPLOY_CONCURRENCY) { + const window = affectedFeatureFlags.slice(i, i + COMPOSITION_DEPLOY_CONCURRENCY); + const composed = await Promise.all( + window.map((featureFlag) => limit(() => this.composeAffectedFeatureFlag(featureFlag, compositionOptions))), + ); + await this.persistAndUploadBatch({ + actorId, + items: composed.flat(), + isFeatureFlagComposition: true, + result, + composer, + limit, + touchedGraphIds, + }); + } + + // Rebuild every affected graph's mapper in parallel. Deferred to the very end so that all router config hashes + // (base + feature flag, across every window) are written first and each rebuild reads the complete set. + const mapperSettled = await Promise.allSettled( + [...touchedGraphIds].map((federatedGraphId) => limit(() => this.updateMapperForFederatedGraph(federatedGraphId))), + ); + const mapperRejected = mapperSettled.find((outcome) => outcome.status === 'rejected'); + if (mapperRejected?.status === 'rejected') { + throw mapperRejected.reason; + } + + return result; + } + + /** + * Compose (no writes) a single affected base federated graph. Mirrors the composition step of + * {@link composeAndDeployFederatedGraph}; the deploy is handled separately by {@link persistAndUploadBatch}. + */ + private async composeAffectedBaseGraph( + federatedGraph: FederatedGraphDTO, + compositionOptions: CompositionOptions, + ): Promise { + const subgraphRepo = new SubgraphRepository(this.logger, this.db, this.organizationId); + const subgraphs = await subgraphRepo.listByFederatedGraph({ + federatedGraphTargetId: federatedGraph.targetId, + published: true, + }); + + let tagOptionsByContractName: SerializedContractTagOptions[]; + if (federatedGraph.contract) { + tagOptionsByContractName = [ + { + contractName: federatedGraph.name, + excludeTags: federatedGraph.contract.excludeTags, + includeTags: federatedGraph.contract.includeTags, + }, + ]; + } else { + const contractRepo = new ContractRepository(this.logger, this.db, this.organizationId); + const contracts = await contractRepo.bySourceFederatedGraphId(federatedGraph.id); + tagOptionsByContractName = contracts.map((contract) => ({ + contractName: contract.downstreamFederatedGraph.target.name, + excludeTags: contract.excludeTags, + includeTags: contract.includeTags, + })); + } + + const { results } = await composeGraphsInWorker({ + federatedGraph, + subgraphsToCompose: [ + { + subgraphs, + isFeatureFlagComposition: false, + featureFlagName: '', + featureFlagId: '', + }, + ], + tagOptionsByContractName, + compositionOptions, + }); + + return { federatedGraph, results }; + } + + /** + * Compose (no writes) a single affected feature flag for every federated graph it targets. Mirrors the composition + * step of {@link composeAndDeployFeatureFlag}; the deploy is handled separately by {@link persistAndUploadBatch}. + */ + private async composeAffectedFeatureFlag( + featureFlag: FeatureFlagDTO, + compositionOptions: CompositionOptions, + ): Promise { + const featureFlagRepo = new FeatureFlagRepository(this.logger, this.db, this.organizationId); + const federatedGraphs = await featureFlagRepo.getFederatedGraphsByFeatureFlag({ + featureFlagId: featureFlag.id, + namespaceId: featureFlag.namespaceId, + excludeDisabled: true, + includeContracts: true, + }); + + if (federatedGraphs.length === 0) { + return []; + } + + const subgraphRepo = new SubgraphRepository(this.logger, this.db, this.organizationId); + const graphAndCompositionResults: FederatedGraphAndCompositionResults[] = []; + for (const graph of federatedGraphs) { + const subgraphs = await subgraphRepo.listByFederatedGraph({ + federatedGraphTargetId: graph.targetId, + published: true, + }); + + const baseCompositionSubgraphs = subgraphs.map((s) => ({ + name: s.name, + url: s.routingUrl, + definitions: parse(s.schemaSDL), + })); + + const subgraphsToCompose = featureFlagRepo.getFeatureFlagRelatedSubgraphsToCompose( + new Map([[featureFlag.id, featureFlag]]), + baseCompositionSubgraphs, + subgraphs, + [], + ); + + const { results } = await composeGraphsInWorker({ + federatedGraph: graph, + subgraphsToCompose: subgraphsToCompose.map((s) => ({ + subgraphs: s.subgraphs, + isFeatureFlagComposition: s.isFeatureFlagComposition, + featureFlagName: s.featureFlagName, + featureFlagId: s.featureFlagId, + })), + tagOptionsByContractName: graph.contract + ? [ + { + contractName: graph.name, + excludeTags: graph.contract.excludeTags, + includeTags: graph.contract.includeTags, + }, + ] + : [], + compositionOptions, + }); + + graphAndCompositionResults.push({ federatedGraph: graph, results }); + } + + return graphAndCompositionResults; + } + + /** + * Deploy step for {@link recomposeAndDeployAffectedBatch}: persist all composition results to the DB sequentially + * (schema versions + router config hashes), then upload the router configs and run the admission webhooks in + * parallel. Does NOT rebuild mappers — the caller does that once, after all hashes are written. Splitting DB writes + * from uploads is what lets the uploads run fully in parallel without any per-graph grouping. + */ + private async persistAndUploadBatch({ + actorId, + items, + isFeatureFlagComposition, + result, + composer, + limit, + touchedGraphIds, + }: { + actorId: string; + items: FederatedGraphAndCompositionResults[]; + isFeatureFlagComposition: boolean; + result: ComposeAndDeployResult; + composer: Composer; + limit: ReturnType; + touchedGraphIds: Set; + }): Promise { + const fedGraphRepo = new FederatedGraphRepository(this.logger, this.db, this.organizationId); + const uploadTasks: Array<() => Promise> = []; + + // --- DB phase (sequential): save schema versions + hashes, queue uploads --- + parentLoop: for (const { federatedGraph, results } of items) { + const baseCompositionData: BaseCompositionData = { + featureFlagRouterExecutionConfigByFeatureFlagName: new Map(), + }; + const contractBaseCompositionDataByContractId = new Map(); + + for (const compositionResult of results) { + const { baseCompositionFailed } = await this.handleCompositionResult({ + actorId, + federatedGraph, + compositionResult, + result, + composer, + baseCompositionData, + }); + + if (baseCompositionFailed) { + continue parentLoop; + } + + if (compositionResult.contracts.length === 0) { + continue; + } + + for (const { contractName, artifact } of compositionResult.contracts) { + const contractGraph = await fedGraphRepo.byName(contractName, federatedGraph.namespace); + if (!contractGraph) { + throw new Error(`The contract graph "${contractName}" was not found.`); + } + + if (!artifact.success) { + result.compositionErrors.push( + ...artifact.errors.map((message) => ({ + federatedGraphName: contractGraph.name, + namespace: contractGraph.namespace, + message, + featureFlag: compositionResult.featureFlagName, + })), + ); + } + + result.compositionWarnings.push( + ...artifact.warnings.map((warning) => ({ + federatedGraphName: contractGraph.name, + namespace: contractGraph.namespace, + message: warning.message, + featureFlag: compositionResult.featureFlagName, + })), + ); + + const contractSchemaVersionId = randomUUID(); + const contractComposedGraph = deserializeComposedGraphArtifact(contractGraph, artifact); + let contractRouterExecutionConfig; + if (artifact.success) { + if (!artifact.routerExecutionConfigJson) { + throw new Error( + `Successful contract composition for federated graph "${contractGraph.name}" does not contain a router execution config.`, + ); + } + + contractRouterExecutionConfig = deserializeRouterExecutionConfig(artifact.routerExecutionConfigJson); + if (!contractRouterExecutionConfig) { + throw new Error( + `Successful contract composition for federated graph "${contractGraph.name}" did not produce a router execution config.`, + ); + } + + contractRouterExecutionConfig.version = contractSchemaVersionId; + } + + const contractComposition = await composer.saveComposition({ + composedGraph: contractComposedGraph, + composedById: actorId, + isFeatureFlagComposition: compositionResult.isFeatureFlagComposition, + federatedSchemaVersionId: contractSchemaVersionId, + routerExecutionConfig: contractRouterExecutionConfig, + featureFlagId: compositionResult.featureFlagId, + }); + + if (!artifact.success || !contractComposition.schemaVersionId) { + continue; + } + + if (!contractRouterExecutionConfig) { + throw new Error( + `Successful contract composition for federated graph "${contractGraph.name}" did not produce a router execution config.`, + ); + } + + if (!compositionResult.isFeatureFlagComposition) { + contractBaseCompositionDataByContractId.set(contractGraph.id, { + schemaVersionId: contractComposition.schemaVersionId, + routerExecutionConfig: contractRouterExecutionConfig, + featureFlagRouterExecutionConfigByFeatureFlagName: new Map(), + }); + + continue; + } + + const existingContractBaseCompositionData = contractBaseCompositionDataByContractId.get(contractGraph.id); + if (!existingContractBaseCompositionData) { + continue; + } + + existingContractBaseCompositionData.featureFlagRouterExecutionConfigByFeatureFlagName.set( + compositionResult.featureFlagName, + routerConfigToFeatureFlagExecutionConfig(contractRouterExecutionConfig), + ); + } + } + + const graph = await fedGraphRepo.byId(federatedGraph.id); + if (!graph) { + throw new Error(`Fatal: The federated graph "${federatedGraph.name}" was not found.`); + } + + if (isFeatureFlagComposition) { + await this.persistAndQueueFeatureFlagUploads({ + actorId, + graph, + featureFlagRouterExecutionConfigByFeatureFlagName: + baseCompositionData.featureFlagRouterExecutionConfigByFeatureFlagName, + composer, + result, + uploadTasks, + }); + } else { + if (!baseCompositionData.routerExecutionConfig) { + throw new Error( + `Fatal: The latest router execution config for federated graph "${federatedGraph.name}" was not generated.`, + ); + } + if (!baseCompositionData.schemaVersionId) { + throw new Error( + `Fatal: The latest base composition for federated graph "${federatedGraph.name}" was not found.`, + ); + } + + await this.persistAndQueueBaseUpload({ + actorId, + graph, + routerExecutionConfig: baseCompositionData.routerExecutionConfig, + schemaVersionId: baseCompositionData.schemaVersionId, + composer, + result, + uploadTasks, + }); + } + touchedGraphIds.add(federatedGraph.id); + + // Contracts (only present for base compositions; feature-flag contracts arrive as their own items). + for (const [contractId, { schemaVersionId, routerExecutionConfig }] of contractBaseCompositionDataByContractId) { + const contractDTO = await fedGraphRepo.byId(contractId); + if (!contractDTO) { + throw new Error(`Unexpected: Contract graph with id "${contractId}" not found after latest composition`); + } + + await this.persistAndQueueBaseUpload({ + actorId, + graph: contractDTO, + routerExecutionConfig, + schemaVersionId, + composer, + result, + uploadTasks, + }); + touchedGraphIds.add(contractDTO.id); + } + } + + // --- Upload phase (parallel): upload configs + run admission webhooks --- + const settled = await Promise.allSettled(uploadTasks.map((task) => limit(task))); + const rejected = settled.find((outcome) => outcome.status === 'rejected'); + if (rejected?.status === 'rejected') { + throw rejected.reason; + } + } + + /** + * Writes the base/contract router config hash (DB) and queues the config upload + admission webhook (deferred, + * parallel). The split-config equivalent of {@link deployGraph}, minus the mapper rebuild. + */ + private async persistAndQueueBaseUpload({ + actorId, + graph, + routerExecutionConfig, + schemaVersionId, + composer, + result, + uploadTasks, + }: { + actorId: string; + graph: FederatedGraphDTO; + routerExecutionConfig: RouterConfig; + schemaVersionId: string; + composer: Composer; + result: ComposeAndDeployResult; + uploadTasks: Array<() => Promise>; + }): Promise { + await this.saveRouterConfigHash(graph.id, undefined, routerExecutionConfig); + + const manifestBasePath = this.getManifestBasePath(graph.id); + const readyPathOverride = this.getLatestPath(graph); + if (!readyPathOverride) { + result.deploymentErrors.push({ + message: `Invalid router compatibility version "${graph.routerCompatibilityVersion}".`, + federatedGraphName: graph.name, + namespace: graph.namespace, + }); + return; + } + + uploadTasks.push(async () => { + const { errors: uploadErrors } = await composer.composeAndUploadRouterConfig({ + admissionConfig: { + cdnBaseUrl: this.admissionConfig.cdnBaseUrl, + jwtSecret: this.admissionConfig.webhookJWTSecret, + }, + baseCompositionRouterExecutionConfig: routerExecutionConfig, + baseCompositionSchemaVersionId: schemaVersionId, + blobStorage: this.blobStorage, + // The router config is split, so feature flags are uploaded separately. + featureFlagRouterExecutionConfigByFeatureFlagName: new Map(), + federatedGraphId: graph.id, + organizationId: this.organizationId, + federatedGraphAdmissionWebhookURL: graph.admissionWebhookURL, + federatedGraphAdmissionWebhookSecret: graph.admissionWebhookSecret, + actorId, + pathOverride: { + ready: `${manifestBasePath}/${readyPathOverride}`, + draft: `${manifestBasePath}/draft.json`, + }, + }); + + result.deploymentErrors.push( + ...uploadErrors + .filter((e) => e instanceof AdmissionError || e instanceof RouterConfigUploadError) + .map((e) => ({ federatedGraphName: graph.name, namespace: graph.namespace, message: e.message ?? '' })), + ); + }); + } + + /** + * Writes each feature-flag router config hash (DB) and queues the config upload + admission webhook (deferred, + * parallel). The deferred equivalent of {@link deployFeatureFlags}. + */ + private async persistAndQueueFeatureFlagUploads({ + actorId, + graph, + featureFlagRouterExecutionConfigByFeatureFlagName, + composer, + result, + uploadTasks, + }: { + actorId: string; + graph: FederatedGraphDTO; + featureFlagRouterExecutionConfigByFeatureFlagName: Map; + composer: Composer; + result: ComposeAndDeployResult; + uploadTasks: Array<() => Promise>; + }): Promise { + const baseManifestPath = this.getManifestBasePath(graph.id); + for (const [ + featureFlagName, + featureFlagRouterExecutionConfig, + ] of featureFlagRouterExecutionConfigByFeatureFlagName) { + const routerExecutionConfig = RouterConfig.fromJson({ + ...(featureFlagRouterExecutionConfig.toJson() as JsonObject), + compatibilityVersion: graph.routerCompatibilityVersion, + }); + + // Hash write stays in the sequential DB phase; only the upload + webhook is deferred to the parallel phase. + await this.saveRouterConfigHash(graph.id, featureFlagName, routerExecutionConfig); + + uploadTasks.push(async () => { + const { errors: uploadErrors } = await composer.composeAndUploadRouterConfig({ + admissionConfig: { + cdnBaseUrl: this.admissionConfig.cdnBaseUrl, + jwtSecret: this.admissionConfig.webhookJWTSecret, + }, + baseCompositionRouterExecutionConfig: routerExecutionConfig, + baseCompositionSchemaVersionId: '', + blobStorage: this.blobStorage, + featureFlagRouterExecutionConfigByFeatureFlagName: new Map(), + federatedGraphId: graph.id, + organizationId: this.organizationId, + federatedGraphAdmissionWebhookURL: graph.admissionWebhookURL, + federatedGraphAdmissionWebhookSecret: graph.admissionWebhookSecret, + actorId, + pathOverride: { + ready: `${baseManifestPath}/feature-flags/${featureFlagName}.json`, + draft: `${baseManifestPath}/feature-flags/${featureFlagName}.draft.json`, + }, + }); + + result.deploymentErrors.push( + ...uploadErrors + .filter((e) => e instanceof AdmissionError || e instanceof RouterConfigUploadError) + .map((e) => ({ federatedGraphName: graph.name, namespace: graph.namespace, message: e.message ?? '' })), + ); + }); + } + } + private async getOrganizationFeatures(): Promise { const orgRepo = new OrganizationRepository(this.logger, this.db); const ignoreExternalKeysFeature = await orgRepo.getFeature({ From cdd6b1b648ed7ec5486085ebc3a5d91adfbfde8a Mon Sep 17 00:00:00 2001 From: JivusAyrus Date: Mon, 15 Jun 2026 11:39:25 +0530 Subject: [PATCH 2/7] feat: enhance parallel processing in composition pipeline for feature flags --- .../src/core/services/CompositionService.ts | 90 +++++++++---------- 1 file changed, 44 insertions(+), 46 deletions(-) diff --git a/controlplane/src/core/services/CompositionService.ts b/controlplane/src/core/services/CompositionService.ts index 634d99376f..82f1948ace 100644 --- a/controlplane/src/core/services/CompositionService.ts +++ b/controlplane/src/core/services/CompositionService.ts @@ -469,7 +469,7 @@ export class CompositionService { for (let i = 0; i < affectedFeatureFlags.length; i += COMPOSITION_DEPLOY_CONCURRENCY) { const window = affectedFeatureFlags.slice(i, i + COMPOSITION_DEPLOY_CONCURRENCY); const composed = await Promise.all( - window.map((featureFlag) => limit(() => this.composeAffectedFeatureFlag(featureFlag, compositionOptions))), + window.map((featureFlag) => this.composeAffectedFeatureFlag(featureFlag, compositionOptions, limit)), ); await this.persistAndUploadBatch({ actorId, @@ -545,13 +545,10 @@ export class CompositionService { return { federatedGraph, results }; } - /** - * Compose (no writes) a single affected feature flag for every federated graph it targets. Mirrors the composition - * step of {@link composeAndDeployFeatureFlag}; the deploy is handled separately by {@link persistAndUploadBatch}. - */ private async composeAffectedFeatureFlag( featureFlag: FeatureFlagDTO, compositionOptions: CompositionOptions, + limit: ReturnType, ): Promise { const featureFlagRepo = new FeatureFlagRepository(this.logger, this.db, this.organizationId); const federatedGraphs = await featureFlagRepo.getFederatedGraphsByFeatureFlag({ @@ -566,50 +563,51 @@ export class CompositionService { } const subgraphRepo = new SubgraphRepository(this.logger, this.db, this.organizationId); - const graphAndCompositionResults: FederatedGraphAndCompositionResults[] = []; - for (const graph of federatedGraphs) { - const subgraphs = await subgraphRepo.listByFederatedGraph({ - federatedGraphTargetId: graph.targetId, - published: true, - }); - - const baseCompositionSubgraphs = subgraphs.map((s) => ({ - name: s.name, - url: s.routingUrl, - definitions: parse(s.schemaSDL), - })); - - const subgraphsToCompose = featureFlagRepo.getFeatureFlagRelatedSubgraphsToCompose( - new Map([[featureFlag.id, featureFlag]]), - baseCompositionSubgraphs, - subgraphs, - [], - ); + return Promise.all( + federatedGraphs.map((graph) => + limit(async () => { + const subgraphs = await subgraphRepo.listByFederatedGraph({ + federatedGraphTargetId: graph.targetId, + published: true, + }); - const { results } = await composeGraphsInWorker({ - federatedGraph: graph, - subgraphsToCompose: subgraphsToCompose.map((s) => ({ - subgraphs: s.subgraphs, - isFeatureFlagComposition: s.isFeatureFlagComposition, - featureFlagName: s.featureFlagName, - featureFlagId: s.featureFlagId, - })), - tagOptionsByContractName: graph.contract - ? [ - { - contractName: graph.name, - excludeTags: graph.contract.excludeTags, - includeTags: graph.contract.includeTags, - }, - ] - : [], - compositionOptions, - }); + const baseCompositionSubgraphs = subgraphs.map((s) => ({ + name: s.name, + url: s.routingUrl, + definitions: parse(s.schemaSDL), + })); + + const subgraphsToCompose = featureFlagRepo.getFeatureFlagRelatedSubgraphsToCompose( + new Map([[featureFlag.id, featureFlag]]), + baseCompositionSubgraphs, + subgraphs, + [], + ); - graphAndCompositionResults.push({ federatedGraph: graph, results }); - } + const { results } = await composeGraphsInWorker({ + federatedGraph: graph, + subgraphsToCompose: subgraphsToCompose.map((s) => ({ + subgraphs: s.subgraphs, + isFeatureFlagComposition: s.isFeatureFlagComposition, + featureFlagName: s.featureFlagName, + featureFlagId: s.featureFlagId, + })), + tagOptionsByContractName: graph.contract + ? [ + { + contractName: graph.name, + excludeTags: graph.contract.excludeTags, + includeTags: graph.contract.includeTags, + }, + ] + : [], + compositionOptions, + }); - return graphAndCompositionResults; + return { federatedGraph: graph, results }; + }), + ), + ); } /** From 9e570a2c314192bf318ee3a9eb29b932ecbf119c Mon Sep 17 00:00:00 2001 From: JivusAyrus Date: Mon, 15 Jun 2026 14:05:54 +0530 Subject: [PATCH 3/7] feat: implement memoization for feature flag queries and subgraph retrieval in batch processing --- .../repositories/FeatureFlagRepository.ts | 49 +++++++++++++------ .../core/repositories/SubgraphRepository.ts | 44 ++++++++++++++--- 2 files changed, 72 insertions(+), 21 deletions(-) diff --git a/controlplane/src/core/repositories/FeatureFlagRepository.ts b/controlplane/src/core/repositories/FeatureFlagRepository.ts index 6984daca6e..d4b2a07f0f 100644 --- a/controlplane/src/core/repositories/FeatureFlagRepository.ts +++ b/controlplane/src/core/repositories/FeatureFlagRepository.ts @@ -65,6 +65,25 @@ export type CheckConstituentFeatureSubgraphsResult = { featureSubgraphIds: Array; }; +export type FeatureFlagCollectCaches = { + featureFlagsByBaseSubgraphId: Map>; + matchedFeatureFlagsByLabelKey: Map>; + featureSubgraphsByFlagId: Map>; +}; + +/** Get-or-compute a cached promise (single-flight). With no cache, just computes. */ +function memoizePromise(cache: Map> | undefined, key: string, compute: () => Promise): Promise { + if (!cache) { + return compute(); + } + let promise = cache.get(key); + if (!promise) { + promise = compute(); + cache.set(key, promise); + } + return promise; +} + @traced export class FeatureFlagRepository { constructor( @@ -1107,26 +1126,28 @@ export class FeatureFlagRepository { baseSubgraphNames, fedGraphLabelMatchers, excludeDisabled, + caches, }: { baseSubgraphId: string; namespaceId: string; baseSubgraphNames: string[]; fedGraphLabelMatchers: string[]; excludeDisabled: boolean; + caches?: FeatureFlagCollectCaches; }): Promise { const featureFlagWithEnabledFeatureGraphs: FeatureFlagWithFeatureSubgraphs[] = []; - const featureFlagsBySubgraphId = await this.getFeatureFlagsByBaseSubgraphId({ + + const featureFlagsBySubgraphId = await memoizePromise( + caches?.featureFlagsByBaseSubgraphId, baseSubgraphId, - namespaceId, - excludeDisabled, - }); + () => this.getFeatureFlagsByBaseSubgraphId({ baseSubgraphId, namespaceId, excludeDisabled }), + ); - // gets all the feature flags that match the label matchers - const matchedFeatureFlags = await this.getMatchedFeatureFlags({ - namespaceId, - fedGraphLabelMatchers, - excludeDisabled, - }); + const matchedFeatureFlags = await memoizePromise( + caches?.matchedFeatureFlagsByLabelKey, + [...fedGraphLabelMatchers].sort().join(';'), + () => this.getMatchedFeatureFlags({ namespaceId, fedGraphLabelMatchers, excludeDisabled }), + ); for (const featureFlag of featureFlagsBySubgraphId) { const matched = matchedFeatureFlags.some((m) => m.id === featureFlag.id); @@ -1134,10 +1155,10 @@ export class FeatureFlagRepository { continue; } - const featureSubgraphsByFlag = await this.getFeatureSubgraphsByFeatureFlagId({ - featureFlagId: featureFlag.id, - namespaceId, - }); + // Feature subgraphs of the flag — memoized by flag id (the same flags recur across the batch). + const featureSubgraphsByFlag = await memoizePromise(caches?.featureSubgraphsByFlagId, featureFlag.id, () => + this.getFeatureSubgraphsByFeatureFlagId({ featureFlagId: featureFlag.id, namespaceId }), + ); // if there are no feature subgraphs in the flag, then skip the flag if (featureSubgraphsByFlag.length === 0) { diff --git a/controlplane/src/core/repositories/SubgraphRepository.ts b/controlplane/src/core/repositories/SubgraphRepository.ts index 9277480fc8..568fa75c7a 100644 --- a/controlplane/src/core/repositories/SubgraphRepository.ts +++ b/controlplane/src/core/repositories/SubgraphRepository.ts @@ -73,7 +73,7 @@ import { OrganizationWebhookService } from '../webhooks/OrganizationWebhookServi import { traced } from '../tracing.js'; import type { CompositionService } from '../services/CompositionService.js'; import { ContractRepository } from './ContractRepository.js'; -import { FeatureFlagRepository } from './FeatureFlagRepository.js'; +import { FeatureFlagCollectCaches, FeatureFlagRepository } from './FeatureFlagRepository.js'; import { FederatedGraphRepository } from './FederatedGraphRepository.js'; import { GraphCompositionRepository } from './GraphCompositionRepository.js'; import { OperationsRepository } from './OperationsRepository.js'; @@ -370,6 +370,12 @@ export class SubgraphRepository { tx: PostgresJsDatabase, data: UpdateSubgraphSchemaData, splitConfigFeature?: Feature, + // When provided (batch path), `listByFederatedGraph` reads are memoized per federated graph across the whole batch. + // Without it, every changed feature subgraph re-loads ALL subgraphs (with their SDL) of the same federated graph, + // making the collect step scale with (changed subgraphs × subgraphs in the graph). + listByFederatedGraphCache?: Map>, + // When provided (batch path), the feature-flag sub-queries are memoized across the whole batch. + featureFlagCaches?: FeatureFlagCollectCaches, ): Promise<{ subgraph: SubgraphDTO | undefined; affectedFederatedGraphById: Map; @@ -568,11 +574,17 @@ export class SubgraphRepository { }); for (const federatedGraphDTO of federatedGraphDTOs) { - // Retrieve all the subgraphs that compose the federated graph to retrieve the feature flags - const subgraphs = await subgraphRepo.listByFederatedGraph({ - federatedGraphTargetId: federatedGraphDTO.targetId, - published: true, - }); + // Retrieve all the subgraphs that compose the federated graph. + // Memoized across the batch (see `listByFederatedGraphCache`), so the same federated graph is loaded once, rather than once per changed subgraph. + let subgraphsPromise = listByFederatedGraphCache?.get(federatedGraphDTO.targetId); + if (!subgraphsPromise) { + subgraphsPromise = subgraphRepo.listByFederatedGraph({ + federatedGraphTargetId: federatedGraphDTO.targetId, + published: true, + }); + listByFederatedGraphCache?.set(federatedGraphDTO.targetId, subgraphsPromise); + } + const subgraphs = await subgraphsPromise; const enabledFeatureFlags = await featureFlagRepo.getFeatureFlagsByBaseSubgraphIdAndLabelMatchers({ baseSubgraphId: baseSubgraph[0].id, @@ -580,6 +592,7 @@ export class SubgraphRepository { fedGraphLabelMatchers: federatedGraphDTO.labelMatchers || [], baseSubgraphNames: subgraphs.map((subgraph) => subgraph.name), excludeDisabled: true, + caches: featureFlagCaches, }); // If an enabled feature flag includes the feature graph that has just been published, push it to the array @@ -674,8 +687,25 @@ export class SubgraphRepository { await this.db.transaction(async (tx) => { // Write every schema version and collect the affected graphs/flags. NO composition happens here. + // Memoize `listByFederatedGraph` across the batch so a federated graph's subgraphs are loaded once, not once per + // changed feature subgraph (the dominant cost when many feature subgraphs of the same graph change at once). + const listByFederatedGraphCache = new Map>(); + // Memoize the feature-flag sub-queries across the batch (matched-flags per graph, feature-subgraphs per flag). + const featureFlagCaches: FeatureFlagCollectCaches = { + featureFlagsByBaseSubgraphId: new Map(), + matchedFeatureFlagsByLabelKey: new Map(), + featureSubgraphsByFlagId: new Map(), + }; const results = await Promise.all( - items.map((item) => this.writeSchemaAndCollectAffected(tx, item, splitConfigFeature)), + items.map((item) => + this.writeSchemaAndCollectAffected( + tx, + item, + splitConfigFeature, + listByFederatedGraphCache, + featureFlagCaches, + ), + ), ); for (const [index, result] of results.entries()) { From 1e21f141766eb8876c5f9b8a686493ec681e9dd8 Mon Sep 17 00:00:00 2001 From: JivusAyrus Date: Mon, 15 Jun 2026 15:12:42 +0530 Subject: [PATCH 4/7] fix: pr suggestions --- controlplane/src/core/repositories/FeatureFlagRepository.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/controlplane/src/core/repositories/FeatureFlagRepository.ts b/controlplane/src/core/repositories/FeatureFlagRepository.ts index d4b2a07f0f..0b0a9aac2e 100644 --- a/controlplane/src/core/repositories/FeatureFlagRepository.ts +++ b/controlplane/src/core/repositories/FeatureFlagRepository.ts @@ -1139,13 +1139,13 @@ export class FeatureFlagRepository { const featureFlagsBySubgraphId = await memoizePromise( caches?.featureFlagsByBaseSubgraphId, - baseSubgraphId, + `${namespaceId}:${excludeDisabled}:${baseSubgraphId}`, () => this.getFeatureFlagsByBaseSubgraphId({ baseSubgraphId, namespaceId, excludeDisabled }), ); const matchedFeatureFlags = await memoizePromise( caches?.matchedFeatureFlagsByLabelKey, - [...fedGraphLabelMatchers].sort().join(';'), + `${namespaceId}:${excludeDisabled}:${[...fedGraphLabelMatchers].sort().join(';')}`, () => this.getMatchedFeatureFlags({ namespaceId, fedGraphLabelMatchers, excludeDisabled }), ); From a0944e0a119610ded15d206a0da20140851714f1 Mon Sep 17 00:00:00 2001 From: JivusAyrus Date: Mon, 15 Jun 2026 15:13:29 +0530 Subject: [PATCH 5/7] fix: lint --- controlplane/src/core/repositories/FeatureFlagRepository.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/controlplane/src/core/repositories/FeatureFlagRepository.ts b/controlplane/src/core/repositories/FeatureFlagRepository.ts index 0b0a9aac2e..fedda0d486 100644 --- a/controlplane/src/core/repositories/FeatureFlagRepository.ts +++ b/controlplane/src/core/repositories/FeatureFlagRepository.ts @@ -72,7 +72,11 @@ export type FeatureFlagCollectCaches = { }; /** Get-or-compute a cached promise (single-flight). With no cache, just computes. */ -function memoizePromise(cache: Map> | undefined, key: string, compute: () => Promise): Promise { +function memoizePromise( + cache: Map> | undefined, + key: string, + compute: () => Promise, +): Promise { if (!cache) { return compute(); } From 6fdce52603fa68327019bc3c832302c181a4dbfb Mon Sep 17 00:00:00 2001 From: JivusAyrus Date: Mon, 15 Jun 2026 17:02:48 +0530 Subject: [PATCH 6/7] fix: restore saveRouterConfigHash call in composeAndDeploy method --- controlplane/src/core/services/CompositionService.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/controlplane/src/core/services/CompositionService.ts b/controlplane/src/core/services/CompositionService.ts index 82f1948ace..25c69f4cdb 100644 --- a/controlplane/src/core/services/CompositionService.ts +++ b/controlplane/src/core/services/CompositionService.ts @@ -836,8 +836,6 @@ export class CompositionService { result: ComposeAndDeployResult; uploadTasks: Array<() => Promise>; }): Promise { - await this.saveRouterConfigHash(graph.id, undefined, routerExecutionConfig); - const manifestBasePath = this.getManifestBasePath(graph.id); const readyPathOverride = this.getLatestPath(graph); if (!readyPathOverride) { @@ -849,6 +847,8 @@ export class CompositionService { return; } + await this.saveRouterConfigHash(graph.id, undefined, routerExecutionConfig); + uploadTasks.push(async () => { const { errors: uploadErrors } = await composer.composeAndUploadRouterConfig({ admissionConfig: { From 0fca6d6a9c3bc94a2ff84124a6e0e349fa024973 Mon Sep 17 00:00:00 2001 From: JivusAyrus Date: Tue, 23 Jun 2026 13:49:33 +0530 Subject: [PATCH 7/7] chore: remove unnecessary comments --- controlplane/src/core/services/CompositionService.ts | 7 ------- 1 file changed, 7 deletions(-) diff --git a/controlplane/src/core/services/CompositionService.ts b/controlplane/src/core/services/CompositionService.ts index 25c69f4cdb..07cf224b75 100644 --- a/controlplane/src/core/services/CompositionService.ts +++ b/controlplane/src/core/services/CompositionService.ts @@ -50,13 +50,6 @@ import { FeatureFlagRepository, SubgraphsToCompose } from './../repositories/Fea import { GraphCompositionRepository } from './../repositories/GraphCompositionRepository.js'; import { SubgraphRepository } from './../repositories/SubgraphRepository.js'; -/** - * Window size for the batch publish pipeline ({@link CompositionService.recomposeAndDeployAffectedBatch}): the number - * of federated graphs / feature flags composed in parallel before being persisted and uploaded, and the max number of - * concurrent uploads / mapper rebuilds. Bounding the window caps how many composition artifacts are held in memory at - * once. Kept below the DB connection pool size (max 10, see `plugins/database.ts`) so concurrent batch publishing - * leaves connections for the rest of the control plane. Set to 1 to effectively restore sequential behavior. - */ const COMPOSITION_DEPLOY_CONCURRENCY = 5; @traced