DEV Community: Federico Bevione

Transactions in Microservices: Part 3 - SAGA Pattern with Orchestration and Temporal.io.

Federico Bevione — Sun, 09 Feb 2025 14:44:46 +0000

In the second article of this series, we explored the Choreography approach to distributed transactions using RabbitMQ. Now, let’s shift our focus to the Orchestration approach, where a central orchestrator manages the entire workflow.

To make this practical, we’ll implement the same healthcare workflow but this time with Temporal.io, a robust orchestration platform for microservices. Each service will have its own logic, but the workflow coordination will be managed centrally.

What is SAGA Orchestration?

Orchestration centralizes the coordination of a distributed workflow. Instead of services emitting and consuming events autonomously, the orchestrator invokes service-specific logic and tracks progress.

Key Benefits:

Centralized Workflow Management: All logic resides in the orchestrator, making workflows easier to visualize and debug.
Stateful Workflows: The orchestrator manages state, enabling retries, timeouts, and compensations.
Error Handling: Built-in mechanisms for compensation and retries reduce boilerplate code.

Challenges:

Single Point of Failure: The orchestrator’s availability is critical.
Complexity in Orchestrator Logic: Centralizing workflow management can lead to complex orchestrator code.
Scalability: The orchestrator may become a bottleneck for high-throughput workflows.

To tackle these challenges, we need a reliable ally—one that can rise to the occasion and handle it all without adding complexity to our lives. Temporal, I choose you!

Who is Temporal?

Temporal is a powerful platform designed to handle complex, long-running workflows with ease. It offers several advantages over traditional approaches to distributed systems and state management.

Before diving into Temporal, it's worth briefly comparing it with other workflow orchestration solutions like AWS Step Functions and Simple Workflow Service (SWF). These tools also provide workflow management, but they have different trade-offs in terms of flexibility, scalability, and cloud dependency.

Comparing Temporal, AWS Step Functions, and SWF

Feature	Temporal.io	AWS Step Functions	AWS SWF
Cloud Agnostic	✅ Yes	❌ No (AWS only)	❌ No (AWS only)
Developer Flexibility	✅ High (code-first workflows)	⚖️ Medium (declarative JSON-based)	🔽 Low (requires AWS SDK)
Scalability	✅ High (self-hosted or managed)	✅ High (AWS-managed)	⚖️ Medium (AWS-managed)
State Management	✅ Durable & Event-driven	⚖️ Managed state, less flexible	⚖️ Managed state but complex
Use Case Suitability	🚀 Microservices, multi-cloud, high-scale	🔹 AWS-native apps, quick automation	🔹 Legacy workflows, low complexity

AWS Step Functions and SWF are easier starting points for teams deeply integrated into AWS. Step Functions, in particular, offer a quick way to define workflows using a declarative JSON structure, making it an appealing option for those who need simple workflow coordination within AWS services.

Bridging the Gap: Why Not Just AWS Step Functions?

For teams deeply integrated into AWS, AWS Step Functions and/or SWF might seem like the default choice. However, as workflows grow in complexity, their limitations become evident:

Limited flexibility: Workflows are defined declaratively, making complex logic harder to maintain.
Tied to AWS: Step Functions are AWS-only, which can be restrictive for multi-cloud or hybrid environments.
State management constraints: While AWS manages state, it lacks the fine-grained control offered by Temporal.

While Step Functions and SWF work well for simple automation and orchestration within AWS, Temporal is a more powerful, long-term solution for microservices workflows.

Why Temporal for SAGA Orchestration?

The SAGA pattern requires a robust workflow orchestrator that can handle long-running processes, retries, and compensation logic. Temporal excels in this area because:

Code-first approach: Unlike Step Functions' JSON-based workflows, Temporal lets developers write workflows in code, making it easier to test, debug, and maintain.
Cloud independence: It works across different cloud environments, avoiding vendor lock-in.
Built-in durability: It provides out-of-the-box state persistence, ensuring long-running workflows remain reliable without extra effort.
Scalability & fault tolerance: Temporal’s architecture is built to scale horizontally and ensures workflows resume seamlessly even after failures.

Key Benefits:

Stateful Workflow Management: Temporal handles workflow state persistence automatically, eliminating the need for external databases or custom solutions.
Automatic Retries and Timeouts: Built-in support for retry policies and timeouts simplifies error handling and ensures reliability without writing additional code.
Simplified Compensation: Temporal provides first-class support for SAGA patterns, making it easy to define and execute compensation logic when workflows fail.
Language-Specific SDKs: Write workflows and activities in your preferred language, using Temporal's SDKs for Go, Java, Node.js, or Python.
Scalability and Resilience: Temporal is designed to scale horizontally, handling thousands of workflows and activities concurrently with minimal configuration.
Event-Driven Architecture: Temporal supports event-driven workflows, enabling seamless integration with other services and systems.
Developer Productivity: Temporal abstracts away complexities like task scheduling, retries, and state management, allowing developers to focus on business logic.

Why Not Traditional Solutions?

Manual State Management: Traditional approaches often require maintaining workflow state in a database, which can be error-prone and hard to scale.
Complex Error Handling: Without Temporal, developers must implement custom retry logic and compensation mechanisms, increasing code complexity.
Tight Coupling: Temporal decouples workflow coordination from business logic, making your system more modular and maintainable.

In short, Temporal simplifies distributed application development by providing a robust, scalable, and developer-friendly framework for managing workflows and activities.

For more details, visit the Temporal documentation.

Practical Example: Healthcare Workflow

We’ll revisit the same healthcare workflow:

Patient Service: Verifies patient details and insurance coverage.
Scheduler Service: Schedules the procedure.
Inventory Service: Reserves medical supplies.
Billing Service: Processes billing.

The workflow will include:

Compensation Logic: In case of failures, services will undo their operations.
Retries: Temporal will retry failed operations automatically based on configured policies.

Setting Up Temporal.io

Deploying the Temporal stack can be challenging due to its requirement for multiple interconnected services. For production-ready environments, I strongly recommend leveraging Temporal Cloud, which simplifies management and ensures reliability by offloading the complexities of running the Temporal service. Additionally, Temporal Cloud offers a very competitive pricing model, making it accessible for projects of various scales.

However, if you enjoy tackling infrastructure challenges, the self-hosted guide provides detailed instructions for setting up Temporal on your own.

For this demo, we’ll keep things simple by using a minimal deployment with Docker Compose.

Start Temporal using Docker Compose:

   git clone https://github.com/fedricobevione/saga_tutorial.git
   docker compose --project-directory ./orchestration/compose/ -f orchestration/compose/docker-compose.yml up

Implementation: Workflow and Activities

What is a Workflow?

A workflow in Temporal is a function that defines the coordination logic for a series of tasks or activities. It represents the overall business process, specifying the order in which activities are executed, how they interact, and how failures are handled.

Key Characteristics:

Deterministic: Workflows must be deterministic, meaning their execution results should not depend on external factors like system time or random values. This ensures that workflows can be replayed consistently.
Long-Running: Workflows can run for extended periods (e.g., days, months, or years) because Temporal automatically handles state persistence.
Fault-Tolerant: Temporal workflows are resilient to failures. If a worker crashes or the system restarts, the workflow resumes from the last persisted state.
Event-Driven: Workflows can respond to external signals and events, enabling real-time interaction with external systems.
Scalable: Temporal workflows can scale horizontally to handle high loads and multiple concurrent executions.

How Workflows Work:

Workflows define the coordination logic, specifying how and when activities are executed.
Temporal ensures that workflows are durable, persisting their state after each decision task.
Developers implement workflows using Temporal's SDKs. For example:

What is an Activity?

An activity in Temporal is a function that performs a specific unit of work, typically involving external systems or resources. Activities are invoked by workflows and are responsible for the actual execution of business logic, such as interacting with APIs, databases, or file systems.

Key Characteristics:

External Work: Activities perform tasks that may involve external systems or I/O operations.
Retryable: Temporal handles automatic retries for failed activities based on configurable retry policies.
Stateless: Activities are stateless and do not maintain persistent state between executions. This ensures they can be retried or executed on any available worker.
Timeouts: Activities have configurable timeouts, ensuring that long-running or stuck tasks do not block the workflow indefinitely.
Language-Specific SDKs: Activities are implemented using Temporal's SDKs (e.g., Go, Java, Node.js, Python).

How Activities Work:

Activities are invoked by workflows as part of the workflow execution.
Temporal schedules activity tasks and assigns them to workers for execution.
Upon completion, the activity's result is sent back to the workflow, which continues execution.

Define the Workflow

The orchestrator will define the sequence of activities.

// workflow.go
package workflow

import (
    "time"

    "github.com/thegoodapi/saga_tutorial/orchestration/activities"
    "go.temporal.io/sdk/temporal"
    "go.temporal.io/sdk/workflow"
    "go.uber.org/multierr"
)

func HealthcareWorkflow(ctx workflow.Context) error {
    logger := workflow.GetLogger(ctx)
    logger.Info("Starting Healthcare Workflow")

    // Define retry options for activities
    retryPolicy := &temporal.RetryPolicy{
        InitialInterval:    time.Second,
        BackoffCoefficient: 2.0,
        MaximumInterval:    time.Minute,
        MaximumAttempts:    5,
    }
    ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
        StartToCloseTimeout: time.Minute,
        RetryPolicy:         retryPolicy,
    })

    // Step 1: Verify Patient
    logger.Info("Verifying Patient")
    err := workflow.ExecuteActivity(ctx, activities.VerifyPatientActivity).Get(ctx, nil)
    if err != nil {
        logger.Error("Patient verification failed", "Error", err)
        return err
    }

    // compensation for patient verification
    defer func() {
        if err != nil {
            errCompensation := workflow.ExecuteActivity(ctx, activities.NotifyProcedureScheduleCancellationActivity).Get(ctx, nil)
            err = multierr.Append(err, errCompensation)
        }
    }()

    // Step 2: Schedule Procedure
    logger.Info("Scheduling Procedure")
    err = workflow.ExecuteActivity(ctx, activities.ScheduleProcedureActivity).Get(ctx, nil)
    if err != nil {
        logger.Error("Procedure scheduling failed", "Error", err)
        return err
    }

    // compensation for procedure scheduling
    defer func() {
        if err != nil {
            errCompensation := workflow.ExecuteActivity(ctx, activities.CancelProcedureScheduleActivity).Get(ctx, nil)
            err = multierr.Append(err, errCompensation)
        }
    }()

    // Step 3: Reserve Supplies
    logger.Info("Reserving Supplies")
    err = workflow.ExecuteActivity(ctx, activities.ReserveSuppliesActivity).Get(ctx, nil)
    if err != nil {
        logger.Error("Supply reservation failed", "Error", err)
        return err
    }

    // compensation for supply reservation
    defer func() {
        if err != nil {
            errCompensation := workflow.ExecuteActivity(ctx, activities.ReleaseReservedSuppliesActivity).Get(ctx, nil)
            err = multierr.Append(err, errCompensation)
        }
    }()

    // Step 4: Process Billing
    logger.Info("Processing Billing")
    err = workflow.ExecuteActivity(ctx, activities.ProcessBillingActivity).Get(ctx, nil)
    if err != nil {
        logger.Error("Billing failed", "Error", err)
        return err
    }

    // compensation for billing
    defer func() {
        if err != nil {
            errCompensation := workflow.ExecuteActivity(ctx, activities.ProcessBillingCompensationActivity).Get(ctx, nil)
            err = multierr.Append(err, errCompensation)
        }
    }()

    logger.Info("Healthcare Workflow completed successfully")
    return nil
}

Define Activities

Each activity encapsulates the logic for a service.

Verify Patient Activity

// patient.go
func VerifyPatientActivity() error {
    // Simulate logic
    fmt.Println("Step 1: Verifying patient...")
    return nil
}

func NotifyProcedureScheduleCancellationActivity() error {
    // Simulate logic
    fmt.Println("Compensation 1: Notify patient of verification failure.")
    return nil
}

Schedule Procedure Activity

// scheduler.go
func ScheduleProcedureActivity() error {
    // Simulate scheduling logic
    fmt.Println("Step 2: Scheduling procedure...")
    return nil // or return an error to simulate failure
}

func CancelProcedureScheduleActivity() error {
    // Simulate cancellation logic
    fmt.Println("Compensation 2: Cancel procedure schedule.")
    return nil
}

Reserve Supplies Activity

// inventory.go
func ReserveSuppliesActivity() error {
    // Simulate scheduling logic
    fmt.Println("Step 3: Reserving medical supplies...")
    return nil // or return an error to simulate failure
}

func ReleaseReservedSuppliesActivity() error {
    // Simulate cancellation logic
    fmt.Println("Compensation 3: Release reserved supplies.")
    return nil
}

Process Billing Activity

// billing.go
func ReserveSuppliesActivity() error {
    // Simulate scheduling logic
    fmt.Println("Step 3: Reserving medical supplies...")
    return nil // or return an error to simulate failure
}

func ReleaseReservedSuppliesActivity() error {
    // Simulate cancellation logic
    fmt.Println("Compensation 3: Release reserved supplies.")
    return nil
}

Create a worker

A Temporal worker is a process that executes workflows and activities in a Temporal application. It serves as the backbone for running the business logic defined in your workflows and activities.

Key Responsibilities:

Executing Workflows: A worker processes the workflow code, including managing workflow state, handling signals, and executing timers.
Running Activities: It performs activities, which are the core units of work that interact with external systems (e.g., databases, APIs, or services).
Retry Logic: Workers handle retries for failed activities or workflows based on the retry policies defined in your code.
Scalability: Multiple workers can be deployed to distribute the load of workflows and activities across different machines or instances.

How It Works:

Workers communicate with the Temporal service, which coordinates task scheduling and state management.
When a workflow or activity task is scheduled, the Temporal service assigns it to an available worker.
Workers listen for tasks, execute them, and report the results back to the Temporal service.

Key Features:

Fault Tolerance: Workers can crash or go offline without affecting workflow execution, as the Temporal service persists state and ensures tasks are reassigned to available workers.
Dynamic Scaling: New worker processes can be added to handle increased load dynamically.

In short, Temporal workers are essential for executing and scaling Temporal workflows and activities, providing the foundation for distributed and reliable task execution.

For more details, visit the Temporal documentation.

package main

import (
    "log"

    "github.com/thegoodapi/saga_tutorial/orchestration/activities"
    "github.com/thegoodapi/saga_tutorial/orchestration/workflow"
    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
)

func main() {
    // Create the client object just once per process
    c, err := client.Dial(client.Options{})
    if err != nil {
        log.Fatalln("unable to create Temporal client", err)
    }
    defer c.Close()
    // This worker hosts both Workflow and Activity functions
    w := worker.New(c, "healthcare-queue", worker.Options{})

    w.RegisterWorkflow(workflow.HealthcareWorkflow)
    w.RegisterActivity(activities.VerifyPatientActivity)
    w.RegisterActivity(activities.NotifyProcedureScheduleCancellationActivity)
    w.RegisterActivity(activities.ScheduleProcedureActivity)
    w.RegisterActivity(activities.CancelProcedureScheduleActivity)
    w.RegisterActivity(activities.ReserveSuppliesActivity)
    w.RegisterActivity(activities.ReleaseReservedSuppliesActivity)
    w.RegisterActivity(activities.ProcessBillingActivity)
    w.RegisterActivity(activities.ProcessBillingCompensationActivity)
    // Start listening to the Task Queue
    err = w.Run(worker.InterruptCh())
    if err != nil {
        log.Fatalln("unable to start Worker", err)
    }
}

Create a starter

What is a Starter?

A starter in Temporal is a client application or process responsible for initiating a workflow execution. It acts as the entry point to Temporal workflows, sending a signal to the Temporal service to begin processing the workflow logic.

Key Characteristics:

Workflow Trigger: The starter initiates a workflow by specifying its type, parameters, and execution options.
Decoupled Invocation: A starter can be a standalone process or integrated into your application, enabling external systems or user actions to trigger workflows.
Stateless: The starter itself does not manage the workflow’s state; it delegates this responsibility to the Temporal service.
Configurable Options: Starters can set workflow-specific configurations like timeouts, IDs, and retry policies.

How a Starter Works:

The starter communicates with the Temporal service via Temporal’s client SDK.
It specifies the workflow type, input arguments, and optional execution parameters.
Once initiated, the Temporal service schedules the workflow, assigns it to a worker, and manages its lifecycle.

package main

import (
    "context"
    "log"

    "github.com/thegoodapi/saga_tutorial/orchestration/workflow"
    "go.temporal.io/sdk/client"
)

func main() {
    // Create the client object just once per process
    c, err := client.Dial(client.Options{})
    if err != nil {
        log.Fatalln("unable to create Temporal client", err)
    }
    defer c.Close()
    options := client.StartWorkflowOptions{
        ID:        "my-id",
        TaskQueue: "healthcare-queue",
    }

    ctx := context.Background()

    // Start the workflow
    we, err := c.ExecuteWorkflow(ctx, options, workflow.HealthcareWorkflow)
    if err != nil {
        log.Fatalln("error starting TransferMoney workflow", err)
    }

    // wait for workflow completion
    err = we.Get(ctx, nil)
    if err != nil {
        log.Fatalln("workflow error", err)
    }
}

Running the Workflow

Start Temporal Worker:

   go run orchestration/worker/worker.go

Start the Workflow:

   go run orchestration/start/start.go

Observe Output:
Logs from the worker and Temporal Web UI will show the workflow execution progress.
The magnificent UI:
With your docker compose running, navigate to http://localhost:8080, you'll find a very usefull ui where you can easily debug your workflow executions

Key Takeaways

Centralized Control: Temporal.io simplifies workflow coordination and state management.
Automatic Retries: Configurable retry policies reduce failure handling overhead.
Compensation Logic: Temporal makes it easy to implement and manage compensations.

What’s Next?

In the next article, we’ll compare Choreography and Orchestration approaches, diving into their strengths, weaknesses, and real-world use cases. We’ll help you understand when to choose one over the other based on your system's requirements, such as scalability, fault tolerance, and ease of debugging.

In future articles of this serie, we’ll also cover some key observability aspects to ensure you can effectively monitor and debug distributed workflows. This will include:

Logging: How to implement structured logging to trace workflow and activity execution.
Metrics: Using tools like Prometheus to track execution times, task queues, and retry rates.
Tracing: Leveraging distributed tracing (e.g., OpenTelemetry) to visualize workflow execution across services.

These insights will provide you with the knowledge to build more robust, scalable, and maintainable distributed systems.

Stay tuned for Part 4, where we bring it all together!

Check out the full repository for this series here. Have questions or feedback? Let’s discuss in the comments!

Transactions in Microservices: Part 2 - SAGA Pattern with Choreography

Federico Bevione — Wed, 22 Jan 2025 16:42:52 +0000

In the first article of this series, we introduced the SAGA pattern and demonstrated how a minimal Orchestration can manage distributed transactions with a central orchestrator.

Let’s get real! This time, we’ll dive into the Choreography approach, where services coordinate workflows by autonomously emitting and consuming events.

To make this practical, we’ll implement a multi-service healthcare workflow using Go and RabbitMQ. Each service will have its own main.go, making it easy to scale, test, and run independently.

What is SAGA Choreography?

Choreography relies on decentralized communication. Each service listens for events and triggers subsequent steps by emitting new events. There’s no central orchestrator; the flow emerges from the interactions of individual services.

Key Benefits:

Decoupled Services: Each service operates independently.
Scalability: Event-driven systems handle high loads efficiently.
Flexibility: Adding new services doesn’t require changing the workflow logic.

Challenges:

Debugging Complexity: Tracking events across multiple services can be tricky. (I'll write an article dedicated to this topic, stay tuned!)
Infrastructure Setup: Services require a robust message broker (e.g., RabbitMQ) to connect all the dots.
Event Storms: Poorly designed workflows can overwhelm the system with events.

Practical Example: Healthcare Workflow

Let’s revisit our healthcare workflow from the first article:

Patient Service: Verifies patient details and insurance coverage.
Scheduler Service: Schedules the procedure.
Inventory Service: Reserves medical supplies.
Billing Service: Processes billing.

Each service will:

Listen for specific events using RabbitMQ.
Emit new events to trigger subsequent steps.

Setting Up RabbitMQ with Docker

We’ll use RabbitMQ as the event queue. Run it locally using Docker:

docker run --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0.5-management

Access the RabbitMQ management interface at http://localhost:15672 (username: guest, password: guest).

Exchanges, Queues, and Bindings Setup

We need to configure RabbitMQ to accommodate our events. Here’s an example init.go file for setting up the RabbitMQ infrastructure:

package main

import (
    "log"

    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    err = ch.ExchangeDeclare("events", "direct", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare an exchange: %v", err)
    }

    _, err = ch.QueueDeclare("PatientVerified", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare a queue: %v", err)
    }

    err = ch.QueueBind("PatientVerified", "PatientVerified", "events", false, nil)
    if err != nil {
        log.Fatalf("Failed to bind a queue: %v", err)
    }
}

Full code here!

Note: In a production setting, you might want to manage this setup using a GitOps approach (e.g., with Terraform) or let each service handle its own queues dynamically.

Implementation: Service Files

Each service will have its own main.go. We’ll also include compensation actions for handling failures gracefully.

1. Patient Service

This service verifies patient details and emits a PatientVerified event. It also compensates by notifying the patient if a downstream failure occurs.

// patient/main.go
package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/amqp091-go"
    "github.com/thegoodapi/saga_tutorial/choreography/common"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    go func() {
        fmt.Println("[PatientService] Waiting for events...")
        msgs, err := common.ConsumeEvent(ch, "ProcedureScheduleCancelled")
        if err != nil {
            log.Fatalf("Failed to consume event: %v", err)
        }

        for range msgs {
            fmt.Println("[PatientService] Processing event: ProcedureScheduleCancelled")
            if err := notifyProcedureScheduleCancellation(); err != nil {
                log.Fatalf("Failed to notify patient: %v", err)
            }
        }
    }()

    common.PublishEvent(ch, "events", "PatientVerified", "Patient details verified")
    fmt.Println("[PatientService] Event published: PatientVerified")

    select {}
}

func notifyProcedureScheduleCancellation() error {
    fmt.Println("Compensation: Notify patient of procedure cancellation.")
    return nil
}

2. Scheduler Service

This service listens for PatientVerified and emits ProcedureScheduled. It compensates by canceling the procedure if a downstream failure occurs.

// scheduler/main.go
package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/amqp091-go"
    "github.com/thegoodapi/saga_tutorial/choreography/common"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    go func() {
        fmt.Println("[SchedulerService] Waiting for events...")
        msgs, err := common.ConsumeEvent(ch, "PatientVerified")
        if err != nil {
            log.Fatalf("Failed to consume event: %v", err)
        }

        for range msgs {
            fmt.Println("[SchedulerService] Processing event: PatientVerified")
            if err := scheduleProcedure(); err != nil {
                common.PublishEvent(ch, "events", "ProcedureScheduleFailed", "Failed to schedule procedure")
                fmt.Println("[SchedulerService] Compensation triggered: ProcedureScheduleFailed")
            } else {
                common.PublishEvent(ch, "events", "ProcedureScheduled", "Procedure scheduled successfully")
                fmt.Println("[SchedulerService] Event published: ProcedureScheduled")
            }
        }
    }()

    select {}
}

func scheduleProcedure() error {
    fmt.Println("Step 2: Scheduling procedure...")
    return nil // or simulate a failure
}

Additional Services

Include Inventory Service and Billing Service implementations, following the same structure as above. Each service listens for the previous event and emits the next one, ensuring compensation logic is in place for failures.

Full code here!

Running the Workflow

Start RabbitMQ:

   docker run --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0.5-management

Run Each Service:
Open separate terminals and run:

   // one-time script to setup rabbitmq
   go run choerography/init/main.go
   // services
   go run choerography/billing/main.go
   go run choerography/inventory/main.go
   go run choerography/scheduler/main.go
   go run choerography/patient/main.go

Observe Output:
Each service processes events in sequence, logging the workflow progress.

What happened?

Let's break it down!

First of all, for the purpose of this article, we are not implementing SuppliesReserveFailed and ProcedureScheduleFailed,l to avoid unseless complexity.

We are implementing the following events

Steps (or transactions):

T1: (init): PatientVerified
T2: ProcedureScheduled
T3: SuppliesReserved
T4: BillingSuccessful

Compensations:

C4: BillingFailed
C3: ReservedSuppliesReleased
C2: ProcedureScheduleCancelled
C1: NotifyFailureToUser (not implemented)

Folowing this implementation diagram

This diagram represents a common approach to documenting choreography. However, I find it somewhat difficult to understand and a bit frustrating, particularly for those who are not familiar with the implementation or the pattern.

Let's break it down!

The diagram above is way more verbose and it breaks down each step making it easier to understand what's going on.

In a nutshell:

Patient service verifies patient details successfully
Patient service emits PatientVerified
Scheduler service consumes PatientVerified
Scheduler service schedule the appintment successfully
Scheduler service emits ProcedureScheduled
Inventory service consumes ProcedureScheduled
Inventory service reserves the supplies successfully
Inventory service emits SuppliesReserved
Billing service consumes SuppliesReserved
Billing service failes to charge the customer and starts the compensation
Billing service emits BillingFailed
Inventory service consumes BillingFailed
Inventory service releases the supplies, reserved in step 7
Inventory service emits ReservedSuppliesReleased
Scheduler service consumes ReservedSuppliesReleased
Scheduler service deletes the appointment scheduled in step 4
Scheduler service emits ProcedureScheduleCancelled
Patient service consumes ProcedureScheduleCancelled
Patient service notifies the customer of the error

Note that we are not implementing failures for steps 1, 4, and 7 for the sake of brevity; however, the approach would be the same. Each of these failures would trigger a rollback of the preceding steps.

Observability

Observability is essential for debugging and monitoring distributed systems. Implementing logs, metrics, and traces ensures that developers can understand system behavior and diagnose issues efficiently.

Logging

Use structured logging (e.g., JSON format) to capture events and metadata.
Include correlation IDs in logs to trace workflows across services.

Metrics

Monitor queue sizes and event processing times.
Use tools like Prometheus to collect and visualize metrics.

Tracing

Implement distributed tracing (e.g., with OpenTelemetry) to track events across services.
Annotate spans with relevant data (e.g., event names, timestamps) for better insights.

We'll dive into observability in choerography later in this serie, stay tuned!

Key Takeaways

Decentralized Control: Choreography enables autonomous collaboration.
Event-Driven Simplicity: RabbitMQ simplifies message exchange.
Scalable Architecture: Adding new services is seamless.
Choerography can be very overwelming at first, but as always: practice make you ~~perfect~~ better!

Stay tuned for the next article, where we’ll explore Orchestration!

Check out the full repository for this series here. Let’s discuss in the comments!

Transactions in Microservices: Part 1 - SAGA Patterns overview.

Federico Bevione — Mon, 20 Jan 2025 14:00:00 +0000

Distributed systems are powerful yet challenging, especially when it comes to ensuring consistency across multiple services. In a world of microservices, traditional database transactions often fall short. This is where distributed transactions step in.

Distributed transactions allow systems to coordinate multiple services while handling failures gracefully. One of the most effective approaches to achieving this is the SAGA pattern, which provides two distinct implementations: Choreography and Orchestration.

This article is the first in a series exploring distributed transactions and the SAGA pattern. By the end, you’ll understand its core concepts, when to use it, and how it can be applied in real-world scenarios. A practical example in Go is included to demonstrate the Orchestration approach.

The Challenge of Distributed Transactions

Imagine building a distributed application where multiple services collaborate to complete a business operation. Here are some examples:

Finance: Managing a multi-step loan approval process.
E-commerce: Coordinating order placement, payment processing, and shipping.
Healthcare: Handling a multi-step workflow for scheduling a medical procedure.

Key Challenges

Partial Failures: One service may fail while others succeed, leaving the system in an inconsistent state.
Data Consistency: The final state must reflect the intended outcome, regardless of individual service failures.
Complex Workflows: Coordinating multiple services in a reliable and maintainable way.

Traditional database transactions are inadequate for solving these challenges in distributed systems, necessitating patterns like SAGA.

The SAGA Pattern: A Reliable Solution

The SAGA pattern breaks a complex workflow into smaller, independent steps. Each step performs a specific task and can be compensated (undone) if something goes wrong.

Implementation Approaches

Choreography: Each service emits events that other services consume to trigger the next step. It is decentralized and event-driven.
Orchestration: A central orchestrator coordinates the sequence of steps and manages compensations. It is more centralized and easier to reason about.

Both approaches have unique strengths, and the choice depends on the system's requirements. In this article, we’ll focus on Orchestration.

Practical Example: Healthcare Workflow with Orchestration

Consider a healthcare system managing a multi-step workflow for scheduling a medical procedure. The services involved might include:

Patient Management: Verifying patient details and insurance coverage.
Appointment Scheduler: Booking an available slot for the procedure.
Inventory Management: Reserving medical supplies for the procedure.
Billing: Charging the patient or insurer.

To ensure consistency across these services, we can use the SAGA pattern with Orchestration. Below is a runnable Go implementation.

Code Example: SAGA with Orchestration in Go

package main

import (
    "fmt"
    "log"
)

// Define step and compensation types
type Step func() error
type Compensation func()

// Saga structure
type Saga struct {
    steps        []Step
    compensations []Compensation
}

func (s *Saga) AddStep(step Step, compensation Compensation) {
    s.steps = append(s.steps, step)
    s.compensations = append([]Compensation{compensation}, s.compensations...)
}

func (s *Saga) Execute() {
    for i, step := range s.steps {
        if err := step(); err != nil {
            log.Printf("Step %d failed: %v. Rolling back...\n", i+1, err)
            for _, compensation := range s.compensations {
                compensation()
            }
            return
        }
    }
    fmt.Println("Transaction completed successfully!")
}

func main() {
    saga := &Saga{}

    // Step 1: Verify patient and insurance details
    saga.AddStep(
        func() error {
            fmt.Println("Step 1: Verifying patient and insurance details...")
            // Simulate success
            return nil
        },
        func() { fmt.Println("Compensation 1: Notify patient of verification failure.") },
    )

    // Step 2: Schedule procedure
    saga.AddStep(
        func() error {
            fmt.Println("Step 2: Scheduling procedure...")
            // Simulate success
            return nil
        },
        func() { fmt.Println("Compensation 2: Cancel procedure schedule.") },
    )

    // Step 3: Reserve medical supplies
    saga.AddStep(
        func() error {
            fmt.Println("Step 3: Reserving medical supplies...")
            // Simulate success
            return nil
        },
        func() { fmt.Println("Compensation 3: Release reserved supplies.") },
    )

    // Step 4: Process billing
    saga.AddStep(
        func() error {
            fmt.Println("Step 4: Processing billing...")
            // Simulate failure
            return fmt.Errorf("billing service unavailable")
        },
        func() { fmt.Println("Compensation 4: Reverse any pending charges.") },
    )

    // Execute the saga
    saga.Execute()
}

The example provided above is a simplified implementation, primarily intended to demonstrate the basic workings of a SAGA with orchestration. While it is not suitable for direct use in real-world scenarios, it should give you a solid understanding of the core concepts.

In the upcoming articles of this series, we will explore more advanced, real-world examples to deepen your understanding.

Stay tuned!

Applications Beyond Healthcare

The SAGA pattern is versatile and widely applicable:

Finance: Automating loan approvals across multiple systems.
Logistics: Managing shipment tracking and order fulfillment.
E-commerce: Handling complex order workflows involving payment, inventory, and delivery.

Key Takeaways

Distributed transactions are vital for maintaining consistency in microservices.
The SAGA pattern simplifies the coordination of these transactions.
Choreography and Orchestration each offer unique advantages depending on system requirements.

In the next article, we’ll explore the Choreography approach, diving into its event-driven nature with another practical example in Go.

Check out the full repository for runnable examples.

Have questions or feedback? Let’s discuss in the comments!