Skip to content

Fixed streaming execution and middleware behavior #612

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
xperiandri merged 3 commits into from
Sep 18, 2026
Merged

Fixed streaming execution and middleware behavior #612

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8d7dfd1
Fix streaming execution and middleware behavior
xperiandri Sep 18, 2026
b57bf2d
Fix review thread regressions
Copilot Sep 18, 2026
9104cb4
Make observable regression test deterministic
Copilot Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions RELEASE_NOTES.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,16 +301,16 @@
* Added `StreamBatching` to group streamed items of a `Define.TaskSeqField` into batches of a fixed size or of a size computed from the sequence, such as a page size kept with a paged SDK sequence. The `preferredBatchSize` argument of `@stream` takes precedence, and the batching function itself is evaluated lazily, only for a `@stream` query that does not supply its own `preferredBatchSize`
* Added `Microsoft.Bcl.AsyncInterfaces` dependency of `FSharp.Data.GraphQL.Shared` for `netstandard2.0`
* Added `Human.friendsStream` field to the Star Wars sample to demonstrate `@stream`
* Fixed a query or mutation whose root field has an invalid inline (literal) argument, such as a custom input object validator failing, being reported as a `Direct` result with `null` data instead of a `RequestError`; inline argument coercion is now checked for every root field before any of them execute, the same as variable coercion, so a mutation no longer executes earlier root fields before rejecting the request over a later one's invalid argument
* Fixed `Define.TaskSeqField` streaming retaining a task for every item already delivered until the sequence ends
* Fixed `graphql-transport-ws` delivery of `@defer` and `@stream` results, which are now sent as soon as they are produced with `path` and `hasNext` instead of after a fixed 5 second delay, followed by a final payload with `hasNext: false`
* Fixed `graphql-transport-ws` failure on deferred and streamed results that are not objects, such as streamed list items and scalars
* Fixed `graphql-transport-ws` dropping errors of the initial payload of a deferred result together with all its deferred results
* Fixed `graphql-transport-ws` discarding the partial `data` of a subscription result that also had field errors, sending `null` instead
* Fixed `graphql-transport-ws` discarding the field errors of a `Direct` (non-subscription) result, sending an empty error list instead
* Fixed `graphql-transport-ws` stranding a subscription id forever when its deferred result completed synchronously, before it was registered
* Fixed `Define.TaskSeqField` streaming retaining a task for every item already delivered until the sequence ends
* Fixed `graphql-transport-ws` leaving a subscription id occupied when subscribing to its result failed synchronously
* Fixed `graphql-transport-ws` addressing a batch of streamed items (grouped by `preferredBatchSize` or `StreamBatching`) with a `path` ending in the list of the batch's own indices, such as `["numbers", [0, 1]]`, which no client can merge into the response tree; a batch is now sent as one independently addressed payload per item instead, in the batch's own order
* Fixed `graphql-transport-ws` never sending `complete` after the `next` of a query or mutation result, as the protocol requires
* Fixed `graphql-transport-ws` sending a request error (rejected before execution: validation, planning, variable coercion, a middleware, or the executor itself failing) as a `next` result followed by `complete`, instead of the terminal `error` message the protocol requires for it; a query or mutation whose non-null root field fails during execution still gets `next` + `complete`, since it is a result, not a request error
* Fixed `graphql-transport-ws` throwing while serializing an `error` message or a `pong` carrying a payload, since neither was written under the `payload` property name `Utf8JsonWriter` requires
* Fixed a query or mutation whose root field has an invalid inline (literal) argument, such as a custom input object validator failing, being reported as a `Direct` result with `null` data instead of a `RequestError`; inline argument coercion is now checked for every root field before any of them execute, the same as variable coercion, so a mutation no longer executes earlier root fields before rejecting the request over a later one's invalid argument
124 changes: 73 additions & 51 deletions src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,18 @@ open FSharp.Data.GraphQL.Ast
open FSharp.Data.GraphQL.Types.Patterns
open FSharp.Data.GraphQL.Types

type internal QueryWeightMiddleware(threshold : float, reportToMetadata : bool) =
type internal QueryWeightMiddleware (threshold : float, reportToMetadata : bool) =

let middleware (threshold : float) (inputContext : InputExecutionContextProvider) (ctx : ExecutionContext) (next : ExecutionContext -> AsyncVal<GQLExecutionResult>) =
let middleware
(threshold : float)
(inputContext : InputExecutionContextProvider)
(ctx : ExecutionContext)
(next : ExecutionContext -> AsyncVal<GQLExecutionResult>)
=
let measureThreshold (threshold : float) (fields : ExecutionInfo list) =
let getWeight f =
if f.ParentDef = upcast ctx.ExecutionPlan.RootDef
then 0.0
if f.ParentDef = upcast ctx.ExecutionPlan.RootDef then
0.0
else
match f.Definition.Metadata.TryFind<float>("queryWeight") with
| ValueSome w -> w
Expand All @@ -34,67 +39,79 @@ type internal QueryWeightMiddleware(threshold : float, reportToMetadata : bool)
| [] -> (true, acc)
| x :: xs ->
let current = acc + (getWeight x)
if current > threshold then (false, current)
else match x.Kind with
| ResolveValue -> checkThreshold current xs
| SelectFields fields ->
if current > threshold then
(false, current)
else
match x.Kind with
| ResolveValue -> checkThreshold current xs
| SelectFields fields ->
let (pass, current) = checkThreshold current fields
if pass then checkThreshold current xs else (false, current)
| ResolveCollection field ->
| ResolveCollection field ->
let (pass, current) = checkThreshold acc [ field ]
if pass then checkThreshold current xs else (false, current)
| ResolveAbstraction typeFields ->
| ResolveAbstraction typeFields ->
let fields = typeFields |> Map.toList |> List.collect (fun (_, v) -> v)
let (pass, current) = checkThreshold current fields
if pass then checkThreshold current xs else (false, current)
| ResolveDeferred info -> checkThreshold current (info :: xs)
| ResolveStreamed (info, _) -> checkThreshold current (info :: xs)
| ResolveLive info -> checkThreshold current (info :: xs)
| ResolveDeferred info -> checkThreshold current (info :: xs)
| ResolveStreamed (info, _) -> checkThreshold current (info :: xs)
| ResolveLive info -> checkThreshold current (info :: xs)
checkThreshold 0.0 fields
let error (ctx : ExecutionContext) =
GQLExecutionResult.ErrorAsync(ctx.ExecutionPlan.DocumentId, "Query complexity exceeds maximum threshold. Please reduce query complexity and try again.", ctx.Metadata)
GQLExecutionResult.ErrorAsync (
ctx.ExecutionPlan.DocumentId,
"Query complexity exceeds maximum threshold. Please reduce query complexity and try again.",
ctx.Metadata
)
let (pass, totalWeight) = measureThreshold threshold ctx.ExecutionPlan.Fields
let ctx =
match reportToMetadata with
| true -> { ctx with Metadata = ctx.Metadata.Add("queryWeightThreshold", threshold).Add("queryWeight", totalWeight) }
| true -> {
ctx with
Metadata = ctx.Metadata.Add("queryWeightThreshold", threshold).Add("queryWeight", totalWeight)
}
| false -> ctx
if pass
then next ctx
else error ctx
if pass then next ctx else error ctx

interface IExecutorMiddleware with
member _.CompileSchema = None
member _.PostCompileSchema = None
member _.PlanOperation = None
member _.ExecuteOperationAsync = Some (middleware threshold)

type internal ObjectListFilterMiddleware<'ObjectType, 'ListType>(reportToMetadata : bool) =
type internal ObjectListFilterMiddleware<'ObjectType, 'ListType> (reportToMetadata : bool) =

let compileMiddleware (ctx : SchemaCompileContext) (next : SchemaCompileContext -> unit) =
let modifyFields (object : ObjectDef<'ObjectType>) (fields : FieldDef<'ObjectType> seq) =
let args = [ Define.Input("filter", Nullable ObjectListFilterType) ]
let args = [ Define.Input ("filter", Nullable ObjectListFilterType) ]
let fields = fields |> Seq.map _.WithArgs(args) |> Seq.toList
object.WithFields(fields)
let typesWithListFields =
ctx.TypeMap.GetTypesWithListFields<'ObjectType, 'ListType>()
if Seq.isEmpty typesWithListFields
then failwith $"No lists with specified type '{typeof<'ObjectType>}' where found on object of type '{typeof<'ListType>}'."
object.WithFields (fields)
let typesWithListFields = ctx.TypeMap.GetTypesWithListFields<'ObjectType, 'ListType>()
if Seq.isEmpty typesWithListFields then
failwith $"No lists with specified type '{typeof<'ObjectType>}' where found on object of type '{typeof<'ListType>}'."
let modifiedTypes =
typesWithListFields
|> Seq.map (fun (object, fields) -> modifyFields object fields)
|> Seq.cast<NamedDef>
ctx.TypeMap.AddTypes(modifiedTypes, overwrite = true)
ctx.TypeMap.AddTypes (modifiedTypes, overwrite = true)
next ctx

let reportMiddleware (inputContext : InputExecutionContextProvider) (ctx : ExecutionContext) (next : ExecutionContext -> AsyncVal<GQLExecutionResult>) =
let rec collectArgs (path: obj list) (acc : KeyValuePair<obj list, ObjectListFilter> list) (fields : ExecutionInfo list) =
let reportMiddleware
(inputContext : InputExecutionContextProvider)
(ctx : ExecutionContext)
(next : ExecutionContext -> AsyncVal<GQLExecutionResult>)
=
let rec collectArgs (path : obj list) (acc : KeyValuePair<obj list, ObjectListFilter> list) (fields : ExecutionInfo list) =
let fieldArgs currentPath field =
let filterResults =
field.Ast.Arguments
|> Seq.map (fun x ->
match x.Name, x.Value with
| "filter", (VariableName variableName) -> Ok (ValueSome (ctx.Variables[variableName] :?> ObjectListFilter))
| "filter", inlineConstant -> ObjectListFilterType.CoerceInput inputContext (InlineConstant inlineConstant) ctx.Variables |> Result.map ValueOption.ofObj
| "filter", inlineConstant ->
ObjectListFilterType.CoerceInput inputContext (InlineConstant inlineConstant) ctx.Variables
|> Result.map ValueOption.ofObj
| _ -> Ok ValueNone)
|> Seq.toList
match filterResults |> splitSeqErrorsList with
Expand All @@ -111,10 +128,8 @@ type internal ObjectListFilterMiddleware<'ObjectType, 'ListType>(reportToMetadat
let currentPath = box x.Ast.AliasOrName :: path
let accResult =
match x.Kind with
| SelectFields fields ->
collectArgs currentPath acc fields
| ResolveCollection field ->
fieldArgs currentPath field
| SelectFields fields -> collectArgs currentPath acc fields
| ResolveCollection field -> fieldArgs currentPath field
| ResolveAbstraction typeFields ->
let fields = typeFields |> Map.toList |> List.collect (fun (_, v) -> v)
collectArgs currentPath acc fields
Expand All @@ -123,18 +138,19 @@ type internal ObjectListFilterMiddleware<'ObjectType, 'ListType>(reportToMetadat
| Error errs -> Error errs
| Ok acc -> collectArgs path acc xs
let ctxResult = result {
let! args = collectArgs [] [] ctx.ExecutionPlan.Fields

match reportToMetadata with
| true ->
let! args = collectArgs [] [] ctx.ExecutionPlan.Fields
let filters = ImmutableDictionary.CreateRange args
return { ctx with Metadata = ctx.Metadata.Add("filters", filters) }
return { ctx with Metadata = ctx.Metadata.Add ("filters", filters) }
| false -> return ctx
}
match ctxResult with
| Ok ctx -> next ctx
| Error errs -> asyncVal {
return GQLExecutionResult.Direct(ctx.ExecutionPlan.DocumentId, null, (errs |> List.map GQLProblemDetails.OfError), ctx.Metadata)
}
return GQLExecutionResult.RequestError (ctx.ExecutionPlan.DocumentId, (errs |> List.map GQLProblemDetails.OfError), ctx.Metadata)
}
interface IExecutorMiddleware with
member _.CompileSchema = Some compileMiddleware
member _.PostCompileSchema = None
Expand All @@ -144,22 +160,25 @@ type internal ObjectListFilterMiddleware<'ObjectType, 'ListType>(reportToMetadat
/// A function that resolves an identity name for a schema object, based on a object definition of it.
type IdentityNameResolver = ObjectDef -> string

type internal LiveQueryMiddleware(identityNameResolver : IdentityNameResolver) =
type internal LiveQueryMiddleware (identityNameResolver : IdentityNameResolver) =

let middleware (ctx : SchemaCompileContext) (next : SchemaCompileContext -> unit) =
let identity (identityName : string) (x : obj) =
x.GetType().GetProperty(identityName).GetValue(x)
let project (fieldName : string) (x : obj) =
x.GetType().GetProperty(fieldName).GetValue(x)
let makeSubscription id typeName fieldName : LiveFieldSubscription =
{ Filter = (fun x y -> identity id x = identity id y); Project = project fieldName; TypeName = typeName; FieldName = fieldName }
let identity (identityName : string) (x : obj) = x.GetType().GetProperty(identityName).GetValue(x)
let project (fieldName : string) (x : obj) = x.GetType().GetProperty(fieldName).GetValue(x)
let makeSubscription id typeName fieldName : LiveFieldSubscription = {
Filter = (fun x y -> identity id x = identity id y)
Project = project fieldName
TypeName = typeName
FieldName = fieldName
}
let getObjDefs (def : FieldDef) =
let rec helper (acc : ObjectDef list) (def : TypeDef) =
match def with
| Object objdef ->
if not (acc |> List.exists (fun x -> x.Name = objdef.Name))
then helper (objdef :: acc) objdef
else acc
if not (acc |> List.exists (fun x -> x.Name = objdef.Name)) then
helper (objdef :: acc) objdef
else
acc
| Nullable innerdef -> helper acc innerdef
| List innerdef -> helper acc innerdef
| Union udef -> (udef.Options |> List.ofArray) @ acc
Expand All @@ -169,14 +188,17 @@ type internal LiveQueryMiddleware(identityNameResolver : IdentityNameResolver) =
|> Map.toSeq
|> Seq.collect (snd >> getObjDefs)
|> Seq.map (fun objdef -> identityNameResolver objdef, objdef)
|> Seq.filter (fun (id, objdef) -> not (isNull (objdef.Type.GetProperty(id))))
|> Seq.filter (fun (id, objdef) -> not (isNull (objdef.Type.GetProperty (id))))
|> Seq.collect (fun (id, objdef) ->
objdef.Fields
|> Map.toSeq
|> Seq.map (snd >> (fun fdef -> makeSubscription id objdef.Name fdef.Name)))
|> Seq.map (
snd
>> (fun fdef -> makeSubscription id objdef.Name fdef.Name)
))
|> Seq.iter (fun x ->
if not (ctx.Schema.LiveFieldSubscriptionProvider.IsRegistered x.TypeName x.FieldName)
then ctx.Schema.LiveFieldSubscriptionProvider.Register x)
if not (ctx.Schema.LiveFieldSubscriptionProvider.IsRegistered x.TypeName x.FieldName) then
ctx.Schema.LiveFieldSubscriptionProvider.Register x)
next ctx

interface IExecutorMiddleware with
Expand Down
2 changes: 1 addition & 1 deletion src/FSharp.Data.GraphQL.Server/Execution.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -552,7 +552,7 @@ let private executeQueryOrMutation (resultSet: (string * ExecutionInfo) []) (ctx
match getArgumentValues argDefs info.Ast.Arguments ctx.GetInputContext ctx.Variables with
| Ok args -> coerced.Add(i, struct (args, []))
| Error errs -> coerced.Add(i, struct (Map.empty, errs)))
let coercionErrors = coerced.Values |> Seq.collect (fun struct (_, errs) -> errs) |> List.ofSeq
let coercionErrors = coerced.Values |> Seq.collect (fun struct (_, errs) -> errs) |> Seq.toList
if not coercionErrors.IsEmpty then
return GQLExecutionResult.Error(documentId, coercionErrors, ctx.Metadata)
else
Expand Down
Loading
Loading