Merge pull request #9 from gigsmart/fix/streaming-resolution-resolve-then-split

refactor(streaming): resolve-then-split strategy for @defer

Author: jwaldrip

lib/absinthe/blueprint/schema/enum_type_definition.ex

@@ -36,7 +36,8 @@ defmodule Absinthe.Blueprint.Schema.EnumTypeDefinition do
       values: values_by(type_def, :identifier),
       values_by_internal_value: values_by(type_def, :value),
       values_by_name: values_by(type_def, :name),
-      applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
+      applied_directives:
+        Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
       definition: type_def.module,
       description: type_def.description
     }
@@ -54,7 +55,8 @@ defmodule Absinthe.Blueprint.Schema.EnumTypeDefinition do
             __private__: value_def.__private__,
             description: value_def.description,
             deprecation: value_def.deprecation,
-            applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(value_def.directives)
+            applied_directives:
+              Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(value_def.directives)
           }
 
           {Map.fetch!(value_def, key), value}

lib/absinthe/blueprint/schema/input_object_type_definition.ex

@@ -37,7 +37,8 @@ defmodule Absinthe.Blueprint.Schema.InputObjectTypeDefinition do
       name: type_def.name,
       fields: build_fields(type_def, schema),
       description: type_def.description,
-      applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
+      applied_directives:
+        Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
       definition: type_def.module
     }
   end
@@ -50,7 +51,8 @@ defmodule Absinthe.Blueprint.Schema.InputObjectTypeDefinition do
         description: field_def.description,
         name: field_def.name,
         type: Blueprint.TypeReference.to_type(field_def.type, schema),
-        applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(field_def.directives),
+        applied_directives:
+          Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(field_def.directives),
         definition: type_def.module,
         __reference__: field_def.__reference__,
         __private__: field_def.__private__,

lib/absinthe/blueprint/schema/interface_type_definition.ex

@@ -44,7 +44,8 @@ defmodule Absinthe.Blueprint.Schema.InterfaceTypeDefinition do
       fields: Blueprint.Schema.ObjectTypeDefinition.build_fields(type_def, schema),
       identifier: type_def.identifier,
       resolve_type: type_def.resolve_type,
-      applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
+      applied_directives:
+        Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
       definition: type_def.module,
       interfaces: type_def.interfaces
     }

lib/absinthe/blueprint/schema/object_type_definition.ex

@@ -103,34 +103,51 @@ defmodule Absinthe.Blueprint.Schema.ObjectTypeDefinition do
     Enum.map(directives, fn directive ->
       %{
         name: directive.name,
-        args: Enum.map(directive.arguments, fn arg ->
-          %{
-            name: arg.name,
-            value: serialize_argument_value(arg.input_value)
-          }
-        end)
+        args:
+          Enum.map(directive.arguments, fn arg ->
+            %{
+              name: arg.name,
+              value: serialize_argument_value(arg.input_value)
+            }
+          end)
       }
     end)
   end
 
   def build_applied_directives(_), do: []
 
-  defp serialize_argument_value(%Absinthe.Blueprint.Input.String{value: value}), do: inspect(value)
-  defp serialize_argument_value(%Absinthe.Blueprint.Input.Integer{value: value}), do: to_string(value)
-  defp serialize_argument_value(%Absinthe.Blueprint.Input.Float{value: value}), do: to_string(value)
-  defp serialize_argument_value(%Absinthe.Blueprint.Input.Boolean{value: value}), do: to_string(value)
+  defp serialize_argument_value(%Absinthe.Blueprint.Input.String{value: value}),
+    do: inspect(value)
+
+  defp serialize_argument_value(%Absinthe.Blueprint.Input.Integer{value: value}),
+    do: to_string(value)
+
+  defp serialize_argument_value(%Absinthe.Blueprint.Input.Float{value: value}),
+    do: to_string(value)
+
+  defp serialize_argument_value(%Absinthe.Blueprint.Input.Boolean{value: value}),
+    do: to_string(value)
+
   defp serialize_argument_value(%Absinthe.Blueprint.Input.Null{}), do: "null"
   defp serialize_argument_value(%Absinthe.Blueprint.Input.Enum{value: value}), do: value
+
   defp serialize_argument_value(%Absinthe.Blueprint.Input.List{items: items}) do
     "[" <> Enum.map_join(items, ", ", &serialize_argument_value/1) <> "]"
   end
+
   defp serialize_argument_value(%Absinthe.Blueprint.Input.Object{fields: fields}) do
-    "{" <> Enum.map_join(fields, ", ", fn field ->
-      "#{field.name}: #{serialize_argument_value(field.input_value)}"
-    end) <> "}"
+    "{" <>
+      Enum.map_join(fields, ", ", fn field ->
+        "#{field.name}: #{serialize_argument_value(field.input_value)}"
+      end) <> "}"
   end
-  defp serialize_argument_value(%Absinthe.Blueprint.Input.RawValue{content: content}), do: serialize_argument_value(content)
-  defp serialize_argument_value(%Absinthe.Blueprint.Input.Value{raw: raw}), do: serialize_argument_value(raw)
+
+  defp serialize_argument_value(%Absinthe.Blueprint.Input.RawValue{content: content}),
+    do: serialize_argument_value(content)
+
+  defp serialize_argument_value(%Absinthe.Blueprint.Input.Value{raw: raw}),
+    do: serialize_argument_value(raw)
+
   defp serialize_argument_value(value), do: inspect(value)
 
   defimpl Inspect do

lib/absinthe/blueprint/schema/scalar_type_definition.ex

@@ -36,7 +36,10 @@ defmodule Absinthe.Blueprint.Schema.ScalarTypeDefinition do
       identifier: type_def.identifier,
       name: type_def.name,
       description: type_def.description,
-      applied_directives: Absinthe.Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
+      applied_directives:
+        Absinthe.Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(
+          type_def.directives
+        ),
       definition: type_def.module,
       serialize: type_def.serialize,
       parse: type_def.parse,

lib/absinthe/blueprint/schema/union_type_definition.ex

@@ -39,7 +39,8 @@ defmodule Absinthe.Blueprint.Schema.UnionTypeDefinition do
       identifier: type_def.identifier,
       types: type_def.types |> atomize_types(schema),
       fields: build_fields(type_def, schema),
-      applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
+      applied_directives:
+        Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(type_def.directives),
       definition: type_def.module,
       resolve_type: type_def.resolve_type
     }
@@ -64,7 +65,8 @@ defmodule Absinthe.Blueprint.Schema.UnionTypeDefinition do
         name: field_def.name,
         type: Blueprint.TypeReference.to_type(field_def.type, schema),
         args: build_args(field_def, schema),
-        applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(field_def.directives),
+        applied_directives:
+          Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(field_def.directives),
         definition: field_def.module,
         __reference__: field_def.__reference__,
         __private__: field_def.__private__
@@ -83,7 +85,8 @@ defmodule Absinthe.Blueprint.Schema.UnionTypeDefinition do
         type: Blueprint.TypeReference.to_type(arg_def.type, schema),
         default_value: arg_def.default_value,
         deprecation: arg_def.deprecation,
-        applied_directives: Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(arg_def.directives),
+        applied_directives:
+          Blueprint.Schema.ObjectTypeDefinition.build_applied_directives(arg_def.directives),
         __reference__: arg_def.__reference__,
         __private__: arg_def.__private__
       }

lib/absinthe/incremental/complexity.ex

@@ -368,7 +368,13 @@ defmodule Absinthe.Incremental.Complexity do
     end
   end
 
-  defp analyze_node(%Blueprint.Document.Fragment.Spread{} = node, schema, config, analysis, depth) do
+  defp analyze_node(
+         %Blueprint.Document.Fragment.Spread{} = node,
+         _schema,
+         config,
+         analysis,
+         depth
+       ) do
     {analysis, _in_defer} = check_defer_directive(node, config, analysis, depth)
     # Would need to look up the fragment definition for full analysis
     analysis

lib/absinthe/incremental/dataloader.ex

@@ -127,7 +127,9 @@ defmodule Absinthe.Incremental.Dataloader do
 
   This allows existing Dataloader resolvers to work with incremental delivery.
   """
-  @spec streaming_dataloader(atom(), any()) :: Resolution.resolver()
+  @spec streaming_dataloader(atom(), any()) ::
+          (Resolution.source(), Resolution.arguments(), Resolution.t() ->
+             {:ok, any()} | {:error, any()} | {:middleware, module(), any()})
   def streaming_dataloader(source, batch_key \\ nil) do
     fn parent, args, %{context: context} = resolution ->
       # Check if we're in a streaming context
@@ -305,7 +307,7 @@ defmodule Absinthe.Incremental.Dataloader do
     }
 
     # Add to the batch queue in the resolution context
-    resolution =
+    _resolution =
       update_in(
         resolution.context[:__dataloader_batch_queue__],
         &[batch_data | &1 || []]

lib/absinthe/incremental/error_handler.ex

@@ -6,7 +6,6 @@ defmodule Absinthe.Incremental.ErrorHandler do
   streaming operations, ensuring robust behavior even when things go wrong.
   """
 
-  alias Absinthe.Incremental.Response
   require Logger
 
   @type error_type ::
@@ -315,18 +314,11 @@ defmodule Absinthe.Incremental.ErrorHandler do
     }
   end
 
-  defp format_exception(exception, stacktrace \\ nil) do
-    formatted_stacktrace =
-      if stacktrace do
-        Exception.format_stacktrace(stacktrace)
-      else
-        "stacktrace not available"
-      end
-
+  defp format_exception(exception, stacktrace) do
     %{
       message: Exception.message(exception),
       type: exception.__struct__,
-      stacktrace: formatted_stacktrace
+      stacktrace: Exception.format_stacktrace(stacktrace)
     }
   end
 

lib/absinthe/incremental/resource_manager.ex

@@ -258,6 +258,16 @@ defmodule Absinthe.Incremental.ResourceManager do
     update_in(state.stream_stats.total_count, &(&1 + 1))
   end
 
+  defp update_stats(state, :stream_timeout) do
+    state
+    |> update_in([:stream_stats, :timeout_count], &(&1 + 1))
+    |> update_in([:stream_stats, :failed_count], &(&1 + 1))
+  end
+
+  defp update_stats(state, :stream_crashed) do
+    update_in(state.stream_stats.failed_count, &(&1 + 1))
+  end
+
   defp update_stats(state, :stream_released, duration) do
     state
     |> update_in([:stream_stats, :completed_count], &(&1 + 1))
@@ -269,16 +279,6 @@ defmodule Absinthe.Incremental.ResourceManager do
     end)
   end
 
-  defp update_stats(state, :stream_timeout) do
-    state
-    |> update_in([:stream_stats, :timeout_count], &(&1 + 1))
-    |> update_in([:stream_stats, :failed_count], &(&1 + 1))
-  end
-
-  defp update_stats(state, :stream_crashed) do
-    update_in(state.stream_stats.failed_count, &(&1 + 1))
-  end
-
   defp schedule_stream_timeout(operation_id, timeout_ms) do
     Process.send_after(self(), {:stream_timeout, operation_id}, timeout_ms)
   end