fix(runtime): close invokable lifecycle gaps

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: ReubenBond

src/Orleans.Core/Networking/Connection.cs

@@ -402,22 +402,11 @@ private async Task ProcessOutgoing()
                     }
 
                     var flushResult = await output.FlushAsync();
+                    DisposeOneWayRequestBodies(inflight);
                     if (flushResult.IsCompleted || flushResult.IsCanceled)
                     {
                         break;
                     }
-
-                    // Dispose one-way request bodies after they've been sent.
-                    // Request-response bodies are disposed when the invocation completes.
-                    foreach (var msg in inflight)
-                    {
-                        if (msg.Direction == Message.Directions.OneWay)
-                        {
-                            msg.DisposeBody();
-                        }
-                    }
-
-                    inflight.Clear();
                 }
             }
             catch (Exception exception)
@@ -431,11 +420,25 @@ private async Task ProcessOutgoing()
             }
             finally
             {
+                DisposeOneWayRequestBodies(inflight);
                 _transport!.Output.Complete();
                 this.StartClosing(error);
             }
         }
 
+        private static void DisposeOneWayRequestBodies(List<Message> messages)
+        {
+            foreach (var message in messages)
+            {
+                if (message.Direction == Message.Directions.OneWay)
+                {
+                    message.DisposeBody();
+                }
+            }
+
+            messages.Clear();
+        }
+
         private void RerouteMessage(Message message)
         {
             LogInformationReroutingMessage(this.Log, message, new EndPointLogValue(this.RemoteEndPoint));

src/Orleans.Core/Runtime/CallbackData.cs

@@ -32,7 +32,7 @@ public CallbackData(
 
         public Message Message { get; } // might hold metadata used by response pipeline
 
-        public bool IsCompleted => this.completed == 1;
+        public bool IsCompleted => Volatile.Read(ref completed) != 0;
 
         public void SubscribeForCancellation(CancellationToken cancellationToken)
         {