diff --git a/README.md b/README.md index b569fe7..0401e0b 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,8 @@ Prerequisites: * [ActivitySimple](src/ActivitySimple) - Simple workflow that runs simple activities. * [ActivityWorker](src/ActivityWorker) - Use .NET activities from a workflow in another language. * [AspNet](src/AspNet) - Demonstration of a generic host worker and an ASP.NET workflow starter. +* [Batch](src/Batch) - Three different best practices for processing a batch of records: an + iterator pattern, a heartbeating activity, and a sliding window of parallel child workflows. * [Bedrock](src/Bedrock) - Orchestrate a chatbot with Amazon Bedrock. * [ClientMtls](src/ClientMtls) - How to use client certificate authentication, e.g. for Temporal Cloud. * [ContextPropagation](src/ContextPropagation) - Context propagation via interceptors. diff --git a/TemporalioSamples.sln b/TemporalioSamples.sln index 4e560c4..34ffb77 100644 --- a/TemporalioSamples.sln +++ b/TemporalioSamples.sln @@ -118,6 +118,7 @@ EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.SampleAppHost", "src\AspireIntegrations\TemporalioSamples.SampleAppHost\TemporalioSamples.SampleAppHost.csproj", "{CA136E75-FC34-44E1-B8B2-6E33D8AF520E}" EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Temporal.Extensions.Aspire.Hosting", "src\AspireIntegrations\Temporal.Extensions.Aspire.Hosting\Temporal.Extensions.Aspire.Hosting.csproj", "{89D196AD-A6CE-42FB-BF46-C80BF579FE20}" +EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "StandaloneActivity", "StandaloneActivity", "{EAB0C45A-7620-D2D2-2901-5E7FCBFFDA77}" EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.StandaloneActivity", "src\StandaloneActivity\TemporalioSamples.StandaloneActivity.csproj", "{240517A1-13B5-4A67-8519-BFCF2C4591B9}" @@ -128,6 +129,17 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "NexusStandaloneOperations", EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.NexusStandaloneOperations", "src\NexusStandaloneOperations\TemporalioSamples.NexusStandaloneOperations.csproj", "{6B6622AE-2970-4AAD-B2E4-A7EE1E2C40EA}" EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Batch", "Batch", "{A95C608E-3127-41B7-BC69-7495768F30C5}" + ProjectSection(SolutionItems) = preProject + src\Batch\README.md = src\Batch\README.md + EndProjectSection +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.Batch.Iterator", "src\Batch\Iterator\TemporalioSamples.Batch.Iterator.csproj", "{5E600381-C245-44F2-996D-4736BB5A6AC5}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.Batch.HeartbeatingActivity", "src\Batch\HeartbeatingActivity\TemporalioSamples.Batch.HeartbeatingActivity.csproj", "{236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.Batch.SlidingWindow", "src\Batch\SlidingWindow\TemporalioSamples.Batch.SlidingWindow.csproj", "{A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -750,6 +762,42 @@ Global {6B6622AE-2970-4AAD-B2E4-A7EE1E2C40EA}.Release|x64.Build.0 = Release|Any CPU {6B6622AE-2970-4AAD-B2E4-A7EE1E2C40EA}.Release|x86.ActiveCfg = Release|Any CPU {6B6622AE-2970-4AAD-B2E4-A7EE1E2C40EA}.Release|x86.Build.0 = Release|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Debug|Any CPU.Build.0 = Debug|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Debug|x64.ActiveCfg = Debug|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Debug|x64.Build.0 = Debug|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Debug|x86.ActiveCfg = Debug|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Debug|x86.Build.0 = Debug|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Release|Any CPU.ActiveCfg = Release|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Release|Any CPU.Build.0 = Release|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Release|x64.ActiveCfg = Release|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Release|x64.Build.0 = Release|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Release|x86.ActiveCfg = Release|Any CPU + {5E600381-C245-44F2-996D-4736BB5A6AC5}.Release|x86.Build.0 = Release|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Debug|Any CPU.Build.0 = Debug|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Debug|x64.ActiveCfg = Debug|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Debug|x64.Build.0 = Debug|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Debug|x86.ActiveCfg = Debug|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Debug|x86.Build.0 = Debug|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Release|Any CPU.ActiveCfg = Release|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Release|Any CPU.Build.0 = Release|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Release|x64.ActiveCfg = Release|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Release|x64.Build.0 = Release|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Release|x86.ActiveCfg = Release|Any CPU + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2}.Release|x86.Build.0 = Release|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Debug|Any CPU.Build.0 = Debug|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Debug|x64.ActiveCfg = Debug|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Debug|x64.Build.0 = Debug|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Debug|x86.ActiveCfg = Debug|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Debug|x86.Build.0 = Debug|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Release|Any CPU.ActiveCfg = Release|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Release|Any CPU.Build.0 = Release|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Release|x64.ActiveCfg = Release|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Release|x64.Build.0 = Release|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Release|x86.ActiveCfg = Release|Any CPU + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -814,5 +862,9 @@ Global {5D493692-53AB-4FAA-BA4D-33B1E54E9A48} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} {17436B0C-8853-030B-9E2F-BAEB65BE59FB} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} {6B6622AE-2970-4AAD-B2E4-A7EE1E2C40EA} = {17436B0C-8853-030B-9E2F-BAEB65BE59FB} + {A95C608E-3127-41B7-BC69-7495768F30C5} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} + {5E600381-C245-44F2-996D-4736BB5A6AC5} = {A95C608E-3127-41B7-BC69-7495768F30C5} + {236FCC5A-9E97-4EC4-9E33-E8FBA98E4DA2} = {A95C608E-3127-41B7-BC69-7495768F30C5} + {A2D3FA9C-EA09-4AEE-B57C-17FB40025EB1} = {A95C608E-3127-41B7-BC69-7495768F30C5} EndGlobalSection EndGlobal diff --git a/src/Batch/HeartbeatingActivity/HeartbeatingActivityBatchWorkflow.workflow.cs b/src/Batch/HeartbeatingActivity/HeartbeatingActivityBatchWorkflow.workflow.cs new file mode 100644 index 0000000..278d397 --- /dev/null +++ b/src/Batch/HeartbeatingActivity/HeartbeatingActivityBatchWorkflow.workflow.cs @@ -0,0 +1,35 @@ +using Temporalio.Workflows; + +namespace TemporalioSamples.Batch.HeartbeatingActivity; + +/// +/// A sample implementation of processing a batch by a single activity. +/// +/// An activity can run for as long as needed. It reports that it is still alive through a +/// heartbeat. If the worker is restarted, the activity is retried after the heartbeat timeout. +/// +/// +[Workflow] +public class HeartbeatingActivityBatchWorkflow +{ + /// + /// Processes the batch of records. + /// + /// Total number of processed records. + [WorkflowRun] + public async Task RunAsync() + { + // No special logic needed here, as the activity is retried automatically by the service. + // + // The start-to-close timeout is set to a high value to support large batch sizes. + // A heartbeat timeout is required to quickly restart the activity in case of failures, + // and to record heartbeat details at the service. + return await Workflow.ExecuteActivityAsync( + () => RecordProcessorActivities.ProcessRecordsAsync(), + new() + { + StartToCloseTimeout = TimeSpan.FromHours(1), + HeartbeatTimeout = TimeSpan.FromSeconds(10), + }); + } +} diff --git a/src/Batch/HeartbeatingActivity/Program.cs b/src/Batch/HeartbeatingActivity/Program.cs new file mode 100644 index 0000000..424208c --- /dev/null +++ b/src/Batch/HeartbeatingActivity/Program.cs @@ -0,0 +1,65 @@ +using Microsoft.Extensions.Logging; +using Temporalio.Client; +using Temporalio.Common.EnvConfig; +using Temporalio.Worker; +using TemporalioSamples.Batch.HeartbeatingActivity; + +const string TaskQueue = "HeartbeatingActivityBatch"; + +var connectOptions = ClientEnvConfig.LoadClientConnectOptions(); +connectOptions.TargetHost ??= "localhost:7233"; +connectOptions.LoggerFactory = LoggerFactory.Create(builder => + builder.AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").SetMinimumLevel(LogLevel.Information)); +var client = await TemporalClient.ConnectAsync(connectOptions); + +async Task RunWorkerAsync() +{ + // Cancellation token cancelled on ctrl+c + using var tokenSource = new CancellationTokenSource(); + Console.CancelKeyPress += (_, eventArgs) => + { + tokenSource.Cancel(); + eventArgs.Cancel = true; + }; + + // Run worker until cancelled. Restart it while the batch is executing to see how the + // activity timeout and retry work. + Console.WriteLine("Running worker"); + using var worker = new TemporalWorker( + client, + new TemporalWorkerOptions(taskQueue: TaskQueue) + .AddAllActivities(typeof(RecordProcessorActivities), null) + .AddWorkflow()); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + Console.WriteLine("Worker cancelled"); + } +} + +async Task ExecuteWorkflowAsync() +{ + var workflowId = "heartbeating-activity-batch-" + Guid.NewGuid(); + Console.WriteLine($"Starting batch workflow with id '{workflowId}'."); + + var handle = await client.StartWorkflowAsync( + (HeartbeatingActivityBatchWorkflow wf) => wf.RunAsync(), + new(workflowId, TaskQueue)); + + Console.WriteLine($"Started batch workflow. WorkflowId={handle.Id}, RunId={handle.ResultRunId}"); +} + +switch (args.ElementAtOrDefault(0)) +{ + case "worker": + await RunWorkerAsync(); + break; + case "workflow": + await ExecuteWorkflowAsync(); + break; + default: + throw new ArgumentException("Must pass 'worker' or 'workflow' as the first argument"); +} diff --git a/src/Batch/HeartbeatingActivity/README.md b/src/Batch/HeartbeatingActivity/README.md new file mode 100644 index 0000000..b0aa089 --- /dev/null +++ b/src/Batch/HeartbeatingActivity/README.md @@ -0,0 +1,23 @@ +# Heartbeating Activity Batch + +A sample implementation of processing a batch by an Activity. + +An Activity can run as long as needed. It reports that it is still alive through heartbeat. + +If the worker is restarted, the Activity is retried after the heartbeat timeout. + +Temporal allows storing data in heartbeat _details_. These details are available to the next +Activity attempt. The progress of the record processing is stored in the details to avoid +reprocessing records from the beginning on failures. + +To run, first see [README.md](../../../README.md) for prerequisites. Then, run the following from +this directory in a separate terminal to start the worker. Restart the worker while the batch is +executing to see how the activity timeout and retry work. + + dotnet run worker + +Then in another terminal, run the workflow from this directory: + + dotnet run workflow + +This will show logs in the worker window of the workflow running. diff --git a/src/Batch/HeartbeatingActivity/RecordProcessorActivities.cs b/src/Batch/HeartbeatingActivity/RecordProcessorActivities.cs new file mode 100644 index 0000000..44105c3 --- /dev/null +++ b/src/Batch/HeartbeatingActivity/RecordProcessorActivities.cs @@ -0,0 +1,78 @@ +using Microsoft.Extensions.Logging; +using Temporalio.Activities; + +namespace TemporalioSamples.Batch.HeartbeatingActivity; + +/// +/// Activity that processes a whole batch of records. +/// +/// It relies on a fake record loader to iterate over the set of records and process them +/// one by one. The heartbeat is used to remember the offset. On activity retry, the data from +/// the last recorded heartbeat is used to minimize the number of records that are reprocessed. +/// Note that not every heartbeat call is sent to the service; the frequency depends on the +/// heartbeat timeout the activity was scheduled with. If no heartbeat timeout is set, no +/// heartbeat is ever sent to the service. +/// +/// The biggest advantage of this approach is efficiency: it uses very few Temporal +/// resources. The biggest limitation is that it cannot deal with individual record processing +/// failures. The only options are either failing the whole batch or skipping the record. While +/// it is possible to build additional logic to record failed records somewhere, the experience +/// is not seamless. +/// +public static class RecordProcessorActivities +{ + // The sample always has 1000 records. The real application would iterate over an existing + // dataset or file. + private const int RecordCount = 1000; + + /// + /// Processes all records in the dataset. + /// + /// The number of records processed. + [Activity] + public static async Task ProcessRecordsAsync() + { + var context = ActivityExecutionContext.Current; + + // On activity retry, load the last reported offset from the heartbeat details. + var offset = context.Info.HeartbeatDetails.Count > 0 + ? await context.Info.HeartbeatDetailAtAsync(0) + : 0; + context.Logger.LogInformation("Activity ProcessRecordsAsync started with offset={Offset}", offset); + + // This sample implementation processes records one by one. If needed, it can be changed + // to use a pool of tasks to process multiple records in parallel. + while (true) + { + var record = GetRecord(offset); + if (record is null) + { + return offset; + } + + await ProcessRecordAsync(record, context.CancellationToken); + + // Report that the activity is still alive. The assumption is that each record takes + // less time to process than the heartbeat timeout. Leverage heartbeat details to + // record the offset. + context.Heartbeat(offset); + offset++; + } + } + + /// + /// Returns the record at the given offset, or null if the offset exceeds the dataset + /// size. + /// + private static SingleRecord? GetRecord(int offset) => + offset < RecordCount ? new SingleRecord(offset) : null; + + /// + /// Fake record processing logic. + /// + private static async Task ProcessRecordAsync(SingleRecord record, CancellationToken cancellationToken) + { + await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); + ActivityExecutionContext.Current.Logger.LogInformation("Processed {Record}", record); + } +} diff --git a/src/Batch/HeartbeatingActivity/SingleRecord.cs b/src/Batch/HeartbeatingActivity/SingleRecord.cs new file mode 100644 index 0000000..f0b98ef --- /dev/null +++ b/src/Batch/HeartbeatingActivity/SingleRecord.cs @@ -0,0 +1,7 @@ +namespace TemporalioSamples.Batch.HeartbeatingActivity; + +/// +/// Record to process. A real application would add use-case-specific data. +/// +/// Id of the record. +public record SingleRecord(int Id); diff --git a/src/Batch/HeartbeatingActivity/TemporalioSamples.Batch.HeartbeatingActivity.csproj b/src/Batch/HeartbeatingActivity/TemporalioSamples.Batch.HeartbeatingActivity.csproj new file mode 100644 index 0000000..52e6553 --- /dev/null +++ b/src/Batch/HeartbeatingActivity/TemporalioSamples.Batch.HeartbeatingActivity.csproj @@ -0,0 +1,7 @@ + + + + Exe + + + diff --git a/src/Batch/Iterator/IteratorBatchWorkflow.workflow.cs b/src/Batch/Iterator/IteratorBatchWorkflow.workflow.cs new file mode 100644 index 0000000..ecb675c --- /dev/null +++ b/src/Batch/Iterator/IteratorBatchWorkflow.workflow.cs @@ -0,0 +1,59 @@ +using Temporalio.Workflows; + +namespace TemporalioSamples.Batch.Iterator; + +/// +/// Implements the iterator workflow pattern. +/// +/// A single workflow run processes a single page of records in parallel. Each record is +/// processed using its own child workflow. +/// +/// After all child workflows complete, a new run of the parent workflow is created using +/// continue-as-new. The new run processes the next page of records. This way a practically +/// unlimited set of records can be processed. +/// +[Workflow] +public class IteratorBatchWorkflow +{ + /// + /// Processes the batch of records. + /// + /// Number of records to process in a single workflow run. + /// Offset of the first record to process. 0 to start the batch + /// processing. + /// Total number of processed records. + [WorkflowRun] + public async Task RunAsync(int pageSize, int offset) + { + // Loads a page of records. + var records = await Workflow.ExecuteActivityAsync( + () => RecordLoaderActivities.GetRecords(pageSize, offset), + new() { StartToCloseTimeout = TimeSpan.FromSeconds(5) }); + + // Starts a child workflow per record asynchronously. + var results = records.Select(record => + { + // Uses a human-friendly child id. + var childId = $"{Workflow.Info.WorkflowId}/{record.Id}"; + return Workflow.ExecuteChildWorkflowAsync( + (RecordProcessorWorkflow wf) => wf.RunAsync(record), + new() { Id = childId }); + }).ToList(); + + // Waits for all children to complete. + await Workflow.WhenAllAsync(results); + + // Skips error handling for the sample's brevity, so failed RecordProcessorWorkflows are + // ignored. + + // No more records in the dataset. Completes the workflow. + if (records.Count == 0) + { + return offset; + } + + // Continues-as-new with the increased offset. + throw Workflow.CreateContinueAsNewException( + (IteratorBatchWorkflow wf) => wf.RunAsync(pageSize, offset + records.Count)); + } +} diff --git a/src/Batch/Iterator/Program.cs b/src/Batch/Iterator/Program.cs new file mode 100644 index 0000000..edd5e81 --- /dev/null +++ b/src/Batch/Iterator/Program.cs @@ -0,0 +1,66 @@ +using Microsoft.Extensions.Logging; +using Temporalio.Client; +using Temporalio.Common.EnvConfig; +using Temporalio.Worker; +using TemporalioSamples.Batch.Iterator; + +const string TaskQueue = "IteratorBatch"; + +var connectOptions = ClientEnvConfig.LoadClientConnectOptions(); +connectOptions.TargetHost ??= "localhost:7233"; +connectOptions.LoggerFactory = LoggerFactory.Create(builder => + builder.AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").SetMinimumLevel(LogLevel.Information)); +var client = await TemporalClient.ConnectAsync(connectOptions); + +async Task RunWorkerAsync() +{ + // Cancellation token cancelled on ctrl+c + using var tokenSource = new CancellationTokenSource(); + Console.CancelKeyPress += (_, eventArgs) => + { + tokenSource.Cancel(); + eventArgs.Cancel = true; + }; + + // Run worker until cancelled. Hosts both the IteratorBatchWorkflow and the + // RecordProcessorWorkflow it starts as children, plus the RecordLoaderActivities activity. + Console.WriteLine("Running worker"); + using var worker = new TemporalWorker( + client, + new TemporalWorkerOptions(taskQueue: TaskQueue) + .AddAllActivities(typeof(RecordLoaderActivities), null) + .AddWorkflow() + .AddWorkflow()); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + Console.WriteLine("Worker cancelled"); + } +} + +async Task ExecuteWorkflowAsync() +{ + var workflowId = "iterator-batch-" + Guid.NewGuid(); + Console.WriteLine($"Starting batch workflow with id '{workflowId}'."); + + var handle = await client.StartWorkflowAsync( + (IteratorBatchWorkflow wf) => wf.RunAsync(5, 0), + new(workflowId, TaskQueue)); + + Console.WriteLine($"Started batch workflow. WorkflowId={handle.Id}, RunId={handle.ResultRunId}"); +} + +switch (args.ElementAtOrDefault(0)) +{ + case "worker": + await RunWorkerAsync(); + break; + case "workflow": + await ExecuteWorkflowAsync(); + break; + default: + throw new ArgumentException("Must pass 'worker' or 'workflow' as the first argument"); +} diff --git a/src/Batch/Iterator/README.md b/src/Batch/Iterator/README.md new file mode 100644 index 0000000..aa33d88 --- /dev/null +++ b/src/Batch/Iterator/README.md @@ -0,0 +1,24 @@ +# Iterator Batch + +A sample implementation of the Workflow iterator pattern. + +A Workflow starts a configured number of child Workflows in parallel. Each child processes a +single record. After all children complete, the parent calls continue-as-new and starts the +children for the next page of records. + +This allows processing a set of records of any size. The advantage of this approach is +simplicity. The main disadvantage is that it processes records in batches, with each batch +waiting for the slowest child Workflow. + +A variation of this pattern runs Activities instead of child Workflows. + +To run, first see [README.md](../../../README.md) for prerequisites. Then, run the following from +this directory in a separate terminal to start the worker: + + dotnet run worker + +Then in another terminal, run the workflow from this directory: + + dotnet run workflow + +This will show logs in the worker window of the workflow running. diff --git a/src/Batch/Iterator/RecordLoaderActivities.cs b/src/Batch/Iterator/RecordLoaderActivities.cs new file mode 100644 index 0000000..66e5536 --- /dev/null +++ b/src/Batch/Iterator/RecordLoaderActivities.cs @@ -0,0 +1,38 @@ +using Temporalio.Activities; + +namespace TemporalioSamples.Batch.Iterator; + +/// +/// Activity used to iterate over a list of records. +/// +/// A specific implementation depends on a use case. For example, it can execute a SQL DB +/// query or read a comma delimited file. More complex use cases would need passing a different +/// type of offset parameter. +/// +public static class RecordLoaderActivities +{ + // The sample always returns 5 pages. The real application would iterate over an existing + // dataset or file. + private const int PageCount = 5; + + /// + /// Returns the next page of records. + /// + /// Maximum number of records to return. + /// Offset of the next page. + /// Empty list if there are no more records to process. + [Activity] + public static IReadOnlyList GetRecords(int pageSize, int offset) + { + var records = new List(pageSize); + if (offset < pageSize * PageCount) + { + for (var i = 0; i < pageSize; i++) + { + records.Add(new SingleRecord(offset + i)); + } + } + + return records; + } +} diff --git a/src/Batch/Iterator/RecordProcessorWorkflow.workflow.cs b/src/Batch/Iterator/RecordProcessorWorkflow.workflow.cs new file mode 100644 index 0000000..12cd5c9 --- /dev/null +++ b/src/Batch/Iterator/RecordProcessorWorkflow.workflow.cs @@ -0,0 +1,23 @@ +using Microsoft.Extensions.Logging; +using Temporalio.Workflows; + +namespace TemporalioSamples.Batch.Iterator; + +/// +/// Fake workflow that implements processing of a single record. +/// +[Workflow] +public class RecordProcessorWorkflow +{ + /// + /// Processes a single record. + /// + /// Record to process. + [WorkflowRun] + public async Task RunAsync(SingleRecord record) + { + // Simulate some processing. + await Workflow.DelayAsync(TimeSpan.FromSeconds(Workflow.Random.Next(30))); + Workflow.Logger.LogInformation("Processed {Record}", record); + } +} diff --git a/src/Batch/Iterator/SingleRecord.cs b/src/Batch/Iterator/SingleRecord.cs new file mode 100644 index 0000000..c79c0c0 --- /dev/null +++ b/src/Batch/Iterator/SingleRecord.cs @@ -0,0 +1,7 @@ +namespace TemporalioSamples.Batch.Iterator; + +/// +/// Record to process. A real application would add use-case-specific data. +/// +/// Id of the record. +public record SingleRecord(int Id); diff --git a/src/Batch/Iterator/TemporalioSamples.Batch.Iterator.csproj b/src/Batch/Iterator/TemporalioSamples.Batch.Iterator.csproj new file mode 100644 index 0000000..52e6553 --- /dev/null +++ b/src/Batch/Iterator/TemporalioSamples.Batch.Iterator.csproj @@ -0,0 +1,7 @@ + + + + Exe + + + diff --git a/src/Batch/README.md b/src/Batch/README.md new file mode 100644 index 0000000..696225e --- /dev/null +++ b/src/Batch/README.md @@ -0,0 +1,16 @@ +# Batch + +These samples show three different best practices for processing a batch of records with a +Temporal Workflow, each trading off simplicity, efficiency, and failure handling differently. + +1. [Iterator](./Iterator/README.md) - Processes a page of records at a time using child + workflows, continuing-as-new between pages. +2. [Heartbeating Activity](./HeartbeatingActivity/README.md) - Processes the whole batch inside a + single heartbeating Activity. +3. [Sliding Window](./SlidingWindow/README.md) - Keeps a fixed number of record processing child + workflows running in parallel at all times, using Signals and continue-as-new to keep the + Workflow history size bounded indefinitely. + +These are .NET ports of the +[Java batch samples](https://github.com/temporalio/samples-java/tree/main/core/src/main/java/io/temporal/samples/batch), +kept as close as possible to the original Java structure and naming. diff --git a/src/Batch/SlidingWindow/BatchProgress.cs b/src/Batch/SlidingWindow/BatchProgress.cs new file mode 100644 index 0000000..ffe2598 --- /dev/null +++ b/src/Batch/SlidingWindow/BatchProgress.cs @@ -0,0 +1,9 @@ +namespace TemporalioSamples.Batch.SlidingWindow; + +/// +/// Result of the query. +/// +/// Count of completed record processing child workflows. +/// Ids of records that are currently being processed by child +/// workflows. +public record BatchProgress(int Progress, IReadOnlySet CurrentRecords); diff --git a/src/Batch/SlidingWindow/BatchWorkflow.workflow.cs b/src/Batch/SlidingWindow/BatchWorkflow.workflow.cs new file mode 100644 index 0000000..1667d59 --- /dev/null +++ b/src/Batch/SlidingWindow/BatchWorkflow.workflow.cs @@ -0,0 +1,60 @@ +using Temporalio.Workflows; + +namespace TemporalioSamples.Batch.SlidingWindow; + +/// +/// Implements batch processing by running multiple +/// instances in parallel. +/// +[Workflow] +public class BatchWorkflow +{ + /// + /// Processes a batch of records using multiple parallel sliding window workflows. + /// + /// Number of records to start processing in a single sliding window + /// workflow run. + /// Number of records to process in parallel by a single + /// sliding window workflow. Can be larger than . + /// Number of SlidingWindowBatchWorkflows to run in parallel. If the + /// number of partitions is too low, the update rate of a single SlidingWindowBatchWorkflow + /// can get too high. + /// Total number of processed records. + [WorkflowRun] + public async Task RunAsync(int pageSize, int slidingWindowSize, int partitions) + { + // The sample partitions the dataset into contiguous ranges. A real application can + // choose any other way to divide the records into multiple collections. + var totalCount = await Workflow.ExecuteActivityAsync( + () => RecordLoaderActivities.GetRecordCount(), + new() { StartToCloseTimeout = TimeSpan.FromSeconds(5) }); + + var partitionSize = (totalCount / partitions) + (totalCount % partitions > 0 ? 1 : 0); + + var partitionResults = new List>(partitions); + for (var i = 0; i < partitions; i++) + { + // Makes the child id more user-friendly. + var childId = $"{Workflow.Info.WorkflowId}/{i}"; + + // Defines the partition boundaries. + var offset = partitionSize * i; + var maximumOffset = Math.Min(offset + partitionSize, totalCount); + + var input = new ProcessBatchInput + { + PageSize = pageSize, + SlidingWindowSize = slidingWindowSize, + Offset = offset, + MaximumOffset = maximumOffset, + }; + + partitionResults.Add(Workflow.ExecuteChildWorkflowAsync( + (SlidingWindowBatchWorkflow wf) => wf.RunAsync(input), + new() { Id = childId })); + } + + var results = await Workflow.WhenAllAsync(partitionResults); + return results.Sum(); + } +} diff --git a/src/Batch/SlidingWindow/ProcessBatchInput.cs b/src/Batch/SlidingWindow/ProcessBatchInput.cs new file mode 100644 index 0000000..4ce788b --- /dev/null +++ b/src/Batch/SlidingWindow/ProcessBatchInput.cs @@ -0,0 +1,49 @@ +namespace TemporalioSamples.Batch.SlidingWindow; + +/// +/// Input of . +/// +public record ProcessBatchInput +{ + /// + /// Gets the number of records to load in a single call. + /// + required public int PageSize { get; init; } + + /// + /// Gets the number of parallel record processing child workflows to run. + /// + required public int SlidingWindowSize { get; init; } + + /// + /// Gets the index of the first record to process. 0 to start the batch processing. + /// + required public int Offset { get; init; } + + /// + /// Gets the maximum offset (exclusive) to process by this workflow. + /// + required public int MaximumOffset { get; init; } + + /// + /// Gets the total number of records processed so far by this workflow. + /// + public int Progress { get; init; } + + /// + /// Gets the ids of records that are currently being processed by child workflows. + /// + /// This puts a limit on the sliding window size, as workflow arguments cannot exceed + /// 2MB in JSON format. Another practical limit is the number of signals a workflow can + /// handle per second. Adjust the number of partitions to keep this rate at a reasonable + /// value. + /// + public HashSet CurrentRecords { get; init; } = new(); + + /// + /// Gets a value indicating whether continue-as-new should be forced well before the real + /// history-size threshold would suggest it, so tests can deterministically exercise that + /// code path without generating a large amount of real history. + /// + public bool TestContinueAsNew { get; init; } +} diff --git a/src/Batch/SlidingWindow/Program.cs b/src/Batch/SlidingWindow/Program.cs new file mode 100644 index 0000000..4ddd32c --- /dev/null +++ b/src/Batch/SlidingWindow/Program.cs @@ -0,0 +1,67 @@ +using Microsoft.Extensions.Logging; +using Temporalio.Client; +using Temporalio.Common.EnvConfig; +using Temporalio.Worker; +using TemporalioSamples.Batch.SlidingWindow; + +const string TaskQueue = "SlidingWindow"; + +var connectOptions = ClientEnvConfig.LoadClientConnectOptions(); +connectOptions.TargetHost ??= "localhost:7233"; +connectOptions.LoggerFactory = LoggerFactory.Create(builder => + builder.AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").SetMinimumLevel(LogLevel.Information)); +var client = await TemporalClient.ConnectAsync(connectOptions); + +async Task RunWorkerAsync() +{ + // Cancellation token cancelled on ctrl+c + using var tokenSource = new CancellationTokenSource(); + Console.CancelKeyPress += (_, eventArgs) => + { + tokenSource.Cancel(); + eventArgs.Cancel = true; + }; + + // Run worker until cancelled. Hosts BatchWorkflow, SlidingWindowBatchWorkflow and + // RecordProcessorWorkflow, plus the RecordLoaderActivities activity. + Console.WriteLine("Running worker"); + using var worker = new TemporalWorker( + client, + new TemporalWorkerOptions(taskQueue: TaskQueue) + .AddAllActivities(typeof(RecordLoaderActivities), null) + .AddWorkflow() + .AddWorkflow() + .AddWorkflow()); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + Console.WriteLine("Worker cancelled"); + } +} + +async Task ExecuteWorkflowAsync() +{ + var workflowId = "sliding-window-batch-" + Guid.NewGuid(); + Console.WriteLine($"Starting batch workflow with id '{workflowId}'."); + + var handle = await client.StartWorkflowAsync( + (BatchWorkflow wf) => wf.RunAsync(10, 25, 3), + new(workflowId, TaskQueue)); + + Console.WriteLine($"Started batch workflow with 3 partitions. WorkflowId={handle.Id}, RunId={handle.ResultRunId}"); +} + +switch (args.ElementAtOrDefault(0)) +{ + case "worker": + await RunWorkerAsync(); + break; + case "workflow": + await ExecuteWorkflowAsync(); + break; + default: + throw new ArgumentException("Must pass 'worker' or 'workflow' as the first argument"); +} diff --git a/src/Batch/SlidingWindow/README.md b/src/Batch/SlidingWindow/README.md new file mode 100644 index 0000000..4f81f7d --- /dev/null +++ b/src/Batch/SlidingWindow/README.md @@ -0,0 +1,42 @@ +# Sliding Window Batch + +A sample implementation of a batch processing Workflow that maintains a sliding window of record +processing Workflows. + +A Workflow starts a configured number of child Workflows in parallel. Each child processes a +single record. When a child completes, a new child is immediately started. + +A parent Workflow calls continue-as-new after starting a preconfigured number of children. A +child completion is reported through a Signal, as a parent cannot directly wait for a child that +was started by a previous run. + +Multiple instances of `SlidingWindowBatchWorkflow` run in parallel, each processing a subset of +records, to support a higher total rate of processing. + +This is the sample that demonstrates the workaround for the race between continue-as-new and a +child Workflow's completion Signal: a Signal can be delivered to the new run before its Workflow +run method has had a chance to restore state from the continue-as-new input. See the comments on +`SlidingWindowBatchWorkflow.recordsToRemove` for the details. + +The Workflow also continues-as-new early whenever `Workflow.ContinueAsNewSuggested` is true, not +just after starting a fixed `pageSize` number of children. A fixed count alone is fragile: a run +that accumulates a lot of other history (for example, many completion Signals) can outgrow the +suggested history size before it reaches `pageSize` children. See the comments where +`Workflow.ContinueAsNewSuggested` is checked in `SlidingWindowBatchWorkflow.RunAsync` for details. + +To run, first see [README.md](../../../README.md) for prerequisites. Then, run the following from +this directory in a separate terminal to start the worker: + + dotnet run worker + +Note that `UnhandledCommand` info messages in the worker output are expected and benign. They +ensure that Signals are not lost when there is a race condition between the workflow calling +continue-as-new and receiving a Signal. If these messages appear too frequently, consider +increasing the number of partitions passed to `BatchWorkflow.RunAsync`. + +Then in another terminal, run the workflow from this directory. Each time the command runs, it +starts a new `BatchWorkflow` execution with 3 partitions. + + dotnet run workflow + +This will show logs in the worker window of the workflow running. diff --git a/src/Batch/SlidingWindow/RecordLoaderActivities.cs b/src/Batch/SlidingWindow/RecordLoaderActivities.cs new file mode 100644 index 0000000..c9bc28f --- /dev/null +++ b/src/Batch/SlidingWindow/RecordLoaderActivities.cs @@ -0,0 +1,44 @@ +using System.Diagnostics.CodeAnalysis; +using Temporalio.Activities; + +namespace TemporalioSamples.Batch.SlidingWindow; + +/// +/// Fake activities used to iterate over a list of records. The real application would iterate +/// over an existing dataset or file. +/// +public static class RecordLoaderActivities +{ + private const int TotalCount = 300; + + /// + /// Returns the next page of records. + /// + /// Maximum number of records to return. + /// Offset of the next page. + /// Empty list if there are no more records to process. + [Activity] + public static IReadOnlyList GetRecords(int pageSize, int offset) + { + var records = new List(pageSize); + if (offset < TotalCount) + { + for (var i = offset; i < Math.Min(offset + pageSize, TotalCount); i++) + { + records.Add(new SingleRecord(i)); + } + } + + return records; + } + + /// + /// Returns the total record count. + /// + /// Used to divide record ranges among partitions. Some applications might choose a + /// completely different approach for partitioning the dataset. + /// + [Activity] + [SuppressMessage("Design", "CA1024:Use properties where appropriate", Justification = "Activities must be methods.")] + public static int GetRecordCount() => TotalCount; +} diff --git a/src/Batch/SlidingWindow/RecordProcessorWorkflow.workflow.cs b/src/Batch/SlidingWindow/RecordProcessorWorkflow.workflow.cs new file mode 100644 index 0000000..3388040 --- /dev/null +++ b/src/Batch/SlidingWindow/RecordProcessorWorkflow.workflow.cs @@ -0,0 +1,35 @@ +using Microsoft.Extensions.Logging; +using Temporalio.Workflows; + +namespace TemporalioSamples.Batch.SlidingWindow; + +/// +/// Fake workflow that implements processing of a single record. Must report completion to a +/// parent through . +/// +[Workflow] +public class RecordProcessorWorkflow +{ + /// + /// Processes a single record. + /// + /// Record to process. + [WorkflowRun] + public async Task RunAsync(SingleRecord record) + { + // Application-specific record processing logic goes here. + await Workflow.DelayAsync(TimeSpan.FromSeconds(Workflow.Random.Next(10))); + Workflow.Logger.LogInformation("Processed {Record}", record); + + // This workflow is always expected to have a parent. But for testing it might be useful + // to skip the notification when run standalone. + var parentId = Workflow.Info.Parent?.WorkflowId; + if (parentId is not null) + { + var parent = Workflow.GetExternalWorkflowHandle(parentId); + + // Notify the parent about record processing completion. + await parent.SignalAsync(wf => wf.ReportCompletionAsync(record.Id)); + } + } +} diff --git a/src/Batch/SlidingWindow/SingleRecord.cs b/src/Batch/SlidingWindow/SingleRecord.cs new file mode 100644 index 0000000..a664c14 --- /dev/null +++ b/src/Batch/SlidingWindow/SingleRecord.cs @@ -0,0 +1,7 @@ +namespace TemporalioSamples.Batch.SlidingWindow; + +/// +/// Record to process. +/// +/// Id of the record. +public record SingleRecord(int Id); diff --git a/src/Batch/SlidingWindow/SlidingWindowBatchWorkflow.workflow.cs b/src/Batch/SlidingWindow/SlidingWindowBatchWorkflow.workflow.cs new file mode 100644 index 0000000..a0dbd5f --- /dev/null +++ b/src/Batch/SlidingWindow/SlidingWindowBatchWorkflow.workflow.cs @@ -0,0 +1,219 @@ +using Temporalio.Workflows; + +namespace TemporalioSamples.Batch.SlidingWindow; + +/// +/// Implements batch processing by running a specified number of record processing child +/// workflows in parallel. A new record processing child workflow is started when a previously +/// started one completes. Child completion is reported through the +/// signal, since it is not possible to passively wait for a +/// child that was started by a previous continue-as-new run. +/// +/// Calls continue-as-new after starting PageSize children. Note that the sliding +/// window size can be larger than PageSize. +/// +[Workflow] +public class SlidingWindowBatchWorkflow +{ + /// + /// Accumulates records to remove for signals delivered before starts + /// executing. This is the workaround for the race between continue-as-new and a child + /// reporting completion: a signal can be delivered to the new run before its workflow run + /// method has had a chance to populate from the input. + /// + private readonly HashSet recordsToRemove = new(); + + /// + /// Ids of records that are being processed by child workflows. Null until the workflow run + /// method has initialized it from the input, which lets + /// detect signals that arrive before that point. + /// + private HashSet? currentRecords; + + /// + /// Count of completed record processing child workflows. + /// + private int progress; + + /// + /// Processes the batch of records. + /// + /// Defines the range of records to process and, on continue-as-new, the + /// progress carried over from the previous run. + /// Total number of processed records. + [WorkflowRun] + public async Task RunAsync(ProcessBatchInput input) + { + progress = input.Progress; + + // Alias as a non-nullable local so it can be captured by the WaitConditionAsync + // lambdas below. It is the same set instance as the currentRecords field, so mutations + // through either reference are visible to both. + var trackedRecords = input.CurrentRecords; + currentRecords = trackedRecords; + + // Remove records for signals delivered before this run started. + var countBefore = trackedRecords.Count; + trackedRecords.ExceptWith(recordsToRemove); + progress += countBefore - trackedRecords.Count; + + var pageSize = input.PageSize; + var offset = input.Offset; + var slidingWindowSize = input.SlidingWindowSize; + + // For ease of testing, forces continue-as-new well before the real history-size + // threshold would suggest it, instead of relying on Workflow.ContinueAsNewSuggested + // alone. Mirrors ClusterManagerWorkflow's maxHistoryLength in the SafeMessageHandlers + // sample. Each child adds several history events of its own (start command, started + // event, completion Signal), so this needs to be well above that per-child footprint or + // a run ends up continuing-as-new after only one or two children. + var maxHistoryLength = input.TestContinueAsNew ? 100 : int.MaxValue; + + var pager = new RecordPager(input.PageSize, input.Offset, input.MaximumOffset); + var childrenStartedByThisRun = new List>>(); + + while (true) + { + // After starting slidingWindowSize children, blocks until a completion signal is + // received. + await Workflow.WaitConditionAsync(() => trackedRecords.Count < slidingWindowSize); + + var record = await pager.NextAsync(); + + // Completes the workflow if there are no more records to process. + if (record is null) + { + // Awaits for all children to complete. By this point every record that was ever + // added to trackedRecords has been removed via ReportCompletionAsync, so progress + // is an accurate count of the records processed by this partition. + await Workflow.WaitConditionAsync(() => trackedRecords.Count == 0); + return progress; + } + + // Uses ParentClosePolicy.Abandon to ensure that children survive continue-as-new of + // the parent. Assigns a user-friendly child workflow id. + var childOptions = new ChildWorkflowOptions + { + Id = $"{Workflow.Info.WorkflowId}/{record.Id}", + ParentClosePolicy = ParentClosePolicy.Abandon, + }; + + // Starts a child workflow asynchronously, ignoring its result. The assumption is + // that the parent doesn't need to deal with child workflow results and failures. + // Another assumption is that a child in any situation calls the + // ReportCompletionAsync signal. + var childStarted = Workflow.StartChildWorkflowAsync( + (RecordProcessorWorkflow wf) => wf.RunAsync(record), + childOptions); + childrenStartedByThisRun.Add(childStarted); + trackedRecords.Add(record.Id); + + // Continues-as-new after starting pageSize children, or earlier if the server + // suggests it (or, for ease of testing, maxHistoryLength is exceeded) because + // history has grown large. Relying on pageSize alone is fragile: if a run also + // accumulates a lot of other history (e.g. many completion Signals), it can outgrow + // the suggested size before reaching pageSize children. + if (childrenStartedByThisRun.Count == pageSize || + Workflow.ContinueAsNewSuggested || + Workflow.CurrentHistoryLength > maxHistoryLength) + { + // Waits for all children to start. Without this wait, workflow completion + // through continue-as-new might lead to a situation where they never start. + // Assumes that they never fail to start, as their automatically generated ids + // are not expected to collide. + await Workflow.WhenAllAsync(childrenStartedByThisRun); + + var newInput = new ProcessBatchInput + { + PageSize = pageSize, + SlidingWindowSize = slidingWindowSize, + Offset = offset + childrenStartedByThisRun.Count, + MaximumOffset = input.MaximumOffset, + Progress = progress, + CurrentRecords = trackedRecords, + TestContinueAsNew = input.TestContinueAsNew, + }; + + throw Workflow.CreateContinueAsNewException( + (SlidingWindowBatchWorkflow wf) => wf.RunAsync(newInput)); + } + } + } + + /// + /// Reports that a child workflow finished processing a record. + /// + /// Id of the record that finished processing. + [WorkflowSignal] + public Task ReportCompletionAsync(int recordId) + { + // Handles the case when the signal is delivered before the workflow run method started. + if (currentRecords is null) + { + recordsToRemove.Add(recordId); + return Task.CompletedTask; + } + + // Dedupes signals, as in some edge cases they can be delivered more than once. + if (currentRecords.Remove(recordId)) + { + progress++; + } + + return Task.CompletedTask; + } + + /// + /// Returns the current progress of the batch. + /// + [WorkflowQuery] + public BatchProgress GetProgress() => new(progress, currentRecords ?? new HashSet()); + + /// + /// Lazily loads records page by page using the activity. + /// + private sealed class RecordPager + { + private readonly int pageSize; + private readonly int maximumOffset; + private IReadOnlyList lastPage = Array.Empty(); + private int offset; + private int index; + + public RecordPager(int pageSize, int initialOffset, int maximumOffset) + { + this.pageSize = pageSize; + this.maximumOffset = maximumOffset; + offset = initialOffset; + } + + /// + /// Returns the next record, loading a new page via activity if the current one is + /// exhausted. Returns null once the maximum offset has been reached. + /// + public async Task NextAsync() + { + if (index >= lastPage.Count) + { + if (offset >= maximumOffset) + { + return null; + } + + var size = Math.Min(pageSize, maximumOffset - offset); + lastPage = await Workflow.ExecuteActivityAsync( + () => RecordLoaderActivities.GetRecords(size, offset), + new() { StartToCloseTimeout = TimeSpan.FromSeconds(5) }); + offset += lastPage.Count; + index = 0; + + if (lastPage.Count == 0) + { + return null; + } + } + + return lastPage[index++]; + } + } +} diff --git a/src/Batch/SlidingWindow/TemporalioSamples.Batch.SlidingWindow.csproj b/src/Batch/SlidingWindow/TemporalioSamples.Batch.SlidingWindow.csproj new file mode 100644 index 0000000..52e6553 --- /dev/null +++ b/src/Batch/SlidingWindow/TemporalioSamples.Batch.SlidingWindow.csproj @@ -0,0 +1,7 @@ + + + + Exe + + + diff --git a/tests/Batch/HeartbeatingActivity/RecordProcessorActivitiesTests.cs b/tests/Batch/HeartbeatingActivity/RecordProcessorActivitiesTests.cs new file mode 100644 index 0000000..f7b7581 --- /dev/null +++ b/tests/Batch/HeartbeatingActivity/RecordProcessorActivitiesTests.cs @@ -0,0 +1,40 @@ +namespace TemporalioSamples.Tests.Batch.HeartbeatingActivity; + +using Temporalio.Converters; +using Temporalio.Testing; +using TemporalioSamples.Batch.HeartbeatingActivity; +using Xunit; +using Xunit.Abstractions; + +public class RecordProcessorActivitiesTests : TestBase +{ + public RecordProcessorActivitiesTests(ITestOutputHelper output) + : base(output) + { + } + + [Fact] + public async Task ProcessRecordsAsync_HeartbeatDetailPresent_ResumesInsteadOfStartingOver() + { + // Seeds the activity context with heartbeat details as if this were a retry after a + // worker restart that got as far as record 995 before dying. + var heartbeats = new List(); + var env = new ActivityEnvironment + { + Info = ActivityEnvironment.DefaultInfo with + { + HeartbeatDetails = new[] { DataConverter.Default.PayloadConverter.ToPayload(995) }, + }, + Heartbeater = details => heartbeats.Add(details), + }; + + var result = await env.RunAsync(() => RecordProcessorActivities.ProcessRecordsAsync()); + + Assert.Equal(1000, result); + + // Resumes from the seeded heartbeat detail rather than reprocessing records 0-994. + Assert.Equal(5, heartbeats.Count); + Assert.Equal(995, heartbeats[0][0]); + Assert.Equal(999, heartbeats[^1][0]); + } +} diff --git a/tests/Batch/Iterator/IteratorBatchWorkflowTests.cs b/tests/Batch/Iterator/IteratorBatchWorkflowTests.cs new file mode 100644 index 0000000..3bffe6c --- /dev/null +++ b/tests/Batch/Iterator/IteratorBatchWorkflowTests.cs @@ -0,0 +1,43 @@ +namespace TemporalioSamples.Tests.Batch.Iterator; + +using Temporalio.Testing; +using Temporalio.Worker; +using TemporalioSamples.Batch.Iterator; +using Xunit; +using Xunit.Abstractions; + +public class IteratorBatchWorkflowTests : TestBase +{ + public IteratorBatchWorkflowTests(ITestOutputHelper output) + : base(output) + { + } + + [TimeSkippingServerFact] + public async Task RunAsync_FullDataset_ProcessesAllPagesAndContinuesAsNew() + { + await using var env = await WorkflowEnvironment.StartTimeSkippingAsync(); + using var worker = new TemporalWorker( + env.Client, + new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). + AddAllActivities(typeof(RecordLoaderActivities), null). + AddWorkflow(). + AddWorkflow()); + await worker.ExecuteAsync(async () => + { + var handle = await env.Client.StartWorkflowAsync( + (IteratorBatchWorkflow wf) => wf.RunAsync(5, 0), + new(id: $"wf-{Guid.NewGuid()}", taskQueue: worker.Options.TaskQueue!)); + + // The fake RecordLoaderActivities always returns 5 pages of 5 records each. + var result = await handle.GetResultAsync(); + Assert.Equal(25, result); + + // Confirm the workflow actually exercised continue-as-new to get there. + var firstRunHistory = await (handle with { RunId = handle.ResultRunId }).FetchHistoryAsync(); + Assert.Contains( + firstRunHistory.Events, + e => e.WorkflowExecutionContinuedAsNewEventAttributes != null); + }); + } +} diff --git a/tests/Batch/SlidingWindow/SlidingWindowBatchWorkflowTests.cs b/tests/Batch/SlidingWindow/SlidingWindowBatchWorkflowTests.cs new file mode 100644 index 0000000..6f51300 --- /dev/null +++ b/tests/Batch/SlidingWindow/SlidingWindowBatchWorkflowTests.cs @@ -0,0 +1,81 @@ +namespace TemporalioSamples.Tests.Batch.SlidingWindow; + +using Temporalio.Testing; +using Temporalio.Worker; +using TemporalioSamples.Batch.SlidingWindow; +using Xunit; +using Xunit.Abstractions; + +public class SlidingWindowBatchWorkflowTests : TestBase +{ + public SlidingWindowBatchWorkflowTests(ITestOutputHelper output) + : base(output) + { + } + + [TimeSkippingServerFact] + public async Task RunAsync_ThreePartitions_ProcessesEveryRecordExactlyOnce() + { + await using var env = await WorkflowEnvironment.StartTimeSkippingAsync(); + using var worker = new TemporalWorker( + env.Client, + new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). + AddAllActivities(typeof(RecordLoaderActivities), null). + AddWorkflow(). + AddWorkflow(). + AddWorkflow()); + await worker.ExecuteAsync(async () => + { + var handle = await env.Client.StartWorkflowAsync( + (BatchWorkflow wf) => wf.RunAsync(10, 25, 3), + new(id: $"wf-{Guid.NewGuid()}", taskQueue: worker.Options.TaskQueue!)); + + // The fake RecordLoaderActivities always has 300 total records, split evenly across + // the 3 partitions. Each partition's SlidingWindowBatchWorkflow must report exactly + // its own share back to the parent, not a cumulative or duplicated count. + var result = await handle.GetResultAsync(); + Assert.Equal(300, result); + }); + } + + [TimeSkippingServerFact] + public async Task RunAsync_TestContinueAsNewSet_ContinuesAsNewBeforePageSizeIsReached() + { + await using var env = await WorkflowEnvironment.StartTimeSkippingAsync(); + using var worker = new TemporalWorker( + env.Client, + new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). + AddAllActivities(typeof(RecordLoaderActivities), null). + AddWorkflow(). + AddWorkflow()); + await worker.ExecuteAsync(async () => + { + // Only processes a small slice of the dataset (rather than the full 300 records) so + // that even in the worst case of continuing-as-new after every single child, the run + // still finishes quickly. PageSize is far larger than that slice, so it would never + // be reached naturally: only the TestContinueAsNew-driven maxHistoryLength check can + // cause a continue-as-new here. + var input = new ProcessBatchInput + { + PageSize = 1000, + SlidingWindowSize = 10, + Offset = 0, + MaximumOffset = 30, + TestContinueAsNew = true, + }; + var handle = await env.Client.StartWorkflowAsync( + (SlidingWindowBatchWorkflow wf) => wf.RunAsync(input), + new(id: $"wf-{Guid.NewGuid()}", taskQueue: worker.Options.TaskQueue!)); + + var result = await handle.GetResultAsync(); + Assert.Equal(30, result); + + // Confirm continue-as-new actually happened, proving the maxHistoryLength branch + // fired rather than the unreachable pageSize branch. + var firstRunHistory = await (handle with { RunId = handle.ResultRunId }).FetchHistoryAsync(); + Assert.Contains( + firstRunHistory.Events, + e => e.WorkflowExecutionContinuedAsNewEventAttributes != null); + }); + } +} diff --git a/tests/TemporalioSamples.Tests.csproj b/tests/TemporalioSamples.Tests.csproj index ef31659..aca5b0c 100644 --- a/tests/TemporalioSamples.Tests.csproj +++ b/tests/TemporalioSamples.Tests.csproj @@ -21,6 +21,9 @@ + + +