Fixed mutation root fields not being executed serially

RELEASE_NOTES.md

@@ -296,6 +296,9 @@
 * **Breaking Change** `BufferedStreamOptions.Interval` and `BufferedStreamOptions.PreferredBatchSize` are now `int voption`
 * **Breaking Change** `ServerMessage.Error` and `ServerRawPayload.ErrorMessages` now carry `GQLProblemDetails list` instead of `NameValueLookup list`, so an `error` message's `payload` is a standard GraphQL error array as the `graphql-transport-ws` protocol requires
 * **Breaking Change** A query or mutation whose non-null root field fails during execution now produces a `Direct` (execution) result with `null` data instead of a `RequestError`, which is now only ever produced for a request rejected before execution (validation, planning, variable or inline argument coercion, a middleware, or the executor itself failing); HTTP and `graphql-transport-ws` responses for such a failure now carry `data: null` as the spec requires, instead of omitting `data` entirely. This also changes the public `GQLResponse.Data`, `GQLResponseContent.Direct.Data`, `DeferredErrors.Data`, and `SubscriptionErrors.Data` signatures to use `voption`
+* **Breaking Change** Validation now rejects `@defer` on an inline fragment or fragment spread at the root of a mutation, as the incremental delivery specification requires: only `@defer` and `@stream` on the mutation root fields themselves were rejected, so a mutation deferring a fragment at its root was accepted and the root fields that fragment selects were executed after the rest of the mutation and concurrently with each other. As on a root field, a directive disabled with a literal `if: false` is still accepted, which the specification does not allow
+* Fixed the root fields of a mutation, or of any operation planned with `ExecutionStrategy.Sequential`, not being executed serially: the resolvers of all root fields were called up front and only their results were awaited in order, so a synchronous resolver of a later field, or a task started by its resolver, ran before an earlier field and its nested fields had completed. A root field is now started only once the previous one has completed, apart from the parts of it deferred with `@defer` or streamed with `@stream`, which are delivered later. Added `AsyncVal.collectSequentialWith`, which maps each item to an `AsyncVal` only once the previous one has completed
+* Fixed a resolver that throws before it returns its value, such as an `AsyncField` resolver throwing instead of returning its `Async`, failing the whole operation with a request error when its field is a root field, discarding the results of the mutation fields already executed, and failing its parent field instead of itself when nested. The exception is now an error of its own field
 * **Breaking Change** Updated `FsToolkit.ErrorHandling` to 5.x, which merged `FsToolkit.ErrorHandling.TaskResult` into the main package: a project referencing `FsToolkit.ErrorHandling.TaskResult` must remove that reference. See the [FsToolkit.ErrorHandling 5.0.0 release notes](https://github.com/demystifyfp/FsToolkit.ErrorHandling/blob/master/RELEASE_NOTES.md#500---june-02-2025) for its other breaking changes, such as `retn` renamed to `singleton`, `returnError` renamed to `error` and `and!` removed from `result { }`
 * Added case-insensitive string comparison support to `ObjectListFilter`, including comparer-aware filter cases and GraphQL filter suffix handling
 * Improved Relay XML documentation comments

src/FSharp.Data.GraphQL.Server/Execution.fs

@@ -108,10 +108,19 @@ let private createFieldContext objdef inputContext argDefs ctx (info : Execution
 }
 
 let private resolveField (execute : ExecuteField) (ctx : ResolveFieldContext) (parentValue : obj) =
+    // A resolver can throw before it returns its value, as an asynchronous resolver throwing instead of returning its
+    // computation does: that is an error of its own field, like a failed value. Thrown on, it would fail the whole
+    // operation for a root field, discarding the root fields of a mutation already executed, and fail the parent field
+    // instead of this one for a nested field
+    let value =
+        try
+            execute ctx parentValue
+        with e ->
+            AsyncVal.Failure e
     if ctx.ExecutionInfo.IsNullable then
-        execute ctx parentValue |> AsyncVal.map (objectOptionCast)
+        value |> AsyncVal.map (objectOptionCast)
     else
-        execute ctx parentValue
+        value
         |> AsyncVal.map (fun v -> if isNull v then ValueNone else ValueSome v)
 
 
@@ -378,7 +387,32 @@ let rec private rootFieldsOfFragment (info : ExecutionInfo) =
     | ResolveDeferredFragment (_, _, _, fragmentFields) -> fragmentFields |> Seq.collect rootFieldsOfFragment
     | _ -> Seq.singleton info
 
+/// The results of the fields merged together in the order of the fields: their data, their deferred parts and their
+/// errors, or the errors of every field that failed.
+let private mergeFieldResults (collected : ResolverResult<KeyValuePair<string, obj>>[]) : ResolverResult<KeyValuePair<string, obj>[]> =
+    let data = Array.zeroCreate (collected.Length)
+
+    let merge r acc =
+        match (r, acc) with
+        | Ok (field, d, e), Ok (i, deferred, errs) ->
+            Array.set data i field
+            // Folded from the last field back, so the current field comes before the ones already merged
+            Ok (i - 1, ValueOption.mergeWith (fun later current -> AnnouncedEvents.merge current later) deferred d, e @ errs)
+        | Error e, Ok (_, _, errs) -> Error (e @ errs)
+        | Ok (_, _, e), Error errs -> Error (e @ errs)
+        | Error e, Error errs -> Error (e @ errs)
+    Array.foldBack merge collected (Ok (data.Length - 1, ValueNone, []))
+    |> ResolverResult.mapValue (fun _ -> data)
+
+/// <summary>
 /// Collect together an array of results using the appropriate execution strategy.
+/// </summary>
+/// <remarks>
+/// The results already exist, so with <see cref="ExecutionStrategy.Sequential"/> they are only awaited one after
+/// another: whatever creating them ran or started is not held back. Fields that must not start before the previous
+/// ones have completed, as the root fields of a mutation, are executed through
+/// <see cref="AsyncVal.collectSequentialWith"/> instead.
+/// </remarks>
 let collectFields
     (strategy : ExecutionStrategy)
     (rs : AsyncVal<ResolverResult<KeyValuePair<string, obj>>>[])
@@ -389,21 +423,7 @@ let collectFields
             match strategy with
             | Parallel -> AsyncVal.collectParallel rs
             | Sequential -> AsyncVal.collectSequential rs
-
-        let data = Array.zeroCreate (collected.Length)
-
-        let merge r acc =
-            match (r, acc) with
-            | Ok (field, d, e), Ok (i, deferred, errs) ->
-                Array.set data i field
-                // Folded from the last field back, so the current field comes before the ones already merged
-                Ok (i - 1, ValueOption.mergeWith (fun later current -> AnnouncedEvents.merge current later) deferred d, e @ errs)
-            | Error e, Ok (_, _, errs) -> Error (e @ errs)
-            | Ok (_, _, e), Error errs -> Error (e @ errs)
-            | Error e, Error errs -> Error (e @ errs)
-        return
-            Array.foldBack merge collected (Ok (data.Length - 1, ValueNone, []))
-            |> ResolverResult.mapValue (fun _ -> data)
+        return mergeFieldResults collected
     }
 
 let rec private direct
@@ -1029,9 +1049,22 @@ let private executeQueryOrMutation
         else
             let operations =
                 coerced
-                |> Seq.map (fun (KeyValue (i, struct (args, _))) -> executeRootOperation rootFields[i] args)
+                |> Seq.map (fun (KeyValue (i, struct (args, _))) -> struct (rootFields[i], args))
                 |> Seq.toArray
-            match! operations |> collectFields ctx.ExecutionPlan.Strategy with
+            let execute (struct (field, args)) = executeRootOperation field args
+            let collected =
+                match ctx.ExecutionPlan.Strategy with
+                | Parallel -> operations |> Array.map execute |> collectFields Parallel
+                // Serial execution, as the root fields of a mutation require: a root field is started, its resolver
+                // called, only once the previous one has completed with its nested fields, apart from those deferred
+                // with @defer or streamed with @stream, which are delivered later. Creating every root field up front
+                // and awaiting them in order would already run the synchronous resolvers and start the tasks of the
+                // later ones while an earlier one is still pending
+                | Sequential ->
+                    operations
+                    |> AsyncVal.collectSequentialWith execute
+                    |> AsyncVal.map mergeFieldResults
+            match! collected with
             | Ok (data, nested, errs) ->
                 let deferred =
                     (nested, deferredFragments)

src/FSharp.Data.GraphQL.Shared/AsyncVal.fs

@@ -125,11 +125,18 @@ module AsyncVal =
             })
         | Failure f -> Failure (f)
 
+    /// <summary>
     /// Converts array of AsyncVals into AsyncVal with array results.
     /// In case when are non-immediate values in provided array, they are
     /// executed asynchronously, one by one with regard to their order in array.
     /// Returned array maintain order of values.
     /// If the array contains a Failure, then the entire array will not resolve
+    /// </summary>
+    /// <remarks>
+    /// The values exist before they are collected, so whatever work creating them did or started (a synchronous
+    /// computation, a running <see cref="Task"/>) is not held back; <see cref="collectSequentialWith"/> creates each
+    /// value only once the previous one has completed.
+    /// </remarks>
     let collectSequential (values : AsyncVal<'T>[]) : AsyncVal<'T[]> =
         if values.Length = 0 then Value [||]
         elif values |> Array.exists isAsync then
@@ -160,6 +167,52 @@ module AsyncVal =
             | 1 -> Failure (exceptions.First ())
             | _ -> Failure (AggregateException exceptions)
 
+    /// <summary>
+    /// Maps the items to AsyncVals one after another and collects their results in the order of the items: an item is
+    /// mapped only once the AsyncVal of the previous one has completed.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// Unlike <see cref="collectSequential"/>, which receives values that already exist, nothing is done for an item
+    /// before the previous one has completed, neither the mapping itself nor any work it starts, such as a
+    /// <see cref="Task"/>.
+    /// </para>
+    /// <para>
+    /// The result is immediate when every item maps to an immediate value. The first item that maps to a Failure, or
+    /// whose asynchronous computation raises an exception, ends the collection with that failure, and the items after
+    /// it are not mapped.
+    /// </para>
+    /// </remarks>
+    let collectSequentialWith (mapping : 'T -> AsyncVal<'U>) (items : 'T[]) : AsyncVal<'U[]> =
+        let immediateResults = Array.zeroCreate items.Length
+
+        /// The results of the items from index on, collected inside an asynchronous computation once the pending
+        /// result of the item at index is available.
+        let collectAsync (index : int) (pending : Async<'U>) =
+            Async (async {
+                // Copied, so that every run of the computation fills an array of its own, as an Async may run more than once
+                let results = Array.copy immediateResults
+                let! result = pending
+                results[index] <- result
+                for i = index + 1 to items.Length - 1 do
+                    let! result = mapping items[i] |> toAsync
+                    results[i] <- result
+                return results
+            })
+
+        let rec collectFrom (index : int) =
+            if index = items.Length then
+                Value immediateResults
+            else
+                match mapping items[index] with
+                | Value result ->
+                    immediateResults[index] <- result
+                    collectFrom (index + 1)
+                | Async pending -> collectAsync index pending
+                | Failure error -> Failure error
+
+        collectFrom 0
+
     /// Converts array of AsyncVals into AsyncVal with array results.
     /// In case when are non-immediate values in provided array, they are
     /// executed all in parallel, in unordered fashion. Order of values

src/FSharp.Data.GraphQL.Shared/Validation.fs

@@ -1634,25 +1634,30 @@ module Ast =
 
     /// <summary>
     /// The <c>@defer</c> and <c>@stream</c> directives applied to the root fields of the selection set, through the
-    /// fragments spread at its root.
+    /// fragments spread at its root, and to the fragments at its root themselves, which have an empty path.
     /// </summary>
     let rec private rootIncrementalDirectiveUsages
         (fragmentDefinitions : FragmentDefinition list)
         (visitedFragments : string list)
         (selectionSet : Selection list)
         : (FieldPath * Directive) list =
+        let usagesOf (directives : Directive list) (path : FieldPath) =
+            directives
+            |> List.filter isIncrementalDirective
+            |> List.map (fun directive -> path, directive)
         selectionSet
         |> List.collect (function
-            | Field field ->
-                field.Directives
-                |> List.filter isIncrementalDirective
-                |> List.map (fun directive -> [ box field.AliasOrName ], directive)
-            | InlineFragment fragment -> rootIncrementalDirectiveUsages fragmentDefinitions visitedFragments fragment.SelectionSet
+            | Field field -> usagesOf field.Directives [ box field.AliasOrName ]
+            | InlineFragment fragment ->
+                usagesOf fragment.Directives []
+                @ rootIncrementalDirectiveUsages fragmentDefinitions visitedFragments fragment.SelectionSet
             | FragmentSpread spread when not (visitedFragments |> List.contains spread.Name) ->
                 match fragmentDefinitions |> List.tryFind (fun fragment -> fragment.Name = ValueSome spread.Name) with
-                | Some fragment -> rootIncrementalDirectiveUsages fragmentDefinitions (spread.Name :: visitedFragments) fragment.SelectionSet
-                | None -> []
-            | FragmentSpread _ -> [])
+                | Some fragment ->
+                    usagesOf spread.Directives []
+                    @ rootIncrementalDirectiveUsages fragmentDefinitions (spread.Name :: visitedFragments) fragment.SelectionSet
+                | None -> usagesOf spread.Directives []
+            | FragmentSpread spread -> usagesOf spread.Directives [])
 
     /// <summary>
     /// The <c>@stream</c> directive may only be applied to list fields
@@ -1695,7 +1700,9 @@ module Ast =
             | _ -> Success)
 
     /// <summary>
-    /// <c>@defer</c> and <c>@stream</c> cannot be applied to the root fields of a mutation, which are executed serially
+    /// <c>@defer</c> and <c>@stream</c> cannot be applied to the root fields of a mutation, which are executed serially,
+    /// nor to the fragments selecting them, whose fields would otherwise be executed after the rest of the mutation and
+    /// concurrently
     /// (<see href="https://github.com/graphql/graphql-spec/pull/1110">Defer And Stream Directives Are Used On Valid Root Field</see>).
     /// </summary>
     let internal validateDeferStreamDirectivesOnRootFields (ctx : ValidationContext) =
@@ -1710,10 +1717,18 @@ module Ast =
                 rootIncrementalDirectiveUsages fragmentDefinitions [] def.SelectionSet
                 |> List.filter (fun (_, directive) -> not (isDisabledIncrementalDirective directive))
                 |> ValidationResult.collect (fun (path, directive) ->
-                    AstError.AsResult (
-                        $"Directive '%s{directive.Name}' cannot be applied to a root field of the mutation type '%s{mutationTypeName}'.",
-                        path
-                    ))
+                    match path with
+                    // Only a fragment at the root has no path of its own
+                    | [] ->
+                        AstError.AsResult (
+                            $"Directive '%s{directive.Name}' cannot be applied to a fragment selecting root fields of the mutation type '%s{mutationTypeName}'.",
+                            path
+                        )
+                    | _ ->
+                        AstError.AsResult (
+                            $"Directive '%s{directive.Name}' cannot be applied to a root field of the mutation type '%s{mutationTypeName}'.",
+                            path
+                        ))
             | _ -> Success)
 
     /// <summary>

tests/FSharp.Data.GraphQL.Tests/AstValidationTests.fs

@@ -1545,6 +1545,30 @@ let ``Validation should grant that defer and stream are not used on mutation roo
             "Directive 'defer' cannot be applied to a root field of the mutation type 'Mutation'."
     )
 
+[<Fact>]
+let ``Validation should grant that defer is not used on fragments selecting mutation root fields`` () =
+    let query =
+        """mutation {
+  ... @defer {
+    first: convert(value: 1)
+  }
+  ...Conversion @defer
+}
+
+fragment Conversion on Mutation {
+  second: convert(value: 2)
+}"""
+    let expected =
+        GQLProblemDetails.CreateValidationFor
+            []
+            "Directive 'defer' cannot be applied to a fragment selecting root fields of the mutation type 'Mutation'."
+    match validateWholeDocument query with
+    | ValidationError errors ->
+        let deferErrors = errors |> List.filter ((=) expected)
+        // Both the inline fragment and the fragment spread are rejected
+        Assert.Equal (2, deferErrors.Length)
+    | Success -> fail $"Expected the validation error '%s{expected.Message}' but the document was accepted"
+
 [<Fact>]
 let ``Validation should grant that defer and stream labels are unique in the document`` () =
     let query =