Manual pagination of ListWorkflows operation

src/Temporalio/Client/ITemporalClient.Workflow.cs

@@ -183,5 +183,24 @@ public IAsyncEnumerable<WorkflowExecution> ListWorkflowsAsync(
         /// <seealso href="https://docs.temporal.io/visibility">Visibility docs.</seealso>
         public Task<WorkflowExecutionCount> CountWorkflowsAsync(
             string query, WorkflowCountOptions? options = null);
+
+        /// <summary>
+        /// List workflows with the given query using manual paging.
+        /// </summary>
+        /// <param name="query">
+        /// Query to use for filtering. Subsequent pages must have the same query as the initial call.
+        /// </param>
+        /// <param name="nextPageToken">
+        /// Set to null for the initial call to retrieve the first page.
+        /// Set to <see cref="WorkflowListPage.NextPageToken"/> returned by a previous call to retrieve the next page.
+        /// </param>
+        /// <param name="options">Options for the list call.</param>
+        /// <returns>
+        /// A single page of a list of workflows.
+        /// Repeat the call using <see cref="WorkflowListPage.NextPageToken"/> to get more pages.
+        /// </returns>
+        /// <seealso href="https://docs.temporal.io/visibility">Visibility docs.</seealso>
+        Task<WorkflowListPage> ListWorkflowsPaginatedAsync(
+            string query, byte[]? nextPageToken, WorkflowListPaginatedOptions? options = null);
     }
 }

src/Temporalio/Client/Interceptors/ClientOutboundInterceptor.Workflow.cs

@@ -99,6 +99,12 @@ public virtual Task<WorkflowHistoryEventPage> FetchWorkflowHistoryEventPageAsync
         /// </summary>
         /// <param name="input">Input details of the call.</param>
         /// <returns>Async enumerator for the workflows.</returns>
+        /// <remarks>
+        /// This method only gets called by <see cref="ITemporalClient.ListWorkflowsAsync"/>.
+        /// It does not get called by <see cref="ITemporalClient.ListWorkflowsPaginatedAsync"/>.
+        /// This method is called before the first page is fetched. Afterwards, before each page fetched
+        /// (including before the first page), <see cref="ListWorkflowsPaginatedAsync"/> is called.
+        /// </remarks>
         public virtual IAsyncEnumerable<WorkflowExecution> ListWorkflowsAsync(
             ListWorkflowsInput input) =>
             Next.ListWorkflowsAsync(input);
@@ -112,5 +118,20 @@ public virtual IAsyncEnumerable<WorkflowExecution> ListWorkflowsAsync(
         public virtual Task<WorkflowExecutionCount> CountWorkflowsAsync(
             CountWorkflowsInput input) =>
             Next.CountWorkflowsAsync(input);
+
+#pragma warning disable CS1574 // ListWorkflowsAsync does not exist in .Net Framework/Standard
+        /// <summary>
+        /// Intercept page fetch for list workflows calls.
+        /// </summary>
+        /// <param name="input">Input details of the call.</param>
+        /// <returns>A single page of query results.</returns>
+        /// <remarks>
+        /// This method is called each time <see cref="ITemporalClient.ListWorkflowsPaginatedAsync"/> is called.
+        /// It also gets called for each page fetched when iterating the enumerable returned by <see cref="ITemporalClient.ListWorkflowsAsync"/>.
+        /// </remarks>
+#pragma warning restore CS1574
+        public virtual Task<WorkflowListPage> ListWorkflowsPaginatedAsync(
+            ListWorkflowsPaginatedInput input) =>
+            Next.ListWorkflowsPaginatedAsync(input);
     }
-}
+}

src/Temporalio/Client/Interceptors/ListWorkflowsPaginatedInput.cs

@@ -0,0 +1,17 @@
+namespace Temporalio.Client.Interceptors
+{
+    /// <summary>
+    /// Input for <see cref="ClientOutboundInterceptor.ListWorkflowsPaginatedAsync" />.
+    /// </summary>
+    /// <param name="Query">List query.</param>
+    /// <param name="NextPageToken">Next page token from a previous response. Null if the request is for the first page.</param>
+    /// <param name="Options">Options passed in to list.</param>
+    /// <remarks>
+    /// WARNING: This constructor may have required properties added. Do not rely on the exact
+    /// constructor, only use "with" clauses.
+    /// </remarks>
+    public record ListWorkflowsPaginatedInput(
+        string Query,
+        byte[]? NextPageToken,
+        WorkflowListPaginatedOptions? Options);
+}

src/Temporalio/Client/TemporalClient.Workflow.cs

@@ -122,9 +122,14 @@ public Task<WorkflowExecutionCount> CountWorkflowsAsync(
             string query, WorkflowCountOptions? options = null) =>
             OutboundInterceptor.CountWorkflowsAsync(new(Query: query, Options: options));
 
+        /// <inheritdoc />
+        public Task<WorkflowListPage> ListWorkflowsPaginatedAsync(
+            string query, byte[]? nextPageToken, WorkflowListPaginatedOptions? options = null) =>
+            OutboundInterceptor.ListWorkflowsPaginatedAsync(new(Query: query, NextPageToken: nextPageToken, Options: options));
+
         internal partial class Impl
         {
-            private static IReadOnlyCollection<HistoryEvent> emptyEvents = new List<HistoryEvent>(0);
+            private static IReadOnlyCollection<HistoryEvent> emptyEvents = new List<HistoryEvent>(0).AsReadOnly();
 
             /// <inheritdoc />
             public override async Task<WorkflowHandle<TWorkflow, TResult>> StartWorkflowAsync<TWorkflow, TResult>(
@@ -621,7 +626,7 @@ public override async Task<WorkflowHistoryEventPage> FetchWorkflowHistoryEventPa
                     if (pageComplete)
                     {
                         return new WorkflowHistoryEventPage(
-                            resp.History?.Events ?? emptyEvents,
+                            resp.History?.Events?.ToList().AsReadOnly() ?? emptyEvents,
                             resp.NextPageToken.IsEmpty ? null : resp.NextPageToken.ToByteArray());
                     }
                     req.NextPageToken = resp.NextPageToken;
@@ -645,39 +650,65 @@ public override async Task<WorkflowExecutionCount> CountWorkflowsAsync(
                 return new(resp);
             }
 
+            /// <inheritdoc />
+            public override async Task<WorkflowListPage> ListWorkflowsPaginatedAsync(ListWorkflowsPaginatedInput input)
+            {
+                var req = new ListWorkflowExecutionsRequest
+                {
+                    Namespace = Client.Options.Namespace,
+                    PageSize = input.Options?.PageSize ?? 0,
+                    Query = input.Query,
+                };
+                if (input.NextPageToken is not null)
+                {
+                    req.NextPageToken = ByteString.CopyFrom(input.NextPageToken);
+                }
+
+                var resp = await Client.Connection.WorkflowService.ListWorkflowExecutionsAsync(
+                    req, DefaultRetryOptions(input.Options?.Rpc)).ConfigureAwait(false);
+
+                return new(
+                    Workflows: resp.Executions
+                        .Select(e => new WorkflowExecution(e, Client.Options.DataConverter, Client.Options.Namespace))
+                        .ToList()
+                        .AsReadOnly(),
+                    NextPageToken: resp.NextPageToken.IsEmpty ? null : resp.NextPageToken.ToByteArray());
+            }
+
 #if NETCOREAPP3_0_OR_GREATER
             private async IAsyncEnumerable<WorkflowExecution> ListWorkflowsInternalAsync(
                 ListWorkflowsInput input,
                 [EnumeratorCancellation] CancellationToken cancellationToken = default)
             {
+                var limit = input.Options?.Limit ?? 0;
+                if (limit < 0)
+                {
+                    throw new ArgumentOutOfRangeException(nameof(input), "Limit cannot be negative");
+                }
+
                 // Need to combine cancellation token
                 var rpcOptsAndCancelSource = DefaultRetryOptions(input.Options?.Rpc).
                     WithAdditionalCancellationToken(cancellationToken);
-                var yielded = 0;
                 try
                 {
-                    var req = new ListWorkflowExecutionsRequest()
-                    {
-                        // TODO(cretz): Allow setting of page size or next page token?
-                        Namespace = Client.Options.Namespace,
-                        Query = input.Query,
-                    };
+                    var pageOpts = new WorkflowListPaginatedOptions { Rpc = rpcOptsAndCancelSource.Item1 };
+                    byte[]? nextPageToken = null;
+                    var yielded = 0;
                     do
                     {
-                        var resp = await Client.Connection.WorkflowService.ListWorkflowExecutionsAsync(
-                            req, rpcOptsAndCancelSource.Item1).ConfigureAwait(false);
-                        foreach (var exec in resp.Executions)
+                        var page = await Client.ListWorkflowsPaginatedAsync(input.Query, nextPageToken, pageOpts).ConfigureAwait(false);
+                        foreach (var exec in page.Workflows)
                         {
-                            if (input.Options != null && input.Options.Limit > 0 &&
-                                yielded++ >= input.Options.Limit)
+                            yield return exec;
+                            yielded++;
+                            if (limit > 0 && yielded >= limit)
                             {
                                 yield break;
                             }
-                            yield return new(exec, Client.Options.DataConverter, Client.Options.Namespace);
                         }
-                        req.NextPageToken = resp.NextPageToken;
+                        nextPageToken = page.NextPageToken;
                     }
-                    while (!req.NextPageToken.IsEmpty);
+                    while (nextPageToken is not null);
                 }
                 finally
                 {

src/Temporalio/Client/WorkflowListPage.cs

@@ -0,0 +1,14 @@
+using System.Collections.Generic;
+
+namespace Temporalio.Client
+{
+    /// <summary>
+    /// Result type of <see cref="ITemporalClient.ListWorkflowsPaginatedAsync"/>.
+    /// </summary>
+    /// <param name="Workflows">A page from the list of workflows matching the query.</param>
+    /// <param name="NextPageToken">
+    /// Token to pass to <see cref="ITemporalClient.ListWorkflowsPaginatedAsync"/> to retrieve the next page.
+    /// Null if there are no more pages.
+    /// </param>
+    public record WorkflowListPage(IReadOnlyCollection<WorkflowExecution> Workflows, byte[]? NextPageToken);
+}

src/Temporalio/Client/WorkflowListPaginatedOptions.cs

@@ -0,0 +1,34 @@
+using System;
+
+namespace Temporalio.Client
+{
+    /// <summary>
+    /// Options for <see cref="ITemporalClient.ListWorkflowsPaginatedAsync"/>.
+    /// </summary>
+    public class WorkflowListPaginatedOptions : ICloneable
+    {
+        /// <summary>
+        /// Gets or sets the number of results per page. Zero means server default.
+        /// </summary>
+        public int PageSize { get; set; }
+
+        /// <summary>
+        /// Gets or sets RPC options for listing workflows.
+        /// </summary>
+        public RpcOptions? Rpc { get; set; }
+
+        /// <summary>
+        /// Create a shallow copy of these options.
+        /// </summary>
+        /// <returns>A shallow copy of these options and any transitive options fields.</returns>
+        public virtual object Clone()
+        {
+            var copy = (WorkflowListPaginatedOptions)MemberwiseClone();
+            if (Rpc != null)
+            {
+                copy.Rpc = (RpcOptions)Rpc.Clone();
+            }
+            return copy;
+        }
+    }
+}

tests/Temporalio.Tests/Client/TemporalClientWorkflowTests.cs

@@ -330,6 +330,35 @@ await AssertMore.EventuallyAsync(async () =>
         });
     }
 
+    [Fact]
+    public async Task ListWorkflowsAsync_ManualPaging_IsAccurate()
+    {
+        var workflowId = $"workflow-{Guid.NewGuid()}";
+        for (var i = 0; i < 5; i++)
+        {
+            var arg = new KSWorkflowParams(new KSAction(Result: new(Value: string.Empty)));
+            await Client.ExecuteWorkflowAsync(
+                (IKitchenSinkWorkflow wf) => wf.RunAsync(arg),
+                new(id: workflowId, taskQueue: Env.KitchenSinkWorkerTaskQueue));
+        }
+
+        var options = new WorkflowListPaginatedOptions { PageSize = 2 };
+
+        var firstPage = await Client.ListWorkflowsPaginatedAsync($"WorkflowId = '{workflowId}'", null, options);
+        Assert.Equal(2, firstPage.Workflows.Count);
+        Assert.NotNull(firstPage.NextPageToken);
+        Assert.NotEmpty(firstPage.NextPageToken);
+
+        var secondPage = await Client.ListWorkflowsPaginatedAsync($"WorkflowId = '{workflowId}'", firstPage.NextPageToken, options);
+        Assert.Equal(2, secondPage.Workflows.Count);
+        Assert.NotNull(secondPage.NextPageToken);
+        Assert.NotEmpty(secondPage.NextPageToken);
+
+        var thirdPage = await Client.ListWorkflowsPaginatedAsync($"WorkflowId = '{workflowId}'", secondPage.NextPageToken, options);
+        Assert.Equal(1, thirdPage.Workflows.Count);
+        Assert.Null(thirdPage.NextPageToken);
+    }
+
     internal record TracingEvent(string Name, object Input);
 
     internal class TracingClientInterceptor : IClientInterceptor