diff --git a/Directory.Build.props b/Directory.Build.props index 83111ee..c0d4184 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -19,13 +19,15 @@ - - - - + + + + - --> + + + diff --git a/README.md b/README.md index d8163e8..f2957e9 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ Prerequisites: * [ClientMtls](src/ClientMtls) - How to use client certificate authentication, e.g. for Temporal Cloud. * [DependencyInjection](src/DependencyInjection) - How to inject dependencies in activities and use generic hosts for workers * [Encryption](src/Encryption) - End-to-end encryption with Temporal payload codecs. +* [Mutex](src/Mutex) - How to implement a mutex as a workflow. Demonstrates how to avoid race conditions or parallel mutually exclusive operations on the same resource. * [Polling](src/Polling) - Recommended implementation of an activity that needs to periodically poll an external resource waiting its successful completion. * [Schedules](src/Schedules) - How to schedule workflows to be run at specific times in the future. * [WorkerVersioning](src/WorkerVersioning) - How to use the Worker Versioning feature to more easily deploy changes to Workflow & other code. diff --git a/TemporalioSamples.sln b/TemporalioSamples.sln index 86885a5..16fb50f 100644 --- a/TemporalioSamples.sln +++ b/TemporalioSamples.sln @@ -47,6 +47,14 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.Dependenc EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkerVersioning", "src\WorkerVersioning\TemporalioSamples.WorkerVersioning.csproj", "{CA3FD1BC-C918-4B15-96F6-D6DDA125E63C}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.Mutex", "src\Mutex\TemporalioSamples.Mutex.csproj", "{3168FB2D-D821-433A-A761-309E0474DE48}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Temporalio", "..\temporal-sdk-dotnet\src\Temporalio\Temporalio.csproj", "{D472F5AE-8C24-4FAB-A60C-667B7A8CDF75}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Temporalio.Extensions.Hosting", "..\temporal-sdk-dotnet\src\Temporalio.Extensions.Hosting\Temporalio.Extensions.Hosting.csproj", "{8B562D41-7525-42A1-928F-C21F1F22EF81}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Temporalio.Extensions.OpenTelemetry", "..\temporal-sdk-dotnet\src\Temporalio.Extensions.OpenTelemetry\Temporalio.Extensions.OpenTelemetry.csproj", "{40607923-C603-4CCB-9674-15894710D1AB}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -128,6 +136,22 @@ Global {CA3FD1BC-C918-4B15-96F6-D6DDA125E63C}.Debug|Any CPU.Build.0 = Debug|Any CPU {CA3FD1BC-C918-4B15-96F6-D6DDA125E63C}.Release|Any CPU.ActiveCfg = Release|Any CPU {CA3FD1BC-C918-4B15-96F6-D6DDA125E63C}.Release|Any CPU.Build.0 = Release|Any CPU + {3168FB2D-D821-433A-A761-309E0474DE48}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {3168FB2D-D821-433A-A761-309E0474DE48}.Debug|Any CPU.Build.0 = Debug|Any CPU + {3168FB2D-D821-433A-A761-309E0474DE48}.Release|Any CPU.ActiveCfg = Release|Any CPU + {3168FB2D-D821-433A-A761-309E0474DE48}.Release|Any CPU.Build.0 = Release|Any CPU + {D472F5AE-8C24-4FAB-A60C-667B7A8CDF75}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {D472F5AE-8C24-4FAB-A60C-667B7A8CDF75}.Debug|Any CPU.Build.0 = Debug|Any CPU + {D472F5AE-8C24-4FAB-A60C-667B7A8CDF75}.Release|Any CPU.ActiveCfg = Release|Any CPU + {D472F5AE-8C24-4FAB-A60C-667B7A8CDF75}.Release|Any CPU.Build.0 = Release|Any CPU + {8B562D41-7525-42A1-928F-C21F1F22EF81}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {8B562D41-7525-42A1-928F-C21F1F22EF81}.Debug|Any CPU.Build.0 = Debug|Any CPU + {8B562D41-7525-42A1-928F-C21F1F22EF81}.Release|Any CPU.ActiveCfg = Release|Any CPU + {8B562D41-7525-42A1-928F-C21F1F22EF81}.Release|Any CPU.Build.0 = Release|Any CPU + {40607923-C603-4CCB-9674-15894710D1AB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {40607923-C603-4CCB-9674-15894710D1AB}.Debug|Any CPU.Build.0 = Debug|Any CPU + {40607923-C603-4CCB-9674-15894710D1AB}.Release|Any CPU.ActiveCfg = Release|Any CPU + {40607923-C603-4CCB-9674-15894710D1AB}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(NestedProjects) = preSolution {7AECC7C6-9A21-4B8A-84D9-AFC4F5840CAF} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} @@ -150,5 +174,9 @@ Global {11A5854B-EE6E-4752-9C46-F466503D853B} = {AE21E7F4-B114-4761-81B1-8FA63E9F6BB8} {10E6F7C9-7F6C-4A8E-94A1-99C10F46BBA4} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} {CA3FD1BC-C918-4B15-96F6-D6DDA125E63C} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} + {3168FB2D-D821-433A-A761-309E0474DE48} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} + {D472F5AE-8C24-4FAB-A60C-667B7A8CDF75} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} + {8B562D41-7525-42A1-928F-C21F1F22EF81} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} + {40607923-C603-4CCB-9674-15894710D1AB} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC} EndGlobalSection EndGlobal diff --git a/src/Mutex/Activities.cs b/src/Mutex/Activities.cs new file mode 100644 index 0000000..a07a13f --- /dev/null +++ b/src/Mutex/Activities.cs @@ -0,0 +1,38 @@ +namespace TemporalioSamples.Mutex; + +using Temporalio.Activities; + +public record NotifyLockedInput(string ResourceId, string ReleaseSignalName); + +public record UseApiThatCantBeCalledInParallelInput(TimeSpan SleepFor); + +public record NotifyUnlockedInput(string ResourceId); + +public static class Activities +{ + [Activity] + public static void NotifyLocked(NotifyLockedInput input) + { + ActivityExecutionContext.Current.Logger.LogInformation( + "Lock for resource '{ResourceId}' acquired, release signal name '{ReleaseSignalName}'", input.ResourceId, input.ReleaseSignalName); + } + + [Activity] + public static async Task UseApiThatCantBeCalledInParallelAsync(UseApiThatCantBeCalledInParallelInput input) + { + var logger = ActivityExecutionContext.Current.Logger; + + logger.LogInformation("Sleeping for '{SleepFor}'...", input.SleepFor); + + await Task.Delay(input.SleepFor); + + logger.LogInformation("Done sleeping!"); + } + + [Activity] + public static void NotifyUnlocked(NotifyUnlockedInput input) + { + ActivityExecutionContext.Current.Logger.LogInformation( + "Lock for resource '{ResourceId}' released", input.ResourceId); + } +} diff --git a/src/Mutex/Impl/ILockHandle.cs b/src/Mutex/Impl/ILockHandle.cs new file mode 100644 index 0000000..d95482a --- /dev/null +++ b/src/Mutex/Impl/ILockHandle.cs @@ -0,0 +1,10 @@ +namespace TemporalioSamples.Mutex.Impl; + +public interface ILockHandle : IAsyncDisposable +{ + public string LockInitiatorId { get; } + + public string ResourceId { get; } + + public string ReleaseSignalName { get; } +} diff --git a/src/Mutex/Impl/ILockHandler.cs b/src/Mutex/Impl/ILockHandler.cs new file mode 100644 index 0000000..3439945 --- /dev/null +++ b/src/Mutex/Impl/ILockHandler.cs @@ -0,0 +1,8 @@ +namespace TemporalioSamples.Mutex.Impl; + +internal interface ILockHandler +{ + public string? CurrentOwnerId { get; } + + public Task HandleAsync(LockRequest lockRequest); +} diff --git a/src/Mutex/Impl/LockRequest.cs b/src/Mutex/Impl/LockRequest.cs new file mode 100644 index 0000000..b095d8f --- /dev/null +++ b/src/Mutex/Impl/LockRequest.cs @@ -0,0 +1,3 @@ +namespace TemporalioSamples.Mutex.Impl; + +internal record LockRequest(string InitiatorId, string AcquireLockSignalName, TimeSpan? Timeout = null); diff --git a/src/Mutex/Impl/MutexActivities.cs b/src/Mutex/Impl/MutexActivities.cs new file mode 100644 index 0000000..eb13c8c --- /dev/null +++ b/src/Mutex/Impl/MutexActivities.cs @@ -0,0 +1,37 @@ +namespace TemporalioSamples.Mutex.Impl; + +using Temporalio.Activities; +using Temporalio.Client; +using Temporalio.Workflows; + +internal record SignalWithStartMutexWorkflowInput(string MutexWorkflowId, string ResourceId, string AcquireLockSignalName, TimeSpan? LockTimeout = null); + +internal class MutexActivities +{ + private static readonly string RequestLockSignalName = + WorkflowSignalDefinition.FromMethod( + typeof(MutexWorkflow).GetMethod(nameof(MutexWorkflow.RequestLockAsync)) + ?? throw new InvalidOperationException($"Method {nameof(MutexWorkflow.RequestLockAsync)} not found on type {typeof(MutexWorkflow)}")) + .Name ?? throw new InvalidOperationException("Signal name is null."); + + private readonly ITemporalClient client; + + public MutexActivities(ITemporalClient client) + { + this.client = client; + } + + [Activity] + public async Task SignalWithStartMutexWorkflowAsync(SignalWithStartMutexWorkflowInput input) + { + var activityInfo = ActivityExecutionContext.Current.Info; + + await this.client.StartWorkflowAsync( + (MutexWorkflow mw) => mw.RunAsync(MutexWorkflowInput.Empty), + new WorkflowOptions(input.MutexWorkflowId, activityInfo.TaskQueue) + { + StartSignal = RequestLockSignalName, + StartSignalArgs = new object[] { new LockRequest(activityInfo.WorkflowId, input.AcquireLockSignalName, input.LockTimeout), }, + }); + } +} diff --git a/src/Mutex/Impl/MutexWorkflow.workflow.cs b/src/Mutex/Impl/MutexWorkflow.workflow.cs new file mode 100644 index 0000000..801f3cf --- /dev/null +++ b/src/Mutex/Impl/MutexWorkflow.workflow.cs @@ -0,0 +1,60 @@ +namespace TemporalioSamples.Mutex.Impl; + +using Temporalio.Workflows; + +internal record MutexWorkflowInput(IReadOnlyCollection InitialRequests) +{ + public static readonly MutexWorkflowInput Empty = new(Array.Empty()); +} + +[Workflow] +internal class MutexWorkflow +{ + private readonly ILockHandler lockHandler = WorkflowMutex.CreateLockHandler(); + private readonly Queue requests = new(); + + [WorkflowRun] + public async Task RunAsync(MutexWorkflowInput input) + { + var logger = Workflow.Logger; + + foreach (var request in input.InitialRequests) + { + requests.Enqueue(request); + } + + while (!Workflow.ContinueAsNewSuggested) + { + if (requests.Count == 0) + { + logger.LogInformation("No lock requests, waiting for more..."); + + await Workflow.WaitConditionAsync(() => requests.Count > 0); + } + + while (requests.TryDequeue(out var lockRequest)) + { + await lockHandler.HandleAsync(lockRequest); + } + } + + if (requests.Count > 0) + { + var newInput = new MutexWorkflowInput(requests); + throw Workflow.CreateContinueAsNewException((MutexWorkflow x) => x.RunAsync(newInput)); + } + } + + [WorkflowQuery] + public string? CurrentOwnerId => lockHandler.CurrentOwnerId; + + [WorkflowSignal] + public Task RequestLockAsync(LockRequest request) + { + requests.Enqueue(request); + + Workflow.Logger.LogInformation("Received lock request. (InitiatorId='{InitiatorId}')", request.InitiatorId); + + return Task.CompletedTask; + } +} diff --git a/src/Mutex/Impl/TemporalWorkerOptionsExtensions.cs b/src/Mutex/Impl/TemporalWorkerOptionsExtensions.cs new file mode 100644 index 0000000..6022aed --- /dev/null +++ b/src/Mutex/Impl/TemporalWorkerOptionsExtensions.cs @@ -0,0 +1,18 @@ +namespace TemporalioSamples.Mutex.Impl; + +using Temporalio.Client; +using Temporalio.Worker; + +public static class TemporalWorkerOptionsExtensions +{ + public static TemporalWorkerOptions AddWorkflowMutex(this TemporalWorkerOptions options, ITemporalClient client) + { + var mutexActivities = new MutexActivities(client); + + options + .AddAllActivities(mutexActivities) + .AddWorkflow(); + + return options; + } +} diff --git a/src/Mutex/Impl/WorkflowMutex.cs b/src/Mutex/Impl/WorkflowMutex.cs new file mode 100644 index 0000000..732cda4 --- /dev/null +++ b/src/Mutex/Impl/WorkflowMutex.cs @@ -0,0 +1,133 @@ +namespace TemporalioSamples.Mutex.Impl; + +using Temporalio.Workflows; + +internal record AcquireLockInput(string ReleaseSignalName); + +/// +/// Represents a mutual exclusion mechanism for Workflows. +/// This part contains API for acquiring locks. +/// +public static class WorkflowMutex +{ + private const string MutexWorkflowIdPrefix = "__wm-lock:"; + + public static async Task LockAsync(string resourceId, TimeSpan? lockTimeout = null) + { + if (!Workflow.InWorkflow) + { + throw new InvalidOperationException("Cannot acquire a lock outside of a workflow."); + } + + var initiatorId = Workflow.Info.WorkflowId; + var lockStarted = Workflow.UtcNow; + + string? releaseSignalName = null; + var acquireLockSignalName = Workflow.NewGuid().ToString(); + var signalDefinition = WorkflowSignalDefinition.CreateWithoutAttribute(acquireLockSignalName, (AcquireLockInput input) => + { + releaseSignalName = input.ReleaseSignalName; + + return Task.CompletedTask; + }); + Workflow.Signals[acquireLockSignalName] = signalDefinition; + try + { + var startMutexWorkflowInput = new SignalWithStartMutexWorkflowInput($"{MutexWorkflowIdPrefix}{resourceId}", resourceId, acquireLockSignalName, lockTimeout); + await Workflow.ExecuteActivityAsync( + act => act.SignalWithStartMutexWorkflowAsync(startMutexWorkflowInput), + new ActivityOptions { StartToCloseTimeout = TimeSpan.FromMinutes(1), }); + + await Workflow.WaitConditionAsync(() => releaseSignalName != null); + + var elapsed = Workflow.UtcNow - lockStarted; + Workflow.Logger.LogInformation( + "Lock for resource '{ResourceId}' acquired in {AcquireTime}ms by '{LockInitiatorId}', release signal name '{ReleaseSignalName}'", + resourceId, + (int)elapsed.TotalMilliseconds, + initiatorId, + releaseSignalName); + + return new LockHandle(initiatorId, startMutexWorkflowInput.MutexWorkflowId, resourceId, releaseSignalName!); + } + finally + { + Workflow.Signals.Remove(acquireLockSignalName); + } + } + + internal static ILockHandler CreateLockHandler() + { + if (!Workflow.InWorkflow) + { + throw new InvalidOperationException("Cannot acquire a lock outside of a workflow."); + } + + return new LockHandler(); + } + + internal sealed class LockHandle : ILockHandle + { + private readonly string mutexWorkflowId; + + public LockHandle(string lockInitiatorId, string mutexWorkflowId, string resourceId, string releaseSignalId) + { + LockInitiatorId = lockInitiatorId; + this.mutexWorkflowId = mutexWorkflowId; + ResourceId = resourceId; + ReleaseSignalName = releaseSignalId; + } + + /// + public string LockInitiatorId { get; } + + /// + public string ResourceId { get; } + + /// + public string ReleaseSignalName { get; } + + /// + public async ValueTask DisposeAsync() + { + var mutexHandle = Workflow.GetExternalWorkflowHandle(mutexWorkflowId); + await mutexHandle.SignalAsync(ReleaseSignalName, Array.Empty()); + } + } + + internal sealed class LockHandler : ILockHandler + { + /// + public string? CurrentOwnerId { get; private set; } + + /// + public async Task HandleAsync(LockRequest lockRequest) + { + var releaseSignalName = Workflow.NewGuid().ToString(); + + var initiator = Workflow.GetExternalWorkflowHandle(lockRequest.InitiatorId); + await initiator.SignalAsync(lockRequest.AcquireLockSignalName, new[] { new AcquireLockInput(releaseSignalName) }); + + var released = false; + Workflow.Signals[releaseSignalName] = WorkflowSignalDefinition.CreateWithoutAttribute(releaseSignalName, () => + { + released = true; + + return Task.CompletedTask; + }); + CurrentOwnerId = lockRequest.InitiatorId; + + if (!await Workflow.WaitConditionAsync(() => released, lockRequest.Timeout ?? Timeout.InfiniteTimeSpan)) + { + Workflow.Logger.LogWarning( + "Lock for resource '{ResourceId}' has been timed out after '{Timeout}'. (LockInitiatorId='{LockInitiatorId}')", + Workflow.Info.WorkflowId[MutexWorkflowIdPrefix.Length..], + lockRequest.Timeout, + lockRequest.InitiatorId); + } + + CurrentOwnerId = null; + Workflow.Signals.Remove(releaseSignalName); + } + } +} diff --git a/src/Mutex/Program.cs b/src/Mutex/Program.cs new file mode 100644 index 0000000..5938215 --- /dev/null +++ b/src/Mutex/Program.cs @@ -0,0 +1,75 @@ +using System.Diagnostics; +using Temporalio.Client; +using Temporalio.Worker; +using TemporalioSamples.Mutex; +using TemporalioSamples.Mutex.Impl; + +var client = await TemporalClient.ConnectAsync(new("localhost:7233") +{ + LoggerFactory = LoggerFactory.Create(builder => + builder.AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").SetMinimumLevel(LogLevel.Information)), +}); + +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 + Console.WriteLine("Running worker"); + + using var worker = new TemporalWorker( + client, + new TemporalWorkerOptions(taskQueue: "workflow-mutex-sample") + .AddWorkflowMutex(client) + .AddAllActivities(typeof(Activities), null) + .AddWorkflow()); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + Console.WriteLine("Worker cancelled"); + } +} + +async Task ExecuteWorkflowsWithMutex(string resourceId) +{ + await Task.WhenAll(Execute(), Execute()); + + return; + + async Task Execute() + { + var workflowId = "test-" + Guid.NewGuid(); + Console.WriteLine($"Starting test workflow with id '{workflowId}'. Connecting to lock workflow '{resourceId}'"); + + var sw = Stopwatch.StartNew(); + var handle = await client.StartWorkflowAsync( + (WorkflowWithMutex wf) => wf.RunAsync(new WorkflowWithMutexInput(resourceId, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(7.5))), + new(workflowId, "workflow-mutex-sample")); + + Console.WriteLine($"Test workflow '{workflowId}' started"); + + await handle.GetResultAsync(); + Console.WriteLine($"Test workflow '{workflowId}' finished after {sw.ElapsedMilliseconds}ms"); + } +} + +switch (args.ElementAtOrDefault(0)) +{ + case "worker": + await RunWorkerAsync(); + break; + case "workflow": + await ExecuteWorkflowsWithMutex(args.ElementAtOrDefault(1) ?? "locked-resource-id"); + break; + default: + throw new ArgumentException("Must pass 'worker' or 'workflow' as the first argument"); +} diff --git a/src/Mutex/README.md b/src/Mutex/README.md new file mode 100644 index 0000000..76b099d --- /dev/null +++ b/src/Mutex/README.md @@ -0,0 +1,37 @@ +# Mutex + +This sample has a MutexWorkflow that receives lock-requested Signals from other Workflows. +It handles the Signals one at a time, first sending a lock-acquired Signal to the sending Workflow, +and then waiting for a Signal from that Workflow indicating that the Workflow is ready to release the lock. Then the MutexWorkflow goes on to the next lock-requested Signal. In this way, you're able to make sure that only one Workflow Execution is performing a certain type of work at a time. + +## Structure +This project is divided into two components: +1. The actual sample, featuring user code that demonstrates the use of `WorkflowMutex`, is located in the `TemporalioSamples.Mutex` namespace. +2. The implementation of `WorkflowMutex` resides in the `TemporalioSamples.Mutex.Impl` namespace, highlighting its potential to be abstracted into a separate library. + +Below is a brief description of the core files: +* [Impl/WorkflowMutex.cs](Impl/WorkflowMutex.cs): WorkflowMutex is a static class that provides mechanism to acquire a lock for a given resource within a workflow context. The primary functionality is encapsulated in the LockAsync() method, which initiates a lock on a resource and returns a handle. This handle then can be used to release the lock. +* [Impl/MutexWorkflow.workflow.cs](Impl/MutexWorkflow.workflow.cs): The MutexWorkflow class is an internal workflow designed to manage and process a queue of lock requests. + Each lock request is handled by a dedicated LockHandler class, encapsulating the mechanics of lock handling, including lock acquisition, release, and timeout. + Requests are processed in the order they are received, and if all have been processed, the workflow waits for new ones. In the event that the workflow has a suggestion to continue as new, all remaining requests will be passed onto the newly created workflow. +* [WorkflowWithMutex.workflow.cs](WorkflowWithMutex.workflow.cs): The WorkflowWithMutex class is an implementation of an asynchronous workflow that acquires a lock on a resource, executes certain activities while the lock is held, and then releases the lock afterward. + +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 start two workflows, which will both try to acquire the lock at the same time. You should see that one Workflow takes slightly longer than the other, because it needs to wait for lock release. + +``` +Starting test workflow with id 'test-c30dded9-dbed-454c-b572-25d4d1222067'. Connecting to lock workflow 'locked-resource-id' +Starting test workflow with id 'test-d2a4642b-6bfd-45ee-aba1-321c08e2cb02'. Connecting to lock workflow 'locked-resource-id' +Test workflow 'test-c30dded9-dbed-454c-b572-25d4d1222067' started +Test workflow 'test-d2a4642b-6bfd-45ee-aba1-321c08e2cb02' started +Test workflow 'test-c30dded9-dbed-454c-b572-25d4d1222067' finished after 5506ms +Test workflow 'test-d2a4642b-6bfd-45ee-aba1-321c08e2cb02' finished after 10444ms +``` diff --git a/src/Mutex/TemporalioSamples.Mutex.csproj b/src/Mutex/TemporalioSamples.Mutex.csproj new file mode 100644 index 0000000..b779533 --- /dev/null +++ b/src/Mutex/TemporalioSamples.Mutex.csproj @@ -0,0 +1 @@ + diff --git a/src/Mutex/WorkflowWithMutex.workflow.cs b/src/Mutex/WorkflowWithMutex.workflow.cs new file mode 100644 index 0000000..b9d0115 --- /dev/null +++ b/src/Mutex/WorkflowWithMutex.workflow.cs @@ -0,0 +1,43 @@ +namespace TemporalioSamples.Mutex; + +using Temporalio.Workflows; +using TemporalioSamples.Mutex.Impl; + +public record WorkflowWithMutexInput(string ResourceId, TimeSpan SleepFor, TimeSpan LockTimeout); + +[Workflow] +public class WorkflowWithMutex +{ + private static readonly ActivityOptions ActivityOptions = new() + { + StartToCloseTimeout = TimeSpan.FromMinutes(1), + }; + + [WorkflowRun] + public async Task RunAsync(WorkflowWithMutexInput input) + { + var currentWorkflowId = Workflow.Info.WorkflowId; + var logger = Workflow.Logger; + + using var ls1 = logger.BeginScope(new KeyValuePair[] + { + new("ResourceId", input.ResourceId), + }); + + logger.LogInformation("Started workflow '{WorkflowId}'!", currentWorkflowId); + + await using (var lockHandle = await WorkflowMutex.LockAsync(input.ResourceId, input.LockTimeout)) + { + logger.LogInformation("[{WorkflowId}]: Acquired lock. Release signal ID: '{ReleaseSignalId}'", currentWorkflowId, lockHandle.ReleaseSignalName); + + var notifyLockedInput = new NotifyLockedInput(input.ResourceId, lockHandle.ReleaseSignalName); + await Workflow.ExecuteActivityAsync(() => Activities.NotifyLocked(notifyLockedInput), ActivityOptions); + + var useApiInput = new UseApiThatCantBeCalledInParallelInput(input.SleepFor); + await Workflow.ExecuteActivityAsync(() => Activities.UseApiThatCantBeCalledInParallelAsync(useApiInput), ActivityOptions); + } + + var notifyUnlockedInput = new NotifyUnlockedInput(input.ResourceId); + await Workflow.ExecuteActivityAsync(() => Activities.NotifyUnlocked(notifyUnlockedInput), ActivityOptions); + } +}