feat: implement worker-based graph composition with Tinypool

controlplane/package.json

@@ -94,6 +94,7 @@
     "stream-json": "^1.8.0",
     "stripe": "^14.19.0",
     "tiny-lru": "^11.2.11",
+    "tinypool": "^2.1.0",
     "uid": "^2.0.2",
     "uuid": "^10.0.0",
     "zod": "^3.25.0",

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

@@ -8,9 +8,8 @@ import {
   CompositionWarning,
   Subgraph,
 } from '@wundergraph/cosmo-connect/dist/platform/v1/platform_pb';
-import { parse } from 'graphql';
 import { COMPOSITION_IGNORE_EXTERNAL_KEYS_FEATURE_ID } from '../../../types/index.js';
-import { composeSubgraphs } from '../../composition/composition.js';
+import { composeGraphsInWorker } from '../../composition/composeGraphs.pool.js';
 import { FederatedGraphRepository } from '../../repositories/FederatedGraphRepository.js';
 import { DefaultNamespace } from '../../repositories/NamespaceRepository.js';
 import { OrganizationRepository } from '../../repositories/OrganizationRepository.js';
@@ -108,19 +107,25 @@ export function checkFederatedGraph(
       featureId: COMPOSITION_IGNORE_EXTERNAL_KEYS_FEATURE_ID,
     });
 
-    const result = composeSubgraphs(
-      subgraphsUsedForComposition.map((s) => ({
-        id: s.id,
-        name: s.name,
-        url: s.routingUrl,
-        definitions: parse(s.schemaSDL),
-      })),
-      federatedGraph.routerCompatibilityVersion,
-      {
+    const { results } = await composeGraphsInWorker({
+      federatedGraph,
+      subgraphsToCompose: [
+        {
+          subgraphs: subgraphsUsedForComposition,
+          isFeatureFlagComposition: false,
+          featureFlagName: '',
+          featureFlagId: '',
+        },
+      ],
+      tagOptionsByContractName: [],
+      compositionOptions: {
         disableResolvabilityValidation: req.disableResolvabilityValidation,
         ignoreExternalKeys: ignoreExternalKeysFeature?.enabled ?? false,
       },
-    );
+      skipRouterConfig: true,
+    });
+
+    const compositionResult = results[0].base;
 
     // If req.limit is not provided, we return all rows
     const returnLimit = req.limit === undefined ? null : clamp(req.limit, 1, maxRowLimitForChecks);
@@ -138,9 +143,9 @@ export function checkFederatedGraph(
     };
 
     const compositionWarnings: PlainMessage<CompositionWarning>[] = [];
-    counts.compositionWarnings = result.warnings.length;
+    counts.compositionWarnings = compositionResult.warnings.length;
 
-    const clampedWarnings = returnLimit ? result.warnings.slice(0, returnLimit) : result.warnings;
+    const clampedWarnings = returnLimit ? compositionResult.warnings.slice(0, returnLimit) : compositionResult.warnings;
     for (const warning of clampedWarnings) {
       compositionWarnings.push({
         message: warning.message,
@@ -150,14 +155,14 @@ export function checkFederatedGraph(
       });
     }
 
-    if (!result.success) {
+    if (!compositionResult.success) {
       const compositionErrors: PlainMessage<CompositionError>[] = [];
-      counts.compositionErrors = result.errors.length;
+      counts.compositionErrors = compositionResult.errors.length;
 
-      const clampedErrors = returnLimit ? result.errors.slice(0, returnLimit) : result.errors;
+      const clampedErrors = returnLimit ? compositionResult.errors.slice(0, returnLimit) : compositionResult.errors;
       for (const error of clampedErrors) {
         compositionErrors.push({
-          message: error.message,
+          message: error,
           federatedGraphName: req.name,
           namespace: federatedGraph.namespace,
           featureFlag: '',

controlplane/src/core/build-server.ts

@@ -53,9 +53,13 @@ import {
   createReactivateOrganizationWorker,
   ReactivateOrganizationQueue,
 } from './workers/ReactivateOrganizationWorker.js';
+import { configureComposeGraphsPool, destroyComposeGraphsPool } from './composition/composeGraphs.pool.js';
 
 export interface BuildConfig {
   logger: LoggerOptions;
+  composition?: {
+    maxThreads: number;
+  };
   database: {
     url: string;
     tls?: {
@@ -157,6 +161,10 @@ const developmentLoggerOpts: LoggerOptions = {
 };
 
 export default async function build(opts: BuildConfig) {
+  configureComposeGraphsPool({
+    maxThreads: opts.composition?.maxThreads ?? 0,
+  });
+
   opts.logger = {
     timestamp: stdTimeFunctions.isoTime,
     formatters: {
@@ -535,6 +543,12 @@ export default async function build(opts: BuildConfig) {
     await Promise.all(bullWorkers.map((worker) => worker.close()));
 
     fastify.log.debug('Bull workers shut down');
+
+    fastify.log.debug('Shutting down composition worker pool');
+
+    await destroyComposeGraphsPool();
+
+    fastify.log.debug('Composition worker pool shut down');
   });
 
   return fastify;

controlplane/src/core/composition/composeGraphs.pool.ts

@@ -0,0 +1,132 @@
+/**
+ * Main-thread bridge for composition worker execution.
+ *
+ * The worker only exchanges plain `Serialized*` payloads so we do not rely on
+ * structured cloning of rich runtime objects across the Tinypool boundary.
+ * Node 22 loads the source `.ts` worker natively in development, and the built
+ * `.js` worker in production.
+ */
+import { existsSync } from 'node:fs';
+import { fileURLToPath } from 'node:url';
+import { availableParallelism } from 'node:os';
+import { Warning } from '@wundergraph/composition';
+import { RouterConfig } from '@wundergraph/cosmo-connect/dist/node/v1/node_pb';
+import WorkerPool from 'tinypool';
+import { FederatedGraphDTO } from '../../types/index.js';
+import { validateRouterCompatibilityVersion } from './composition.js';
+import { ComposedFederatedGraph, CompositionSubgraphRecord } from './composer.js';
+import {
+  ComposeGraphsTaskInput,
+  ComposeGraphsTaskResult,
+  SerializedComposedGraphArtifact,
+} from './composeGraphs.types.js';
+
+let composeGraphsPool: WorkerPool | undefined;
+const composeGraphsPoolConfig = {
+  maxThreads: 0,
+};
+
+export interface ConfigureComposeGraphsPoolOptions {
+  maxThreads: number;
+}
+
+function getWorkerFilename() {
+  const sourceWorker = new URL('composeGraphs.worker.ts', import.meta.url);
+  if (existsSync(fileURLToPath(sourceWorker))) {
+    return {
+      filename: sourceWorker.href,
+    };
+  }
+
+  return {
+    filename: new URL('composeGraphs.worker.js', import.meta.url).href,
+  };
+}
+
+function getMaxThreads() {
+  if (composeGraphsPoolConfig.maxThreads > 0) {
+    return composeGraphsPoolConfig.maxThreads;
+  }
+
+  return Math.max(1, availableParallelism());
+}
+
+function getComposeGraphsPool() {
+  if (composeGraphsPool) {
+    return composeGraphsPool;
+  }
+
+  const { filename } = getWorkerFilename();
+
+  composeGraphsPool = new WorkerPool({
+    filename,
+    minThreads: 1,
+    maxThreads: getMaxThreads(),
+    concurrentTasksPerWorker: 2,
+  });
+
+  return composeGraphsPool;
+}
+
+function deserializeWarning(message: string, subgraphName?: string) {
+  return new Warning({
+    message,
+    subgraph: {
+      name: subgraphName || '',
+    },
+  });
+}
+
+export type DeserializedComposedGraph = Omit<ComposedFederatedGraph, 'subgraphs'> & {
+  subgraphs: CompositionSubgraphRecord[];
+};
+
+export function deserializeComposedGraphArtifact(
+  federatedGraph: Pick<FederatedGraphDTO, 'id' | 'targetId' | 'name' | 'namespace' | 'namespaceId'>,
+  artifact: SerializedComposedGraphArtifact,
+): DeserializedComposedGraph {
+  return {
+    id: federatedGraph.id,
+    targetID: federatedGraph.targetId,
+    name: federatedGraph.name,
+    namespace: federatedGraph.namespace,
+    namespaceId: federatedGraph.namespaceId,
+    composedSchema: artifact.composedSchema,
+    federatedClientSchema: artifact.federatedClientSchema,
+    shouldIncludeClientSchema: artifact.shouldIncludeClientSchema,
+    errors: artifact.errors.map((message) => new Error(message)),
+    fieldConfigurations: artifact.fieldConfigurations,
+    subgraphs: artifact.subgraphs,
+    warnings: artifact.warnings.map((warning) => deserializeWarning(warning.message, warning.subgraphName)),
+  };
+}
+
+export function deserializeRouterExecutionConfig(routerExecutionConfigJson?: ReturnType<RouterConfig['toJson']>) {
+  if (!routerExecutionConfigJson) {
+    return;
+  }
+
+  return RouterConfig.fromJson(routerExecutionConfigJson);
+}
+
+export function composeGraphsInWorker(task: Omit<ComposeGraphsTaskInput, 'routerCompatibilityVersion'>) {
+  const fullTask: ComposeGraphsTaskInput = {
+    ...task,
+    routerCompatibilityVersion: validateRouterCompatibilityVersion(task.federatedGraph.routerCompatibilityVersion),
+  };
+  return getComposeGraphsPool().run(fullTask) as Promise<ComposeGraphsTaskResult>;
+}
+
+export function configureComposeGraphsPool(options: ConfigureComposeGraphsPoolOptions) {
+  composeGraphsPoolConfig.maxThreads = options.maxThreads;
+}
+
+export async function destroyComposeGraphsPool() {
+  if (!composeGraphsPool) {
+    return;
+  }
+
+  const pool = composeGraphsPool;
+  composeGraphsPool = undefined;
+  await pool.destroy();
+}

controlplane/src/core/composition/composeGraphs.types.ts

@@ -0,0 +1,80 @@
+/**
+ * These types define the thread boundary for compose-and-deploy work.
+ *
+ * The `Serialized*` prefix is intentional: these payloads are flattened to
+ * structured-clone-safe data before crossing the Tinypool worker boundary.
+ * Rich runtime objects such as GraphQL schema instances, Maps, protobuf
+ * classes, and custom Error/Warning instances are reconstructed outside the
+ * worker when needed.
+ */
+import type {
+  CompositionOptions,
+  FieldConfiguration,
+  SupportedRouterCompatibilityVersion,
+} from '@wundergraph/composition';
+import { RouterConfig } from '@wundergraph/cosmo-connect/dist/node/v1/node_pb';
+import { FederatedGraphDTO, SubgraphDTO } from '../../types/index.js';
+
+export interface SerializedContractTagOptions {
+  contractName: string;
+  excludeTags: string[];
+  includeTags: string[];
+}
+
+export interface SerializedCompositionWarning {
+  message: string;
+  subgraphName?: string;
+}
+
+export interface SerializedComposedSubgraph {
+  id: string;
+  isFeatureSubgraph: boolean;
+  name: string;
+  sdl: string;
+  schemaVersionId: string;
+  targetId: string;
+}
+
+export interface SerializedComposedGraphArtifact {
+  success: boolean;
+  errors: string[];
+  warnings: SerializedCompositionWarning[];
+  composedSchema?: string;
+  federatedClientSchema?: string;
+  shouldIncludeClientSchema: boolean;
+  fieldConfigurations: FieldConfiguration[];
+  subgraphs: SerializedComposedSubgraph[];
+  routerExecutionConfigJson?: ReturnType<RouterConfig['toJson']>;
+}
+
+export interface SerializedContractCompositionArtifact {
+  contractName: string;
+  artifact: SerializedComposedGraphArtifact;
+}
+
+export interface ComposeGraphsTaskInput {
+  federatedGraph: FederatedGraphDTO;
+  /** Pre-validated on the main thread before dispatching to the worker. */
+  routerCompatibilityVersion: SupportedRouterCompatibilityVersion;
+  subgraphsToCompose: {
+    subgraphs: SubgraphDTO[];
+    isFeatureFlagComposition: boolean;
+    featureFlagName: string;
+    featureFlagId: string;
+  }[];
+  tagOptionsByContractName: SerializedContractTagOptions[];
+  compositionOptions?: CompositionOptions;
+  skipRouterConfig?: boolean;
+}
+
+export interface ComposeGraphsTaskResultItem {
+  isFeatureFlagComposition: boolean;
+  featureFlagName: string;
+  featureFlagId: string;
+  base: SerializedComposedGraphArtifact;
+  contracts: SerializedContractCompositionArtifact[];
+}
+
+export interface ComposeGraphsTaskResult {
+  results: ComposeGraphsTaskResultItem[];
+}