saga

README

saga

A Saga pattern in Go

Documentation

Overview

Package saga is an implementation of saga design pattern for error handling. It aims to solve distributed (business) transactions without two-phase-commit as this does not scale in distributed systems. The saga has responsibility to gain eventual consistency by calling compensation steps in reverse order.

This library uses orchestration approach by a centralized state machine. The implementation is inspired by this great talk https://www.youtube.com/watch?v=xDuwrtwYHu8. More documents could be found at https://www.reactivedesignpatterns.com/patterns/saga.html or https://msdn.microsoft.com/en-us/library/jj591569.aspx

If you have any suggestion or comment, please feel free to open an issue on this github

Index

• type Activity
• type Aggregator
• type Config
• type Executor
  • func NewExecutor(c Config) (*Executor, error)
  • func (e *Executor) Execute(ctx context.Context, tx Transaction) (Transaction, error)
• type Logger
• type MockAggregator
  • func (_m *MockAggregator) Execute(ctx context.Context, tx Transaction) (Transaction, error)
• type MockTransaction
  • func (_m *MockTransaction) State() State
• type State
• type Transaction

Examples

• Executor
• Executor (Rollback)

Types

type Activity

type Activity struct {
	SuccessState    State
	FailureState    State
	RolledBackState State
	Aggregator      Aggregator
	Compensation    Aggregator
}

Activity includes details of an activity in a saga - Aggregator - SuccessState when it is executed successfully - FailureState when it is failed to be executed. - RolledBackState when it is successfulled executed but rolled back - Compensataion aggregator to rollback

type Aggregator

type Aggregator interface {
	Execute(ctx context.Context, tx Transaction) (Transaction, error)
}

Aggregator is an interface of a agregator which execute a single step in a list of steps in a saga transaction.

Execute process the current step of a transaction and returns another object of the transaction in next state. It returns an error when fails to process.

type Config

type Config struct {
	InitState  State
	Activities []Activity
	Logger     Logger
}

Config includes details of a saga - InitState is the initial state - Activities list of activities in this saga in order - Logger is a logger to log state of a saga transaction

type Executor

type Executor struct {
	// contains filtered or unexported fields
}

Executor to execute a saga transaction

Example

package main

import (
	"context"
	"fmt"

	"github.com/bongnv/saga"
)

const (
	stateInit saga.State = iota
	stateBookedHotel
	stateCanceledHotel
	stateBookHotelFailed
	stateBookedFlight
	stateCanceledFlight
	stateBookFlightFailed
)

type exampleTransaction struct {
	state saga.State
}

func (t *exampleTransaction) State() saga.State {
	return t.state
}

type aggBookHotel struct{}

func (a *aggBookHotel) Execute(ctx context.Context, tx saga.Transaction) (saga.Transaction, error) {
	fmt.Println("Book hotel")
	return &exampleTransaction{
		state: stateBookedHotel,
	}, nil
}

type aggBookFlight struct {
	nextState saga.State
}

func (a *aggBookFlight) Execute(ctx context.Context, tx saga.Transaction) (saga.Transaction, error) {
	fmt.Println("Book flight")
	return &exampleTransaction{
		state: a.nextState,
	}, nil
}

func main() {
	exampleConfig := saga.Config{
		InitState: stateInit,
		Activities: []saga.Activity{
			{
				Aggregator:      &aggBookHotel{},
				SuccessState:    stateBookedHotel,
				FailureState:    stateBookHotelFailed,
				RolledBackState: stateCanceledHotel,
			},
			{
				Aggregator: &aggBookFlight{
					nextState: stateBookedFlight,
				},
				SuccessState:    stateBookedFlight,
				FailureState:    stateBookFlightFailed,
				RolledBackState: stateCanceledFlight,
			},
		},
	}

	executor, err := saga.NewExecutor(exampleConfig)
	if err != nil {
		fmt.Printf("Failed to create saga executor, err: %v.\n", err)
		return
	}

	finalTx, err := executor.Execute(context.Background(), &exampleTransaction{
		state: stateInit,
	})
	if err != nil {
		fmt.Printf("Failed to execute saga transaction, err: %v.\n", err)
		return
	}
	if finalTx.State() == stateBookedFlight {
		fmt.Println("OK")
	}
}

Output:

Book hotel
Book flight
OK

Example (Rollback)

package main

import (
	"context"
	"fmt"

	"github.com/bongnv/saga"
)

const (
	stateInit saga.State = iota
	stateBookedHotel
	stateCanceledHotel
	stateBookHotelFailed
	stateBookedFlight
	stateCanceledFlight
	stateBookFlightFailed
)

type exampleTransaction struct {
	state saga.State
}

func (t *exampleTransaction) State() saga.State {
	return t.state
}

type aggBookHotel struct{}

func (a *aggBookHotel) Execute(ctx context.Context, tx saga.Transaction) (saga.Transaction, error) {
	fmt.Println("Book hotel")
	return &exampleTransaction{
		state: stateBookedHotel,
	}, nil
}

type aggBookFlight struct {
	nextState saga.State
}

func (a *aggBookFlight) Execute(ctx context.Context, tx saga.Transaction) (saga.Transaction, error) {
	fmt.Println("Book flight")
	return &exampleTransaction{
		state: a.nextState,
	}, nil
}

type aggCancelHotel struct{}

func (a *aggCancelHotel) Execute(ctx context.Context, tx saga.Transaction) (saga.Transaction, error) {
	fmt.Println("Cancel hotel")
	return &exampleTransaction{
		state: stateCanceledHotel,
	}, nil
}

func main() {
	exampleConfig := saga.Config{
		InitState: stateInit,
		Activities: []saga.Activity{
			{
				Aggregator:      &aggBookHotel{},
				SuccessState:    stateBookedHotel,
				FailureState:    stateBookHotelFailed,
				Compensation:    &aggCancelHotel{},
				RolledBackState: stateCanceledHotel,
			},
			{
				Aggregator: &aggBookFlight{
					nextState: stateBookFlightFailed,
				},
				SuccessState:    stateBookedFlight,
				FailureState:    stateBookFlightFailed,
				RolledBackState: stateCanceledFlight,
			},
		},
	}

	executor, err := saga.NewExecutor(exampleConfig)
	if err != nil {
		fmt.Printf("Failed to create saga executor, err: %v.\n", err)
		return
	}

	finalTx, err := executor.Execute(context.Background(), &exampleTransaction{
		state: stateInit,
	})
	if err != nil {
		fmt.Printf("Failed to execute saga transaction, err: %v.\n", err)
		return
	}
	if finalTx.State() == stateCanceledHotel {
		fmt.Println("OK")
	}
}

Output:

Book hotel
Book flight
Cancel hotel
OK

func NewExecutor

func NewExecutor(c Config) (*Executor, error)

NewExecutor creates a new Executor for a saga

func (*Executor) Execute

func (e *Executor) Execute(ctx context.Context, tx Transaction) (Transaction, error)

Execute continue to process a saga transaction based on current state It returns a transaction object after it's executed successfully It returns error when failed to execute it

type Logger

type Logger interface {
	Log(ctx context.Context, tx Transaction) error
}

Logger is an interface that wraps Log function.

Log persists a transaction normally for auditing or recovering It returns an error while fails to store.

type MockAggregator

type MockAggregator struct {
	mock.Mock
}

MockAggregator is an autogenerated mock type for the Aggregator type

func (*MockAggregator) Execute

func (_m *MockAggregator) Execute(ctx context.Context, tx Transaction) (Transaction, error)

Execute provides a mock function with given fields: ctx, tx

type MockTransaction

type MockTransaction struct {
	mock.Mock
}

MockTransaction is an autogenerated mock type for the Transaction type

func (*MockTransaction) State

func (_m *MockTransaction) State() State

State provides a mock function with given fields:

type State

type State int16

State of a saga transaction

type Transaction

type Transaction interface {
	State() State
}

Transaction is an interface of a saga transaction.

State returns current state of a transaction.