SHAKA.DWE 1.1.0

SHAKA.DWE Core Abstractions Documentation

SHAKA.DWE is a core infrastructure-agnostic library for distributed workflow management and it provide the foundation for defining, executing, and managing distributed workflows.

Core Abstractions

First part of the library defines abstractions for Workflow Engine, Messaging, Persistence, Observability and Resilience. Here we have bunch of interfaces that define the contracts for each component. Some of this abstractions that doesn't require any specific infrastructure is implemented within SHAKA.DWE itself, but the abstractions are still maintain to allow for custom implementations and replacement if needed.

Before jumping in the implementations of those abstractions I want to mention some of the core models that make up the workflow engine.

Core Models

• WorkflowContext: Carries data, state and correlation context through the workflow execution.

• WorkflowInstance: Represents a running instance of a workflow, including its current state and progress.

• IWorkflowStep: Represents a single step in a workflow. It contains the Action that will be executed to publishing the event for the subscriber microservice, upon which that microservice will perform the actual business logic.

• WorkflowStepHistory: Tracks execution history of each step (attempts, results, timestamps, retry counts, error messages and the index of the step in the workflow).

• StepResult: Represents the outcome of the publishing a workflow event that should trigger the step in the subscriber microservice. It can be that publishing failed due to connection issues or other in which can error message will be also maintained.

• WorkflowMetaData: Metadata about the workflow definition (version, author, etc.).

• WorkflowDefinition: Describes the structure and metadata of a workflow. It contains information about the WorkflowSteps which can be used by the workflow engine to orchestrate the execution. It also provides a InitiatorCompensation that is a specific compensation policy that will be executed during the workflow failure if the initial step of the workflow was not initiated from the workflow engine( If the orchestrator is separated service and microservice itself doesn't implement the orchestration).

• ResponseEventType: Each step should receive an event from the subscriber microservice that indicates the result of the step execution. the result can be the success or failure and we allow the users to define so called ResponseEvents for each step specifying wether this event will indicate a success or failure of the step. This is important for the workflow engine to determine wether it needs to proceed to the next step or trigger the compensation logic.

Both WorkflowContext and WorkflowInstance has properties named Data and TypedBusinessData, the data property is used to store the information about the progress, state and execution of the workflow, while the TypedBusinessData is a representation of the business data that the workflow is processing, this business data will be used in every step of the workflow and they needs to be propagated properly. Since the sturcture of the businessData is unknown in advance for the workflow engine we store it as an object and we also and we also store its type information, then we have a method GetBusinessData<TData> that allow us to retrieve the business data in a strongly typed manner. TypedBusinessDataType property names can be used to serialize/deserialize the business data in the infrastructure specific implementations

Core Implementations

First of all we have the resilience abstractions with the implementations within this library.

• IRetryPolicy: Defines retry logic for steps.
  • SimpleRetryPolicy: Implements a fixed delay retry strategy.
  • ExponentialBackoffRetryPolicy: Implements an exponential backoff retry strategy.
  • CircuitBreakerRetryPolicy: Implements a circuit breaker pattern for retries.
• ICompensationPolicy: Defines compensation logic for failed steps.
  • CompensationPolicy which stores the action that should be executed as compensation. This action will be the method that publish the compensation trigger event for the subscriber microservice.

We also have some Registry abstractions with in-implementations for WorkflowDefinitions and ResponseEvents.

• IWorkflowRegistry: Stores and manages registered workflow definitions.

  • WorkflowRegistry: Uses in-memory Dictionary to store the workflow definitions considering the versions also.

• IResponseEventRegistry: Contains all the response events that have been registered and their corresponding ResponseEventType. It also allows to query the event type by the event name and registration of new response events.

  • ResponseEventRegistry: Uses In-memory Dictionary to store the response events and their types.

• IWorkflowEngine: Main interface for starting and managing workflows. It defines the methods like StartWorkflow and ResumeWorkflow

  • WorkflowEngine: Implements the workflow orchestration logic, including step execution, retries, and compensation.
    • In the StartWorkflow method, it retrieves a workflow definition using the IWorkflowRegistry and uses it to create an persist a new WorkflowInstance. It then initiates the first step of the workflow using ExecuteNextStepWithSimpleRetryForPublishing method.
    • ExecuteNextStepWithSimpleRetryForPublishing within itself uses the simple retry policy for executing the actual workflow step action this action is just publishing the workflow event that will trigger the subscriber microservice to perform the actual business logic and previously mentioned reslience policies are not for this action but for the actual business logic that will be performed by the subscriber microservice.
    • ExecuteNextStep within itself tries to check if there is a next step at all and if not it'll complete the workflow otherwise
    • it'll initialize the retry state for the business action of this step and executes the action.
    • HandleExecuteNextStepResult: method always called after the action retrial execution and it checks if the result is not successful it'll declare the step as failed and persist it, otherwise it'll declare the step as Suspended , after which the workflow can be resumed upon receiving the response event from the subscriber microservice.
    • ResumeWorkflow: method is called when the workflow engine receives a response event from the subscriber microservice and it retrieves the corresponding workflow information using the CorrelationId, her it also handles the business logic failure scenario and firstly applies previously mentioned retry policies(if configured) after which if the business action still fail it'll execute compensation actions for each of the previous steps and declare the workflow as failed. If the subscriber microservice indicated the success scenario this method will also use the ExecuteNextStepWithSimpleRetryForPublishing method to proceed to the next step of the workflow.

• ServiceCollectionExtensions: Extension methods for registering workflows and related services in the DI container.

  • AddWorkflowEngine: Method registers WorkflowEngine and eagerly registers singleton instances for WorkflowRegistry and ResponseEventRegistry.

Infrastructure Specific Abstractions

Messaging Abstraction

• IWorkflowEventBus: Abstraction for publishing workflow events.
• IWorkflowEvent: Base interface for workflow-related events.
• IWorkflowEventHandler<TEvent>: Handles workflow events of type TEvent.

Persistence Abstraction

• IWorkflowRepository: CRUD and query operations for workflow instances and state.
• IUnitOfWork: Ensures atomicity for workflow state and event log updates.

Example: ORDER_FULFILLMENT Workflow (Usage Simulation)

Below is a example of how these library could be used to define a workflow similar to the OrderFulfillment.

public const string ORDER_FULFILLMENT_WORKFLOW_NAME = nameof(OrderFulfillment);
public static IServiceCollection AddOrderFulfillment(this IServiceCollection services)
{
    services.AddWorkflowDefinition(
        name: ORDER_FULFILLMENT_WORKFLOW_NAME,
        version: "1.0",
        configureWorkflow: builder =>
        {
            builder.WithInitiatorCompensation(RejectOrder);
            builder.AddStep(ReserveInventory, step => step
                .WithExponentialBackoffRetry(
                    maxAttempts: 3,
                    initialDelay: TimeSpan.FromSeconds(1),
                    backoffFactor: 2.0,
                    addJitter: true,
                    maxDelay: TimeSpan.FromSeconds(30))
                .WithPossibleResponse<InventoryReserved>(ResponseEventType.Success)
                .WithPossibleResponse<InventoryReservationFailed>(ResponseEventType.Failure)
                .WithCompensation(ReleaseInventory)
            );
            builder.AddStep(ProcessPayment, step => step
                .WithCircuitBreakerRetry(
                    maxAttempts: 3,
                    delay: TimeSpan.FromSeconds(2),
                    failureThreshold: 2,
                    durationOfBreak: TimeSpan.FromSeconds(30),
                    successThreshold: 2)
                .WithPossibleResponse<PaymentCreated>(ResponseEventType.Success)
                .WithPossibleResponse<PaymentFailed>(ResponseEventType.Failure)
                .WithCompensation(RefundPayment)
            );
            builder.AddStep(CompleteOrder, step => step
                .WithSimpleRetry(3, TimeSpan.FromSeconds(2))
                .WithPossibleResponse<OrderCompleted>(ResponseEventType.Success)
                .WithPossibleResponse<OrderCompletionFailed>(ResponseEventType.Failure)
            );
        });

    return services;

    public class OrderCreatedEventHandler(IWorkflowEngine workflowEngine, ILogger<OrderCreatedEventHandler> logger)
    : IWorkflowEventHandler<OrderCreated>
    {
        public async Task Handle(OrderCreated @event, CancellationToken cancellationToken = default)
        {
            var data = new OrderFulfillmentData
            {
                OrderId = @event.OrderId,
                CustomerId = @event.CustomerId,
                Items = @event.Items
            };

            logger.LogInformation("Starting OrderFulfillment workflow for CustomerId: {CustomerId} with OrderItems: {@OrderItems}",
                @event.CustomerId, @event.Items);

            await workflowEngine.StartWorkflow(OrderFulfillment.ORDER_FULFILLMENT_WORKFLOW_NAME, data, cancellationToken);
        }
    }

}

This example demonstrates how the abstractions enable workflow definition, registration, and execution in a way that is completely decoupled from any specific infrastructure.

• Observability & Middleware

• IWorkflowMiddleware: Allows injection of custom logic (logging, tracing, auditing) into workflow execution.(not yet implemented)
• ICorrelationContext: Provides correlation IDs for distributed tracing(not yet implemented).