Avoid channel closed exception when stopping batching producer (#2262)

rogeralsing · web-flow · commit dfc120687b25 · 2025-08-07T11:15:59.000+02:00
diff --git a/src/Proto.Cluster/PubSub/BatchingProducer.cs b/src/Proto.Cluster/PubSub/BatchingProducer.cs
@@ -214,10 +214,9 @@ private void RemoveCancelledFromBatch(PubSubBatchWithReceipts batchWrapper)
 
     private void StopAcceptingNewMessages()
     {
-        if (!_publisherChannel.Reader.Completion.IsCompleted)
-        {
-            _publisherChannel.Writer.Complete();
-        }
+        // Attempt to complete the channel without throwing if it has already been closed
+        // The publisher loop and publish error handling might both call this concurrently
+        _publisherChannel.Writer.TryComplete();
     }
 
     private async Task PublishBatch(PubSubBatchWithReceipts batchWrapper)

Original file line number	Diff line number	Diff line change
`@@ -214,10 +214,9 @@ private void RemoveCancelledFromBatch(PubSubBatchWithReceipts batchWrapper)`
`214`	`214`
`215`	`215`	`private void StopAcceptingNewMessages()`
`216`	`216`	`{`
`217`		`- if (!_publisherChannel.Reader.Completion.IsCompleted)`
`218`		`- {`
`219`		`- _publisherChannel.Writer.Complete();`
`220`		`- }`
	`217`	`+ // Attempt to complete the channel without throwing if it has already been closed`
	`218`	`+ // The publisher loop and publish error handling might both call this concurrently`
	`219`	`+ _publisherChannel.Writer.TryComplete();`
`221`	`220`	`}`
`222`	`221`
`223`	`222`	`private async Task PublishBatch(PubSubBatchWithReceipts batchWrapper)`