Eagerly init subgraph operation for subscription primary nodes

.changesets/fix_tninesling_subscription_operation_init.md

@@ -0,0 +1,5 @@
+### Eagerly init subgraph operation for subscription primary nodes ([PR #6509](https://github.com/apollographql/router/pull/6509))
+
+When subgraph operations are deserialized, typically from a query plan cache, they are not automatically parsed into a full document. Instead, each node needs to initialize its operation(s) prior to execution. With this change, the primary node inside SubscriptionNode is initialized in the same way as other nodes in the plan.
+
+By [@tninesling](https://github.com/tninesling) in https://github.com/apollographql/router/pull/6509

apollo-router/src/plugins/demand_control/cost_calculator/fixtures/subscription_query.graphql

@@ -0,0 +1,6 @@
+subscription MessageSubscription {
+  messages {
+    subject
+    content
+  }
+}

apollo-router/src/plugins/demand_control/cost_calculator/fixtures/subscription_schema.graphql

@@ -0,0 +1,92 @@
+schema
+  @link(url: "https://specs.apollo.dev/link/v1.0")
+  @link(url: "https://specs.apollo.dev/join/v0.4", for: EXECUTION) {
+  query: Query
+  mutation: Mutation
+  subscription: Subscription
+}
+
+directive @join__directive(
+  graphs: [join__Graph!]
+  name: String!
+  args: join__DirectiveArguments
+) repeatable on SCHEMA | OBJECT | INTERFACE | FIELD_DEFINITION
+
+directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
+
+directive @join__field(
+  graph: join__Graph
+  requires: join__FieldSet
+  provides: join__FieldSet
+  type: String
+  external: Boolean
+  override: String
+  usedOverridden: Boolean
+  overrideLabel: String
+) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
+
+directive @join__graph(name: String!, url: String!) on ENUM_VALUE
+
+directive @join__implements(
+  graph: join__Graph!
+  interface: String!
+) repeatable on OBJECT | INTERFACE
+
+directive @join__type(
+  graph: join__Graph!
+  key: join__FieldSet
+  extension: Boolean! = false
+  resolvable: Boolean! = true
+  isInterfaceObject: Boolean! = false
+) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
+
+directive @join__unionMember(
+  graph: join__Graph!
+  member: String!
+) repeatable on UNION
+
+directive @link(
+  url: String
+  as: String
+  for: link__Purpose
+  import: [link__Import]
+) repeatable on SCHEMA
+
+scalar join__DirectiveArguments
+
+scalar join__FieldSet
+
+enum join__Graph {
+  SUBGRAPH @join__graph(name: "subgraph", url: "http://localhost:4001")
+}
+
+scalar link__Import
+
+enum link__Purpose {
+  """
+  `SECURITY` features provide metadata necessary to securely resolve fields.
+  """
+  SECURITY
+
+  """
+  `EXECUTION` features provide metadata necessary for operation execution.
+  """
+  EXECUTION
+}
+
+type Message @join__type(graph: SUBGRAPH) {
+  subject: String
+  content: String
+}
+
+type Mutation @join__type(graph: SUBGRAPH) {
+  addMessage(subject: String, content: String): Message
+}
+
+type Query @join__type(graph: SUBGRAPH) {
+  allMessages: [Message]
+}
+
+type Subscription @join__type(graph: SUBGRAPH) {
+  messages: Message
+}

apollo-router/src/plugins/demand_control/cost_calculator/static_cost.rs

@@ -1143,4 +1143,15 @@ mod tests {
 
         assert_eq!(estimated_cost(schema, query, variables), 1.0);
     }
+
+    #[test(tokio::test)]
+    async fn subscription_request() {
+        let schema = include_str!("./fixtures/subscription_schema.graphql");
+        let query = include_str!("./fixtures/subscription_query.graphql");
+        let variables = "{}";
+
+        assert_eq!(estimated_cost(schema, query, variables), 1.0);
+        assert_eq!(planned_cost_js(schema, query, variables).await, 1.0);
+        assert_eq!(planned_cost_rust(schema, query, variables), 1.0);
+    }
 }

apollo-router/src/query_planner/plan.rs

@@ -354,7 +354,8 @@ impl PlanNode {
                     }
                 }
             }
-            PlanNode::Subscription { primary: _, rest } => {
+            PlanNode::Subscription { primary, rest } => {
+                primary.init_parsed_operation(subgraph_schemas)?;
                 if let Some(node) = rest.as_mut() {
                     node.init_parsed_operations(subgraph_schemas)?;
                 }

apollo-router/src/query_planner/subscription.rs

@@ -12,10 +12,12 @@ use tower::ServiceExt;
 use tracing_futures::Instrument;
 
 use super::execution::ExecutionParameters;
+use super::fetch::SubgraphSchemas;
 use super::fetch::Variables;
 use super::rewrites;
 use super::OperationKind;
 use crate::error::FetchError;
+use crate::error::ValidationErrors;
 use crate::graphql::Error;
 use crate::graphql::Request;
 use crate::graphql::Response;
@@ -272,4 +274,13 @@ impl SubscriptionNode {
 
         Ok(response.errors)
     }
+
+    pub(crate) fn init_parsed_operation(
+        &mut self,
+        subgraph_schemas: &SubgraphSchemas,
+    ) -> Result<(), ValidationErrors> {
+        let schema = &subgraph_schemas[self.service_name.as_ref()];
+        self.operation.init_parsed(schema)?;
+        Ok(())
+    }
 }