Hardened WebSocket streaming and incremental delivery (#608)

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

Author: 3 people

RELEASE_NOTES.md

@@ -288,7 +288,7 @@
 
 * **Breaking Change** Migrated to .NET 10
 * **Breaking Change** Made Relay `Edge` a read-only struct
-* **Breaking Change** `SubscriptionExecutionResult.Data` is now `obj Skippable`, and the record has new `Path` and `HasNext` fields for incremental delivery
+* **Breaking Change** `SubscriptionExecutionResult.Data` is now `obj voption Skippable`, and the record has new `Path` and `HasNext` fields for incremental delivery
 * **Breaking Change** `BufferedStreamOptions.Interval` and `BufferedStreamOptions.PreferredBatchSize` are now `int voption`
 * **Breaking Change** `ServerMessage.Error` and `ServerRawPayload.ErrorMessages` now carry `GQLProblemDetails list` instead of `NameValueLookup list`, so an `error` message's `payload` is a standard GraphQL error array as the `graphql-transport-ws` protocol requires
 * **Breaking Change** A query or mutation whose non-null root field fails during execution now produces a `Direct` (execution) result with `null` data instead of a `RequestError`, which is now only ever produced for a request rejected before execution (validation, planning, variable or inline argument coercion, a middleware, or the executor itself failing); HTTP and `graphql-transport-ws` responses for such a failure now carry `data: null` as the spec requires, instead of omitting `data` entirely. This also changes the public `GQLResponse.Data`, `GQLResponseContent.Direct.Data`, `DeferredErrors.Data`, and `SubscriptionErrors.Data` signatures to use `voption`

src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLSubscriptionsManagement.fs

@@ -1,14 +1,16 @@
 module internal FSharp.Data.GraphQL.Server.AspNetCore.GraphQLSubscriptionsManagement
 
+open System
+
 open FSharp.Data.GraphQL.Shared.WebSockets
 
 let addSubscription
     (id : SubscriptionId, unsubscriber : SubscriptionUnsubscriber, onUnsubscribe : OnUnsubscribeAction)
     (subscriptions : SubscriptionsDict)
     =
-    subscriptions.Add (id, (unsubscriber, onUnsubscribe))
+    lock subscriptions (fun () -> subscriptions.Add (id, (unsubscriber, onUnsubscribe)))
 
-let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = subscriptions.ContainsKey (id)
+let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = lock subscriptions (fun () -> subscriptions.ContainsKey (id))
 
 let executeOnUnsubscribeAndDispose (id : SubscriptionId) (subscription : SubscriptionUnsubscriber * OnUnsubscribeAction) =
     match subscription with
@@ -19,15 +21,37 @@ let executeOnUnsubscribeAndDispose (id : SubscriptionId) (subscription : Subscri
             unsubscriber.Dispose ()
 
 let removeSubscription (id : SubscriptionId) (subscriptions : SubscriptionsDict) =
-    match subscriptions.TryGetValue id with
-    | true, sub ->
-        sub |> executeOnUnsubscribeAndDispose id
-        subscriptions.Remove (id) |> ignore
-    | false, _ -> ()
+    let subscription =
+        lock subscriptions (fun () ->
+            match subscriptions.TryGetValue id with
+            | true, sub ->
+                subscriptions.Remove (id) |> ignore
+                ValueSome sub
+            | false, _ -> ValueNone)
+
+    match subscription with
+    | ValueSome sub -> sub |> executeOnUnsubscribeAndDispose id
+    | ValueNone -> ()
 
 let removeAllSubscriptions (subscriptions : SubscriptionsDict) =
-    subscriptions
-    |> Seq.iter (fun subscription ->
-        subscription.Value
-        |> executeOnUnsubscribeAndDispose subscription.Key)
-    subscriptions.Clear ()
+    let subscriptionsToDispose =
+        lock subscriptions (fun () ->
+            let snapshot =
+                subscriptions
+                |> Seq.map (fun subscription -> struct (subscription.Key, subscription.Value))
+                |> Seq.toArray
+
+            subscriptions.Clear ()
+            snapshot)
+
+    let exceptions = ResizeArray ()
+
+    subscriptionsToDispose
+    |> Array.iter (fun struct (id, subscription) ->
+        try
+            subscription |> executeOnUnsubscribeAndDispose id
+        with ex ->
+            exceptions.Add ex)
+
+    if exceptions.Count > 0 then
+        raise (AggregateException ("One or more subscriptions failed to unsubscribe.", exceptions))