feat(controlplane): run federated graph composition in worker pool to unblock main event loop

controlplane/package.json

@@ -90,6 +90,7 @@
     "stream-json": "^1.8.0",
     "stripe": "^14.19.0",
     "tiny-lru": "^11.2.11",
+    "tinypool": "^2.0.0",
     "uid": "^2.0.2",
     "uuid": "^10.0.0",
     "zod": "^3.22.4",

controlplane/src/core/bufservices/contract/createContract.ts

@@ -206,6 +206,7 @@ export function createContract(
         chClient: opts.chClient!,
         compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
         federatedGraphs: [{ ...contractGraph, contract }],
+        composeWorkerPool: opts.composeWorkerPool,
       });
 
       compositionErrors.push(...composition.compositionErrors);

controlplane/src/core/bufservices/contract/updateContract.ts

@@ -140,6 +140,7 @@ export function updateContract(
       },
       labelMatchers: [],
       chClient: opts.chClient!,
+      composeWorkerPool: opts.composeWorkerPool,
       compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
     });
 
@@ -155,6 +156,7 @@ export function updateContract(
       },
       blobStorage: opts.blobStorage,
       chClient: opts.chClient!,
+      composeWorkerPool: opts.composeWorkerPool,
       compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
       federatedGraphs: [
         {

controlplane/src/core/bufservices/feature-flag/createFeatureFlag.ts

@@ -205,6 +205,7 @@ export function createFeatureFlag(
         },
         blobStorage: opts.blobStorage,
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
         compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
         federatedGraphs,
       });

controlplane/src/core/bufservices/feature-flag/deleteFeatureFlag.ts

@@ -135,6 +135,7 @@ export function deleteFeatureFlag(
         },
         blobStorage: opts.blobStorage,
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
         compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
         federatedGraphs,
       });

controlplane/src/core/bufservices/feature-flag/enableFeatureFlag.ts

@@ -119,6 +119,7 @@ export function enableFeatureFlag(
         },
         blobStorage: opts.blobStorage,
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
         compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
         federatedGraphs,
       });

controlplane/src/core/bufservices/feature-flag/updateFeatureFlag.ts

@@ -174,6 +174,7 @@ export function updateFeatureFlag(
         },
         blobStorage: opts.blobStorage,
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
         compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
         federatedGraphs: allFederatedGraphsToCompose,
       });

controlplane/src/core/bufservices/federated-graph/createFederatedGraph.ts

@@ -232,6 +232,7 @@ export function createFederatedGraph(
         },
         blobStorage: opts.blobStorage,
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
         compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
         federatedGraphs: [federatedGraph],
       });

controlplane/src/core/bufservices/federated-graph/migrateFromApollo.ts

@@ -156,6 +156,7 @@ export function migrateFromApollo(
           webhookJWTSecret: opts.admissionWebhookJWTSecret,
         },
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
         compositionOptions: {
           disableResolvabilityValidation: true,
         },

controlplane/src/core/bufservices/federated-graph/moveFederatedGraph.ts

@@ -125,6 +125,7 @@ export function moveFederatedGraph(
           jwtSecret: opts.admissionWebhookJWTSecret,
         },
         opts.chClient!,
+        opts.composeWorkerPool,
       );
 
       const allDeploymentErrors: PlainMessage<DeploymentError>[] = [];
@@ -163,6 +164,7 @@ export function moveFederatedGraph(
             jwtSecret: opts.admissionWebhookJWTSecret,
           },
           opts.chClient!,
+          opts.composeWorkerPool,
         );
 
         allCompositionErrors.push(...contractErrors);

controlplane/src/core/bufservices/federated-graph/updateFederatedGraph.ts

@@ -119,6 +119,7 @@ export function updateFederatedGraph(
       admissionWebhookURL: req.admissionWebhookURL,
       blobStorage: opts.blobStorage,
       chClient: opts.chClient!,
+      composeWorkerPool: opts.composeWorkerPool,
       compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
       labelMatchers: req.labelMatchers,
       namespaceId: federatedGraph.namespaceId,

controlplane/src/core/bufservices/graph/setGraphRouterCompatibilityVersion.ts

@@ -140,6 +140,7 @@ export function setGraphRouterCompatibilityVersion(
         },
         blobStorage: opts.blobStorage,
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
         compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
         federatedGraphs: [federatedGraph],
       });

controlplane/src/core/bufservices/monograph/publishMonograph.ts

@@ -152,6 +152,7 @@ export function publishMonograph(
           webhookJWTSecret: opts.admissionWebhookJWTSecret,
         },
         opts.chClient!,
+        opts.composeWorkerPool,
       );
 
     for (const graph of updatedFederatedGraphs) {

controlplane/src/core/bufservices/monograph/updateMonograph.ts

@@ -150,6 +150,7 @@ export function updateMonograph(
         admissionWebhookURL: req.admissionWebhookURL,
         admissionWebhookSecret: req.admissionWebhookSecret,
         chClient: opts.chClient!,
+        composeWorkerPool: opts.composeWorkerPool,
       });
 
       await subgraphRepo.update(
@@ -173,6 +174,7 @@ export function updateMonograph(
           webhookJWTSecret: opts.admissionWebhookJWTSecret,
         },
         opts.chClient!,
+        opts.composeWorkerPool,
       );
 
       await auditLogRepo.addAuditLog({

controlplane/src/core/bufservices/subgraph/deleteFederatedSubgraph.ts

@@ -186,6 +186,7 @@ export function deleteFederatedSubgraph(
           chClient: opts.chClient!,
           compositionOptions: newCompositionOptions(req.disableResolvabilityValidation),
           federatedGraphs: affectedFederatedGraphs,
+          composeWorkerPool: opts.composeWorkerPool,
         });
 
         return { affectedFederatedGraphs, compositionErrors, deploymentErrors, compositionWarnings };

controlplane/src/core/bufservices/subgraph/fixSubgraphSchema.ts

@@ -182,7 +182,7 @@ export function fixSubgraphSchema(
       if (composition.errors.length > 0) {
         for (const error of composition.errors) {
           compositionErrors.push({
-            message: error.message,
+            message: error,
             federatedGraphName: composition.name,
             namespace: composition.namespace,
             featureFlag: '',

controlplane/src/core/bufservices/subgraph/moveSubgraph.ts

@@ -117,6 +117,7 @@ export function moveSubgraph(
               jwtSecret: opts.admissionWebhookJWTSecret,
             },
             opts.chClient!,
+            opts.composeWorkerPool,
             newCompositionOptions(req.disableResolvabilityValidation),
           );
 

controlplane/src/core/bufservices/subgraph/publishFederatedSubgraph.ts

@@ -574,6 +574,7 @@ export function publishFederatedSubgraph(
           webhookJWTSecret: opts.admissionWebhookJWTSecret,
         },
         opts.chClient!,
+        opts.composeWorkerPool,
         newCompositionOptions(req.disableResolvabilityValidation),
       );
 

controlplane/src/core/bufservices/subgraph/updateSubgraph.ts

@@ -184,6 +184,7 @@ export function updateSubgraph(
           webhookJWTSecret: opts.admissionWebhookJWTSecret,
         },
         opts.chClient!,
+        opts.composeWorkerPool,
         newCompositionOptions(req.disableResolvabilityValidation),
       );
 

controlplane/src/core/build-server.ts

@@ -1,3 +1,4 @@
+import { join } from 'node:path';
 import Fastify, { FastifyBaseLogger } from 'fastify';
 import { S3Client } from '@aws-sdk/client-s3';
 import { fastifyConnectPlugin } from '@connectrpc/connect-fastify';
@@ -8,6 +9,7 @@ import { compressionBrotli, compressionGzip } from '@connectrpc/connect-node';
 import fastifyGracefulShutdown from 'fastify-graceful-shutdown';
 import { App } from 'octokit';
 import { Worker } from 'bullmq';
+import { Tinypool } from 'tinypool';
 import routes from './routes.js';
 import fastifyHealth from './plugins/health.js';
 import fastifyMetrics, { MetricsPluginOptions } from './plugins/metrics.js';
@@ -136,6 +138,9 @@ export interface BuildConfig {
       key?: string; // e.g. string or '/path/to/my/client-key.pem'
     };
   };
+  composeWorkers?: {
+    maxCount: number;
+  };
 }
 
 export interface MetricsOptions {
@@ -469,6 +474,11 @@ export default async function build(opts: BuildConfig) {
     keycloakRealm: opts.keycloak.realm,
   });
 
+  const composeWorkerPool = new Tinypool({
+    filename: join(process.cwd(), 'dist/workers/compose.js'),
+    maxThreads: opts.composeWorkers?.maxCount,
+  });
+
   // Must be registered after custom fastify routes
   // Because it registers an all-catch route for connect handlers
 
@@ -502,6 +512,7 @@ export default async function build(opts: BuildConfig) {
       stripeSecretKey: opts.stripe?.secret,
       admissionWebhookJWTSecret: opts.admissionWebhook.secret,
       cdnBaseUrl: opts.cdnBaseUrl,
+      composeWorkerPool,
     }),
     contextValues(req) {
       return createContextValues().set<FastifyBaseLogger>({ id: fastifyLoggerId, defaultValue: req.log }, req.log);
@@ -522,6 +533,12 @@ export default async function build(opts: BuildConfig) {
   });
 
   fastify.gracefulShutdown(async () => {
+    fastify.log.debug('Shutting down compose worker pool');
+
+    await composeWorkerPool.destroy();
+
+    fastify.log.debug('Compose worker pool shut down');
+
     fastify.log.debug('Shutting down bull workers');
 
     await Promise.all(bullWorkers.map((worker) => worker.close()));

controlplane/src/core/composition/composer.ts

@@ -222,10 +222,10 @@ export function mapResultToComposedGraph(
     composedSchema: result.success ? printSchemaWithDirectives(result.federatedGraphSchema) : undefined,
     federatedClientSchema: result.success ? printSchemaWithDirectives(result.federatedGraphClientSchema) : undefined,
     shouldIncludeClientSchema: result.success ? result.shouldIncludeClientSchema : false,
-    errors: result.success ? [] : result.errors,
+    errors: result.success ? [] : result.errors.map((e) => e.message),
     subgraphs: subgraphDTOsToComposedSubgraphs(federatedGraph.organizationId, subgraphs, result),
     fieldConfigurations: result.success ? result.fieldConfigurations : [],
-    warnings: result.warnings,
+    warnings: result.warnings.map((e) => e.message),
   };
 }
 
@@ -236,12 +236,12 @@ export interface ComposedFederatedGraph {
   namespace: string;
   namespaceId: string;
   composedSchema?: string;
-  errors: Error[];
+  errors: string[];
   subgraphs: ComposedSubgraph[];
   fieldConfigurations: FieldConfiguration[];
   federatedClientSchema?: string;
   shouldIncludeClientSchema?: boolean;
-  warnings: Warning[];
+  warnings: string[];
 }
 
 export interface CompositionDeployResult {