saga

README

Saga manager

In distributed systems, it can be difficult to ensure that a complex operation is completed successfully without encountering errors or failures. A saga is a way of managing these types of operations by breaking them down into smaller, independent steps that can be executed and managed separately.

Saga steps of state:

• INIT -> WAIT -> RUN -> DONE/REJECT
• REJECT -> FAIL/ROLLBACK

graph LR
  INIT[Initialization] -->|Initialization Complete| WAIT[Waiting]
  WAIT -->|Waiting Complete| RUN[Running]
  RUN -->|Running Complete| DONE[Done]
  RUN -->|Error Occurred| REJECT[Rejection]
  REJECT -->|Rejection Handled| FAIL[Failure]
  REJECT -->|Rollback Triggered| ROLLBACK[Rollback]

  style INIT fill:#FFADAD
  style WAIT fill:#FFD6A5
  style RUN fill:#FDFFB6
  style DONE fill:#CAFFBF
  style REJECT fill:#9BF6FF
  style FAIL fill:#A0C4FF
  style ROLLBACK fill:#BDB2FF

Example

package main

import (
	"context"
	
  "github.com/shortlink-org/shortlink/pkg/pattern/saga"
)

func (l *linkUseCase) addLinkSaga(ctx, link link.Link) error {
	const SAGA_NAME = "Add link"
	const SAGA_STEP_SAVE_LINK = "Save link in store"
	const SAGA_STEP_GET_METADATA = "Get metadata by link"

  // create a new saga for add link
  sagaAddLink, errs := saga.
    New(SAGA_NAME).   // name saga
    WithContext(ctx). // ctx for tracing
    Build()
  
  if len(errs) > 0 {
    // check err...
  }
  
  // step: save to store
  _, err = saga.AddStep(SAGA_STEP_SAVE_LINK).
    Then(func(context.Context) error {
      err := l.Store.Add(link)
      return err
    }).
    Reject(func(context.Context, thenErr error) error {
      err := l.Store.Delete(link)
      return err
    }).
    Build()
  
  if len(errs) > 0 {
    // check err...
  }

  // step: get metadata
  saga.AddStep(SAGA_STEP_GET_METADATA).
    Then(addFunc).
    Reject(cancelAddFunc)

  // step: send notify
  saga.AddStep("send notify").
    Needs(SAGA_STEP_SAVE_LINK, SAGA_STEP_GET_METADATA).
    Then(youNotifyFunc)
  
  // Run saga
  err := sagaAddLink.Play(nil)
  return err
}

Options

• SetLogger - add logger for saga
• SetLimiter - set limiter for goroutines (default unlimited)

OpenTracing

References

[!NOTE] Alternatives: cff - Concurrency toolkit for Go (uber)

• Pattern: Saga
• DAG - for build pipeline steps
• libs:
  • go-saga - example go-library
  • goflow
  • https://github.com/danielgerlag/workflow-core
  • https://github.com/VladKopanev/cats-saga

Documentation

Index

• Constants
• func GetError(ctx context.Context) error
• func WithError(ctx context.Context, err error) context.Context
• type Builder
  • func New(name string, setters ...Option) *Builder
  • func (s *Builder) Build() (*Saga, []error)
  • func (s *Builder) WithContext(ctx context.Context) *Builder
• type BuilderStep
  • func (s *BuilderStep) Build() (*Step, []error)
  • func (s *BuilderStep) Needs(keys ...string) *BuilderStep
  • func (s *BuilderStep) Reject(f func(ctx context.Context, thenError error) error) *BuilderStep
  • func (s *BuilderStep) Then(f func(ctx context.Context) error) *BuilderStep
• type DublicateStepError
  • func (e *DublicateStepError) Error() string
• type Error
• type Option
  • func SetLimiter(limiter int) Option
  • func SetLogger(log logger.Logger) Option
• type Options
• type Saga
  • func (s *Saga) AddStep(name string, setters ...Option) *BuilderStep
  • func (s *Saga) Play(initSteps map[string]*Step) error
  • func (s *Saga) Reject(rejectSteps map[string]*Step) error
• type Step
  • func (s *Step) Reject() error
  • func (s *Step) Run() error
• type StepState

Constants

const ContextErrorKey = Error("saga-error")

Functions

func GetError

func GetError(ctx context.Context) error

func WithError

func WithError(ctx context.Context, err error) context.Context

Types

type Builder

type Builder struct {
	*Saga
	// contains filtered or unexported fields
}

func New

func New(name string, setters ...Option) *Builder

func (*Builder) Build

func (s *Builder) Build() (*Saga, []error)

func (*Builder) WithContext

func (s *Builder) WithContext(ctx context.Context) *Builder

type BuilderStep

type BuilderStep struct {
	*Step
	// contains filtered or unexported fields
}

func (*BuilderStep) Build

func (s *BuilderStep) Build() (*Step, []error)

func (*BuilderStep) Needs

func (s *BuilderStep) Needs(keys ...string) *BuilderStep

func (*BuilderStep) Reject

func (s *BuilderStep) Reject(f func(ctx context.Context, thenError error) error) *BuilderStep

func (*BuilderStep) Then

func (s *BuilderStep) Then(f func(ctx context.Context) error) *BuilderStep

type DublicateStepError

type DublicateStepError struct {
	Name string
}

func (*DublicateStepError) Error

func (e *DublicateStepError) Error() string

type Error

type Error string

type Option

type Option func(*Options)

func SetLimiter

func SetLimiter(limiter int) Option

func SetLogger

func SetLogger(log logger.Logger) Option

type Options

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

type Saga

type Saga struct {
	Options
	// contains filtered or unexported fields
}

func (*Saga) AddStep

func (s *Saga) AddStep(name string, setters ...Option) *BuilderStep

func (*Saga) Play

func (s *Saga) Play(initSteps map[string]*Step) error

func (*Saga) Reject

func (s *Saga) Reject(rejectSteps map[string]*Step) error

type Step

type Step struct {
	Options
	// contains filtered or unexported fields
}

func (*Step) Reject

func (s *Step) Reject() error

func (*Step) Run

func (s *Step) Run() error

type StepState

type StepState int

const (
	INIT StepState = iota + 1
	WAIT
	RUN
	DONE
	REJECT
	FAIL
	ROLLBACK
)

Success chain: INIT -> WAIT -> RUN -> DONE Fail chain: INIT -> WAIT -> RUN -> REJECT -> FAIL or ROLLBACK