snatch-dev/chronicle

Implementation of saga pattern for .NET Core

Chronicle

Chronicle is simple process manager/saga pattern implementation for .NET Core that helps you manage long-living and distirbuted transactions.

	master	develop
AppVeyor
CodeCov

Installation

Chornicle is available on NuGet

Package manager

Install-Package Chronicle_ -Version 3.2.1

.NET CLI

dotnet add package Chronicle_ --version 3.2.1

Getting started

In order to create and process a saga you need to go through a few steps:

1. Create a class that dervies from either Saga or Saga<TData>.
2. Inside your saga implemention, inherit from one or several ISagaStartAction<TMessage> and ISagaAction<TMessage> to implement HandleAsync() and CompensateAsync() methods for each message type. An initial step must be implemented as an ISagaStartAction<TMessage>, while the rest can be ISagaAction<TMessage>. It's worth mentioning that you can implement as many ISagaStartAction<TMessage> as you want. In this case, the first incoming message is going to initialize the saga and any subsequent ISagaStartAction<TMessage> or ISagaAction<TMessage> will only update the current saga state.
3. Register all your sagas in Startup.cs by calling services.AddChronicle(). By default, AddChronicle() will use the InMemorySagaStateRepository and InMemorySagaLog for maintaining SagaState and for logging SagaLogData in the SagaLog. The SagaLog maintains a historical record of which message handlers have been executed. Optionally, AddChronicle() accepts an Action<ChronicleBuilder> parameter which provides access to UseSagaStateRepository<ISagaStateRepository>() and UseSagaLog<ISagaLog>() for custom implementations of ISagaStateRepository and ISagaLog. If either method is called, then both methods need to be called.
4. Inject ISagaCoordinator and invoke ProcessAsync() methods passing a message. The coordinator will take care of everything by looking for all implemented sagas that can handle a given message.
5. To complete a successful saga, call CompleteSaga() or CompleteSagaAsync(). This will update the SagaState to Completed. To flag a saga which has failed or been rejected, call the Reject() or RejectAsync() methods to update the SagaState to Rejected. Doing so will utilize the SagaLog to call each message type's CompensateAsync() in the reverse order of their respective HandleAsync() method was called. Additionally, an unhanded exception thrown from a HandleAsync() method will cause Reject() to be called and begin the compensation.

Below is the very simple example of saga that completes once both messages (Message1 and Message2) are received:

public class Message1
{
    public string Text { get; set; }
}

public class Message2
{
    public string Text { get; set; }
}

public class SagaData
{
    public bool IsMessage1Received { get; set; }
    public bool IsMessage2Received { get; set; }
}

public class SampleSaga : Saga<SagaData>, ISagaStartAction<Message1>, ISagaAction<Message2>
{
    public Task HandleAsync(Message1 message, ISagaContext context)
    {
        Data.IsMessage1Received = true;
        Console.WriteLine($"Received message1 with message: {message.Text}");
        CompleteSaga();
        return Task.CompletedTask;
    }
    
    public Task HandleAsync(Message2 message, ISagaContext context)
    {
        Data.IsMessage2Received = true;
        Console.WriteLine($"Received message2 with message: {message.Text}");
        CompleteSaga();
        return Task.CompletedTask;
    }

    public Task CompensateAsync(Message1 message, ISagaContext context)
        => Task.CompletedTask;

    public Task CompensateAsync(Message2 message, ISagaContext context)
        => Task.CompletedTask;

    private void CompleteSaga()
    {
        if(Data.IsMessage1Received && Data.IsMessage2Received)
        {
            Complete();
            Console.WriteLine("SAGA COMPLETED");
        }
    }
}

Both messages are processed by mentioned coordinator:

var coordinator = app.ApplicationServices.GetService<ISagaCoordinator>();

var context = SagaContext
    .Create()
    .WithCorrelationId(Guid.NewGuid())
    .Build();

coordinator.ProcessAsync(new Message1 { Text = "Hello" }, context);
coordinator.ProcessAsync(new Message2 { Text = "World" }, context);

The result looks as follows:

Documentation

If you're looking for documentation, you can find it here.

Icon

Icon made by Smashicons from www.flaticon.com is licensed by Creative Commons BY 3.0

chronicle's Issues

RegisterSagas also in dependent assemblies

Hi @GooRiOn,
I have a small request with adding possibility of registering Sagas also from dependent assemblies. Currently in Chronicle.Extensions class in method RegisterSagas(...) you use Scrutor scan with method FromAssemblies(..), I think adding also FromAssemblyDependencies(...) can extend the possible usage of Chronicle.
What you think about it?

btw. Chronicle is great and powerful weapon!

Add support for peristence with MongoDB, Redis, SQL Server

More properties in ISagaContext

Hi is it possible to add more properties in ISagaContext interface?

Compensate exception handling and ILogger<> to Chronicle Builder

https://github.com/chronicle-stack/Chronicle/blob/128bdeb31efc4e1deaf9e709fe2be916b8a04e44/src/Chronicle/Managers/SagaPostProcessor.cs#L42

When an exception gets thrown in a HandleAsync(), the saga is rejected. Do you think there is a preferred way of handling an unhandled exception thrown from a CompensateAsync() method that would allow the saga to be recovered and continue compensations at a later date?

Maybe update the log to maintain a status of successfully executed compensations and an additional state of 'RejectCompleted'?

Either way, I'm going to start considering adding an optional UseLogging(ILogger<>) to the ChronicleBuilder and expand on logging.

Add saga versioning

Saga class needs version property to avoid concurrent saving in repository.

Add optional IDataProtectionProvider to encrypt SagaLogData and SagaState at rest

CompensateAsync not fired in When using saga inside Azure Service Bus handler

I have tried to use Chronicle ProcessAsync from Azure Service Bus message handler, like below:
`
public void RegisterQueueHandler(string primaryKey, string queueName) where T : class
{
var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
{
MaxConcurrentCalls = 1,
AutoComplete = false
};
_queueClient = new QueueClient(primaryKey, queueName);
_queueClient.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions);
}

    private async Task ProcessMessagesAsync<T>(Message message, CancellationToken token) where T : class
    {
        using var serviceProviderScope = _serviceProvider.CreateScope();
        var payload = JsonConvert.DeserializeObject<T>(Encoding.UTF8.GetString(message.Body));

        if (payload is ISagaCommand)
        {
            var sagaCoordinator = serviceProviderScope.ServiceProvider.GetRequiredService<ISagaCoordinator>();
            await sagaCoordinator.ProcessAsync(payload, null);
        }
        else
        {
            var mediator = serviceProviderScope.ServiceProvider.GetRequiredService<IMediator>();
            await mediator.Send(payload, token);
        }

        await _queueClient.CompleteAsync(message.SystemProperties.LockToken);
    }

`

When any exception happens inside saga HandleAsync, CompensateAsync is ignored and ExceptionReceivedHandler fired.

Add documentation

Can you please provide more documentation

1. I would like to know if the compensate logic was called inside the saga or not after calling complete? so that I can respond to the end-user with the right message how can I do that?

ISagaCoordinator.ProcessAsync is ambiguous

Suppose you don't have any context to pass to the Saga Coordinator:

sagaCoordinator.ProcessAsync(myMessage);

Unfortunately this call is ambiguous because the two method overloads have only optional parameters.

Task ProcessAsync<TMessage>(TMessage message, ISagaContext context = null) where TMessage : class;
Task ProcessAsync<TMessage>(TMessage message, Func<TMessage, ISagaContext, Task> onCompleted = null, Func<TMessage, ISagaContext, Task> onRejected = null, ISagaContext context = null) where TMessage : class;

So in the end you get this error:

CS0121	The call is ambiguous between the following methods or properties: 'ISagaCoordinator.ProcessAsync<TMessage>(TMessage, ISagaContext)' and 'ISagaCoordinator.ProcessAsync<TMessage>(TMessage, Func<TMessage, ISagaContext, Task>, Func<TMessage, ISagaContext, Task>, ISagaContext)'

Get all the pending saga of a given type

What about adding a method to get all the pending saga of a given type?

Something like this:

    public interface ISagaStateRepository
    {
        // ...

        Task<IEnumerable<ISagaState>> ReadPendingAsync(Type type);
        // or...
        Task<IEnumerable<ISagaState>> ReadAsync(Type type, SagaStates state);
    }

Upgrade to .NET 6.0

Superseding issue #62

Reject method does not work using Chronicle_ 2.0.1 NuGet

Running TestApp (repository's test application) with the Chronicle source code, the application works fine. If you use Chronicle_ NuGet and run TestApp, Reject() does not work as expected.

public Task HandleAsync(Message2 message, ISagaContext context)
        {
            Reject();
            Data.IsMessage2 = true;
            Console.WriteLine("M2 reached!");
            CompleteSaga();
            return Task.CompletedTask;
        }

Consider renaming the project

The Chronicle_ package name is a bit confusing, sinche there is another Chronicle on NuGet.
Have you considered renaming it?

Support use of MongoClientSettings to configure mongodb persistence

Chronicle.Integrations.MongoDb only supports configuration of a MongoClient by use of a connection string.

MongoClient has an overload that accepts an instance of MongoClientSettings, which supports many additional properties that aren't configurable by use of a simple connectionstring.

Unhandled exception in HandleAsync does not call Reject state

Additionally, an unhandled exception thrown from a HandleAsync() method will cause Reject() to be called and begin the compensation.

But this does not happen in this version (or maybe in the newer version as well) I had to catch the exception in the HandleAsync method and explicitly call reject. This looks not so clean at the moment - is this how it is supposed to work? This is how my code looks now. I was expecting not to handle any exceptions and let your awesome package take care of calling compensation.

 /// <summary>
        /// Handles the asynchronous.
        /// </summary>
        /// <param name="task">The task.</param>
        /// <param name="sagaContext">The saga context.</param>
        public async Task HandleAsync(CompleteTask task, ISagaContext sagaContext)
        {
             try
            {
                // service call here. 
                this.CompleteSaga();
                await Task.CompletedTask;
            }
            catch (System.Exception ex)
            {
                // log exception here
                await RejectAsync();              
            }
        }

Wait until Saga is Completed

Is there any "legal" way to wait until the Saga is Completed?
Sometimes need to start Saga and block, for example, UI operation until Saga is Completed or Rejected.

var coordinator = app.ApplicationServices.GetService<ISagaCoordinator>();

var context = SagaContext
    .Create()
    .WithCorrelationId(Guid.NewGuid())
    .Build();

// Sending initial Message/Event
coordinator.ProcessAsync(new Message1 { Text = "Hello" }, context);

// Desired behavior: block until Saga is Completed or Rejected and return ISagaState
var state = coordinator.WaitAsync(context);

Now I can do it with a few stupid lines of code:

// Helper extensions
public static class SagaExtensions
    {
        public static async Task<ISagaState> WaitAsync(this ISagaCoordinator coordinator, ISagaContext context, ISagaStateRepository sagaStateRepository)
        {
            var sagaState = await sagaStateRepository.GetStateAsync(context);
            while (sagaState is null || sagaState.State == SagaStates.Pending)
            {
                await Task.Delay(TimeSpan.FromSeconds(1));
                sagaState = await sagaStateRepository.GetStateAsync(context);
            }

            return sagaState;
        }

        public static async Task<ISagaState> GetStateAsync(this ISagaStateRepository sagaStateRepo, ISagaContext context)
        {
            return await sagaStateRepo.ReadAsync(context.SagaId, (Type) context.Metadata.First(md => md.Key == "sagaType").Value);
        }
    }

// Then in some place
var coordinator = app.ApplicationServices.GetService<ISagaCoordinator>();

var context = SagaContext
    .Create()
    .WithSagaId(SagaId.NewSagaId())
    .WithOriginator("Test")
    .WithMetadata("key", "lulz")
    .WithMetadata("sagaType", typeof(SampleSaga))
    .Build();

// Sending initial Message/Event
coordinator.ProcessAsync(new Message1 { Text = "Hello" }, context);

// Block until Saga is Completed or Rejected and return ISagaState
var sagaStateRepo = app.ApplicationServices.GetService<ISagaStateRepository>();
var state = await coordinator.WaitAsync(context, sagaStateRepo);

if (sagaState.State is SagaStates.Rejected)
{
// onRejected()
}

Saga parallelism

It would be good to be able to perform optimistic concurrency check to avoid data corruption when running a few instances at the same time.

I propose to add Revision property of type uint to the ISagaState.

Add support for outbound queue

There's a possibility that once the HandleAsync method finishes the state of the saga will not be persisted into DB. This leads to the state where we have inconsistent transaction. Outbound queue might be handy in this case.

need some sample for integration of redis or mongo in test project for custom implementation of ISagaStateRepository

need some sample for integration of redis or mongo in test project for custom implementation of ISagaStateRepository

Add support to .NET Core 3.1

Reusing Saga Coordinator

How do I reuse the coordinator. I tried to use the Reject functionality and it's working fine. But the next time unable to use the coordinator in subsequent request. I can see the state as Rejected.

Add support for .net core 2.2

We are using this library for some time and we want to continue using - is there a plan to upgrade to .net core 2.2?

Link to documentation is broken

Link from Readme.MD is broken :

https://github.com/chronicle-stack/Chronicle/blob/master/chronicle.readthedocs.io

Saga timeout

It would be nice if the saga could set a timeout in the message handlers.

For example, the saga could implement such an interface, similar to ISagaAction:

    public interface ISagaTimeoutAction
    {
        Task HandleTimeoutAsync(ISagaContext context);
        Task CompensateTimeoutAsync(ISagaContext context);
    }

(I'm supposing that a timeout doesn't necessarily complete the Saga.)

Timeout could be scheduled in message handler through a method of ISaga

    public interface ISaga
    {
        // ...
        void ScheduleTimeout(TimeSpan timeSpan, ISagaContext context);
    }

Source of inspiration: https://docs.particular.net/nservicebus/sagas/timeouts

How do get a event on SAGA

I saw your the DNC-DShop video, and the bus that you use to publish event i guess is RabbitMQ Raw, i want to know if i can use Chronicle library with mediatoR, the current lib for bus i 'm using

Typo in readme

"disitrbuted "

Migrate to .NET 5

Option to remove SagaLog and SagaState from configured persistence.

https://github.com/chronicle-stack/Chronicle/blob/128bdeb31efc4e1deaf9e709fe2be916b8a04e44/src/Chronicle/Managers/SagaPostProcessor.cs#L28-L31

I'm working on a RedisSagaStateRepository and RedisSagaLog. Do you think a configuration option to immediately remove the SagaLog and SagaState on SagaStatus.Completed would be appropriate or leave that for the onComplete() and onRejected() methods?

I think it would require adding void Remove(SagaId id, Type type) to ISagaLog and ISagaStateRepository.