Fixed postgres subscription issues v14

src/HotChocolate/Core/src/Subscriptions.Postgres/PostgresChannelWriter.cs

@@ -1,7 +1,7 @@
-using System.Data;
 using System.Threading.Channels;
 using HotChocolate.Subscriptions.Diagnostics;
 using Npgsql;
+using NpgsqlTypes;
 using static HotChocolate.Subscriptions.Postgres.PostgresResources;
 
 namespace HotChocolate.Subscriptions.Postgres;
@@ -69,6 +69,7 @@
     private async Task HandleMessage(NpgsqlConnection connection, CancellationToken ct)
     {
         PostgresMessageEnvelope firstItem;
+
         while (!_channel.Reader.TryRead(out firstItem))
         {
             if (ct.IsCancellationRequested)
@@ -79,7 +80,7 @@
             await _channel.Reader.WaitToReadAsync(ct);
         }
 
-        var messages = new List<PostgresMessageEnvelope> { firstItem, };
+        var messages = new List<PostgresMessageEnvelope> { firstItem };
         while (!ct.IsCancellationRequested &&
                _maxSendBatchSize > messages.Count &&
                _channel.Reader.TryRead(out var item))
@@ -93,40 +94,60 @@
             // firstMessage that was already read from the channel
             ct.ThrowIfCancellationRequested();
 
-            await using var batch = connection.CreateBatch();
+            var payloads = new string[messages.Count];
 
-            foreach (var message in messages)
+            for (var i = 0; i < messages.Count; i++)
             {
-                var command = batch.CreateBatchCommand();
+                payloads[i] = messages[i].FormattedPayload;
+            }
 
-                command.CommandText = "SELECT pg_notify(@channel, @message);";
+            const string sql =
+                """
+                SELECT pg_notify(t.channel, t.message)
+                FROM (SELECT @channel AS channel, unnest(@messages) AS message ) AS t;
+                """;
 
-                var channel = new NpgsqlParameter("channel", DbType.String)
-                {
-                    Value = _channelName,
-                };
-                var msg = new NpgsqlParameter("message", DbType.String)
+            await using var command = connection.CreateCommand();
+
+            command.CommandText = sql;
+
+            command.Parameters.Add(
+                new NpgsqlParameter("channel", NpgsqlDbType.Text)
                 {
-                    Value = message.FormattedPayload,
-                };
-                command.Parameters.Add(channel);
-                command.Parameters.Add(msg);
+                    Value = _channelName
+                });
 
-                batch.BatchCommands.Add(command);
-            }
+            command.Parameters.Add(
+                new NpgsqlParameter("messages", NpgsqlDbType.Array | NpgsqlDbType.Varchar)
+                {
+                    Value = payloads
+                });
 
-            await batch.PrepareAsync(ct);
-            await batch.ExecuteNonQueryAsync(ct);
+            await command.PrepareAsync(ct);
+            await command.ExecuteNonQueryAsync(ct);
         }
         catch (Exception ex)
         {
             var msg = string.Format(ChannelWriter_FailedToSend, messages.Count, ex.Message);
             _diagnosticEvents.ProviderInfo(msg);
 
             // if we cannot send the message we put it back into the channel
+            // however as the channel is bounded, we might not able to requeue the message and will
+            // be forced to drop them if they can't be written
+            var failedCount = 0;
+
             foreach (var message in messages)
             {
-                await _channel.Writer.WriteAsync(message, ct);
+                if (!_channel.Writer.TryWrite(message))
+                {
+                    failedCount++;
+                }
+            }
+
+            if (failedCount > 0)
+            {
+                _diagnosticEvents.ProviderInfo(
+                    string.Format(ChannelWriter_FailedToRequeueMessage, failedCount));
             }
         }
     }

src/HotChocolate/Core/src/Subscriptions.Postgres/Properties/PostgresResources.resx

@@ -55,4 +55,7 @@
   <data name="PostgresMessageEnvelope_PayloadTooLarge" xml:space="preserve">
     <value>Payload is too long to we written to Postgres. Serialized message is {0} bytes but limit is {1} bytes</value>
   </data>
+  <data name="ChannelWriter_FailedToRequeueMessage" xml:space="preserve">
+    <value>The postgres writer was unable to requeue messages. {0} messages have been lost</value>
+  </data>
 </root>