feat: add compatibility handshake between router and execution config

composition/package.json

@@ -18,7 +18,8 @@
     "test:watch": "vitest test",
     "test": "vitest run",
     "lint": "prettier --check src tests",
-    "lint:fix": "prettier --write src tests"
+    "lint:fix": "prettier --write src tests",
+    "postinstall": "node ./scripts/get-composition-version.mjs"
   },
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",

composition/src/index.ts

@@ -17,7 +17,8 @@ export * from './schema-building/type-definition-data';
 export * from './schema-building/type-merging';
 export * from './schema-building/utils';
 export * from './subgraph/subgraph';
-export * from './utils/utils';
+export * from './utils/composition-version';
 export * from './utils/constants';
+export * from './utils/utils';
 export * from './utils/string-constants';
 export * from './warnings/warnings';

composition/src/scripts/get-composition-version.mjs

@@ -0,0 +1,20 @@
+import fs from 'fs';
+import { fileURLToPath } from 'node:url';
+import path from 'node:path';
+
+const __filename = fileURLToPath(import.meta.url);
+const __dirname = path.dirname(__filename);
+const compositionVersion = '{{$COMPOSITION__VERSION}}';
+
+// From pnpm v10+, hooks will not be supported.
+if (process.argv[1] === __filename) {
+  const json = JSON.parse(fs.readFileSync(path.join(__dirname, '../package.json')).toString());
+  const version = json.version;
+  const varFilePath = path.join(__dirname, '../dist/utils/composition-version.js');
+  let content = fs.readFileSync(varFilePath).toString();
+  if (content.indexOf(compositionVersion) < 0) {
+    throw new Error(`"${compositionVersion}" string not found in dist/utils/composition-version.js.`);
+  }
+  content = content.replace(compositionVersion, version);
+  fs.writeFileSync(varFilePath, content);
+}

composition/src/utils/composition-version.ts

@@ -0,0 +1 @@
+export const COMPOSITION_VERSION = '{{$COMPOSITION__VERSION}}';

composition/tsconfig.json

@@ -3,7 +3,8 @@
   "compilerOptions": {
     "declaration": true,
     "outDir": "./dist",
-    "module": "commonjs",
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
   },
   "include": ["src/**/*"],
   "exclude": ["node_modules"]

connect/src/wg/cosmo/node/v1/node_pb.ts

@@ -330,6 +330,11 @@ export class RouterConfig extends Message<RouterConfig> {
    */
   featureFlagConfigs?: FeatureFlagRouterExecutionConfigs;
 
+  /**
+   * @generated from field: string compatibility_version = 5;
+   */
+  compatibilityVersion = "";
+
   constructor(data?: PartialMessage<RouterConfig>) {
     super();
     proto3.util.initPartial(data, this);
@@ -342,6 +347,7 @@ export class RouterConfig extends Message<RouterConfig> {
     { no: 2, name: "version", kind: "scalar", T: 9 /* ScalarType.STRING */ },
     { no: 3, name: "subgraphs", kind: "message", T: Subgraph, repeated: true },
     { no: 4, name: "feature_flag_configs", kind: "message", T: FeatureFlagRouterExecutionConfigs, opt: true },
+    { no: 5, name: "compatibility_version", kind: "scalar", T: 9 /* ScalarType.STRING */ },
   ]);
 
   static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): RouterConfig {

proto/wg/cosmo/node/v1/node.proto

@@ -26,6 +26,7 @@ message RouterConfig {
   repeated Subgraph subgraphs = 3;
   // The map of feature flag router execution configs requires a wrapper to be non-breaking
   optional FeatureFlagRouterExecutionConfigs feature_flag_configs = 4;
+  string compatibility_version = 5;
 }
 
 message Response {

router/core/router.go

@@ -10,6 +10,8 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -56,6 +58,8 @@ type IPAnonymizationMethod string
 const (
 	Hash   IPAnonymizationMethod = "hash"
 	Redact IPAnonymizationMethod = "redact"
+	// CompatibilityThreshold should ONLY be updated if there is a breaking change in the router execution config.
+	CompatibilityThreshold = 1
 )
 
 var CompressibleContentTypes = []string{
@@ -1105,6 +1109,16 @@ func (r *Router) Start(ctx context.Context) error {
 					return nil
 				}
 
+				/* Older versions of composition will not populate a compatibility version.
+				 * Currently, all "old" router execution configurations are compatible as there have been no breaking
+				 * changes.
+				 * Upon the first breaking change to the execution config, an unpopulated compatibility version will
+				 * also be unsupported (and the logic for isRouterCompatibleWithExecutionConfig will need to be updated).
+				 */
+				if !r.isRouterCompatibleWithExecutionConfig(cfg.CompatibilityVersion) {
+					return nil
+				}
+
 				if err := r.newServer(ctx, cfg); err != nil {
 					r.logger.Error("Failed to update server with new config", zap.Error(err))
 					return nil
@@ -1334,6 +1348,36 @@ func (r *Router) Shutdown(ctx context.Context) (err error) {
 	return err
 }
 
+func (r *Router) isRouterCompatibleWithExecutionConfig(compatibilityVersion string) bool {
+	if compatibilityVersion == "" {
+		return true
+	}
+	/* A compatibility version is composed thus: <compatibility threshold>:<composition version>
+	 * A router version will support a range of compatibility thresholds.
+	 * In the event the execution config exceeds the compatibility threshold supported by the router,
+	 * an error will request the router version be upgraded.
+	 */
+	segments := strings.Split(compatibilityVersion, ":")
+	if len(segments) != 2 {
+		r.logger.Error("Failed to parse compatibility version", zap.String("compatibility_version", compatibilityVersion))
+		return false
+	}
+	threshold, err := strconv.ParseInt(segments[0], 10, 32)
+	if err != nil {
+		r.logger.Error("Failed to parse compatibility threshold of compatibility version", zap.String("compatibility_version", compatibilityVersion))
+		return false
+	}
+	if threshold > CompatibilityThreshold {
+		r.logger.Error(
+			fmt.Sprintf("This router version only supports the router execution configuration compatibility threshold <= %d. Please upgrade your router version.", CompatibilityThreshold),
+			zap.Int64("compatibility_threshold", threshold),
+			zap.String("composition_version", segments[1]),
+		)
+		return false
+	}
+	return true
+}
+
 func WithListenerAddr(addr string) Option {
 	return func(r *Router) {
 		r.listenAddr = addr