avoid stackoverflow around ParseRedisStreamEntries

StackExchange.Redis/StackExchange/Redis/Interfaces/ISubscriber.cs

@@ -56,7 +56,7 @@ public interface ISubscriber : IRedis
         Task<long> PublishAsync(RedisChannel channel, RedisValue message, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
-        /// Subscribe to perform some operation when a change to the preferred/active node is broadcast.
+        /// Subscribe to perform some operation when a message to the preferred/active node is broadcast, without any guarantee of ordered handling.
         /// </summary>
         /// <param name="channel">The channel to subscribe to.</param>
         /// <param name="handler">The handler to invoke when a message is received on <paramref name="channel"/>.</param>
@@ -66,7 +66,7 @@ public interface ISubscriber : IRedis
         void Subscribe(RedisChannel channel, Action<RedisChannel, RedisValue> handler, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
-        /// Subscribe to perform some operation when a change to the preferred/active node is broadcast, as a channel.
+        /// Subscribe to perform some operation when a message to the preferred/active node is broadcast, as a queue that guarantees ordered handling.
         /// </summary>
         /// <param name="channel">The redis channel to subscribe to.</param>
         /// <param name="flags">The command flags to use.</param>

StackExchange.Redis/StackExchange/Redis/RawResult.cs

@@ -390,14 +390,6 @@ internal bool TryGetInt64(out long value)
             _payload.CopyTo(span);
             return RedisValue.TryParseInt64(span, out value);
         }
-
-        internal static RawResult CreateMultiBulk(RawResult rawResult0, RawResult rawResult1)
-        {
-            var arr = ArrayPool<RawResult>.Shared.Rent(2);
-            arr[0] = rawResult0;
-            arr[1] = rawResult1;
-            return new RawResult(arr, 2);
-        }
     }
 }
 

StackExchange.Redis/StackExchange/Redis/ResultProcessor.cs

 using System;
+using System.Buffers;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
@@ -1584,9 +1585,14 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
                 // Items 9-12 represent the first and last entry in the stream. The values will
                 // be nil (stored in index 9 & 11) if the stream length is 0.
 
-                var tmp = RawResult.CreateMultiBulk(arr[9], arr[11]);
+                var leased = ArrayPool<RawResult>.Shared.Rent(2);
+                leased[0] = arr[9];
+                leased[1] = arr[11];
+                var tmp = new RawResult(leased, 2);
                 var entries = ParseRedisStreamEntries(tmp);
-                tmp.Recycle();
+                // note: don't .Recycle(), would be a stack overflow because
+                // it would bridge the fake and real result set
+                ArrayPool<RawResult>.Shared.Return(leased); 
 
                 var streamInfo = new StreamInfo(length: (int)arr[1].AsRedisValue(),
                     radixTreeKeys: (int)arr[3].AsRedisValue(),