From 8d7dfd1b959fea78c34852527cc117a39d1a3cde Mon Sep 17 00:00:00 2001 From: Andrii Chebukin Date: Fri, 18 Sep 2026 19:32:44 +0200 Subject: [PATCH 1/3] Fix streaming execution and middleware behavior Bring in the remaining non-WebSocket, non-typo changes from the original branch on top of the response-data-skippable contract: generic streaming/async-enumerator fixes and reformatting in Execution.fs, Executor.fs and ObservableExtensions.fs; schema/ type-system additions and documentation in SchemaDefinitions.fs and TypeSystem.fs; the ObjectListFilter middleware behavior change and reformatting in MiddlewareDefinitions.fs; the ErrorMessages.fs 'indetifier' rename and the 'indicies' rename in Execution.fs/ TaskSeqFieldTests.fs (incidental identifier renames, not documentation typo fixes); RELEASE_NOTES.md bullet-list reordering; and non-WebSocket test additions/reformatting across DeferredTests.fs, ExecutionTests.fs, ExecutorMiddlewareTests.fs, MutationTests.fs, Relay/ConnectionTests.fs, LazyEnumerationExceptionTests.fs, MiddlewareTests.fs, ObservableExtensionsTests.fs, ErrorHelpers.fs and TaskSeqFieldTests.fs. Excludes WebSocket transport/lifecycle/serialization changes (kept for the next commit) and typo/wording corrections (there are none in the original diff; see the Fix streaming-related typos commit for details). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- RELEASE_NOTES.md | 4 +- .../MiddlewareDefinitions.fs | 121 +++-- src/FSharp.Data.GraphQL.Server/Execution.fs | 2 +- .../ObservableExtensions.fs | 125 +++-- .../SchemaDefinitions.fs | 152 ++++++ src/FSharp.Data.GraphQL.Shared/TypeSystem.fs | 26 + .../FSharp.Data.GraphQL.Tests/ErrorHelpers.fs | 36 +- .../ObservableExtensionsTests.fs | 367 ++++++++++--- .../MiddlewareTests.fs | 503 +++++++++++++----- .../TaskSeqFieldTests.fs | 86 ++- 10 files changed, 1074 insertions(+), 348 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 718363c9c..8508dc49f 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -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 diff --git a/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs b/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs index b28b757ed..458844d10 100644 --- a/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs +++ b/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs @@ -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) = + let middleware + (threshold : float) + (inputContext : InputExecutionContextProvider) + (ctx : ExecutionContext) + (next : ExecutionContext -> AsyncVal) + = 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("queryWeight") with | ValueSome w -> w @@ -34,33 +39,40 @@ 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 @@ -68,33 +80,38 @@ type internal QueryWeightMiddleware(threshold : float, reportToMetadata : bool) 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 - ctx.TypeMap.AddTypes(modifiedTypes, overwrite = true) + ctx.TypeMap.AddTypes (modifiedTypes, overwrite = true) next ctx - let reportMiddleware (inputContext : InputExecutionContextProvider) (ctx : ExecutionContext) (next : ExecutionContext -> AsyncVal) = - let rec collectArgs (path: obj list) (acc : KeyValuePair list) (fields : ExecutionInfo list) = + let reportMiddleware + (inputContext : InputExecutionContextProvider) + (ctx : ExecutionContext) + (next : ExecutionContext -> AsyncVal) + = + let rec collectArgs (path : obj list) (acc : KeyValuePair 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 @@ -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 @@ -127,14 +142,14 @@ type internal ObjectListFilterMiddleware<'ObjectType, 'ListType>(reportToMetadat | 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 @@ -144,22 +159,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 @@ -169,14 +187,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 diff --git a/src/FSharp.Data.GraphQL.Server/Execution.fs b/src/FSharp.Data.GraphQL.Server/Execution.fs index f9a5fc7f7..194e70969 100644 --- a/src/FSharp.Data.GraphQL.Server/Execution.fs +++ b/src/FSharp.Data.GraphQL.Server/Execution.fs @@ -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 diff --git a/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs b/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs index d1063fa98..45fa3d11b 100644 --- a/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs +++ b/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs @@ -13,32 +13,33 @@ module internal Observable = let ofAsyncVal x = x |> AsyncVal.toAsync |> ofAsync - let toSeq (o : IObservable<'T>) : 'T seq = Observable.ToEnumerable(o) + let toSeq (o : IObservable<'T>) : 'T seq = Observable.ToEnumerable (o) /// Projects each element of an observable sequence into consecutive non-overlapping buffers /// which are produced based on timing information. let bufferMilliseconds (ms : int) x = - let span = TimeSpan.FromMilliseconds(float ms) - Observable.Buffer(x, span) + let span = TimeSpan.FromMilliseconds (float ms) + Observable.Buffer (x, span) /// Projects each element of an observable sequence into consecutive non-overlapping buffers /// which are produced based on timing and element count information. let bufferMillisecondsCount (ms : int) (count : int) x = - let span = TimeSpan.FromMilliseconds(float ms) - Observable.Buffer(x, span, count) + let span = TimeSpan.FromMilliseconds (float ms) + Observable.Buffer (x, span, count) - let ofAsyncSeq (items : Async<'Item> seq) = - items |> Seq.map ofAsync |> Observable.Merge + let ofAsyncSeq (items : Async<'Item> seq) = items |> Seq.map ofAsync |> Observable.Merge - let ofAsyncValSeq (items : AsyncVal<'Item> seq) = - items |> Seq.map ofAsyncVal |> Observable.Merge + let ofAsyncValSeq (items : AsyncVal<'Item> seq) = items |> Seq.map ofAsyncVal |> Observable.Merge let singleton (value : 'T) = { new IObservable<'T> with - member _.Subscribe(observer) = + member _.Subscribe (observer) = observer.OnNext value - observer.OnCompleted() - { new IDisposable with member _.Dispose() = () } + observer.OnCompleted () + { + new IDisposable with + member _.Dispose () = () + } } /// @@ -57,20 +58,6 @@ module internal Observable = return failure |> ValueOption.orElse (ValueSome ex) } - /// - /// Matches an that was not requested on - /// . - /// - /// - /// Distinguishes a linked-token cancellation, typically caused by a resolution failure, from the - /// subscription itself being disposed. - /// - [] - let (|CanceledIndependently|_|) (cancellationToken : CancellationToken) (ex : exn) = - match ex with - | :? OperationCanceledException when not cancellationToken.IsCancellationRequested -> ValueSome () - | _ -> ValueNone - /// /// Creates a cold observable, which enumerates the asynchronous sequence for every subscription. /// @@ -84,7 +71,7 @@ module internal Observable = // a subscriber's synchronization context must neither be captured by the loop nor be needed to pump it let enumerate (observer : IObserver<'T>) (cancellationToken : CancellationToken) : Task = backgroundTask { let mutable enumerator = ValueNone - let mutable failure = ValueNone + let mutable enumerationFailure = ValueNone try // Acquired inside the try, because a source may throw when asked for its enumerator let acquired = source.GetAsyncEnumerator cancellationToken @@ -93,17 +80,19 @@ module internal Observable = // The token is checked explicitly, because a sequence is not obliged to observe the token it was given while hasNext && not cancellationToken.IsCancellationRequested do let! moved = acquired.MoveNextAsync () - if moved then observer.OnNext acquired.Current - else hasNext <- false + if moved then + observer.OnNext acquired.Current + else + hasNext <- false with ex -> - failure <- ValueSome ex - let! failure = disposeEnumerator enumerator failure + enumerationFailure <- ValueSome ex + let! failure = disposeEnumerator enumerator enumerationFailure match failure with // A failure caused by disposing the subscription has no observer left to be delivered to | ValueSome ex when not cancellationToken.IsCancellationRequested -> observer.OnError ex | _ -> () } - Observable.Create<'T> (Func, CancellationToken, Task> (fun observer cancellationToken -> enumerate observer cancellationToken)) + Observable.Create<'T>(Func, CancellationToken, Task>(fun observer cancellationToken -> enumerate observer cancellationToken)) /// /// Enumerates the sequence, resolving each item into a result with . At most @@ -145,12 +134,14 @@ module internal Observable = // concurrency slot and, at the end, the resolutions still draining on the thread pool - a subscriber's // synchronization context that has to be pumped for those continuations would be a deadlock waiting to happen let enumerate (observer : IObserver<'Result>) (cancellationToken : CancellationToken) : Task = backgroundTask { - use linkedCancellation = CancellationTokenSource.CreateLinkedTokenSource cancellationToken + use enumerationCancellation = CancellationTokenSource.CreateLinkedTokenSource cancellationToken use slots = new SemaphoreSlim (maxConcurrency, maxConcurrency) // Observer calls are not required to be thread-safe, but resolutions complete on arbitrary threads let sync = obj () let emit (result : 'Result) = - lock sync (fun () -> if not cancellationToken.IsCancellationRequested then observer.OnNext result) + lock sync (fun () -> + if not cancellationToken.IsCancellationRequested then + observer.OnNext result) // Only the number of resolutions still in flight is tracked, not the tasks themselves, so a long-running // source does not retain one task per item; the last resolution to settle after the enumeration has ended // completes drained. Ref cells, because the resolutions run on other threads. @@ -160,19 +151,22 @@ module internal Observable = let resolutionFailure = ref ValueNone let failed () = lock sync (fun () -> resolutionFailure.Value.IsSome) let stopped () = cancellationToken.IsCancellationRequested || failed () - let reportResolutionFailure (ex : exn) = - let shouldCancel = + let recordResolutionFailure (ex : exn) = + let shouldCancelEnumeration = lock sync (fun () -> if resolutionFailure.Value.IsNone then resolutionFailure.Value <- ValueSome ex true - else false) - if shouldCancel then - linkedCancellation.Cancel () + else + false) + + if shouldCancelEnumeration then + enumerationCancellation.Cancel () let settle () = lock sync (fun () -> inFlight.Value <- inFlight.Value - 1 - if enumerationEnded.Value && inFlight.Value = 0 then drained.TrySetResult () |> ignore) + if enumerationEnded.Value && inFlight.Value = 0 then + drained.TrySetResult () |> ignore) let resolveInBackground (pendingResult : AsyncVal<'Result>) = lock sync (fun () -> inFlight.Value <- inFlight.Value + 1) // backgroundTask, not task: a resolution must never resume on a caller's synchronization context, @@ -185,7 +179,7 @@ module internal Observable = emit result with ex -> // The first failure stops the enumeration; it is delivered once every started resolution has settled - reportResolutionFailure ex + recordResolutionFailure ex finally // Released whatever happened, otherwise the enumeration would wait for this slot forever. // Released before settling, because settling lets the enumeration finish and dispose the semaphore. @@ -194,10 +188,10 @@ module internal Observable = } |> ignore let mutable enumerator = ValueNone - let mutable failure = ValueNone + let mutable enumerationFailure = ValueNone try // Acquired inside the try, because a source may throw when asked for its enumerator - let acquired = source.GetAsyncEnumerator linkedCancellation.Token + let acquired = source.GetAsyncEnumerator enumerationCancellation.Token enumerator <- ValueSome acquired let mutable index = 0 let mutable hasNext = true @@ -228,27 +222,40 @@ module internal Observable = finally slots.Release () |> ignore | pendingResult -> resolveInBackground pendingResult - with - | CanceledIndependently cancellationToken when failed () -> () - | ex -> failure <- ValueSome ex + with ex -> + if failed () && not cancellationToken.IsCancellationRequested then + () + else + enumerationFailure <- ValueSome ex // Captured items no longer need the enumerator, so it is disposed before waiting for their resolutions - let! failure = disposeEnumerator enumerator failure + let failureBeforeDispose = + enumerationFailure + |> ValueOption.orElse resolutionFailure.Value + let! failureAfterDispose = disposeEnumerator enumerator failureBeforeDispose + let disposalFailure = + match failureBeforeDispose, failureAfterDispose with + | ValueNone, ValueSome ex -> ValueSome ex + | _ -> ValueNone // Resolutions still in flight neither need the enumerator nor the loop, only their slots lock sync (fun () -> enumerationEnded.Value <- true - if inFlight.Value = 0 then drained.TrySetResult () |> ignore) + if inFlight.Value = 0 then + drained.TrySetResult () |> ignore) do! drained.Task - let failureToReport = - match resolutionFailure.Value, failure with - | ValueSome resolutionFailure, ValueSome (CanceledIndependently cancellationToken) -> - ValueSome resolutionFailure - | _ -> failure |> ValueOption.orElse resolutionFailure.Value - match failureToReport with + match + enumerationFailure + |> ValueOption.orElse resolutionFailure.Value + |> ValueOption.orElse disposalFailure + with // A failure caused by disposing the subscription has no observer left to be delivered to | ValueSome ex when not cancellationToken.IsCancellationRequested -> emit (onFailure ex) | _ -> () + if not cancellationToken.IsCancellationRequested then + lock sync (fun () -> observer.OnCompleted ()) } - Observable.Create<'Result> (Func, CancellationToken, Task> (fun observer cancellationToken -> enumerate observer cancellationToken)) + Observable.Create<'Result>( + Func, CancellationToken, Task>(fun observer cancellationToken -> enumerate observer cancellationToken) + ) /// /// Wraps every element into and emits when the source completes. @@ -273,7 +280,7 @@ module internal AsyncEnumerable = // backgroundTask, not task: the async workflow around it may have been started on a caller's synchronization // context, which the drain has no reason to capture let enumerate () : Task> = backgroundTask { - let items = ResizeArray<'T> () + let items = ResizeArray<'T>() let mutable enumerator = ValueNone let mutable failure = ValueNone try @@ -284,8 +291,10 @@ module internal AsyncEnumerable = while hasNext do cancellationToken.ThrowIfCancellationRequested () let! moved = acquired.MoveNextAsync () - if moved then items.Add acquired.Current - else hasNext <- false + if moved then + items.Add acquired.Current + else + hasNext <- false with ex -> failure <- ValueSome ex diff --git a/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs b/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs index b67a642c3..112bf27fa 100644 --- a/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs +++ b/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs @@ -1364,6 +1364,158 @@ module SchemaDefinitions = DeprecationReason = deprecationReason Metadata = Metadata.Empty } + /// + /// Creates a struct nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a block that uses + /// or must be defined in a separate function called from + /// the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq voption>, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> voption>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq voption> = + upcast { FieldDefinition.Name = name + Description = None + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = [||] + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a struct nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a block that uses + /// or must be defined in a separate function called from + /// the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Field description. Useful for generating documentation. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq voption>, description : string, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> voption>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq voption> = + upcast { FieldDefinition.Name = name + Description = Some description + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = [||] + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a struct nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a block that uses + /// or must be defined in a separate function called from + /// the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// List of field arguments used to parametrize resolve expression output. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq voption>, args : InputFieldDef list, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> voption>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq voption> = + upcast { FieldDefinition.Name = name + Description = None + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = args |> List.toArray + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a struct nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a block that uses + /// or must be defined in a separate function called from + /// the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Field description. Useful for generating documentation. + /// List of field arguments used to parametrize resolve expression output. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq voption>, description : string, args : InputFieldDef list, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> voption>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq voption> = + upcast { FieldDefinition.Name = name + Description = Some description + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = args |> List.toArray + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + /// /// Creates a custom defined field using a custom field execution function. /// diff --git a/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs b/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs index 2ac61474d..915416223 100644 --- a/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs +++ b/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs @@ -874,6 +874,7 @@ and Resolve = /// output defines .NET type of the sequence items /// expr is untyped version of Expr'Input->IAsyncEnumerable<'Output>> /// or Expr'Input->IAsyncEnumerable<'Output> option> + /// or Expr'Input->IAsyncEnumerable<'Output> voption> /// streaming defines how items are grouped and how many are resolved concurrently when the field is streamed | TaskSeq of input : Type * output : Type * expr : Expr * streaming : TaskSeqStreamingOptions @@ -2423,6 +2424,16 @@ module Resolve = else ValueNone + [] + let private (|FSharpValueOption|_|) (typ : Type) = + if + typ.GetTypeInfo().IsGenericType + && typ.GetGenericTypeDefinition () = typedefof> + then + ValueSome (typ.GenericTypeArguments |> Array.head) + else + ValueNone + [] let private (|FSharpAsync|_|) (typ : Type) = if @@ -2489,6 +2500,17 @@ module Resolve = | None -> null resolve + let private boxifyTaskSeqValueOption<'T, 'U> (streaming : TaskSeqStreamingOptions) (f : ResolveFieldContext -> 'T -> IAsyncEnumerable<'U> voption) : ResolveFieldContext -> obj -> obj = + let valueNone : obj voption = ValueNone + let resolve (ctx : ResolveFieldContext) (x : obj) = + match f ctx (x :?> 'T) with + | ValueSome source -> + match wrapAsyncEnumerable streaming source with + | null -> null + | wrapped -> box (ValueSome wrapped) + | ValueNone -> box valueNone + resolve + let private getRuntimeMethod name = let methods = typeof.DeclaringType.GetRuntimeMethods () methods |> Seq.find (fun m -> m.Name.Equals name) @@ -2505,6 +2527,8 @@ module Resolve = let private runtimeBoxifyTaskSeqOption = getRuntimeMethod (nameof boxifyTaskSeqOption) + let private runtimeBoxifyTaskSeqValueOption = getRuntimeMethod (nameof boxifyTaskSeqValueOption) + let private unwrapExpr = function | WithValue (resolver, _, _) -> (resolver, resolver.GetType ()) @@ -2557,6 +2581,8 @@ module Resolve = match unwrapExpr expr with | resolver, FSharpFunc (_, FSharpFunc (d, AsyncEnumerable (c))) -> invoke runtimeBoxifyTaskSeq d c resolver | resolver, FSharpFunc (_, FSharpFunc (d, FSharpOption (AsyncEnumerable (c)))) -> invoke runtimeBoxifyTaskSeqOption d c resolver + | resolver, FSharpFunc (_, FSharpFunc (d, FSharpValueOption (AsyncEnumerable (c)))) -> + invoke runtimeBoxifyTaskSeqValueOption d c resolver | resolver, _ -> failwithf "Unsupported signature for TaskSeq Resolve %A" (resolver.GetType ()) let (|BoxedSync|_|) = diff --git a/tests/FSharp.Data.GraphQL.Tests/ErrorHelpers.fs b/tests/FSharp.Data.GraphQL.Tests/ErrorHelpers.fs index 9231f0780..da65172b2 100644 --- a/tests/FSharp.Data.GraphQL.Tests/ErrorHelpers.fs +++ b/tests/FSharp.Data.GraphQL.Tests/ErrorHelpers.fs @@ -11,9 +11,12 @@ type ErrorSource = | Variable of Name : string | Argument of Name : string -let ensureDeferred (result : GQLExecutionResult) (onDeferred : Output -> GQLProblemDetails list -> IObservable -> unit) : unit = +let ensureDeferred + (result : GQLExecutionResult) + (onDeferred : Output -> GQLProblemDetails list -> IObservable -> unit) + : unit = match result.Content with - | Deferred(data, errors, deferred) -> onDeferred data errors deferred + | Deferred (data, errors, deferred) -> onDeferred data errors deferred | response -> fail $"Expected a 'Deferred' GQLResponse but got\n{response}" let ensureDirect (result : GQLExecutionResult) (onDirect : Output -> GQLProblemDetails list -> unit) : unit = @@ -38,16 +41,14 @@ let ensureValidationError (message : string) (path : FieldPath) (error : GQLProb equals (Include path) error.Path match error.Extensions with | Skip -> fail "Expected extensions to be present" - | Include extensions -> - equals Validation (unbox extensions[CustomErrorFields.Kind]) + | Include extensions -> equals Validation (unbox extensions[CustomErrorFields.Kind]) let ensureExecutionError (message : string) (path : FieldPath) (error : GQLProblemDetails) = equals message error.Message equals (Include path) error.Path match error.Extensions with | Skip -> fail "Expected extensions to be present" - | Include extensions -> - equals Execution (unbox extensions[CustomErrorFields.Kind]) + | Include extensions -> equals Execution (unbox extensions[CustomErrorFields.Kind]) let ensureInputCoercionError (errorSource : ErrorSource) (message : string) (``type`` : string) (error : GQLProblemDetails) = equals message error.Message @@ -59,11 +60,18 @@ let ensureInputCoercionError (errorSource : ErrorSource) (message : string) (``t | Variable name -> equals name (unbox extensions[CustomErrorFields.VariableName]) equals ``type`` (unbox extensions[CustomErrorFields.VariableType]) - | Argument name -> + | Argument name -> equals name (unbox extensions[CustomErrorFields.ArgumentName]) equals ``type`` (unbox extensions[CustomErrorFields.ArgumentType]) -let ensureInputObjectFieldCoercionError (errorSource : ErrorSource) (message : string) (inputObjectPath : FieldPath) (objectType : string) (fieldType : string) (error : GQLProblemDetails) = +let ensureInputObjectFieldCoercionError + (errorSource : ErrorSource) + (message : string) + (inputObjectPath : FieldPath) + (objectType : string) + (fieldType : string) + (error : GQLProblemDetails) + = equals message error.Message match error.Extensions with | Skip -> fail "Expected extensions to be present" @@ -71,13 +79,19 @@ let ensureInputObjectFieldCoercionError (errorSource : ErrorSource) (message : s equals InputCoercion (unbox extensions[CustomErrorFields.Kind]) match errorSource with | Variable name -> equals name (unbox extensions[CustomErrorFields.VariableName]) - | Argument name -> equals name (unbox extensions[CustomErrorFields.ArgumentName]) + | Argument name -> equals name (unbox extensions[CustomErrorFields.ArgumentName]) if not inputObjectPath.IsEmpty then equals inputObjectPath (unbox extensions[CustomErrorFields.Path]) equals objectType (unbox extensions[CustomErrorFields.ObjectType]) equals fieldType (unbox extensions[CustomErrorFields.FieldType]) -let ensureInputObjectValidationError (errorSource : ErrorSource) (message : string) (inputObjectPath : FieldPath) (objectType : string) (error : GQLProblemDetails) = +let ensureInputObjectValidationError + (errorSource : ErrorSource) + (message : string) + (inputObjectPath : FieldPath) + (objectType : string) + (error : GQLProblemDetails) + = equals message error.Message match error.Extensions with | Skip -> fail "Expected extensions to be present" @@ -85,7 +99,7 @@ let ensureInputObjectValidationError (errorSource : ErrorSource) (message : stri equals InputObjectValidation (unbox extensions[CustomErrorFields.Kind]) match errorSource with | Variable name -> equals name (unbox extensions[CustomErrorFields.VariableName]) - | Argument name -> equals name (unbox extensions[CustomErrorFields.ArgumentName]) + | Argument name -> equals name (unbox extensions[CustomErrorFields.ArgumentName]) if not inputObjectPath.IsEmpty then equals inputObjectPath (unbox extensions[CustomErrorFields.Path]) equals objectType (unbox extensions[CustomErrorFields.ObjectType]) diff --git a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs index ee8757d50..7a31d503f 100644 --- a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs @@ -8,22 +8,32 @@ open FSharp.Data.GraphQL open Helpers open System +open System.Threading +open System.Threading.Tasks open FSharp.Control.Reactive [] let ``ofSeq should call OnComplete and return items in expected order`` () = - let source = seq { for x in 1 .. 5 do yield x } + let source = seq { + for x in 1..5 do + yield x + } let obs = Observable.ofSeq source use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals source [] let ``bind should call OnComplete and return items in expected order`` () = - let source = seq { for x in 1 .. 5 do yield x } - let obs = Observable.ofSeq source |> Observable.bind (fun x -> Observable.ofSeq [x; x]) + let source = seq { + for x in 1..5 do + yield x + } + let obs = + Observable.ofSeq source + |> Observable.bind (fun x -> Observable.ofSeq [ x; x ]) use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 1; 2; 2; 3; 3; 4; 4; 5; 5 ] [] @@ -31,7 +41,7 @@ let ``ofAsync should call OnComplete and return items in expected order`` () = let source = async { return "test" } let obs = Observable.ofAsync source use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ "test" ] @@ -40,12 +50,15 @@ let ``ofAsyncVal should call OnComplete and return items in expected order`` () let source = async { return "test" } |> AsyncVal.ofAsync let obs = Observable.ofAsyncVal source use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ "test" ] [] let ``toSeq on a finite sequence should generate a finite sequence`` () = - let source = seq { for x in 1 .. 5 do yield x } + let source = seq { + for x in 1..5 do + yield x + } let obs = Observable.ofSeq source let result = Observable.toSeq obs result |> seqEquals source @@ -55,7 +68,7 @@ let ``ofSeq on an empty sequence should call OnComplete and return items in expe let source = Seq.empty let obs = Observable.ofSeq source use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals source [] @@ -63,10 +76,11 @@ let ``ofAsyncSeq should call OnComplete and return items in expected order`` () let source = seq { yield delay 300 2 yield delay 100 1 - yield delay 200 3 } + yield delay 200 3 + } let obs = Observable.ofAsyncSeq source use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 3; 2 ] [] @@ -74,10 +88,11 @@ let ``ofAsyncValSeq should call OnComplete and return items in expected order`` let source = seq { yield delay 300 2 |> AsyncVal.ofAsync yield delay 100 1 |> AsyncVal.ofAsync - yield delay 200 3 |> AsyncVal.ofAsync } + yield delay 200 3 |> AsyncVal.ofAsync + } let obs = Observable.ofAsyncValSeq source use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 3; 2 ] [] @@ -85,22 +100,30 @@ let ``bufferByTiming should call OnComplete and return items in expected order`` let source = seq { yield delay 400 2 yield delay 100 1 - yield delay 200 3 } - let obs = Observable.ofAsyncSeq source |> Observable.bufferMilliseconds (ms 300) |> Observable.map List.ofSeq + yield delay 200 3 + } + let obs = + Observable.ofAsyncSeq source + |> Observable.bufferMilliseconds (ms 300) + |> Observable.map List.ofSeq use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) - sub.Received |> seqEquals [ [1; 3]; [2] ] + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ [ 1; 3 ]; [ 2 ] ] [] let ``bufferByElementCount should call OnComplete and return items in expected order`` () = let source = seq { yield delay 400 2 yield delay 100 1 - yield delay 200 3 } - let obs = Observable.ofAsyncSeq source |> Observable.bufferCount 2 |> Observable.map List.ofSeq + yield delay 200 3 + } + let obs = + Observable.ofAsyncSeq source + |> Observable.bufferCount 2 + |> Observable.map List.ofSeq use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) - sub.Received |> seqEquals [ [1; 3]; [2] ] + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ [ 1; 3 ]; [ 2 ] ] [] let ``bufferByTimingAndElementCount should call OnComplete and return items in expected order`` () = @@ -108,34 +131,47 @@ let ``bufferByTimingAndElementCount should call OnComplete and return items in e yield delay 500 2 yield delay 50 1 yield delay 100 3 - yield delay 150 4 } - let obs = Observable.ofAsyncSeq source |> Observable.bufferMillisecondsCount (ms 300) 2 |> Observable.map List.ofSeq + yield delay 150 4 + } + let obs = + Observable.ofAsyncSeq source + |> Observable.bufferMillisecondsCount (ms 300) 2 + |> Observable.map List.ofSeq use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) - sub.Received |> seqEquals [ [1; 3]; [4]; [2] ] + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ [ 1; 3 ]; [ 4 ]; [ 2 ] ] -type IndexException(index : int) = - inherit exn(sprintf "Error at index %i." index) +type IndexException (index : int) = + inherit exn (sprintf "Error at index %i." index) member _.Index = index [] let ``catch should call OnComplete and return items in expected order`` () = - let source : int seq = seq { for x in 1 .. 5 do yield raise <| IndexException(x) } + let source : int seq = seq { + for x in 1..5 do + yield raise <| IndexException (x) + } let obs = Observable.ofSeq source |> Observable.catchWith (fun (ex : IndexException) -> ex.Index |> Observable.singleton) use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1 ] [] let ``choose should cal OnComplete`` () = - let source = seq { for x in 1 .. 5 do yield x } + let source = seq { + for x in 1..5 do + yield x + } let obs = Observable.ofSeq source - |> Observable.choose (fun x -> match x % 2 with | 0 -> Some x | _ -> None) + |> Observable.choose (fun x -> + match x % 2 with + | 0 -> Some x + | _ -> None) use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 2; 4 ] [] @@ -143,17 +179,23 @@ let ``concatInner should call OnComplete and return items in expected order`` () let source1 = seq { yield delay 500 2 yield delay 100 1 - yield delay 200 3 } + yield delay 200 3 + } let source2 = seq { yield delay 400 4 - yield delay 300 5 } - let source = seq { yield Seq.empty; yield source1; yield source2 } + yield delay 300 5 + } + let source = seq { + yield Seq.empty + yield source1 + yield source2 + } let obs = Observable.ofSeq source |> Observable.map Observable.ofAsyncSeq |> Observable.concatInner use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 3; 2; 5; 4 ] [] @@ -161,15 +203,17 @@ let ``concat should call OnComplete and return items in expected order`` () = let source1 = seq { yield delay 500 2 yield delay 100 1 - yield delay 200 3 } + yield delay 200 3 + } let source2 = seq { yield delay 400 4 - yield delay 300 5 } + yield delay 300 5 + } let obs = Observable.ofAsyncSeq source1 |> Observable.concat (Observable.ofAsyncSeq source2) use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 3; 2; 5; 4 ] [] @@ -177,17 +221,23 @@ let ``mergeInner should call OnComplete and return items in expected order`` () let source1 = seq { yield delay 500 2 yield delay 100 1 - yield delay 200 3 } + yield delay 200 3 + } let source2 = seq { yield delay 400 4 - yield delay 300 5 } - let source = seq { yield Seq.empty; yield source1; yield source2 } + yield delay 300 5 + } + let source = seq { + yield Seq.empty + yield source1 + yield source2 + } let obs = Observable.ofSeq source |> Observable.map Observable.ofAsyncSeq |> Observable.mergeInner use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 3; 5; 4; 2 ] [] @@ -195,31 +245,48 @@ let ``merge should call OnComplete and return items in expected order`` () = let source1 = seq { yield delay 500 2 yield delay 100 1 - yield delay 200 3 } + yield delay 200 3 + } let source2 = seq { yield delay 400 4 - yield delay 300 5 } + yield delay 300 5 + } let obs = Observable.ofAsyncSeq source1 |> Observable.merge (Observable.ofAsyncSeq source2) use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 3; 5; 4; 2 ] [] let ``concatSeq should call OnComplete and return items in expected order`` () = - let source = seq { for x in 1 .. 5 do yield x } + let source = seq { + for x in 1..5 do + yield x + } let obs = Observable.ofSeq source use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals source -[] +[] let ``mapAsync should call OnComplete and return items in expected order`` () = - let source = seq { "a"; "b"; "c"; "d"; "e"; "f"; "g" } - let obs = Observable.ofSeq source |> Observable.flatmapAsync (fun x -> async { return x }) |> Observable.map (fun x -> x) + let source = seq { + "a" + "b" + "c" + "d" + "e" + "f" + "g" + } + let obs = + Observable.ofSeq source + |> Observable.flatmapAsync (fun x -> async { return x }) + |> Observable.map (fun x -> x) use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals source // This test tries to ensure that flatmapAsync always generates the output sequence in // the same order as the input sequence. @@ -245,7 +312,7 @@ let ``mapAsync should call OnComplete and return items in expected order`` () = let ``singleton should call OnComplete and return item`` () = let obs = Observable.singleton 1 use sub = Observer.create obs - sub.WaitCompleted(timeout = ms 10) + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals (Seq.singleton 1) open System.Threading @@ -254,8 +321,10 @@ open FSharp.Control [] let ``ofAsyncEnumerable should call OnComplete and return items in expected order`` () = - use sub = Observable.ofAsyncEnumerable (asyncItems [ 1 .. 5 ]) |> Observer.create - sub.WaitCompleted(timeout = ms 10) + use sub = + Observable.ofAsyncEnumerable (asyncItems [ 1..5 ]) + |> Observer.create + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; 2; 3; 4; 5 ] [] @@ -264,8 +333,11 @@ let ``ofAsyncEnumerable should deliver items produced before an enumeration erro yield 1 failwith "Boom" } - use sub = Observable.ofAsyncEnumerable source |> Observable.materialize |> Observer.create - sub.WaitCompleted(timeout = ms 10) + use sub = + Observable.ofAsyncEnumerable source + |> Observable.materialize + |> Observer.create + sub.WaitCompleted (timeout = ms 10) Assert.Collection ( sub.Received, (fun (notification : System.Reactive.Notification) -> @@ -282,7 +354,9 @@ let ``ofAsyncEnumerable should stop the enumeration when the subscription is dis let disposed = TaskCompletionSource () let received = TaskCompletionSource () let source = endlessNumbers pulled disposed - let subscription = Observable.ofAsyncEnumerable source |> Observable.subscribe (fun _ -> received.TrySetResult () |> ignore) + let subscription = + Observable.ofAsyncEnumerable source + |> Observable.subscribe (fun _ -> received.TrySetResult () |> ignore) do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Expected an item before the subscription is disposed" received.Task subscription.Dispose () do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Expected the enumerator to be disposed with the subscription" disposed.Task @@ -295,7 +369,7 @@ let ``ofAsyncEnumerable should stop the enumeration when the subscription is dis [] let ``ofAsyncEnumerableResolved should emit synchronously resolved results in order`` () = use sub = - Observable.ofAsyncEnumerableResolved 3 (fun _ (n : int) -> AsyncVal.wrap (n * 10)) (fun _ -> -1) (asyncItems [ 1 .. 5 ]) + Observable.ofAsyncEnumerableResolved 3 (fun _ (n : int) -> AsyncVal.wrap (n * 10)) (fun _ -> -1) (asyncItems [ 1..5 ]) |> Observer.create sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 10; 20; 30; 40; 50 ] @@ -308,16 +382,23 @@ let ``ofAsyncEnumerableResolved should never resolve more than maxConcurrency it async { let current = Interlocked.Increment inFlight let mutable observed = maxObserved.Value - while current > observed && Interlocked.CompareExchange (maxObserved, current, observed) <> observed do + while current > observed + && Interlocked.CompareExchange (maxObserved, current, observed) + <> observed do observed <- maxObserved.Value do! Async.Sleep (ms 50) Interlocked.Decrement inFlight |> ignore return n } |> AsyncVal.ofAsync - use sub = Observable.ofAsyncEnumerableResolved 2 resolve (fun _ -> -1) (asyncItems [ 1 .. 6 ]) |> Observer.create + use sub = + Observable.ofAsyncEnumerableResolved 2 resolve (fun _ -> -1) (asyncItems [ 1..6 ]) + |> Observer.create sub.WaitCompleted (timeout = ms 10) - sub.Received |> Seq.toList |> List.sort |> seqEquals [ 1; 2; 3; 4; 5; 6 ] + sub.Received + |> Seq.toList + |> List.sort + |> seqEquals [ 1; 2; 3; 4; 5; 6 ] Assert.True (maxObserved.Value <= 2, $"Expected at most 2 concurrent resolutions, but observed {maxObserved.Value}") [] @@ -332,7 +413,9 @@ let ``ofAsyncEnumerableResolved should emit the failure after a slower earlier i |> AsyncVal.ofAsync else AsyncVal.wrap n - use sub = Observable.ofAsyncEnumerableResolved 4 resolve (fun _ -> -1) source |> Observer.create + use sub = + Observable.ofAsyncEnumerableResolved 4 resolve (fun _ -> -1) source + |> Observer.create sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ 1; -1 ] @@ -359,7 +442,10 @@ let ``ofAsyncEnumerable should deliver OnError when GetAsyncEnumerator throws`` // Regression test: acquiring the enumerator happens before the try, so a throwing source must not bypass // the failure handling and fault the returned Task in a way that skips OnError let source = ThrowingAsyncEnumerable "Boom acquiring the enumerator" - use sub = Observable.ofAsyncEnumerable source |> Observable.materialize |> Observer.create + use sub = + Observable.ofAsyncEnumerable source + |> Observable.materialize + |> Observer.create sub.WaitCompleted (timeout = ms 10) Assert.Collection ( sub.Received, @@ -371,7 +457,10 @@ let ``ofAsyncEnumerable should deliver OnError when GetAsyncEnumerator throws`` [] let ``ofAsyncEnumerable should deliver OnError when DisposeAsync throws`` () = let source = itemThenDisposalFailure 1 - use sub = Observable.ofAsyncEnumerable source |> Observable.materialize |> Observer.create + use sub = + Observable.ofAsyncEnumerable source + |> Observable.materialize + |> Observer.create sub.WaitCompleted (timeout = ms 10) Assert.Collection ( sub.Received, @@ -408,7 +497,9 @@ let ``ofAsyncEnumerableResolved should stop and deliver the failure when a resol // Regression test: a failed resolution used to leave its concurrency slot held forever, so with // maxConcurrency = 1 the enumeration would deadlock instead of ever reaching onFailure or OnCompleted let resolve _ (_ : int) = AsyncVal.Failure (exn "Boom resolving") - use sub = Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) (asyncItems [ 1 ]) |> Observer.create + use sub = + Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) (asyncItems [ 1 ]) + |> Observer.create sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ -1 ] @@ -426,7 +517,9 @@ let ``ofAsyncEnumerableResolved should not pull another item after a resolution |> AsyncVal.ofAsync else AsyncVal.wrap n - use sub = Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) (asyncItems [ 1; 2; 3 ]) |> Observer.create + use sub = + Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) (asyncItems [ 1; 2; 3 ]) + |> Observer.create sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ -1 ] @@ -436,15 +529,75 @@ let ``ofAsyncEnumerableResolved should not resolve an item pulled after a resolu // when that move completed the code used to go straight to resolving it without rechecking the failure, so a // synchronously resolved item 2 was pulled and emitted before the failure that already happened let source = - SuspendingAsyncEnumerable (fun cancellationToken index -> - task { + SuspendingAsyncEnumerable(fun _ index -> task { + match index with + | 0 -> return ValueSome 1 + | 1 -> + do! Task.Delay (ms 150) + return ValueSome 2 + | _ -> return ValueNone + }) + let resolve _ (n : int) = + if n = 1 then + async { + do! Async.Sleep (ms 50) + return failwith "Boom resolving" + } + |> AsyncVal.ofAsync + else + AsyncVal.wrap n + use sub = + Observable.ofAsyncEnumerableResolved 2 resolve (fun _ -> -1) source + |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ -1 ] + +[] +let ``ofAsyncEnumerableResolved should cancel a pending MoveNextAsync after a resolution fails`` () = + // Regression test: with maxConcurrency > 1 the loop can already be suspended in MoveNextAsync for the next item + // when an earlier background resolution fails. That failure must cancel the in-progress move so the stream can + // finish with onFailure instead of hanging forever in the source. + let source = + SuspendingAsyncEnumerable(fun cancellationToken index -> task { + match index with + | 0 -> return ValueSome 1 + | 1 -> + do! Task.Delay (Timeout.Infinite, cancellationToken) + return ValueSome 2 + | _ -> return ValueNone + }) + let resolve _ (n : int) = + if n = 1 then + async { + do! Async.Sleep (ms 50) + return failwith "Boom resolving" + } + |> AsyncVal.ofAsync + else + AsyncVal.wrap n + use sub = + Observable.ofAsyncEnumerableResolved 2 resolve (fun _ -> -1) source + |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ -1 ] + +[] +let ``ofAsyncEnumerableResolved should preserve a resolution failure over DisposeAsync after canceling MoveNextAsync`` () = + // Regression test: a resolution failure can cancel an in-progress MoveNextAsync, whose cancellation is suppressed + // as expected; if DisposeAsync then throws, the original resolution failure must still win over the later + // disposal failure because it is what stopped the stream. + let source = + SuspendingAsyncEnumerable( + (fun cancellationToken index -> task { match index with | 0 -> return ValueSome 1 | 1 -> - do! Task.Delay (ms 150, cancellationToken) + do! Task.Delay (Timeout.Infinite, cancellationToken) return ValueSome 2 | _ -> return ValueNone - }) + }), + fun () -> failwith "Boom disposing" + ) let resolve _ (n : int) = if n = 1 then async { @@ -454,9 +607,57 @@ let ``ofAsyncEnumerableResolved should not resolve an item pulled after a resolu |> AsyncVal.ofAsync else AsyncVal.wrap n - use sub = Observable.ofAsyncEnumerableResolved 2 resolve (fun _ -> -1) source |> Observer.create + let observedFailure = ref "" + use sub = + Observable.ofAsyncEnumerableResolved + 2 + resolve + (fun ex -> + observedFailure.Value <- ex.Message + -1) + source + |> Observer.create sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ -1 ] + Assert.Equal ("Boom resolving", observedFailure.Value) + +[] +let ``ofAsyncEnumerableResolved should preserve a resolution failure over a later enumeration exception after canceling MoveNextAsync`` () = + // Regression test: once a resolution failure has already stopped the stream, a source that reacts to the linked + // cancellation by throwing a different exception from MoveNextAsync must not replace that original failure. + let moveFailed = TaskCompletionSource () + let source = + SuspendingAsyncEnumerable(fun cancellationToken index -> task { + match index with + | 0 -> return ValueSome 1 + | 1 -> + use _ = cancellationToken.Register (fun () -> moveFailed.TrySetResult () |> ignore) + do! moveFailed.Task + return failwith "Boom during enumeration" + | _ -> return ValueNone + }) + let resolve _ (n : int) = + if n = 1 then + async { + do! Async.Sleep (ms 50) + return failwith "Boom resolving" + } + |> AsyncVal.ofAsync + else + AsyncVal.wrap n + let observedFailure = ref "" + use sub = + Observable.ofAsyncEnumerableResolved + 2 + resolve + (fun ex -> + observedFailure.Value <- ex.Message + -1) + source + |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ -1 ] + Assert.Equal ("Boom resolving", observedFailure.Value) [] let ``ofAsyncEnumerableResolved should release the slot and not hang when the observer throws`` () : Task = task { @@ -467,13 +668,14 @@ let ``ofAsyncEnumerableResolved should release the slot and not hang when the ob // of the enumeration hanging forever on the concurrency slot the throwing resolution never released. let disposed = TaskCompletionSource () let source = - SuspendingAsyncEnumerable ( + SuspendingAsyncEnumerable( (fun _ index -> task { return if index = 0 then ValueSome 1 else ValueNone }), fun () -> disposed.TrySetResult () |> ignore ) let resolve _ (n : int) = async { return n } |> AsyncVal.ofAsync let onReceived (_ : TestObserver) (value : int) = - if value = 1 then failwith "Boom in observer" + if value = 1 then + failwith "Boom in observer" use sub = Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) source |> Observer.createWithCallback onReceived @@ -483,12 +685,19 @@ let ``ofAsyncEnumerableResolved should release the slot and not hang when the ob [] let ``withCompletionMarker should emit the items and then the marker when the source completes`` () = - use sub = Observable.ofSeq [ 1; 2 ] |> Observable.withCompletionMarker |> Observer.create - sub.WaitCompleted(timeout = ms 10) - sub.Received |> seqEquals [ ValueSome 1; ValueSome 2; ValueNone ] + use sub = + Observable.ofSeq [ 1; 2 ] + |> Observable.withCompletionMarker + |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received + |> seqEquals [ ValueSome 1; ValueSome 2; ValueNone ] [] let ``withCompletionMarker should emit only the marker for an empty source`` () = - use sub = Observable.ofSeq Seq.empty |> Observable.withCompletionMarker |> Observer.create - sub.WaitCompleted(timeout = ms 10) + use sub = + Observable.ofSeq Seq.empty + |> Observable.withCompletionMarker + |> Observer.create + sub.WaitCompleted (timeout = ms 10) sub.Received |> seqEquals [ ValueNone ] diff --git a/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs b/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs index 0584313d1..cc39831fd 100644 --- a/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs @@ -24,8 +24,9 @@ let private parseGuidId (value : string) = | true, guid -> Ok (ValueObjectId guid) | false, _ -> Error [ - { new IGQLError with - member _.Message = $"Cannot coerce '{value}' to GuidID" + { + new IGQLError with + member _.Message = $"Cannot coerce '{value}' to GuidID" } ] @@ -40,8 +41,9 @@ let ValueObjectType = | InputParameterValue.InlineConstant (StringValue value) -> parseGuidId value | _ -> Error [ - { new IGQLError with - member _.Message = "ValueObject must be provided as string" + { + new IGQLError with + member _.Message = "ValueObject must be provided as string" } ]), coerceOutput = @@ -86,16 +88,66 @@ type Property = | Community of Community let getExecutor (expectedFilter : ObjectListFilter voption) = - let a1 : A = { Id = 1; Value = "A1"; GuidValue = Guid.Parse "11111111-1111-1111-1111-111111111111"; ValueObject = ValueObjectId (Guid.Parse "11111111-1111-1111-1111-111111111111"); Subjects = [ 2; 6 ] } - let a2 : A = { Id = 2; Value = "A2"; GuidValue = Guid.Parse "22222222-2222-2222-2222-222222222222"; ValueObject = ValueObjectId (Guid.Parse "22222222-2222-2222-2222-222222222222"); Subjects = [ 1; 3; 5 ] } - let a3 : A = { Id = 3; Value = "A3"; GuidValue = Guid.Parse "33333333-3333-3333-3333-333333333333"; ValueObject = ValueObjectId (Guid.Parse "33333333-3333-3333-3333-333333333333"); Subjects = [ 1; 2; 4 ] } - let b1 = { Id = 4; Value = "1000"; GuidValue = Guid.Parse "44444444-4444-4444-4444-444444444444"; ValueObject = ValueObjectId (Guid.Parse "44444444-4444-4444-4444-444444444444"); Subjects = [ 1; 5 ] } - let b2 = { Id = 5; Value = "2000"; GuidValue = Guid.Parse "55555555-5555-5555-5555-555555555555"; ValueObject = ValueObjectId (Guid.Parse "55555555-5555-5555-5555-555555555555"); Subjects = [ 3; 4; 6 ] } - let b3 = { Id = 6; Value = "3000"; GuidValue = Guid.Parse "66666666-6666-6666-6666-666666666666"; ValueObject = ValueObjectId (Guid.Parse "66666666-6666-6666-6666-666666666666"); Subjects = [ 1; 3; 5 ] } + let a1 : A = { + Id = 1 + Value = "A1" + GuidValue = Guid.Parse "11111111-1111-1111-1111-111111111111" + ValueObject = ValueObjectId (Guid.Parse "11111111-1111-1111-1111-111111111111") + Subjects = [ 2; 6 ] + } + let a2 : A = { + Id = 2 + Value = "A2" + GuidValue = Guid.Parse "22222222-2222-2222-2222-222222222222" + ValueObject = ValueObjectId (Guid.Parse "22222222-2222-2222-2222-222222222222") + Subjects = [ 1; 3; 5 ] + } + let a3 : A = { + Id = 3 + Value = "A3" + GuidValue = Guid.Parse "33333333-3333-3333-3333-333333333333" + ValueObject = ValueObjectId (Guid.Parse "33333333-3333-3333-3333-333333333333") + Subjects = [ 1; 2; 4 ] + } + let b1 = { + Id = 4 + Value = "1000" + GuidValue = Guid.Parse "44444444-4444-4444-4444-444444444444" + ValueObject = ValueObjectId (Guid.Parse "44444444-4444-4444-4444-444444444444") + Subjects = [ 1; 5 ] + } + let b2 = { + Id = 5 + Value = "2000" + GuidValue = Guid.Parse "55555555-5555-5555-5555-555555555555" + ValueObject = ValueObjectId (Guid.Parse "55555555-5555-5555-5555-555555555555") + Subjects = [ 3; 4; 6 ] + } + let b3 = { + Id = 6 + Value = "3000" + GuidValue = Guid.Parse "66666666-6666-6666-6666-666666666666" + ValueObject = ValueObjectId (Guid.Parse "66666666-6666-6666-6666-666666666666") + Subjects = [ 1; 3; 5 ] + } let al = [ a1; a2; a3 ] let bl = [ b1; b2; b3 ] - let p1 = Complex{ Id = 1; Name = "Complex 1"; Discriminator = "Complex"; Communities = [ 5 ]; Buildings = [ 3 ] } - let p2 = Complex{ Id = 2; Name = "Complex 2"; Discriminator = "Complex"; Communities = [ 6 ]; Buildings = [ 4 ] } + let p1 = + Complex { + Id = 1 + Name = "Complex 1" + Discriminator = "Complex" + Communities = [ 5 ] + Buildings = [ 3 ] + } + let p2 = + Complex { + Id = 2 + Name = "Complex 2" + Discriminator = "Complex" + Communities = [ 6 ] + Buildings = [ 4 ] + } let p3 = Building { Id = 3; Name = "Building 1"; Discriminator = "Building" } let p4 = Building { Id = 4; Name = "Building 2"; Discriminator = "Building" } let p5 = Community { Id = 5; Name = "Community 1"; Discriminator = "Community" } @@ -126,7 +178,7 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = | B _ -> upcast BType) ) and AType = - DefineRec.Object ( + DefineRec.Object( name = "A", isTypeOf = (fun o -> o :? A), fieldsFn = @@ -145,11 +197,11 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = |> ValueOption.iter (fun _ -> equals expectedFilter ctx.Filter) a.Subjects |> List.map getSubject |> List.toSeq |> Some ) - .WithQueryWeight (1.0) + .WithQueryWeight(1.0) ] ) and BType = - DefineRec.Object ( + DefineRec.Object( name = "B", isTypeOf = (fun o -> o :? B), fieldsFn = @@ -168,11 +220,11 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = |> ValueOption.iter (fun _ -> equals expectedFilter ctx.Filter) b.Subjects |> List.map getSubject |> List.toSeq |> Some ) - .WithQueryWeight (1.0) + .WithQueryWeight(1.0) ] ) and ComplexType = - DefineRec.Object ( + DefineRec.Object( name = "Complex", isTypeOf = (fun o -> o :? Complex), fieldsFn = @@ -185,7 +237,7 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = ] ) and BuildingType = - Define.Object ( + Define.Object( name = "Building", isTypeOf = (fun o -> o :? Building), fields = [ @@ -195,7 +247,7 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = ] ) and CommunityType = - Define.Object ( + Define.Object( name = "Community", isTypeOf = (fun o -> o :? Community), fields = [ @@ -205,7 +257,7 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = ] ) and PropertyType = - Define.Union<_, _> ( + Define.Union<_, _>( name = "Property", options = [ ComplexType; BuildingType; CommunityType ], resolveValue = @@ -220,7 +272,7 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = | Community _ -> upcast CommunityType) ) let Query = - Define.Object ( + Define.Object( name = "Query", fields = [ Define.Field ("A", Nullable AType, "A Field", [ Define.Input ("id", IntType) ], resolve = (fun ctx _ -> getA (ctx.Arg ("id")))) @@ -242,8 +294,8 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = let schema = Schema (Query) let middleware = [ Define.QueryWeightMiddleware (2.0, true) - Define.ObjectListFilterMiddleware (true) - Define.ObjectListFilterMiddleware (true) + Define.ObjectListFilterMiddleware(true) + Define.ObjectListFilterMiddleware(true) ] Executor (schema, middleware) @@ -257,7 +309,8 @@ let executeWithVariables (query : Document, variables : ImmutableDictionary, filterToVerify : ObjectListFilter) = let ex = getExecutor (ValueSome filterToVerify) - ex.AsyncExecute (ast = query, getInputContext = getMockInputContext, variables = variables) |> sync + ex.AsyncExecute (ast = query, getInputContext = getMockInputContext, variables = variables) + |> sync let expectedThresholdErrors : GQLProblemDetails list = [ GQLProblemDetails.Create ("Query complexity exceeds maximum threshold. Please reduce query complexity and try again.") @@ -303,11 +356,14 @@ let ``Simple query: Must pass when below threshold`` () = ] let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 1.0) + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 1.0) [] let ``Simple query: Must not pass when above threshold`` () = @@ -363,9 +419,12 @@ let ``Simple query: Must not pass when above threshold`` () = }""" let result = execute query - ensureRequestError result <| fun errors -> errors |> equals expectedThresholdErrors - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 3.0) + ensureRequestError result + <| fun errors -> errors |> equals expectedThresholdErrors + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 3.0) [] let ``Deferred queries : Must pass when below threshold`` () = @@ -403,14 +462,17 @@ let ``Deferred queries : Must pass when below threshold`` () = ) let result = execute query - ensureDeferred result <| fun data errors deferred -> + ensureDeferred result + <| fun data errors deferred -> empty errors data |> equals (upcast expected) use sub = Observer.create deferred sub.WaitCompleted () sub.Received |> single |> equals expectedDeferred - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 2.0) [] let ``Streamed queries : Must pass when below threshold`` () = @@ -444,7 +506,8 @@ let ``Streamed queries : Must pass when below threshold`` () = DeferredResult ([| NameValueLookup.ofList [ "id", upcast 6; "value", upcast "3000" ] |], [ "A"; "subjects"; 1 ]) let result = execute query - ensureDeferred result <| fun data errors deferred -> + ensureDeferred result + <| fun data errors deferred -> empty errors data |> equals (upcast expected) use sub = Observer.create deferred @@ -454,8 +517,10 @@ let ``Streamed queries : Must pass when below threshold`` () = |> contains expectedDeferred1 |> contains expectedDeferred2 |> ignore - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 2.0) [] let ``Deferred and Streamed queries : Must not pass when above threshold`` () = @@ -512,9 +577,12 @@ let ``Deferred and Streamed queries : Must not pass when above threshold`` () = asts query |> Seq.map execute |> Seq.iter (fun result -> - ensureRequestError result <| fun errors -> errors |> equals expectedThresholdErrors - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 3.0)) + ensureRequestError result + <| fun errors -> errors |> equals expectedThresholdErrors + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 3.0)) [] let ``Inline fragment query : Must pass when below threshold`` () = @@ -552,11 +620,14 @@ let ``Inline fragment query : Must pass when below threshold`` () = ] let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 1.0) + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 1.0) [] let ``Inline fragment query : Must not pass when above threshold`` () = @@ -604,9 +675,38 @@ let ``Inline fragment query : Must not pass when above threshold`` () = }""" let result = execute query - ensureRequestError result <| fun errors -> errors |> equals expectedThresholdErrors - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 3.0) + ensureRequestError result + <| fun errors -> errors |> equals expectedThresholdErrors + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 3.0) + +[] +let ``Object list filter: inline coercion failure is request error`` () = + let query = + parse + """query testQuery { + A (id : 1) { + subjects (filter : 123) { ...Value } + } + } + + fragment Value on Subject { + ...on A { + id + value + } + ...on B { + id + value + } + }""" + + let result = execute query + + ensureRequestError result + <| fun errors -> Assert.Single errors |> ignore [] let ``Object list filter: must return filter information in Metadata`` () = @@ -646,15 +746,25 @@ let ``Object list filter: must return filter information in Metadata`` () = ] ] let expectedFilter : KeyValuePair = - kvp ([ "A"; "s" ]) (And (Equals ({ FieldName = "id"; Value = 2L }, null), StartsWith ({ FieldName = "value"; Value = "A" }, StringComparer.CurrentCultureIgnoreCase))) + kvp + ([ "A"; "s" ]) + (And ( + Equals ({ FieldName = "id"; Value = 2L }, null), + StartsWith ({ FieldName = "value"; Value = "A" }, StringComparer.CurrentCultureIgnoreCase) + )) let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("queryWeightThreshold") |> equals (ValueSome 2.0) - result.Metadata.TryFind ("queryWeight") |> equals (ValueSome 1.0) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("queryWeightThreshold") + |> equals (ValueSome 2.0) + result.Metadata.TryFind("queryWeight") + |> equals (ValueSome 1.0) + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must return AND filter information in Metadata`` () = @@ -694,13 +804,21 @@ let ``Object list filter: Must return AND filter information in Metadata`` () = ] ] let expectedFilter : KeyValuePair = - kvp ([ "A"; "subjects" ]) (And (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase), Equals ({ FieldName = "id"; Value = 6L }, null))) + kvp + ([ "A"; "subjects" ]) + (And ( + StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase), + Equals ({ FieldName = "id"; Value = 6L }, null) + )) let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must return OR filter information in Metadata`` () = @@ -740,13 +858,21 @@ let ``Object list filter: Must return OR filter information in Metadata`` () = ] ] let expectedFilter : KeyValuePair = - kvp ([ "A"; "subjects" ]) (Or (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase), Equals ({ FieldName = "id"; Value = 6L }, null))) + kvp + ([ "A"; "subjects" ]) + (Or ( + StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase), + Equals ({ FieldName = "id"; Value = 6L }, null) + )) let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must return IN filter information in Metadata`` () = @@ -789,10 +915,13 @@ let ``Object list filter: Must return IN filter information in Metadata`` () = kvp ([ "A"; "subjects" ]) (In { FieldName = "value"; Value = [ "3000"; "A2" ] }) let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must return Contains filter information in Metadata`` () = @@ -835,10 +964,13 @@ let ``Object list filter: Must return Contains filter information in Metadata`` kvp ([ "A"; "subjects" ]) (Contains ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase)) let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must return NOT filter information in Metadata`` () = @@ -881,10 +1013,13 @@ let ``Object list filter: Must return NOT filter information in Metadata`` () = kvp ([ "A"; "subjects" ]) (Not (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase))) let result = execute query - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must return filter information in Metadata when supplied as variable and parse all filter operators`` () = @@ -924,172 +1059,263 @@ let ``Object list filter: Must return filter information in Metadata when suppli ] ] do - let notStartsFilter = """{ "not": { "value_starts_with": "3" } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notStartsFilter) - let filter = Not (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase)) + let notStartsFilter = + """{ "not": { "value_starts_with": "3" } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notStartsFilter) + let filter = + Not (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase)) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notEndsFilter = """{ "not": { "value_ends_with": "2" } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notEndsFilter) - let filter = Not (EndsWith ({ FieldName = "value"; Value = "2" }, StringComparer.CurrentCultureIgnoreCase)) + let notEndsFilter = + """{ "not": { "value_ends_with": "2" } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notEndsFilter) + let filter = + Not (EndsWith ({ FieldName = "value"; Value = "2" }, StringComparer.CurrentCultureIgnoreCase)) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notStartsFilter = """{ "not": { "value_sw": "3" } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notStartsFilter) - let filter = Not (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase)) + let notStartsFilter = + """{ "not": { "value_sw": "3" } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notStartsFilter) + let filter = + Not (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase)) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notEndsFilter = """{ "not": { "value_ew": "2" } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notEndsFilter) - let filter = Not (EndsWith ({ FieldName = "value"; Value = "2" }, StringComparer.CurrentCultureIgnoreCase)) + let notEndsFilter = + """{ "not": { "value_ew": "2" } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notEndsFilter) + let filter = + Not (EndsWith ({ FieldName = "value"; Value = "2" }, StringComparer.CurrentCultureIgnoreCase)) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notGreaterThanOrEqualFilter = """{ "not": { "id_greater_than_or_equal": 2 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notGreaterThanOrEqualFilter) + let notGreaterThanOrEqualFilter = + """{ "not": { "id_greater_than_or_equal": 2 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = + ImmutableDictionary.Empty.Add("filter", notGreaterThanOrEqualFilter) let filter = Not (GreaterThanOrEqual { FieldName = "id"; Value = 2.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notLessThanOrEqualFilter = """{ "not": { "id_less_than_or_equal": 4 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notLessThanOrEqualFilter) + let notLessThanOrEqualFilter = + """{ "not": { "id_less_than_or_equal": 4 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notLessThanOrEqualFilter) let filter = Not (LessThanOrEqual { FieldName = "id"; Value = 4.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notGreaterThanFilter = """{ "not": { "id_greater_than": 2 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notGreaterThanFilter) + let notGreaterThanFilter = + """{ "not": { "id_greater_than": 2 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notGreaterThanFilter) let filter = Not (GreaterThan { FieldName = "id"; Value = 2.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notLessThanFilter = """{ "not": { "id_less_than": 4 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notLessThanFilter) + let notLessThanFilter = + """{ "not": { "id_less_than": 4 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notLessThanFilter) let filter = Not (LessThan { FieldName = "id"; Value = 4.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notGreaterThanOrEqualFilter = """{ "not": { "id_gte": 2 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notGreaterThanOrEqualFilter) + let notGreaterThanOrEqualFilter = + """{ "not": { "id_gte": 2 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = + ImmutableDictionary.Empty.Add("filter", notGreaterThanOrEqualFilter) let filter = Not (GreaterThanOrEqual { FieldName = "id"; Value = 2.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notLessThanOrEqualFilter = """{ "not": { "id_lte": 4 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notLessThanOrEqualFilter) + let notLessThanOrEqualFilter = + """{ "not": { "id_lte": 4 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notLessThanOrEqualFilter) let filter = Not (LessThanOrEqual { FieldName = "id"; Value = 4.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notGreaterThanFilter = """{ "not": { "id_gt": 2 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notGreaterThanFilter) + let notGreaterThanFilter = + """{ "not": { "id_gt": 2 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notGreaterThanFilter) let filter = Not (GreaterThan { FieldName = "id"; Value = 2.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notLessThanFilter = """{ "not": { "id_lt": 4 } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notLessThanFilter) + let notLessThanFilter = + """{ "not": { "id_lt": 4 } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notLessThanFilter) let filter = Not (LessThan { FieldName = "id"; Value = 4.0 }) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notContainsFilter = """{ "not": { "value_contains": "A" } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notContainsFilter) - let filter = Not (Contains ({ FieldName = "value"; Value = "A" }, StringComparer.CurrentCultureIgnoreCase)) + let notContainsFilter = + """{ "not": { "value_contains": "A" } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notContainsFilter) + let filter = + Not (Contains ({ FieldName = "value"; Value = "A" }, StringComparer.CurrentCultureIgnoreCase)) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] do - let notEqualsFilter = """{ "not": { "value": "A2" } }""" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", notEqualsFilter) + let notEqualsFilter = + """{ "not": { "value": "A2" } }""" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("filter", notEqualsFilter) let filter = Not (Equals ({ FieldName = "value"; Value = "A2" }, null)) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must parse filter that references variables`` () = @@ -1130,15 +1356,19 @@ let ``Object list filter: Must parse filter that references variables`` () = ] do let filterValue = "3" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", filterValue) - let filter = (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase)) + let variables = ImmutableDictionary.Empty.Add("filter", filterValue) + let filter = + (StartsWith ({ FieldName = "value"; Value = "3" }, StringComparer.CurrentCultureIgnoreCase)) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) (filter) let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must parse inline filter variable backed by Guid scalar`` () = @@ -1180,15 +1410,18 @@ let ``Object list filter: Must parse inline filter variable backed by Guid scala let guidText = "22222222-2222-2222-2222-222222222222" let filterValue = $"\"{guidText}\"" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("filter", filterValue) + let variables = ImmutableDictionary.Empty.Add("filter", filterValue) let filter = Equals ({ FieldName = "guidvalue"; Value = guidText }, null) let expectedFilter : KeyValuePair = kvp ([ "A"; "subjects" ]) filter let result = executeAndVerifyFilter (query, variables, filter) - ensureDirect result <| fun data errors -> + ensureDirect result + <| fun data errors -> empty errors data |> equals (upcast expected) - result.Metadata.TryFind ("filters") |> wantValueSome |> seqEquals [ expectedFilter ] + result.Metadata.TryFind("filters") + |> wantValueSome + |> seqEquals [ expectedFilter ] [] let ``Object list filter: Must parse inline filter variable backed by wrapped value object`` () = @@ -1214,12 +1447,14 @@ let ``Object list filter: Must parse inline filter variable backed by wrapped va }""" let valueObjectText = "22222222-2222-2222-2222-222222222222" - let valueObjectVariable = $"\"{valueObjectText}\"" |> JsonDocument.Parse |> _.RootElement - let variables = ImmutableDictionary.Empty.Add ("valueObject", valueObjectVariable) + let valueObjectVariable = + $"\"{valueObjectText}\"" + |> JsonDocument.Parse + |> _.RootElement + let variables = ImmutableDictionary.Empty.Add("valueObject", valueObjectVariable) let result = executeWithVariables (query, variables) - ensureDirect result <| fun _ errors -> - empty errors + ensureDirect result <| fun _ errors -> empty errors [] let ``Object list filter: Must return empty filter when all discriminated union types are specified`` () = diff --git a/tests/FSharp.Data.GraphQL.Tests/TaskSeqFieldTests.fs b/tests/FSharp.Data.GraphQL.Tests/TaskSeqFieldTests.fs index ef1115f29..fb2440e7f 100644 --- a/tests/FSharp.Data.GraphQL.Tests/TaskSeqFieldTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/TaskSeqFieldTests.fs @@ -31,6 +31,18 @@ let gatedNumbers (gate : Task) = }) :> IAsyncEnumerable +let signaledGatedNumbers (reachedGate : TaskCompletionSource) (gate : Task) = + SuspendingAsyncEnumerable(fun _ index -> task { + match index with + | 0 -> return ValueSome 1 + | 1 -> + reachedGate.TrySetResult () |> ignore + do! gate + return ValueSome 2 + | _ -> return ValueNone + }) + :> IAsyncEnumerable + let failingNumbers () = taskSeq { yield 1 yield 2 @@ -137,15 +149,30 @@ let ``TaskSeq field without directives returns the whole sequence as a list`` () data |> equals (upcast expectedData) [] -let ``TaskSeq field without directives waits for a sequence that suspends`` () = +let ``TaskSeq field without directives waits for a sequence that suspends`` () : Task = task { + let gate = TaskCompletionSource () + let reachedGate = TaskCompletionSource () let executor = - executorFor [ Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> gatedNumbers Task.CompletedTask) ] + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> signaledGatedNumbers reachedGate gate.Task) + ] let expectedData = NameValueLookup.ofList [ "numbers", upcast [| box 1; box 2 |] ] - let result = executeQuery executor "{ numbers }" + let execution = + executor.AsyncExecute (parse "{ numbers }", getMockInputContext, ()) + |> Async.StartImmediateAsTask + do! + waitForTask + (TimeSpan.FromSeconds (float (ms 5))) + "Timeout while waiting for the non-stream execution to reach the suspended second item" + reachedGate.Task + Assert.False (execution.IsCompleted, "The non-stream execution must wait for the sequence to produce its last item") + gate.SetResult () + let! result = execution ensureDirect result <| fun data errors -> empty errors data |> equals (upcast expectedData) +} [] let ``TaskSeq field with defer directive delivers the whole list in one deferred payload`` () = @@ -163,6 +190,22 @@ let ``TaskSeq field with defer directive delivers the whole list in one deferred |> single |> equals (DeferredResult ([| box 1; box 2; box 3 |], [ box "numbers" ])) +[] +let ``TaskSeq field with defer directive supports struct nullable lists`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", StructNullable (ListOf IntType), fun _ _ -> ValueSome (asyncItems [ 1; 2; 3 ])) + ] + let expectedData = NameValueLookup.ofList [ "numbers", null ] + let result = executeQuery executor "{ numbers @defer }" + ensureDeferred result + <| fun data errors deferred -> + empty errors + data |> equals (upcast expectedData) + waitForCompletion deferred + |> single + |> equals (DeferredResult ([| box 1; box 2; box 3 |], [ box "numbers" ])) + [] let ``TaskSeq field with stream directive delivers items before the sequence completes`` () = let gate = TaskCompletionSource () @@ -362,14 +405,16 @@ let ``Batching from source runs only for a stream query that does not override t callCount |> equals 0 executeQuery executor "{ deferrable @defer }" |> ignore callCount |> equals 0 - executeQuery executor "{ numbers @stream(preferredBatchSize: 1) }" |> ignore + executeQuery executor "{ numbers @stream(preferredBatchSize: 1) }" + |> ignore callCount |> equals 0 executeQuery executor "{ numbers @stream }" |> ignore callCount |> equals 1 [] let ``Throwing batching callback does not affect a query that does not stream the field`` () = - let throwingBatching = StreamBatching.FromSource (fun _ -> failwith "Batching must not run for this query") + let throwingBatching = + StreamBatching.FromSource (fun _ -> failwith "Batching must not run for this query") let executor = executorFor [ Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1; 2; 3 ]), batching = throwingBatching) @@ -431,7 +476,11 @@ let ``Streamed TaskSeq field that fails acquiring the enumerator still delivers // this field's DeferredErrors, which would drop sibling deferred results and the final completion payload let executor = executorFor [ - Define.TaskSeqField ("failing", ListOf IntType, fun _ _ -> ThrowingAsyncEnumerable "Boom acquiring the enumerator" :> IAsyncEnumerable) + Define.TaskSeqField ( + "failing", + ListOf IntType, + fun _ _ -> ThrowingAsyncEnumerable "Boom acquiring the enumerator" :> IAsyncEnumerable + ) Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> asyncItems [ 10; 20 ]) ] let expectedData = @@ -470,7 +519,10 @@ let ``Streamed TaskSeq field delivers an item's own resolver error and keeps str // streaming operator is concerned, so it must not be mistaken for a failure of the source or of the enumeration // itself: the item's error is delivered on its own path and later items keep streaming, exactly like @stream on // an ordinary list (see DeferredTests."Resolver list error") - let items = [ { Id = 1; Value = async { return failwith "Boom resolving the item" } }; { Id = 2; Value = async { return "two" } } ] + let items = [ + { Id = 1; Value = async { return failwith "Boom resolving the item" } } + { Id = 2; Value = async { return "two" } } + ] let executor = executorFor [ Define.TaskSeqField ("items", ListOf StreamItemType, (fun _ _ -> asyncItems items), maxConcurrency = 1) @@ -498,7 +550,10 @@ let ``A batch containing a failed item alongside a succeeding one is delivered a // slot in `data`, so GraphQLWebsocketMiddleware.splitBatch's List.map2 would throw on a mixed success/error // batch. It does not: both arms of `merge` prepend the item's index, so `indices` and `data` always end up the // same length as the chunk, with the failed item's slot left null. This pins that shape end to end. - let items = [ { Id = 1; Value = async { return failwith "Boom resolving item 0" } }; { Id = 2; Value = async { return "two" } } ] + let items = [ + { Id = 1; Value = async { return failwith "Boom resolving item 0" } } + { Id = 2; Value = async { return "two" } } + ] let executor = executorFor [ Define.TaskSeqField ("items", ListOf StreamItemType, (fun _ _ -> asyncItems items), batching = StreamBatching.Fixed 2, maxConcurrency = 1) @@ -529,7 +584,9 @@ let ``Disposing the stream subscription stops the enumeration of the TaskSeq fie match result.Content with | Deferred (_, errors, deferred) -> empty errors - let subscription = deferred |> Observable.subscribe (fun _ -> received.TrySetResult () |> ignore) + let subscription = + deferred + |> Observable.subscribe (fun _ -> received.TrySetResult () |> ignore) do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Timeout while waiting for the first streamed item" received.Task subscription.Dispose () do! @@ -551,14 +608,16 @@ let ``TaskSeq field with stream directive never resolves more than maxConcurrenc let trackConcurrency (work : Async<'T>) : Async<'T> = async { let current = Interlocked.Increment inFlight let mutable observed = maxObserved.Value - while current > observed && Interlocked.CompareExchange (maxObserved, current, observed) <> observed do + while current > observed + && Interlocked.CompareExchange (maxObserved, current, observed) + <> observed do observed <- maxObserved.Value try return! work finally Interlocked.Decrement inFlight |> ignore } - let items = [ for id in 1 .. 6 -> { Id = id; Value = trackConcurrency (delay 100 (string id)) } ] + let items = [ for id in 1..6 -> { Id = id; Value = trackConcurrency (delay 100 (string id)) } ] let executor = executorFor [ Define.TaskSeqField ("items", ListOf StreamItemType, (fun _ _ -> asyncItems items), maxConcurrency = 2) @@ -573,8 +632,9 @@ let ``TaskSeq field with stream directive never resolves more than maxConcurrenc [] let ``TaskSeqField with a non-positive maxConcurrency fails at definition time`` () = - throws (fun () -> - Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1 ]), maxConcurrency = 0) |> ignore) + throws(fun () -> + Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1 ]), maxConcurrency = 0) + |> ignore) [] let ``TaskSeq field resolved as null reports a non-null field error`` () = From b57bf2d5d86358dba6e19895f24af163b626db95 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 18 Sep 2026 22:09:20 +0000 Subject: [PATCH 2/3] Fix review thread regressions Co-authored-by: xperiandri <2365592+xperiandri@users.noreply.github.com> --- .../MiddlewareDefinitions.fs | 3 +- .../ObservableExtensionsTests.fs | 15 ++++++---- .../MiddlewareTests.fs | 30 +++++++++++++------ 3 files changed, 33 insertions(+), 15 deletions(-) diff --git a/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs b/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs index 458844d10..7b354beff 100644 --- a/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs +++ b/src/FSharp.Data.GraphQL.Server.Middleware/MiddlewareDefinitions.fs @@ -138,9 +138,10 @@ type internal ObjectListFilterMiddleware<'ObjectType, 'ListType> (reportToMetada | 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) } | false -> return ctx diff --git a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs index 7a31d503f..e3fc7462c 100644 --- a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs @@ -557,22 +557,23 @@ let ``ofAsyncEnumerableResolved should cancel a pending MoveNextAsync after a re // Regression test: with maxConcurrency > 1 the loop can already be suspended in MoveNextAsync for the next item // when an earlier background resolution fails. That failure must cancel the in-progress move so the stream can // finish with onFailure instead of hanging forever in the source. + let secondMoveEntered = TaskCompletionSource () let source = SuspendingAsyncEnumerable(fun cancellationToken index -> task { match index with | 0 -> return ValueSome 1 | 1 -> + secondMoveEntered.TrySetResult () |> ignore do! Task.Delay (Timeout.Infinite, cancellationToken) return ValueSome 2 | _ -> return ValueNone }) let resolve _ (n : int) = if n = 1 then - async { - do! Async.Sleep (ms 50) + asyncVal { + do! secondMoveEntered.Task return failwith "Boom resolving" } - |> AsyncVal.ofAsync else AsyncVal.wrap n use sub = @@ -586,12 +587,14 @@ let ``ofAsyncEnumerableResolved should preserve a resolution failure over Dispos // Regression test: a resolution failure can cancel an in-progress MoveNextAsync, whose cancellation is suppressed // as expected; if DisposeAsync then throws, the original resolution failure must still win over the later // disposal failure because it is what stopped the stream. + let secondMoveEntered = TaskCompletionSource () let source = SuspendingAsyncEnumerable( (fun cancellationToken index -> task { match index with | 0 -> return ValueSome 1 | 1 -> + secondMoveEntered.TrySetResult () |> ignore do! Task.Delay (Timeout.Infinite, cancellationToken) return ValueSome 2 | _ -> return ValueNone @@ -601,7 +604,7 @@ let ``ofAsyncEnumerableResolved should preserve a resolution failure over Dispos let resolve _ (n : int) = if n = 1 then async { - do! Async.Sleep (ms 50) + do! secondMoveEntered.Task |> Async.AwaitTask return failwith "Boom resolving" } |> AsyncVal.ofAsync @@ -626,12 +629,14 @@ let ``ofAsyncEnumerableResolved should preserve a resolution failure over a late // Regression test: once a resolution failure has already stopped the stream, a source that reacts to the linked // cancellation by throwing a different exception from MoveNextAsync must not replace that original failure. let moveFailed = TaskCompletionSource () + let secondMoveEntered = TaskCompletionSource () let source = SuspendingAsyncEnumerable(fun cancellationToken index -> task { match index with | 0 -> return ValueSome 1 | 1 -> use _ = cancellationToken.Register (fun () -> moveFailed.TrySetResult () |> ignore) + secondMoveEntered.TrySetResult () |> ignore do! moveFailed.Task return failwith "Boom during enumeration" | _ -> return ValueNone @@ -639,7 +644,7 @@ let ``ofAsyncEnumerableResolved should preserve a resolution failure over a late let resolve _ (n : int) = if n = 1 then async { - do! Async.Sleep (ms 50) + do! secondMoveEntered.Task |> Async.AwaitTask return failwith "Boom resolving" } |> AsyncVal.ofAsync diff --git a/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs b/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs index cc39831fd..4dbacb2fd 100644 --- a/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/MiddlewareTests.fs @@ -87,7 +87,18 @@ type Property = | Building of Building | Community of Community -let getExecutor (expectedFilter : ObjectListFilter voption) = +let private getFilterMiddlewares (reportToMetadata : bool voption) = + match reportToMetadata with + | ValueSome reportToMetadata -> [ + Define.ObjectListFilterMiddleware(reportToMetadata) + Define.ObjectListFilterMiddleware(reportToMetadata) + ] + | ValueNone -> [ + Define.ObjectListFilterMiddleware() + Define.ObjectListFilterMiddleware() + ] + +let getExecutor (expectedFilter : ObjectListFilter voption) (reportToMetadata : bool voption) = let a1 : A = { Id = 1 Value = "A1" @@ -292,23 +303,24 @@ let getExecutor (expectedFilter : ObjectListFilter voption) = ] ) let schema = Schema (Query) - let middleware = [ - Define.QueryWeightMiddleware (2.0, true) - Define.ObjectListFilterMiddleware(true) - Define.ObjectListFilterMiddleware(true) - ] + let middleware = [ Define.QueryWeightMiddleware (2.0, true); yield! getFilterMiddlewares reportToMetadata ] Executor (schema, middleware) -let executor = getExecutor (ValueNone) +let executor = getExecutor ValueNone (ValueSome true) +let executorWithDefaultFilterMetadataSetting = getExecutor ValueNone ValueNone let execute (query : Document) = executor.AsyncExecute (query, getMockInputContext) |> sync +let executeWithDefaultFilterMetadataSetting (query : Document) = + executorWithDefaultFilterMetadataSetting.AsyncExecute (query, getMockInputContext) + |> sync + let executeWithVariables (query : Document, variables : ImmutableDictionary) = executor.AsyncExecute (ast = query, getInputContext = getMockInputContext, variables = variables) |> sync let executeAndVerifyFilter (query : Document, variables : ImmutableDictionary, filterToVerify : ObjectListFilter) = - let ex = getExecutor (ValueSome filterToVerify) + let ex = getExecutor (ValueSome filterToVerify) (ValueSome true) ex.AsyncExecute (ast = query, getInputContext = getMockInputContext, variables = variables) |> sync @@ -703,7 +715,7 @@ let ``Object list filter: inline coercion failure is request error`` () = } }""" - let result = execute query + let result = executeWithDefaultFilterMetadataSetting query ensureRequestError result <| fun errors -> Assert.Single errors |> ignore From 9104cb4a1173dfbfbe7dbb0f9c2cd37cc74a368d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 18 Sep 2026 23:19:17 +0000 Subject: [PATCH 3/3] Make observable regression test deterministic Co-authored-by: xperiandri <2365592+xperiandri@users.noreply.github.com> --- .../ObservableExtensionsTests.fs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs index e3fc7462c..eb4ba42ef 100644 --- a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs @@ -528,22 +528,25 @@ let ``ofAsyncEnumerableResolved should not resolve an item pulled after a resolu // Regression test: a background resolution can fail while MoveNextAsync for the next item is still suspended; // when that move completed the code used to go straight to resolving it without rechecking the failure, so a // synchronously resolved item 2 was pulled and emitted before the failure that already happened + let secondMoveEntered = TaskCompletionSource () + let secondMoveCanceled = TaskCompletionSource () let source = - SuspendingAsyncEnumerable(fun _ index -> task { + SuspendingAsyncEnumerable(fun cancellationToken index -> task { match index with | 0 -> return ValueSome 1 | 1 -> - do! Task.Delay (ms 150) + use _ = cancellationToken.Register (fun () -> secondMoveCanceled.TrySetResult () |> ignore) + secondMoveEntered.TrySetResult () |> ignore + do! secondMoveCanceled.Task return ValueSome 2 | _ -> return ValueNone }) let resolve _ (n : int) = if n = 1 then - async { - do! Async.Sleep (ms 50) + asyncVal { + do! secondMoveEntered.Task return failwith "Boom resolving" } - |> AsyncVal.ofAsync else AsyncVal.wrap n use sub =