Fix for FT.CURSOR in cluster

.gitignore

@@ -410,3 +410,8 @@ tests/NRedisStack.Tests/redis_credentials/redis_user.crt
 # global.json
 global.json
 tests/NRedisStack.Tests/lcov.net8.0.info
+
+# docker containers
+tests/dockers/cluster/
+tests/dockers/standalone/
+tests/dockers/all/

Directory.Packages.props

@@ -12,7 +12,7 @@
     <!-- primary library -->
     <PackageVersion Include="NetTopologySuite" Version="2.6.0" />
     <PackageVersion Include="System.Text.Json" Version="9.0.7" />
-    <PackageVersion Include="StackExchange.Redis" Version="2.8.58" />
+    <PackageVersion Include="StackExchange.Redis" Version="2.9.17" />
     <!-- tests, etc -->
     <PackageVersion Include="BouncyCastle.Cryptography" Version="2.6.1" />
     <PackageVersion Include="coverlet.collector" Version="6.0.4" />

src/NRedisStack/Auxiliary.cs

@@ -67,12 +67,22 @@ public static RedisResult Execute(this IDatabase db, SerializedCommand command)
         return db.Execute(command.Command, command.Args);
     }
 
+    internal static RedisResult Execute(this IServer server, int? db, SerializedCommand command)
+    {
+        return server.Execute(db, command.Command, command.Args);
+    }
+
     public static async Task<RedisResult> ExecuteAsync(this IDatabaseAsync db, SerializedCommand command)
     {
         ((IDatabase)db).SetInfoInPipeline();
         return await db.ExecuteAsync(command.Command, command.Args);
     }
 
+    internal static async Task<RedisResult> ExecuteAsync(this IServer server, int? db, SerializedCommand command)
+    {
+        return await server.ExecuteAsync(db, command.Command, command.Args);
+    }
+
     public static List<RedisResult> ExecuteBroadcast(this IDatabase db, string command)
         => db.ExecuteBroadcast(new SerializedCommand(command));
 

src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt

@@ -1 +1,13 @@
 #nullable enable
+NRedisStack.ISearchCommands.AggregateEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.ISearchCommands.CursorDel(NRedisStack.Search.AggregationResult! result) -> bool
+NRedisStack.ISearchCommands.CursorRead(NRedisStack.Search.AggregationResult! result, int? count = null) -> NRedisStack.Search.AggregationResult!
+NRedisStack.ISearchCommandsAsync.AggregateAsyncEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IAsyncEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.ISearchCommandsAsync.CursorDelAsync(NRedisStack.Search.AggregationResult! result) -> System.Threading.Tasks.Task<bool>!
+NRedisStack.ISearchCommandsAsync.CursorReadAsync(NRedisStack.Search.AggregationResult! result, int? count = null) -> System.Threading.Tasks.Task<NRedisStack.Search.AggregationResult!>!
+NRedisStack.SearchCommands.AggregateEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.SearchCommands.CursorDel(NRedisStack.Search.AggregationResult! result) -> bool
+NRedisStack.SearchCommands.CursorRead(NRedisStack.Search.AggregationResult! result, int? count = null) -> NRedisStack.Search.AggregationResult!
+NRedisStack.SearchCommandsAsync.AggregateAsyncEnumerable(string! index, NRedisStack.Search.AggregationRequest! query) -> System.Collections.Generic.IAsyncEnumerable<NRedisStack.Search.Aggregation.Row>!
+NRedisStack.SearchCommandsAsync.CursorDelAsync(NRedisStack.Search.AggregationResult! result) -> System.Threading.Tasks.Task<bool>!
+NRedisStack.SearchCommandsAsync.CursorReadAsync(NRedisStack.Search.AggregationResult! result, int? count = null) -> System.Threading.Tasks.Task<NRedisStack.Search.AggregationResult!>!

src/NRedisStack/ResponseParser.cs

@@ -737,6 +737,20 @@ public static AggregationResult ToAggregationResult(this RedisResult result, Agg
         }
     }
 
+    internal static AggregationResult ToAggregationResult(this RedisResult result, string indexName, AggregationRequest query, IServer? server, int? database)
+    {
+        if (query.IsWithCursor())
+        {
+            var results = (RedisResult[])result!;
+
+            return new AggregationResult.WithCursorAggregationResult(indexName, results[0], (long)results[1], server, database);
+        }
+        else
+        {
+            return new(result);
+        }
+    }
+
     public static Dictionary<string, RedisResult>[] ToDictionarys(this RedisResult result)
     {
         var resArr = (RedisResult[])result!;

src/NRedisStack/Search/AggregationRequest.cs

@@ -128,6 +128,7 @@ public AggregationRequest Cursor(int? count = null, long? maxIdle = null)
 
         if (count != null)
         {
+            Count = count;
             args.Add(SearchArgs.COUNT);
             args.Add(count);
         }
@@ -139,6 +140,7 @@ public AggregationRequest Cursor(int? count = null, long? maxIdle = null)
         }
         return this;
     }
+    internal int? Count { get; set; }
 
     public AggregationRequest Params(Dictionary<string, object> nameValue)
     {

src/NRedisStack/Search/AggregationResult.cs

@@ -3,15 +3,29 @@
 
 namespace NRedisStack.Search;
 
-public sealed class AggregationResult
+public class AggregationResult
 {
+    // internal subclass for WITHCURSOR calls, which need to be issued to the same connection
+    internal sealed class WithCursorAggregationResult : AggregationResult
+    {
+        internal WithCursorAggregationResult(string indexName, RedisResult result, long cursorId, IServer? server,
+            int? database) : base(result, cursorId)
+        {
+            IndexName = indexName;
+            Server = server;
+            Database = database;
+        }
+        public string IndexName { get; }
+        public IServer? Server { get; }
+        public int? Database { get; }
+    }
+
     public long TotalResults { get; }
     private readonly Dictionary<string, object>[] _results;
     private Dictionary<string, RedisValue>[]? _resultsAsRedisValues;
 
     public long CursorId { get; }
 
-
     internal AggregationResult(RedisResult result, long cursorId = -1)
     {
         var arr = (RedisResult[])result!;
@@ -45,7 +59,6 @@ internal AggregationResult(RedisResult result, long cursorId = -1)
         CursorId = cursorId;
     }
 
-
     /// <summary>
     /// takes a Redis multi-bulk array represented by a RedisResult[] and recursively processes its elements.
     /// For each element in the array, it checks if it's another multi-bulk array, and if so, it recursively calls itself.

src/NRedisStack/Search/ISearchCommands.cs

@@ -1,4 +1,6 @@
+using System.ComponentModel;
 using NRedisStack.Search;
+using NRedisStack.Search.Aggregation;
 using NRedisStack.Search.DataTypes;
 using StackExchange.Redis;
 
@@ -18,11 +20,20 @@ public interface ISearchCommands
     /// Run a search query on an index, and perform aggregate transformations on the results.
     /// </summary>
     /// <param name="index">The index name.</param>
-    /// <param name="query">The query</param>
+    /// <param name="query">The query.</param>
     /// <returns>An <see langword="AggregationResult"/> object</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
     AggregationResult Aggregate(string index, AggregationRequest query);
 
+    /// <summary>
+    /// Run a search query on an index, and perform aggregate transformations on the results.
+    /// </summary>
+    /// <param name="index">The index name.</param>
+    /// <param name="query">The query.</param>
+    /// <returns>A sequence of <see langword="Row"/> values.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
+    IEnumerable<Row> AggregateEnumerable(string index, AggregationRequest query);
+
     /// <summary>
     /// Add an alias to an index.
     /// </summary>
@@ -92,22 +103,43 @@ public interface ISearchCommands
     /// <summary>
     /// Delete a cursor from the index.
     /// </summary>
-    /// <param name="indexName">The index name</param>
+    /// <param name="indexName">The index name.</param>
     /// <param name="cursorId">The cursor's ID.</param>
     /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    [Obsolete("When possible, use CursorDel(AggregationResult) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     bool CursorDel(string indexName, long cursorId);
 
+    /// <summary>
+    /// Delete a cursor from the index.
+    /// </summary>
+    /// <param name="result">The result of a previous call to Aggregate or CursorRead.</param>
+    /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    bool CursorDel(AggregationResult result);
+
     /// <summary>
     /// Read next results from an existing cursor.
     /// </summary>
-    /// <param name="indexName">The index name</param>
+    /// <param name="indexName">The index name.</param>
     /// <param name="cursorId">The cursor's ID.</param>
     /// <param name="count">Limit the amount of returned results.</param>
     /// <returns>A AggregationResult object with the results</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    [Obsolete("When possible, use AggregateEnumerable or CursorRead(AggregationResult, int?) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     AggregationResult CursorRead(string indexName, long cursorId, int? count = null);
 
+    /// <summary>
+    /// Read next results from an existing cursor.
+    /// </summary>
+    /// <param name="result">The result of a previous call to Aggregate or CursorRead.</param>
+    /// <param name="count">Limit the amount of returned results.</param>
+    /// <returns>A AggregationResult object with the results</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    public AggregationResult CursorRead(AggregationResult result, int? count = null);
+
     /// <summary>
     /// Add terms to a dictionary.
     /// </summary>

src/NRedisStack/Search/ISearchCommandsAsync.cs

@@ -1,4 +1,6 @@
+using System.ComponentModel;
 using NRedisStack.Search;
+using NRedisStack.Search.Aggregation;
 using NRedisStack.Search.DataTypes;
 using StackExchange.Redis;
 
@@ -14,14 +16,25 @@ public interface ISearchCommandsAsync
     Task<RedisResult[]> _ListAsync();
 
     /// <summary>
-    /// Run a search query on an index, and perform aggregate transformations on the results.
+    /// Run a search query on an index, and perform aggregate transformations on the results. This operates
+    /// as a cursor and may involve multiple commands to the server.
     /// </summary>
     /// <param name="index">The index name.</param>
     /// <param name="query">The query</param>
     /// <returns>An <see langword="AggregationResult"/> object</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
     Task<AggregationResult> AggregateAsync(string index, AggregationRequest query);
 
+    /// <summary>
+    /// Run a search query on an index, and perform aggregate transformations on the results. This operates
+    /// as a cursor and may involve multiple commands to the server.
+    /// </summary>
+    /// <param name="index">The index name.</param>
+    /// <param name="query">The query.</param>
+    /// <returns>A sequence of <see langword="Row"/> values.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.aggregate"/></remarks>
+    IAsyncEnumerable<Row> AggregateAsyncEnumerable(string index, AggregationRequest query);
+
     /// <summary>
     /// Add an alias to an index.
     /// </summary>
@@ -95,8 +108,18 @@ public interface ISearchCommandsAsync
     /// <param name="cursorId">The cursor's ID.</param>
     /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    [Obsolete("When possible, use CursorDelAsync(AggregationResult, int?) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     Task<bool> CursorDelAsync(string indexName, long cursorId);
 
+    /// <summary>
+    /// Delete a cursor from the index.
+    /// </summary>
+    /// <param name="result">The result of a previous call to AggregateAsync or CursorReadAsync.</param>
+    /// <returns><see langword="true"/> if it has been deleted, <see langword="false"/> if it did not exist.</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-del/"/></remarks>
+    Task<bool> CursorDelAsync(AggregationResult result);
+
     /// <summary>
     /// Read next results from an existing cursor.
     /// </summary>
@@ -105,8 +128,19 @@ public interface ISearchCommandsAsync
     /// <param name="count">Limit the amount of returned results.</param>
     /// <returns>A AggregationResult object with the results</returns>
     /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    [Obsolete("When possible, use AggregateAsyncEnumerable or CursorReadAsync(AggregationResult, int?) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     Task<AggregationResult> CursorReadAsync(string indexName, long cursorId, int? count = null);
 
+    /// <summary>
+    /// Read next results from an existing cursor.
+    /// </summary>
+    /// <param name="result">The result of a previous AggregateAsync or CursorReadAsync call.</param>
+    /// <param name="count">Limit the amount of returned results.</param>
+    /// <returns>A AggregationResult object with the results</returns>
+    /// <remarks><seealso href="https://redis.io/commands/ft.cursor-read/"/></remarks>
+    Task<AggregationResult> CursorReadAsync(AggregationResult result, int? count = null);
+
     /// <summary>
     /// Add terms to a dictionary.
     /// </summary>

src/NRedisStack/Search/SearchCommands.cs

@@ -1,4 +1,6 @@
+using System.ComponentModel;
 using NRedisStack.Search;
+using NRedisStack.Search.Aggregation;
 using NRedisStack.Search.DataTypes;
 using StackExchange.Redis;
 namespace NRedisStack;
@@ -16,8 +18,54 @@ public RedisResult[] _List()
     public AggregationResult Aggregate(string index, AggregationRequest query)
     {
         SetDefaultDialectIfUnset(query);
-        var result = db.Execute(SearchCommandBuilder.Aggregate(index, query));
-        return result.ToAggregationResult(query);
+        IServer? server = null;
+        int? database = null;
+
+        var command = SearchCommandBuilder.Aggregate(index, query);
+        if (query.IsWithCursor())
+        {
+            // we can issue this anywhere, but follow-up calls need to be on the same server
+            server = GetRandomServerForCluster(db, out database);
+        }
+
+        RedisResult result;
+        if (server is not null)
+        {
+            result = server.Execute(database, command);
+        }
+        else
+        {
+            result = db.Execute(command);
+        }
+
+        return result.ToAggregationResult(index, query, server, database);
+    }
+
+    public IEnumerable<Row> AggregateEnumerable(string index, AggregationRequest query)
+    {
+        if (!query.IsWithCursor()) query.Cursor();
+
+        var result = Aggregate(index, query);
+        try
+        {
+            while (true)
+            {
+                var count = checked((int)result.TotalResults);
+                for (int i = 0; i < count; i++)
+                {
+                    yield return result.GetRow(i);
+                }
+                if (result.CursorId == 0) break;
+                result = CursorRead(result, query.Count);
+            }
+        }
+        finally
+        {
+            if (result.CursorId != 0)
+            {
+                CursorDel(result);
+            }
+        }
     }
 
     /// <inheritdoc/>
@@ -72,18 +120,52 @@ public bool Create(string indexName, Schema schema)
     }
 
     /// <inheritdoc/>
+    [Obsolete("When possible, use CursorDel(AggregationResult) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     public bool CursorDel(string indexName, long cursorId)
     {
         return db.Execute(SearchCommandBuilder.CursorDel(indexName, cursorId)).OKtoBoolean();
     }
 
+    public bool CursorDel(AggregationResult result)
+    {
+        if (result is not AggregationResult.WithCursorAggregationResult withCursor)
+        {
+            throw new ArgumentException(
+                message: $"{nameof(CursorDelAsync)} must be called with a value returned from a previous call to {nameof(AggregateAsync)} with a cursor.",
+                paramName: nameof(result));
+        }
+
+        var command = SearchCommandBuilder.CursorDel(withCursor.IndexName, withCursor.CursorId);
+        var resp = withCursor.Server is { } server
+            ? server.Execute(withCursor.Database, command)
+            : db.Execute(command);
+        return resp.OKtoBoolean();
+    }
+
     /// <inheritdoc/>
+    [Obsolete("When possible, use CusorReadEnumerable or CursorRead(AggregationResult, int?) instead. This legacy API will not work correctly on CLUSTER environments, but will continue to work for single-node deployments.")]
+    [Browsable(false), EditorBrowsable(EditorBrowsableState.Never)]
     public AggregationResult CursorRead(string indexName, long cursorId, int? count = null)
     {
         var resp = db.Execute(SearchCommandBuilder.CursorRead(indexName, cursorId, count)).ToArray();
         return new(resp[0], (long)resp[1]);
     }
 
+    public AggregationResult CursorRead(AggregationResult result, int? count = null)
+    {
+        if (result is not AggregationResult.WithCursorAggregationResult withCursor)
+        {
+            throw new ArgumentException(message: $"{nameof(CursorReadAsync)} must be called with a value returned from a previous call to {nameof(AggregateAsync)} with a cursor.", paramName: nameof(result));
+        }
+        var command = SearchCommandBuilder.CursorRead(withCursor.IndexName, withCursor.CursorId, count);
+        var rawResult = withCursor.Server is { } server
+            ? server.Execute(withCursor.Database, command)
+            : db.Execute(command);
+        var resp = rawResult.ToArray();
+        return new AggregationResult.WithCursorAggregationResult(withCursor.IndexName, resp[0], (long)resp[1], withCursor.Server, withCursor.Database);
+    }
+
     /// <inheritdoc/>
     public long DictAdd(string dict, params string[] terms)
     {