Added Define.TaskSeqField with streaming of IAsyncEnumerable results

[xperiandri]

Summary

Adds native support for list fields resolved from IAsyncEnumerable<'T>, such as taskSeq { }, C# async iterators or Azure SDK AsyncPageable<T>, and makes @defer/@stream results usable over graphql-transport-ws.

Define.TaskSeqField

Define.TaskSeqField ("orders", ListOf OrderType, (fun _ customer -> getOrders customer.Id), batching = StreamBatching.Fixed 50, maxConcurrency = 8)

Query	Behavior
no directive	The sequence is enumerated into a regular list
@defer	The whole list is delivered in one deferred payload (requires the Nullable (ListOf …) overloads, which are included)
@stream	Every item is delivered as soon as the sequence produces it and its fields are resolved, at most maxConcurrency items resolving at the same time

• The sequence is pulled lazily, and the enumeration is cancelled when the subscriber disposes.
• An exception thrown acquiring the enumerator, while enumerating, or disposing it:
  • without directives behaves like a lazy seq today: nullable field → null + field error, non-nullable → error propagates;
  • with @stream it is delivered as DeferredErrors for the field after every item already pulled has been resolved and delivered, so a slower item can never be overtaken by the error that follows it, and buffered items and sibling deferred streams are not lost. An item resolution that throws stops the enumeration and is delivered the same way, once every resolution already started has settled, and no item pulled after such a failure is resolved; its concurrency slot is always released, whether the resolution itself failed or delivering its result did.
• An item whose own fields fail (an ordinary GraphQL resolver error, as opposed to an exception escaping the resolution machinery) is delivered as that item's own DeferredErrors, and streaming continues with the items after it — exactly as @stream already behaves on an ordinary list.
• Existing @stream on ordinary lists keeps its behavior.
• maxConcurrency (optional, Environment.ProcessorCount by default) bounds how many items are pulled from the source and resolved at the same time for @stream; pulling the next item waits for a free slot, and only items still in flight are tracked, so a long or infinite source does not retain what it already delivered. It applies to @stream only.

Batching of streamed items

StreamBatching<'Item> is set on the field:

• StreamBatching.Fixed size
• StreamBatching.FromSource (IAsyncEnumerable<'Item> -> int voption) computes the size from the resolved sequence instance.

The preferredBatchSize argument of @stream takes precedence over the field-level policy. The batching callback itself is now evaluated lazily, only when a query actually streams the field without supplying its own preferredBatchSize — it used to run eagerly for every query shape (including @defer and no directive), so a throwing FromSource callback could break a non-streamed query.

Note: Azure AsyncPageable<T> does not expose a page size (it is only a hint passed to AsPages), so FromSource can batch by pages only when the application keeps the hint, for example in a subclass of AsyncPageable<T>. Tests cover both a plain AsyncPageable<T> and one that keeps the hint.

A batch is grouped by the engine into one internal event, but over graphql-transport-ws it is still sent as one independently addressed next payload per item, in the batch's own order — a payload addressed by a list of indices (path: ["numbers", [0, 1]]) isn't mergeable by any client. The splitting logic (IncrementalPayloadSplitting, internal) is unit tested directly.

WebSocket (graphql-transport-ws) delivery

• Deferred and streamed payloads are sent immediately instead of after a hard-coded 5 second delay.
• Incremental payloads carry path and hasNext; the initial payload has hasNext: true, and a final {"hasNext": false} payload precedes complete.
• Payloads are no longer cast to Dictionary<string, obj>, which failed for streamed items and scalars.
• Errors of the initial deferred payload are no longer dropped together with all deferred results.
• A subscription that produces a SubscriptionErrors result no longer discards its partial data; it is forwarded alongside the errors instead of always sending null.
• A Direct (non-subscription) result no longer discards its field errors; they are now logged and forwarded like the other branches.
• Fixed a race where a subscription whose deferred result completed synchronously — before its id was registered — stranded that id forever, leaking the subscription entry. The unsubscriber is now registered as a placeholder (SingleAssignmentDisposable) before Subscribe runs, so a synchronous completion can still find and clean up the id.
• Fixed the same id ever being leaked the other way: if Subscribe itself threw synchronously, the placeholder used to stay registered forever too. It is now removed (or left alone if a synchronous completion already removed it) before the failure is rethrown for the existing per-message error handling to report.
• Fixed a batch of streamed items being addressed by a path ending in the list of the batch's own indices, which no client could merge; each item of a batch is now sent as its own independently addressed payload instead (see above).
• Fixed complete never being sent after the single next of a Direct (query/mutation) result.
• Fixed a request rejected before execution (validation, planning, variable or inline argument coercion, a middleware, or the executor itself failing) being sent as a next result followed by complete, instead of the terminal error message the protocol requires for it; error's payload is now a standard GraphQL error array (GQLProblemDetails list, see Breaking changes) rather than an arbitrary object. A query or mutation whose non-null root field fails during execution is not a request error — it is still sent as next (with data: null) + complete, like any other result.
• Fixed error messages and pong messages carrying a payload failing to serialize at all (Utf8JsonWriter throwing), because neither was written under the payload property name.

Example with the Star Wars sample (Human.friendsStream was added to demonstrate @stream):

{"type":"next","id":"1","payload":{"data":{"hero":{"name":"Luke Skywalker","friendsStream":[]}},"errors":[],"hasNext":true}}
{"type":"next","id":"1","payload":{"data":[{"name":"Han Solo"}],"errors":[],"path":["hero","friendsStream",0],"hasNext":true}}
…
{"type":"next","id":"1","payload":{"data":[{"name":"R2-D2"}],"errors":[],"path":["hero","friendsStream",3],"hasNext":true}}
{"type":"next","id":"1","payload":{"errors":[],"hasNext":false}}
{"type":"complete","id":"1"}

Follow-up (separate branch): the wire format above (data/path/hasNext) predates this PR and is not the format any current client speaks. A newer incremental-delivery format (pending/incremental/completed, the one graphql-js 17 and Apollo Client's GraphQL17Alpha9Handler use, where a client appends streamed items itself and no path index is needed at all) is worth adopting, but requires a per-field completion signal threaded through the engine's whole merged deferred/streamed/live observable — which touches roughly 25 pre-existing tests in DeferredTests.fs that predate this PR and aren't specific to TaskSeqField. Given that size and blast radius, it's being done separately rather than folded into this PR.

Breaking changes

• SubscriptionExecutionResult.Data is now obj Skippable, and the record has new Path and HasNext fields. Use the Create, CreateErrors, CreateInitial, CreateIncremental and CreateCompleted factory members.
• BufferedStreamOptions.Interval and BufferedStreamOptions.PreferredBatchSize are now int voption. Construct the options with ValueSome/ValueNone, for example SchemaConfig.DefaultWithBufferedStream { Interval = ValueSome 2000; PreferredBatchSize = ValueNone }.
• ServerMessage.Error and ServerRawPayload.ErrorMessages now carry GQLProblemDetails list instead of NameValueLookup list.
• A query or mutation whose non-null root field fails during execution now produces a GQLResponseContent.Direct (execution) result with null data instead of RequestError, which is now only ever produced for a request rejected before execution (validation, planning, variable or inline argument coercion, a middleware, or the executor itself failing). HTTP and graphql-transport-ws responses for such a failure now carry data: null as the spec requires, instead of omitting data entirely. Inline argument/input object coercion failures on a root field remain RequestError and are now checked for every root field before any of them execute, the same as variable coercion — a mutation no longer runs an earlier root field's resolver before rejecting the request over a later field's invalid literal argument.

Dependencies

• FSharp.Data.GraphQL.Shared references Microsoft.Bcl.AsyncInterfaces for netstandard2.0 only.
• Tests reference FSharp.Control.TaskSeq, Azure.Core and IcedTasks explicitly.

SDK

dev now pins SDK 10.0.303 instead of 10.0.401. The F# compiler in SDK 10.0.4xx (verified on 10.0.400 and 10.0.401) compiles resumable state machines incorrectly in Debug, so a taskSeq { } that awaits returns no further items: dotnet/fsharp#20466, fixed in main by dotnet/fsharp#20469 but not shipped yet (also fsprojects/FSharp.Control.TaskSeq#473). 10.0.303 is the latest SDK without the regression. The --always-inline+ workaround and Optimize=true alone did not help on 10.0.401.

The code in this PR still builds with SDK 10.0.4xx. Passing a voption to a struct optional parameter through ?name = value is supported only by the 10.0.4xx compiler, so that call is written as a match.

Known limitations

• Resolvers are captured as quotations, so a taskSeq { } block using let! or yield! must live in a separate function called from the resolver. This is documented.
• WithResolveMiddleware is not supported for TaskSeqField and throws NotSupportedException.
• Consumers building with SDK 10.0.4xx in Debug hit the compiler regression above in their own taskSeq { } blocks. Tests that need a real await use a hand-written suspending IAsyncEnumerable, so they pass on any SDK.
• The HTTP handler still returns only the initial payload of a deferred result; incremental delivery over HTTP is out of scope.
• Integration test introspection snapshots were not regenerated for the new sample field.
• Neither addClientSubscription fix (synchronous-completion race, synchronous Subscribe failure) is covered by an automated test: it is a private middleware function and there is no in-process graphql-transport-ws test harness in this repo yet. The disposal semantics they rely on (SingleAssignmentDisposable disposing an already-assigned/already-disposed instance) were verified in isolation instead.
• If an observer's OnNext itself throws while a streamed item is delivered (for example a WebSocket send failing mid-stream), System.Reactive tears the subscription down as soon as that happens — this is standard Subscribe(IObserver<T>) behavior, not specific to this operator. The concurrency slot is still always released (no deadlock), but no onFailure/OnCompleted follows, since the subscription is already gone by then.
• The wire format is still the pre-existing data/path/hasNext shape, not a format any current client library implements as-is; see the Follow-up note above.

Testing

• New TaskSeqFieldTests: draining, @defer, @stream ordering and early delivery, query/fixed/source batching and precedence, enumeration errors (nullable, non-nullable, streamed, including failures acquiring the enumerator), an item's own resolver error not ending the stream, a batch mixing a failed item with a succeeding one, cancellation, null sequence, Azure AsyncPageable<T>, maxConcurrency bounding and validation, lazy evaluation of FromSource.
• ObservableExtensionsTests for ofAsyncEnumerable, ofAsyncEnumerableResolved (ordering, bounded concurrency, disposal, failures raised acquiring/enumerating/disposing the source, a failed resolution not overtaken by another item pulled right after it — whether that item was waiting on a concurrency slot or still being produced by the source — and an observer throwing while a result is delivered) and withCompletionMarker; IncrementalPayloadSplittingTests for the batch-splitting logic (out-of-order batches, item-level error attribution, a failed item's null data slot); SerializationTests for the new WebSocket payload shapes, including error messages and a payload-carrying pong.
• Updated ExecutionTests, LazyEnumerationExceptionTests and TaskSeqFieldTests (6 tests total) to assert Direct with null data for a failed non-null root field, instead of RequestError. A follow-up commit added a regression test (ExecutionTests) pinning that an inline argument/input object coercion failure on one root field stays RequestError and pre-empts every root field's resolver, since that case was initially — and incorrectly — folded into the same Direct (null, ...) change; InputObjectValidatorTests's existing inline-validation test is the same scenario at the single-field level. The other pre-existing ensureRequestError assertions in the suite are genuine pre-execution failures and are unchanged.
• A tenth review raised a possible crash in splitBatch (GraphQLWebsocketMiddleware.fs) on a batch mixing a successful and a failed item; tracing Execution.collectItems showed the claimed length mismatch between indices and data cannot occur (both are always exactly chunk.Length, a failed item just leaves a null slot), and two new regression tests (TaskSeqFieldTests, IncrementalPayloadSplittingTests, both listed above) confirm this end to end and at the splitBatch level.
• CI (GitHub Actions, SDK 10.0.303, ubuntu/windows/macOS) is green on ea3709d7: full solution build (dotnet build FSharp.Data.GraphQL.slnx) and the full unit test suite (dotnet test, 660 tests) pass on all three runners, verified locally with the same commands before pushing.
• The concurrency-related fixes across the review rounds (in-flight tracking, slot release on resolution/delivery failure, rechecking stop conditions after each await in the pull loop), the batch-splitting logic, the Utf8JsonWriter payload-property fix, and the IcedTasks-based valueTask CE were all verified with isolated dotnet fsi repros against the real packages before being wired in or committed — the concurrency ones running the exact race hundreds of times and confirming both the failure before and the fix after, since these are timing-sensitive and easy to get subtly wrong.
• Manually verified WebSocket streaming against the Star Wars sample (before the second review-fix round).

🤖 Generated with Claude Code

[github-actions]

Test Results

    9 files      9 suites   12m 54s ⏱️
  765 tests   760 ✅  5 💤 0 ❌
2 295 runs  2 280 ✅ 15 💤 0 ❌

Results for commit dc28bff.

♻️ This comment has been updated with latest results.

[Copilot]

🟡 Changes recommended

Stream failures can overtake prior items, batching callbacks run for non-streamed queries, and item resolution concurrency is unbounded.

Get a fresh assessment by requesting another Copilot review.

Pull request overview

Adds native IAsyncEnumerable field resolution, streaming/batching support, and incremental WebSocket delivery.

Changes:

• Introduces Define.TaskSeqField and configurable stream batching.
• Adds async-enumerable execution, cancellation, and error handling.
• Updates WebSocket payloads, documentation, samples, and tests.

File summaries

File	Description
TaskSeqFieldTests.fs	Tests async sequence fields and streaming.
Helpers.fs	Adds a suspending async-enumerable test helper.
ObservableExtensionsTests.fs	Tests async-enumerable observables.
FSharp.Data.GraphQL.Tests.fsproj	Adds test dependencies and source file.
SerializationTests.fs	Tests incremental payload serialization.
WebSockets.fs	Expands WebSocket execution payloads.
TypeSystem.fs	Adds async-sequence resolver and batching types.
SchemaDefinitionsExtensions.fs	Rejects unsupported resolver middleware.
SchemaDefinitions.fs	Adds TaskSeqField overloads.
FSharp.Data.GraphQL.Shared.fsproj	Adds async-interface compatibility dependency.
ObservableExtensions.fs	Adds async-enumerable observable adapters.
Execution.fs	Executes and streams async sequence items.
ErrorMessages.fs	Updates enumerable type error text.
GraphQLWebsocketMiddleware.fs	Sends incremental payloads immediately.
star-wars-api.fsproj	Adds TaskSeq dependency.
Schema.fs	Demonstrates streamed friends.
RELEASE_NOTES.md	Documents features and breaking changes.
Packages.props	Centrally versions new dependencies.
docs/type-system.md	Documents asynchronous sequence fields.

Review details

• Files reviewed: 19/19 changed files
• Comments generated: 3
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

🟡 Changes recommended

Async-enumerator lifecycle failures and synchronous WebSocket completion can terminate or strand incremental operations, while some WebSocket errors are still discarded.

Get a fresh assessment by requesting another Copilot review.

Review details

Suppressed comments (2)

src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs:205

• Direct executions may contain partial data together with field errors, but the errors are ignored here even though Create accepts them. Forward the returned error list so WebSocket clients receive the same execution result as other transports.

            | Direct (data, _) -> do! SubscriptionExecutionResult.Create (data, []) |> sendOutput id

src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs:129

• DisposeAsync can fail, but this await is outside the captured-failure path. Such a failure faults the observable and terminates the entire deferred result stream rather than producing the field-level StreamFailure; this is inconsistent with the non-streamed path and can suppress sibling incremental results. Capture disposal failures (while preserving an earlier enumeration failure) and route them through onFailure.

            // Captured items no longer need the enumerator, so it is disposed before waiting for their resolutions
            do! enumerator.DisposeAsync ()

• Files reviewed: 23/23 changed files
• Comments generated: 3
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Pending resolution tasks can leak memory or deadlock, and synchronous subscription failures can strand operation IDs.

Get a fresh assessment by requesting another Copilot review.

Review details

• Files reviewed: 23/23 changed files
• Comments generated: 3
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

It changes public APIs, concurrent execution, and WebSocket protocol behavior without a successful full regression run.

Review details

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

RELEASE_NOTES.md:305

• The previous Direct branch already forwarded data; it only discarded the accompanying field errors. This release note incorrectly says that partial data was discarded for both result kinds. Describe the subscription partial-data fix and the direct-result error fix separately.

• Files reviewed: 23/23 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

WebSocket disconnects do not dispose registered deferred subscriptions, allowing infinite asynchronous sequences to continue running.

Get a fresh assessment by requesting another Copilot review.

Review details

• Files reviewed: 29/29 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Concurrent subscription-map cleanup can throw and leave active streams undisposed.

Get a fresh assessment by requesting another Copilot review.

Review details

• Files reviewed: 29/29 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Observable errors can leak operation IDs, and the public API omits the established struct-nullable field variant.

Review details

Suppressed comments (2)

Previously missed (2) — in code that hasn't changed since the last review.

src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs:198

• A source that terminates through OnError never invokes OnCompleted, and the current error callback only logs. The newly registered placeholder therefore remains in subscriptions, so the operation ID cannot be reused until the whole socket disconnects. Remove the subscription from the error callback (in a finally so logging cannot prevent cleanup).
  src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs:1242
• These nullable overloads only accept option, so TaskSeqField cannot be used with the repository's established StructNullable (ListOf ...) wrapper even though ordinary fields support it and voption is the preferred nullable representation. Add corresponding IAsyncEnumerable<'Item> voption overloads and teach the TaskSeq quotation boxifier to unwrap ValueSome/ValueNone.

• Files reviewed: 30/30 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[xperiandri]

@copilot fix that

[Copilot]

@copilot fix that

Fixed in 3af2b73.

[Copilot]

🟡 Changes recommended

A resolver failure can leave streaming permanently blocked in an outstanding MoveNextAsync.

Get a fresh assessment by requesting another Copilot review.

Review details

• Files reviewed: 30/30 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Public response types declare data as non-nullable while the new execution paths intentionally produce null values.

Get a fresh assessment by requesting another Copilot review.

Review details

• Files reviewed: 30/30 changed files
• Comments generated: 2
• Review effort level: Balanced