wehub-resource-sync
db620d33df
CodeQL / Analyze (csharp) (push) Waiting to run
CodeQL / Analyze (python) (push) Waiting to run
dotnet-build-and-test / dotnet-test-functions (push) Has been cancelled
dotnet-build-and-test / paths-filter (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Debug, windows-latest, net9.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Release, ubuntu-latest, net10.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Release, ubuntu-latest, net8.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-build (Release, windows-latest, net472) (push) Has been cancelled
dotnet-build-and-test / dotnet-test (Release, integration, true, ubuntu-latest, net10.0) (push) Has been cancelled
dotnet-build-and-test / dotnet-test (Release, integration, true, windows-latest, net472) (push) Has been cancelled
dotnet-build-and-test / dotnet-foundry-hosted-it (push) Has been cancelled
dotnet-build-and-test / dotnet-build-and-test-check (push) Has been cancelled
dotnet-build-and-test / Integration Test Report (push) Has been cancelled
770 lines
36 KiB
C#
770 lines
36 KiB
C#
// Copyright (c) Microsoft. All rights reserved.
|
|
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FluentAssertions;
|
|
using Microsoft.Agents.AI.Workflows.InProc;
|
|
using Microsoft.Agents.AI.Workflows.Sample;
|
|
|
|
namespace Microsoft.Agents.AI.Workflows.UnitTests;
|
|
|
|
/// <summary>
|
|
/// Regression tests for GH-2485: pending <see cref="RequestInfoEvent"/> objects must be
|
|
/// re-emitted after resuming a workflow from a checkpoint.
|
|
/// </summary>
|
|
public class CheckpointResumeTests
|
|
{
|
|
/// <summary>
|
|
/// Verifies that a resumed workflow re-emits <see cref="RequestInfoEvent"/>s for
|
|
/// pending external requests that existed at the time of the checkpoint.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_WithPendingRequests_RepublishesRequestInfoEventsAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
|
|
ForwardMessageExecutor<string> processor = new("Processor");
|
|
|
|
Workflow workflow = new WorkflowBuilder(requestPort)
|
|
.AddEdge(requestPort, processor)
|
|
.Build();
|
|
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
// Act 1: Run workflow, collect pending requests and a checkpoint.
|
|
List<ExternalRequest> originalRequests = [];
|
|
CheckpointInfo? checkpoint = null;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "Hello"))
|
|
{
|
|
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
|
|
{
|
|
if (evt is RequestInfoEvent requestInfo)
|
|
{
|
|
originalRequests.Add(requestInfo.Request);
|
|
}
|
|
|
|
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
|
|
{
|
|
checkpoint = cp;
|
|
}
|
|
}
|
|
|
|
originalRequests.Should().NotBeEmpty("the workflow should have created at least one external request");
|
|
checkpoint.Should().NotBeNull("a checkpoint should have been created");
|
|
}
|
|
|
|
// Act 2: Resume from the checkpoint.
|
|
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
|
|
.ResumeStreamingAsync(workflow, checkpoint!);
|
|
|
|
// Assert: The pending requests should be re-emitted.
|
|
List<ExternalRequest> reEmittedRequests = [];
|
|
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
|
|
|
|
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
|
|
{
|
|
if (evt is RequestInfoEvent requestInfo)
|
|
{
|
|
reEmittedRequests.Add(requestInfo.Request);
|
|
}
|
|
}
|
|
|
|
reEmittedRequests.Should().HaveCount(originalRequests.Count,
|
|
"all pending requests from the checkpoint should be re-emitted after resume");
|
|
reEmittedRequests.Select(r => r.RequestId)
|
|
.Should().BeEquivalentTo(originalRequests.Select(r => r.RequestId),
|
|
"the re-emitted request IDs should match the original pending request IDs");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that <see cref="RunStatus"/> transitions to <see cref="RunStatus.PendingRequests"/>
|
|
/// after resuming from a checkpoint with pending external requests (not stuck at NotStarted).
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_WithPendingRequests_RunStatusIsPendingRequestsAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
|
|
ForwardMessageExecutor<string> processor = new("Processor");
|
|
|
|
Workflow workflow = new WorkflowBuilder(requestPort)
|
|
.AddEdge(requestPort, processor)
|
|
.Build();
|
|
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
// First run: collect a checkpoint with pending requests.
|
|
CheckpointInfo? checkpoint = null;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "Hello"))
|
|
{
|
|
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
|
|
{
|
|
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
|
|
{
|
|
checkpoint = cp;
|
|
}
|
|
}
|
|
|
|
checkpoint.Should().NotBeNull();
|
|
}
|
|
|
|
// Act: Resume from the checkpoint and consume events so the run loop processes.
|
|
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
|
|
.ResumeStreamingAsync(workflow, checkpoint!);
|
|
|
|
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
|
|
await foreach (WorkflowEvent _ in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
|
|
{
|
|
// Consume all events until the stream completes.
|
|
}
|
|
|
|
// Assert
|
|
RunStatus status = await resumed.GetStatusAsync();
|
|
status.Should().Be(RunStatus.PendingRequests,
|
|
"the resumed workflow should report PendingRequests after rehydration");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the full roundtrip: resume from checkpoint, observe the re-emitted request,
|
|
/// send a response, and verify the workflow completes without duplicating the request.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_RespondToPendingRequest_CompletesWithoutDuplicateAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
|
|
ForwardMessageExecutor<string> processor = new("Processor");
|
|
|
|
Workflow workflow = new WorkflowBuilder(requestPort)
|
|
.AddEdge(requestPort, processor)
|
|
.Build();
|
|
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
// First run: collect checkpoint + pending request.
|
|
ExternalRequest? pendingRequest = null;
|
|
CheckpointInfo? checkpoint = null;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "Hello"))
|
|
{
|
|
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
|
|
{
|
|
if (evt is RequestInfoEvent requestInfo)
|
|
{
|
|
pendingRequest = requestInfo.Request;
|
|
}
|
|
|
|
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
|
|
{
|
|
checkpoint = cp;
|
|
}
|
|
}
|
|
|
|
pendingRequest.Should().NotBeNull();
|
|
checkpoint.Should().NotBeNull();
|
|
}
|
|
|
|
// Act: Resume and respond to the restored request.
|
|
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
|
|
.ResumeStreamingAsync(workflow, checkpoint!);
|
|
|
|
int requestEventCount = 0;
|
|
|
|
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
|
|
|
|
// Use blockOnPendingRequest: false for the first pass to see the re-emitted requests.
|
|
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
|
|
{
|
|
if (evt is RequestInfoEvent requestInfo)
|
|
{
|
|
requestEventCount++;
|
|
requestInfo.Request.RequestId.Should().Be(pendingRequest!.RequestId,
|
|
"the re-emitted request should match the original");
|
|
}
|
|
}
|
|
|
|
requestEventCount.Should().Be(1,
|
|
"the pending request should be emitted exactly once (no duplicates)");
|
|
|
|
// Assert intermediate state before responding: the run should be in PendingRequests
|
|
// and we should have observed the re-emitted request. If the first WatchStreamAsync
|
|
// didn't complete or yielded nothing, these assertions catch it with a clear message.
|
|
RunStatus statusBeforeResponse = await resumed.GetStatusAsync();
|
|
statusBeforeResponse.Should().Be(RunStatus.PendingRequests,
|
|
"the run should be in PendingRequests state before we send a response");
|
|
|
|
// Now send the response and verify the workflow processes it.
|
|
ExternalResponse response = pendingRequest!.CreateResponse("World");
|
|
await resumed.SendResponseAsync(response);
|
|
|
|
// Consume the resulting events to verify the workflow progresses without errors.
|
|
List<WorkflowEvent> postResponseEvents = [];
|
|
|
|
using CancellationTokenSource cts2 = new(TimeSpan.FromSeconds(10));
|
|
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts2.Token))
|
|
{
|
|
postResponseEvents.Add(evt);
|
|
}
|
|
|
|
postResponseEvents.Should().NotBeEmpty(
|
|
"the workflow should process the response and produce events");
|
|
postResponseEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
|
|
"no errors should occur when processing the restored request's response");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that restoring a live run to a checkpoint re-emits pending requests and allows
|
|
/// the workflow to continue from that restored point.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Restore_WithPendingRequests_RepublishesRequestInfoEventsAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
Workflow workflow = CreateSimpleRequestWorkflow();
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
await using StreamingRun run = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "Hello");
|
|
|
|
(ExternalRequest pendingRequest, CheckpointInfo checkpoint) = await CapturePendingRequestAndCheckpointAsync(run);
|
|
|
|
// Advance the run past the checkpoint so the restore has meaningful work to undo.
|
|
await run.SendResponseAsync(pendingRequest.CreateResponse("World"));
|
|
|
|
List<WorkflowEvent> firstCompletionEvents = await ReadToHaltAsync(run);
|
|
firstCompletionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
|
|
"the workflow should continue cleanly before we restore");
|
|
RunStatus statusAfterFirstResponse = await run.GetStatusAsync();
|
|
statusAfterFirstResponse.Should().Be(RunStatus.Idle,
|
|
"the workflow should finish processing the first response before we restore");
|
|
|
|
// Act
|
|
await run.RestoreCheckpointAsync(checkpoint);
|
|
|
|
// Assert
|
|
List<WorkflowEvent> restoredEvents = await ReadToHaltAsync(run);
|
|
ExternalRequest[] replayedRequests = [.. restoredEvents.OfType<RequestInfoEvent>().Select(evt => evt.Request)];
|
|
|
|
replayedRequests.Should().ContainSingle("runtime restore should re-emit the restored pending request");
|
|
replayedRequests[0].RequestId.Should().Be(pendingRequest.RequestId,
|
|
"the replayed request should match the request captured at the checkpoint");
|
|
|
|
await run.SendResponseAsync(replayedRequests[0].CreateResponse("Again"));
|
|
|
|
List<WorkflowEvent> secondCompletionEvents = await ReadToHaltAsync(run);
|
|
secondCompletionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
|
|
"runtime restore replay should not introduce workflow errors");
|
|
RunStatus statusAfterRestoreResponse = await run.GetStatusAsync();
|
|
statusAfterRestoreResponse.Should().Be(RunStatus.Idle,
|
|
"the workflow should be able to continue after the runtime restore replay");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that restoring a live run clears any queued external responses from the
|
|
/// superseded timeline before importing checkpoint state.
|
|
/// </summary>
|
|
[Fact]
|
|
internal async Task Checkpoint_Restore_ClearsQueuedExternalResponsesBeforeImportAsync()
|
|
{
|
|
Workflow workflow = CreateSimpleRequestWorkflow();
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = ExecutionEnvironment.InProcess_Lockstep.ToWorkflowExecutionEnvironment();
|
|
|
|
await using StreamingRun run = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "Hello");
|
|
|
|
(ExternalRequest pendingRequest, CheckpointInfo checkpoint) = await CapturePendingRequestAndCheckpointAsync(run);
|
|
|
|
await run.SendResponseAsync(pendingRequest.CreateResponse("World"));
|
|
await run.RestoreCheckpointAsync(checkpoint);
|
|
|
|
List<WorkflowEvent> restoredEvents = await ReadToHaltAsync(run);
|
|
ExternalRequest replayedRequest = restoredEvents.OfType<RequestInfoEvent>()
|
|
.Select(evt => evt.Request)
|
|
.Should()
|
|
.ContainSingle("the restored run should still be waiting for the checkpointed request")
|
|
.Subject;
|
|
|
|
restoredEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
|
|
"a queued response from the superseded timeline should not be processed after restore");
|
|
RunStatus statusAfterRestore = await run.GetStatusAsync();
|
|
statusAfterRestore.Should().Be(RunStatus.PendingRequests,
|
|
"the restored run should remain pending until a post-restore response is sent");
|
|
|
|
await run.SendResponseAsync(replayedRequest.CreateResponse("Again"));
|
|
|
|
List<WorkflowEvent> completionEvents = await ReadToHaltAsync(run);
|
|
completionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
|
|
"the restored request should complete cleanly once a new response is provided");
|
|
RunStatus finalStatus = await run.GetStatusAsync();
|
|
finalStatus.Should().Be(RunStatus.Idle,
|
|
"the workflow should finish once the replayed request receives a fresh response");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that fan-in edge state buffered before a checkpoint is still present after resume.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_PreservesFanInBarrierBufferedMessagesAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
Workflow workflow = CreateFanInBarrierWorkflow();
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
ExternalRequest pendingRequest;
|
|
CheckpointInfo checkpoint;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "start"))
|
|
{
|
|
(pendingRequest, checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
|
|
}
|
|
|
|
// Act + Assert
|
|
ExternalRequest replayedRequest = await ResumeAndAssertBarrierReleasesAsync(
|
|
env, checkpointManager, workflow, checkpoint, ["before", "after"]);
|
|
|
|
pendingRequest.RequestId.Should().Be(replayedRequest.RequestId,
|
|
"the replayed request should be the one from the checkpointed superstep");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that fan-in barrier state is preserved across resume when more than two sources
|
|
/// participate, and multiple contributions are buffered before the checkpoint.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_PreservesFanInBarrierBufferedMessages_MultiSourceAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange: a fan-out start broadcasts a trigger to two early barrier sources and to a
|
|
// request-port kickoff. Both early sources contribute pre-checkpoint and only the port
|
|
// response path is unseen at the checkpoint - the barrier must hold both buffered
|
|
// contributions across resume and only release once the third source contributes.
|
|
const string RequestPortId = "Approval";
|
|
const string SinkId = "Sink";
|
|
|
|
ForwardMessageExecutor<string> start = new("Start");
|
|
ExecutorBinding earlyA = new BarrierContributor("EarlyA", SinkId, "before-1");
|
|
ExecutorBinding earlyB = new BarrierContributor("EarlyB", SinkId, "before-2");
|
|
ExecutorBinding kickoff = new RequestPortKickoff("Kickoff", RequestPortId);
|
|
ExecutorBinding afterResume = new PostCheckpointBarrierSource("AfterResume", SinkId);
|
|
ExecutorBinding sink = new BarrierSink(SinkId);
|
|
RequestPort<ApprovalRequest, ApprovalReply> requestPort = RequestPort.Create<ApprovalRequest, ApprovalReply>(RequestPortId);
|
|
|
|
Workflow workflow = new WorkflowBuilder(start)
|
|
.AddEdge(start, earlyA)
|
|
.AddEdge(start, earlyB)
|
|
.AddEdge(start, kickoff)
|
|
.AddEdge(kickoff, requestPort)
|
|
.AddEdge(requestPort, afterResume)
|
|
.AddFanInBarrierEdge([earlyA, earlyB, afterResume], sink)
|
|
.Build();
|
|
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
ExternalRequest pendingRequest;
|
|
CheckpointInfo checkpoint;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "start"))
|
|
{
|
|
(pendingRequest, checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
|
|
}
|
|
|
|
// Act + Assert
|
|
ExternalRequest replayedRequest = await ResumeAndAssertBarrierReleasesAsync(
|
|
env, checkpointManager, workflow, checkpoint, ["before-1", "before-2", "after"]);
|
|
|
|
pendingRequest.RequestId.Should().Be(replayedRequest.RequestId,
|
|
"the replayed request should match the one captured at checkpoint time");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that the same checkpoint can be resumed independently more than once - i.e. that
|
|
/// completing one resumed run does not mutate state inside the stored checkpoint.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Without a snapshot at the export/import boundary, <c>FanInEdgeRunner</c> would hand the
|
|
/// in-memory <c>CheckpointManager</c> a live reference to the mutable <c>FanInEdgeState</c>.
|
|
/// The first resume would then reset the buffer back to "all unseen" while completing, and a
|
|
/// second resume from the same <see cref="CheckpointInfo"/> would deadlock waiting for the
|
|
/// pre-checkpoint contribution that no longer exists.
|
|
/// </remarks>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_FanInBarrierCheckpointCanBeResumedTwiceAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
CheckpointInfo checkpoint;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(CreateFanInBarrierWorkflow(), "start"))
|
|
{
|
|
(_, checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
|
|
}
|
|
|
|
// Act + Assert: each resume needs a fresh Workflow object because the previous run takes
|
|
// ownership of it. Resuming the same CheckpointInfo more than once must yield identical
|
|
// results - the first resume must not mutate state the checkpoint store is still holding.
|
|
for (int attempt = 0; attempt < 2; attempt++)
|
|
{
|
|
await ResumeAndAssertBarrierReleasesAsync(
|
|
env, checkpointManager, CreateFanInBarrierWorkflow(), checkpoint, ["before", "after"]);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that fan-in barrier state buffered inside a subworkflow is preserved across a
|
|
/// checkpoint/resume cycle of the parent workflow.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_PreservesFanInBarrierBufferedMessages_InSubworkflowAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange: the fan-in barrier lives inside a subworkflow; the fix has to apply to the
|
|
// subworkflow's runner context too.
|
|
const string InnerRequestPortId = "InnerApproval";
|
|
const string ForwardedRequestId = "ForwardedInnerApproval";
|
|
|
|
Workflow BuildOuter()
|
|
{
|
|
ExecutorBinding subworkflow = CreateFanInBarrierWorkflow(
|
|
requestPortId: InnerRequestPortId, sinkId: "InnerSink")
|
|
.BindAsExecutor("InnerSubworkflow");
|
|
|
|
return new WorkflowBuilder(subworkflow)
|
|
.AddExternalRequest<ApprovalRequest, ApprovalReply>(subworkflow, id: ForwardedRequestId)
|
|
.Build();
|
|
}
|
|
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
ExternalRequest pendingRequest;
|
|
CheckpointInfo checkpoint;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(BuildOuter(), "start"))
|
|
{
|
|
(pendingRequest, checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
|
|
}
|
|
|
|
// Act + Assert
|
|
ExternalRequest replayedRequest = await ResumeAndAssertBarrierReleasesAsync(
|
|
env, checkpointManager, BuildOuter(), checkpoint, ["before", "after"]);
|
|
|
|
pendingRequest.RequestId.Should().Be(replayedRequest.RequestId,
|
|
"the replayed subworkflow request should match the one captured at checkpoint time");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that a resumed parent workflow re-emits pending requests that originated in a subworkflow.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_SubworkflowWithPendingRequests_RepublishesQualifiedRequestInfoEventsAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
Workflow workflow = CreateCheckpointedSubworkflowRequestWorkflow();
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
ExternalRequest pendingRequest;
|
|
CheckpointInfo checkpoint;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "Hello"))
|
|
{
|
|
(pendingRequest, checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
|
|
}
|
|
|
|
// Act
|
|
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
|
|
.ResumeStreamingAsync(workflow, checkpoint);
|
|
|
|
// Assert
|
|
List<WorkflowEvent> resumedEvents = await ReadToHaltAsync(resumed);
|
|
ExternalRequest[] replayedRequests = [.. resumedEvents.OfType<RequestInfoEvent>().Select(evt => evt.Request)];
|
|
|
|
replayedRequests.Should().ContainSingle("the resumed parent workflow should surface the subworkflow request once");
|
|
replayedRequests[0].RequestId.Should().Be(pendingRequest.RequestId,
|
|
"the replayed subworkflow request should match the checkpointed request");
|
|
replayedRequests[0].PortInfo.PortId.Should().Be(pendingRequest.PortInfo.PortId,
|
|
"the replayed request should remain qualified through the subworkflow boundary");
|
|
|
|
await resumed.SendResponseAsync(replayedRequests[0].CreateResponse("World"));
|
|
|
|
List<WorkflowEvent> completionEvents = await ReadToHaltAsync(resumed);
|
|
completionEvents.OfType<RequestInfoEvent>().Should().BeEmpty(
|
|
"the resumed subworkflow request should not be replayed twice");
|
|
completionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
|
|
"subworkflow replay should not introduce workflow errors");
|
|
RunStatus statusAfterSubworkflowResponse = await resumed.GetStatusAsync();
|
|
statusAfterSubworkflowResponse.Should().Be(RunStatus.Idle,
|
|
"the resumed subworkflow should continue after responding to the replayed request");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that when <c>republishPendingEvents</c> is <see langword="false"/>,
|
|
/// no <see cref="RequestInfoEvent"/> is re-emitted after resuming from a checkpoint.
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
|
|
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
|
|
internal async Task Checkpoint_Resume_WithRepublishDisabled_DoesNotEmitRequestInfoEventsAsync(ExecutionEnvironment environment)
|
|
{
|
|
// Arrange
|
|
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
|
|
ForwardMessageExecutor<string> processor = new("Processor");
|
|
|
|
Workflow workflow = new WorkflowBuilder(requestPort)
|
|
.AddEdge(requestPort, processor)
|
|
.Build();
|
|
|
|
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
|
|
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
|
|
|
|
// First run: collect a checkpoint with pending requests.
|
|
CheckpointInfo? checkpoint = null;
|
|
|
|
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
|
|
.RunStreamingAsync(workflow, "Hello"))
|
|
{
|
|
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
|
|
{
|
|
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
|
|
{
|
|
checkpoint = cp;
|
|
}
|
|
}
|
|
|
|
checkpoint.Should().NotBeNull();
|
|
}
|
|
|
|
// Act: Resume with republishPendingEvents: false via the internal API.
|
|
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
|
|
.ResumeStreamingInternalAsync(workflow, checkpoint!, republishPendingEvents: false);
|
|
|
|
// Assert: No RequestInfoEvent should appear in the event stream.
|
|
int requestEventCount = 0;
|
|
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
|
|
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
|
|
{
|
|
if (evt is RequestInfoEvent)
|
|
{
|
|
requestEventCount++;
|
|
}
|
|
}
|
|
|
|
requestEventCount.Should().Be(0,
|
|
"no RequestInfoEvent should be emitted when republishPendingEvents is false");
|
|
}
|
|
|
|
private static Workflow CreateSimpleRequestWorkflow(
|
|
string requestPortId = "TestPort",
|
|
string processorId = "Processor")
|
|
{
|
|
RequestPort<string, string> requestPort = RequestPort.Create<string, string>(requestPortId);
|
|
ForwardMessageExecutor<string> processor = new(processorId);
|
|
|
|
return new WorkflowBuilder(requestPort)
|
|
.AddEdge(requestPort, processor)
|
|
.Build();
|
|
}
|
|
|
|
private static Workflow CreateCheckpointedSubworkflowRequestWorkflow()
|
|
{
|
|
ExecutorBinding subworkflow = CreateSimpleRequestWorkflow(
|
|
requestPortId: "InnerTestPort",
|
|
processorId: "InnerProcessor")
|
|
.BindAsExecutor("Subworkflow");
|
|
|
|
return new WorkflowBuilder(subworkflow)
|
|
.AddExternalRequest<string, string>(subworkflow, id: "ForwardedSubworkflowRequest")
|
|
.Build();
|
|
}
|
|
|
|
private static async ValueTask<(ExternalRequest PendingRequest, CheckpointInfo Checkpoint)> CapturePendingRequestAndCheckpointAsync(StreamingRun run)
|
|
{
|
|
ExternalRequest? pendingRequest = null;
|
|
CheckpointInfo? checkpoint = null;
|
|
|
|
await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false))
|
|
{
|
|
if (evt is RequestInfoEvent requestInfo)
|
|
{
|
|
pendingRequest ??= requestInfo.Request;
|
|
}
|
|
|
|
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
|
|
{
|
|
checkpoint = cp;
|
|
}
|
|
}
|
|
|
|
pendingRequest.Should().NotBeNull("the workflow should have emitted a pending request");
|
|
checkpoint.Should().NotBeNull("the workflow should have produced a checkpoint");
|
|
return (pendingRequest!, checkpoint!);
|
|
}
|
|
|
|
private static async ValueTask<List<WorkflowEvent>> ReadToHaltAsync(StreamingRun run)
|
|
{
|
|
List<WorkflowEvent> events = [];
|
|
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
|
|
|
|
await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
|
|
{
|
|
events.Add(evt);
|
|
}
|
|
|
|
return events;
|
|
}
|
|
|
|
private static Workflow CreateFanInBarrierWorkflow(
|
|
string requestPortId = "Approval",
|
|
string sinkId = "Sink")
|
|
{
|
|
ExecutorBinding before = new PreCheckpointBarrierSource("BeforePause", requestPortId, sinkId);
|
|
ExecutorBinding after = new PostCheckpointBarrierSource("AfterResume", sinkId);
|
|
ExecutorBinding sink = new BarrierSink(sinkId);
|
|
RequestPort<ApprovalRequest, ApprovalReply> requestPort =
|
|
RequestPort.Create<ApprovalRequest, ApprovalReply>(requestPortId);
|
|
|
|
return new WorkflowBuilder(before)
|
|
.AddEdge(before, requestPort)
|
|
.AddEdge(requestPort, after)
|
|
.AddFanInBarrierEdge([before, after], sink)
|
|
.Build();
|
|
}
|
|
|
|
private static async ValueTask<ExternalRequest> ResumeAndAssertBarrierReleasesAsync(
|
|
InProcessExecutionEnvironment env,
|
|
CheckpointManager checkpointManager,
|
|
Workflow workflow,
|
|
CheckpointInfo checkpoint,
|
|
IEnumerable<string> expectedBarrierSources)
|
|
{
|
|
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
|
|
.ResumeStreamingAsync(workflow, checkpoint);
|
|
|
|
List<WorkflowEvent> resumedEvents = await ReadToHaltAsync(resumed);
|
|
ExternalRequest replayedRequest = resumedEvents.OfType<RequestInfoEvent>()
|
|
.Select(evt => evt.Request)
|
|
.Should()
|
|
.ContainSingle("resume should replay exactly one pending request")
|
|
.Subject;
|
|
|
|
await resumed.SendResponseAsync(replayedRequest.CreateResponse(new ApprovalReply("yes")));
|
|
|
|
List<WorkflowEvent> completionEvents = await ReadToHaltAsync(resumed);
|
|
|
|
completionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
|
|
"resuming across a partially satisfied fan-in barrier should not raise workflow errors");
|
|
|
|
string[] outputs = [.. completionEvents.OfType<BarrierReleasedEvent>().Select(evt => evt.Source)];
|
|
outputs.Should().BeEquivalentTo(expectedBarrierSources,
|
|
"the barrier should release every expected contribution");
|
|
|
|
RunStatus status = await resumed.GetStatusAsync();
|
|
status.Should().Be(RunStatus.Idle,
|
|
"the resumed run should halt cleanly once every barrier source has contributed");
|
|
|
|
return replayedRequest;
|
|
}
|
|
|
|
private sealed record BarrierContribution(string Source);
|
|
|
|
private sealed record ApprovalRequest(string Prompt);
|
|
|
|
private sealed record ApprovalReply(string Value);
|
|
|
|
private sealed class BarrierReleasedEvent(string source) : WorkflowEvent
|
|
{
|
|
public string Source { get; } = source;
|
|
}
|
|
|
|
private sealed class PreCheckpointBarrierSource(string id, string requestPortId, string sinkId) : Executor(id)
|
|
{
|
|
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
|
=> protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler<string>(this.HandleAsync))
|
|
.SendsMessage<BarrierContribution>()
|
|
.SendsMessage<ApprovalRequest>();
|
|
|
|
private async ValueTask HandleAsync(string input, IWorkflowContext ctx)
|
|
{
|
|
await ctx.SendMessageAsync(new BarrierContribution("before"), sinkId).ConfigureAwait(false);
|
|
await ctx.SendMessageAsync(new ApprovalRequest("continue?"), requestPortId).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private sealed class BarrierContributor(string id, string sinkId, string label) : Executor(id)
|
|
{
|
|
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
|
=> protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler<string>(this.HandleAsync))
|
|
.SendsMessage<BarrierContribution>();
|
|
|
|
private ValueTask HandleAsync(string input, IWorkflowContext ctx)
|
|
=> ctx.SendMessageAsync(new BarrierContribution(label), sinkId);
|
|
}
|
|
|
|
private sealed class RequestPortKickoff(string id, string requestPortId) : Executor(id)
|
|
{
|
|
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
|
=> protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler<string>(this.HandleAsync))
|
|
.SendsMessage<ApprovalRequest>();
|
|
|
|
private ValueTask HandleAsync(string input, IWorkflowContext ctx)
|
|
=> ctx.SendMessageAsync(new ApprovalRequest("continue?"), requestPortId);
|
|
}
|
|
|
|
private sealed class PostCheckpointBarrierSource(string id, string sinkId) : Executor(id)
|
|
{
|
|
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
|
=> protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler<ApprovalReply>(this.HandleAsync))
|
|
.SendsMessage<BarrierContribution>();
|
|
|
|
private ValueTask HandleAsync(ApprovalReply reply, IWorkflowContext ctx)
|
|
=> ctx.SendMessageAsync(new BarrierContribution("after"), sinkId);
|
|
}
|
|
|
|
private sealed class BarrierSink(string id) : Executor(id)
|
|
{
|
|
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
|
=> protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler<BarrierContribution>(this.HandleAsync));
|
|
|
|
private ValueTask HandleAsync(BarrierContribution contribution, IWorkflowContext ctx)
|
|
=> ctx.AddEventAsync(new BarrierReleasedEvent(contribution.Source));
|
|
}
|
|
}
|