workflow

package module
v0.2.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 27, 2025 License: BSD-3-Clause Imports: 25 Imported by: 19

README

Workflow Logo
Go Reference Mentioned in Awesome Go

Workflow

Workflow is a distributed event driven workflow framework that runs robust, durable, and scalable sequential business logic on your services.

Workflow uses a RoleScheduler to distribute the work across your instances through a role assignment process (similar to a leadership election process, but with more than a single role of leader).

Workflow expects to be run on multiple instances but can also be run on single instances. Using the above-mentioned RoleScheduler, Workflow is able to make sure each process only runs once at any given time regardless if you are running 40 instances of your service or 1 instance.

Features

  • Tech stack agnostic: Use Kafka, Cassandra, Redis, MongoDB, Postgresql, MySQL, RabbitM, or Reflex - the choice is yours!
  • Graph based (Directed Acyclic Graph - DAG): Design the workflow by defining small units of work called "Steps".
  • TDD: Workflow was built using TDD and remains well-supported through a suit of tools.
  • Callbacks: Allow for manual callbacks from webhooks or manual triggers from consoles to progress the workflow, such as approval buttons or third-party webhooks.
  • Event fusion: Add event connectors to your workflow to consume external event streams (even if it's from a different event streaming platform).
  • Hooks: Write hooks that execute on core changes in a workflow Run.
  • Schedule: Allows standard cron spec to schedule workflows
  • Timeouts: Set either a dynamic or static time for a workflow to wait for. Once the timeout finishes everything continues as it was.
  • Parallel consumers: Specify how many step consumers should run or specify the default for all consumers.
  • Consumer management: Consumer management and graceful shutdown of all processes making sure there is no goroutine leaks!

Installation

To start using workflow you will need to add the workflow module to your project. You can do this by running:

go get github.com/luno/workflow

Adapters

Adapters enable Workflow to be tech stack agnostic by placing an interface / protocol between Workflow and the tech stack. Workflow uses adapters to understand how to use that specific tech stack.

For example, the Kafka adapter enables workflow to produce messages to a topic as well as consume them from a topic using a set of predefined methods that wrap the kafka client. Reflex is an event streaming framework that works very differently to Kafka and the adapter pattern allows for the differences to be contained and localised in the adapter and not spill into the main implementation.

Event Streamer

The EventStreamer adapter interface defines what is needed to be satisfied in order for an event streaming platform or framework to be used by Workflow.

All implementations of the EventStreamer interface should be tested using adaptertest.TestEventStreamer

Record Store

The RecordStore adapter interface defines what is needed to satisfied in order for a storage solution to be used by Workflow.

All implementations of the RecordStore interface should be tested using adaptertest.RunRecordStoreTest

Role Scheduler

The RoleScheduler adapter interface defines what is needed to satisfied in order for a role scheduling solution to be used by Workflow.

All implementations of the RoleScheduler interface should be tested using adaptertest.RunRoleSchedulerTest

There are more adapters available but only the above 3 are core requirements to use Workflow. To start, use the in-memory implementations as that is the simplest way to experiment and get used to Workflow. For testing other adapter types be sure to look at adaptertest which are tests written for adapters to ensure that they meet the specification.

Adapters, except for the in-memory implementations, don't come with the core Workflow module such as kafkastreamer, reflexstreamer, sqlstore, sqltimeout, rinkrolescheduler, webui and many more. If you wish to use these you need to add them individually based on your needs or build out your own adapter.

Kafka
go get github.com/luno/workflow/adapters/kafkastreamer
Reflex
go get github.com/luno/workflow/adapters/reflexstreamer
SQL Store
go get github.com/luno/workflow/adapters/sqlstore
SQL Timeout
go get github.com/luno/workflow/adapters/sqltimeout
Rink Role Scheduler
go get github.com/luno/workflow/adapters/rinkrolescheduler
WebUI
go get github.com/luno/workflow/adapters/webui

Connectors

Connectors allow Workflow to consume events from an event streaming platform or framework and either trigger a workflow run or provide a callback to the workflow run. This means that Connectors can act as a way for Workflow to connect with the rest of the system.

Connectors are implemented as adapters as they would share a lot of the same code as implementations of an EventStreamer and can be seen as a subsection of an adapter.

An example can be found here.

Basic Usage

Step 1: Define the workflow
package usage

import (
	"context"

	"github.com/luno/workflow"
)

type Step int

func (s Step) String() string {
	switch s {
	case StepOne:
		return "One"
	case StepTwo:
		return "Two"
	case StepThree:
		return "Three"
	default:
		return "Unknown"
	}
}

const (
	StepUnknown Step = 0
	StepOne Step = 1
	StepTwo Step = 2
	StepThree Step = 3
)

type MyType struct {
	Field string
}

func Workflow() *workflow.Workflow[MyType, Step] {
	b := workflow.NewBuilder[MyType, Step]("my workflow name")

	b.AddStep(StepOne, func(ctx context.Context, r *workflow.Run[MyType, Step]) (Step, error) {
		r.Object.Field = "Hello,"
		return StepTwo, nil
	}, StepTwo)

	b.AddStep(StepTwo, func(ctx context.Context, r *workflow.Run[MyType, Step]) (Step, error) {
		r.Object.Field += " world!"
		return StepThree, nil
	}, StepThree)

	return b.Build(...)
}

---
title: The above defined workflow creates the below Directed Acyclic Graph
---
stateDiagram-v2
	direction LR
    
	[*]-->One
    One-->Two
    Two-->Three
    Three-->[*]
Step 2: Run the workflow
wf := usage.Workflow()

ctx := context.Background()
wf.Run(ctx)

Stop: To stop all processes and wait for them to shut down correctly call

wf.Stop()
Step 3: Trigger the workflow
foreignID := "82347982374982374"
runID, err := wf.Trigger(ctx, foreignID, StepOne)
if err != nil {
	...
}

Awaiting results: If appropriate and desired you can wait for the workflow to complete. Using context timeout (cancellation) is advised.

foreignID := "82347982374982374"
runID, err := wf.Trigger(ctx, foreignID, StepOne)
if err != nil {
	...
}

ctx, cancel := context.WithTimeout(ctx, 10 * time.Second)
defer cancel()

record, err := wf.Await(ctx, foreignID, runID, StepThree)
if err != nil {
	...
}
Detailed examples

Head on over to ./_examples to get familiar with callbacks, timeouts, testing, connectors and more about the syntax in depth 😊

What is a workflow Run

When a Workflow is triggered it creates an individual workflow instance called a Run. This is represented as workflow.Run in Workflow. Each run has a lifecycle which is a finite set of states - commonly referred to as Finite State Machine. Each workflow Run has the following of states (called RunState in Workflow):

Run State Value (int) Description
Unknown 0 Has no meaning. Protects against default zero value.
Initiated 1 State assinged at creation of Run and is yet to be processed.
Running 2 Has begun to be processed and is currently still being processed by a step in the workflow.
Paused 3 Temporary stoppage that can be resumed or cancelled. Will prevent any new triggers of the same Foreign ID.
Completed 4 Finished all the steps configured at time of execution.
Cancelled 5 Did not complete all the steps and was terminated before completion.
Data Deleted 6 Run Object has been modified to remove data or has been entirely removed. Likely for PII scrubbing reasons.
Requested Data Deleted 7 Request state for the workflow to apply the default or custom provided delete operation to the Run Object.

A Run can only exist in one state at any given time and the RunState allows for control over the Run.

---
title: Diagram the run states of a workflow
---
stateDiagram-v2
	direction LR
    
    Initiated-->Running
    
    Running-->Completed
    Running-->Paused

    Paused-->Running
    
    Running --> Cancelled
    Paused --> Cancelled
    
    state Finished {
        Completed --> RequestedDataDeleted
        Cancelled --> RequestedDataDeleted
            
        DataDeleted-->RequestedDataDeleted
        RequestedDataDeleted-->DataDeleted
    }

Hooks

Hooks allow for you to write some functionality for Runs that enter a specific RunState. For example when using PauseAfterErrCount the usage of the OnPause hook can be used to send a notification to a team to notify them that a specific Run has errored to the threshold and now has been paused and should be investigated. Another example is handling a known sentinel error in a Workflow Run and cancelling the Run by calling (where r is *Run) r.Cancel(ctx) or if a Workflow Run is manually cancelled from a UI then a notifgication can be sent to the team for visibility.

Hooks run in an event consumer. This means that it will retry until a nil error has been returned and is durable across deploys and interruptions. At-least-once delivery is guaranteed, and it is advised to use the RunID as an idempotency key to ensure that the operation is idempotent.

Available Hooks:
Hook Parameter(s) Return(s) Description Is Event Driven?
OnPause workflow.RunStateChangeHookFunc error Fired when a Run enters RunStatePaused Yes
OnCancelled workflow.RunStateChangeHookFunc error Fired when a Run enters RunStateCancelled Yes
OnCompleted workflow.RunStateChangeHookFunc error Fired when a Run enters RunStateCompleted Yes

Configuration Options

This package provides several options to configure the behavior of the workflow process. You can use these options to customize the instance count, polling frequency, error handling, lag settings, and more. Each option is defined as a function that takes a pointer to an options struct and modifies it accordingly. Below is a description of each available option:

ParallelCount
func ParallelCount(instances int) Option
  • Description: Defines the number of instances of the workflow process. These instances are distributed consistently, each named to reflect its position (e.g., "consumer-1-of-5"). This helps in managing parallelism in workflow execution.
  • Parameters:
    • instances: The total number of parallel instances to create.
  • Usage Example:
b.AddStep(
    StepOne,
    ...,
    StepTwo,
).WithOptions(
    workflow.ParallelCount(5)
)
PollingFrequency
func PollingFrequency(d time.Duration) Option
  • Description: Sets the duration at which the workflow process polls for changes. Adjust this to control how frequently the process checks for new events or updates.
  • Parameters:
    • d: The polling frequency as a time.Duration.
  • Usage Example:
b.AddStep(
    StepOne,
    ...,
    StepTwo,
).WithOptions(
    workflow.PollingFrequency(10 * time.Second)
)
ErrBackOff
func ErrBackOff(d time.Duration) Option
  • Description: Defines the duration for which the workflow process will back off after encountering an error. This is useful for managing retries and avoiding rapid repeated failures.
  • Parameters:
    • d: The backoff duration as a time.Duration.
  • Usage Example:
b.AddStep(
    StepOne,
    ...,
    StepTwo,
).WithOptions(
    workflow.ErrBackOff(5 * time.Minute)
)
LagAlert
func LagAlert(d time.Duration) Option
  • Description: Specifies the time threshold before a Prometheus metric switches to true, indicating that the workflow consumer is struggling to keep up. This can signal the need to convert to a parallel consumer.
  • Parameters:
    • d: The duration of the lag alert as a time.Duration.
  • Usage Example:
b.AddStep(
    StepOne,
    ...,
    StepTwo,
).WithOptions(
    workflow.LagAlert(15 * time.Minute),
)
ConsumeLag
func ConsumeLag(d time.Duration) Option
  • Description: Defines the maximum age of events that the consumer will process. Events newer than the specified duration will be held until they are older than the lag period.
  • Parameters:
    • d: The lag duration as a time.Duration.
  • Usage Example:
b.AddStep(
    StepOne,
    ...,
    StepTwo,
).WithOptions(
    workflow.ConsumeLag(10 * time.Minute),
)
PauseAfterErrCount
func PauseAfterErrCount(count int) Option
  • Description: Sets the number of errors allowed before a record is updated to RunStatePaused. This mechanism acts similarly to a Dead Letter Queue, preventing further processing of problematic records and allowing for investigation and retry.
  • Parameters:
    • count: The maximum number of errors before pausing.
  • Usage Example:
b.AddStep(
    StepOne,
    ...,
    StepTwo,
).WithOptions(
    workflow.PauseAfterErrCount(3),
)

Glossary

Term Description
Builder A struct type that facilitates the construction of workflows. It provides methods for adding steps, callbacks, timeouts, and connecting workflows.
Callback A method in the workflow API that can be used to trigger a callback function for a specified status. It passes data from a reader to the specified callback function.
Consumer A component that consumes events from an event stream. In this context, it refers to the background consumer goroutines launched by the workflow.
EventStreamer An interface representing a stream for workflow events. It includes methods for producing and consuming events.
Graph A representation of the workflow's structure, showing the relationships between different statuses and transitions.
Hooks An event driven process that take place on a Workflow's Run's lifecycle defined in a finite number of states called RunState.
Producer A component that produces events to an event stream. It is responsible for sending events to the stream.
Record Is the "wire format" and representation of a Run that can be stored and retrieved. The RecordStore is used for storing and retrieving records.
RecordStore An interface representing a store for Record(s). It defines the methods needed for storing and retrieving records. The RecordStore's underlying technology must support transactions in order to prevent dual-writes.
RoleScheduler An interface representing a scheduler for roles in the workflow. It is responsible for coordinating the execution of different roles.
Run A Run is the representation of the instance that is created and processed by the Workflow. Each time Trigger is called a new "Run" is created.
RunState RunState defines the finite number of states that a Run can be in. This is used to control and monitor the lifecycle of Runs.
Topic A method that generates a topic for producing events in the event streamer based on the workflow name and status.
Trigger A method in the workflow API that initiates a workflow for a specified foreignID and starting status. It returns a Run ID and allows for additional configuration options.

Best practices

  1. Break up complex business logic into small steps.
  2. Workflow can be used to produce new meaningful data and not just be used to execute logic. If it is used for this, it's suggested to implement a CQRS pattern where the workflow acts as the "Command" and the data is persisted into a more queryable manner.
  3. Changes to workflows must be backwards compatible. If you need to introduce a non-backwards compatible change then the non-backwards compatible workflow should be added alongside the existing workflow with the non-backwards compatible workflow receiving all the incoming triggers. The old workflow should be given time to finish processing any workflows it started and once it has finished processing all the existing non-finished Runs then it may be safely removed. Alternatively versioning can be added internally to your Object type that you provide, but this results in changes to the workflow's Directed Acyclic Graph (map of steps connecting together).
  4. Workflow is not intended for low-latency. Asynchronous event driven systems are not meant to be low-latency but prioritise decoupling, durability, distribution of workload, and breakdown of complex logic (to name a few).
  5. Ensure that the prometheus metrics that come with Workflow are being used for monitoring and alerting.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrRecordNotFound       = errors.New("record not found")
	ErrTimeoutNotFound      = errors.New("timeout not found")
	ErrWorkflowInProgress   = errors.New("current workflow still in progress - retry once complete")
	ErrOutboxRecordNotFound = errors.New("outbox record not found")
	ErrInvalidTransition    = errors.New("invalid transition")
)

Functions

func AwaitTimeoutInsert 

func AwaitTimeoutInsert[Type any, Status StatusType](
	t testing.TB,
	w *Workflow[Type, Status],
	foreignID, runID string,
	waitFor Status,
)

func CreateDiagram added in v0.2.0 

func CreateDiagram[Type any, Status StatusType](a API[Type, Status], path string, d MermaidDirection) error

CreateDiagram creates a diagram in a md file for communicating a workflow's set of steps in an easy-to-understand manner.

func DeleteForever added in v0.1.2 

func DeleteForever(
	ctx context.Context,
	workflowName string,
	processName string,
	c Consumer,
	store storeFunc,
	lookup lookupFunc,
	customDeleteFn customDelete,
	lagAlert time.Duration,
	clock clock.Clock,
) error

func DeleteTopic added in v0.1.2 

func DeleteTopic(workflowName string) string

func FilterConnectorEventUsing added in v0.1.2 

func FilterConnectorEventUsing(e *ConnectorEvent, filters ...ConnectorEventFilter) bool

func FilterUsing added in v0.1.2 

func FilterUsing(e *Event, filters ...EventFilter) bool

func MakeFilter added in v0.1.2 

func MakeFilter(filters ...RecordFilter) *recordFilters

func Marshal 

func Marshal[T any](t *T) ([]byte, error)

Marshal create a single point of change if the encoding changes.

func Require 

func Require[Type any, Status StatusType](
	t testing.TB,
	w *Workflow[Type, Status],
	foreignID string,
	waitForStatus Status,
	expected Type,
)

func RunStateChangeTopic added in v0.2.0 

func RunStateChangeTopic(workflowName string) string

func Topic 

func Topic(workflowName string, statusType int) string

func TriggerCallbackOn 

func TriggerCallbackOn[Type any, Status StatusType, Payload any](
	t testing.TB,
	w *Workflow[Type, Status],
	foreignID, runID string,
	waitForStatus Status,
	p Payload,
)

func Unmarshal 

func Unmarshal[T any](b []byte, t *T) error

Unmarshal create a single point of change if the decoding changes.

func WaitFor added in v0.2.0 

func WaitFor[Type any, Status StatusType](
	t testing.TB,
	w *Workflow[Type, Status],
	foreignID string,
	fn func(r *Run[Type, Status]) (bool, error),
)

Types

type API 

type API[Type any, Status StatusType] interface {
	// Name returns the name of the implemented workflow.
	Name() string

	// Trigger will kickstart a workflow for the provided foreignID starting from the provided starting status. There
	// is no limitation as to where you start the workflow from. For workflows that have data preceding the initial
	// trigger that needs to be used in the workflow, using WithInitialValue will allow you to provide pre-populated
	// fields of Type that can be accessed by the consumers.
	//
	// foreignID should not be random and should be deterministic for the thing that you are running the workflow for.
	// This especially helps when connecting other workflows as the foreignID is the only way to connect the streams. The
	// same goes for Callback as you will need the foreignID to connect the callback back to the workflow instance that
	// was run.
	Trigger(
		ctx context.Context,
		foreignID string,
		startingStatus Status,
		opts ...TriggerOption[Type, Status],
	) (runID string, err error)

	// Schedule takes a cron spec and will call Trigger at the specified intervals. Schedule is a blocking call and all
	// schedule errors will be retried indefinitely. The same options are available for Schedule as they are
	// for Trigger.
	Schedule(foreignID string, startingStatus Status, spec string, opts ...ScheduleOption[Type, Status]) error

	// Await is a blocking call that returns the typed Run when the workflow of the specified run ID reaches the
	// specified status.
	Await(ctx context.Context, foreignID, runID string, status Status, opts ...AwaitOption) (*Run[Type, Status], error)

	// Callback can be used if Builder.AddCallback has been defined for the provided status. The data in the reader
	// will be passed to the CallbackFunc that you specify and so the serialisation and deserialisation is in the
	// hands of the user.
	Callback(ctx context.Context, foreignID string, status Status, payload io.Reader) error

	// Run must be called in order to start up all the background consumers / consumers required to run the workflow. Run
	// only needs to be called once. Any subsequent calls to run are safe and are noop.
	Run(ctx context.Context)

	// Stop tells the workflow to shut down gracefully.
	Stop()
}

type Ack 

type Ack func() error

Ack is used for the event streamer to safeUpdate its cursor of what messages have been consumed. If Ack is not called then the event streamer, depending on implementation, will likely not keep track of which records / events have been consumed.

type AwaitOption 

type AwaitOption func(o *awaitOpts)

func WithAwaitPollingFrequency added in v0.1.2 

func WithAwaitPollingFrequency(d time.Duration) AwaitOption

type BuildOption 

type BuildOption func(w *buildOptions)

func WithClock 

func WithClock(c clock.Clock) BuildOption

WithClock allows the configuring of workflow's use and access of time. Instead of using time.Now() and other associated functionality from the time package a clock is used instead in order to make it testable.

func WithCustomDelete added in v0.1.2 

func WithCustomDelete[Type any](fn func(object *Type) error) BuildOption

WithCustomDelete allows for specifying a custom deleter function for scrubbing PII data when a workflow Run enters RunStateRequestedDataDeleted and is the function that once executed successfully allows for the RunState to move to RunStateDataDeleted.

func WithDebugMode 

func WithDebugMode() BuildOption

WithDebugMode enabled debug mode for a workflow which results in increased logs such as when processes ar launched, shutdown, events are skipped etc.

func WithDefaultOptions added in v0.1.2 

func WithDefaultOptions(opts ...Option) BuildOption

WithDefaultOptions applies the provided options to the entire workflow and not just to an individual process.

func WithLogger added in v0.2.0 

func WithLogger(l Logger) BuildOption

WithLogger allows for specifying a custom logger. The default is to use a wrapped version of log/slog's Logger.

func WithOutboxErrBackoff added in v0.1.2 

func WithOutboxErrBackoff(d time.Duration) BuildOption

func WithOutboxLagAlert added in v0.1.2 

func WithOutboxLagAlert(d time.Duration) BuildOption

func WithOutboxLookupLimit added in v0.1.2 

func WithOutboxLookupLimit(limit int64) BuildOption

func WithOutboxParallelCount added in v0.1.2 

func WithOutboxParallelCount(count int) BuildOption

func WithOutboxPollingFrequency added in v0.1.2 

func WithOutboxPollingFrequency(d time.Duration) BuildOption

func WithTimeoutStore added in v0.1.2 

func WithTimeoutStore(s TimeoutStore) BuildOption

WithTimeoutStore allows the configuration of a TimeoutStore which is required when using timeouts in a workflow. It is not required by default as timeouts are less common of a feature requirement but when needed the abstraction of complexity of handling scheduling, expiring, and executing are incredibly useful and is included as one of the three key feature offerings of workflow which are sequential steps, callbacks, and timeouts.

type Builder 

type Builder[Type any, Status StatusType] struct {
	// contains filtered or unexported fields
}

func NewBuilder 

func NewBuilder[Type any, Status StatusType](name string) *Builder[Type, Status]

func (*Builder[Type, Status]) AddCallback 

func (b *Builder[Type, Status]) AddCallback(from Status, fn CallbackFunc[Type, Status], allowedDestinations ...Status)

func (*Builder[Type, Status]) AddConnector 

func (b *Builder[Type, Status]) AddConnector(
	name string,
	csc ConnectorConstructor,
	cf ConnectorFunc[Type, Status],
) *connectorUpdater[Type, Status]

func (*Builder[Type, Status]) AddStep 

func (b *Builder[Type, Status]) AddStep(
	from Status,
	c ConsumerFunc[Type, Status],
	allowedDestinations ...Status,
) *stepUpdater[Type, Status]

func (*Builder[Type, Status]) AddTimeout 

func (b *Builder[Type, Status]) AddTimeout(
	from Status,
	timer TimerFunc[Type, Status],
	tf TimeoutFunc[Type, Status],
	allowedDestinations ...Status,
) *timeoutUpdater[Type, Status]

func (*Builder[Type, Status]) Build 

func (b *Builder[Type, Status]) Build(
	eventStreamer EventStreamer,
	recordStore RecordStore,
	roleScheduler RoleScheduler,
	opts ...BuildOption,
) *Workflow[Type, Status]

func (*Builder[Type, Status]) OnCancel added in v0.2.0 

func (b *Builder[Type, Status]) OnCancel(hook RunStateChangeHookFunc[Type, Status])

func (*Builder[Type, Status]) OnComplete added in v0.2.0 

func (b *Builder[Type, Status]) OnComplete(hook RunStateChangeHookFunc[Type, Status])

func (*Builder[Type, Status]) OnPause added in v0.2.0 

func (b *Builder[Type, Status]) OnPause(hook RunStateChangeHookFunc[Type, Status])

type CallbackFunc 

type CallbackFunc[Type any, Status StatusType] func(ctx context.Context, r *Run[Type, Status], reader io.Reader) (Status, error)

type ConnectorConstructor added in v0.1.2 

type ConnectorConstructor interface {
	Make(ctx context.Context, consumerName string) (ConnectorConsumer, error)
}

type ConnectorConsumer added in v0.1.2 

type ConnectorConsumer interface {
	Recv(ctx context.Context) (*ConnectorEvent, Ack, error)
	Close() error
}

type ConnectorEvent added in v0.1.2 

type ConnectorEvent struct {
	// ID is a unique ID for the event.
	ID string
	// ForeignID refers to the ID of the element that the event relates to.
	ForeignID string
	// Type relates to the StatusType that the associated record changed to.
	Type string
	// Headers stores meta-data in a simple and easily queryable way.
	Headers map[string]string
	// CreatedAt is the time that the event was produced and is generated by the event streamer.
	CreatedAt time.Time
}

ConnectorEvent defines a schema that is inline with how workflow uses an event notification pattern. This means that events only tell us what happened and do not transmit the state change. ConnectorEvent differs slightly from Event in that all fields, except for CreatedAt, are string based and allows representation relations to elements with string identifiers and string based types.

type ConnectorEventFilter added in v0.1.2 

type ConnectorEventFilter func(e *ConnectorEvent) bool

ConnectorEventFilter can be passed to the event streaming implementation to allow specific consumers to have an earlier on filtering process. True is returned when the event should be skipped.

type ConnectorFunc 

type ConnectorFunc[Type any, Status StatusType] func(ctx context.Context, w API[Type, Status], e *ConnectorEvent) error

type Consumer 

type Consumer interface {
	Recv(ctx context.Context) (*Event, Ack, error)
	Close() error
}

type ConsumerFunc 

type ConsumerFunc[Type any, Status StatusType] func(ctx context.Context, r *Run[Type, Status]) (Status, error)

ConsumerFunc provides a record that is expected to be modified if the data needs to change. If true is returned with a nil error then the record, along with its modifications, will be stored. If false is returned with a nil error then the record will not be stored and the event will be skipped and move onto the next event. If a non-nil error is returned then the consumer will back off and try again until a nil error occurs or the retry max has been reached if a Dead Letter Queue has been configured for the workflow.

type ConsumerOption 

type ConsumerOption func(*ConsumerOptions)

func WithConsumerPollFrequency 

func WithConsumerPollFrequency(d time.Duration) ConsumerOption

type ConsumerOptions 

type ConsumerOptions struct {
	PollFrequency time.Duration
	Lag           time.Duration
}

type Event 

type Event struct {
	// ID is a unique ID for the event generated by the event streamer.
	ID int64

	// ForeignID refers to the ID of a record in the record store.
	ForeignID string

	// Type relates to the StatusType that the associated record changed to.
	Type int

	// Headers stores meta-data in a simple and easily queryable way.
	Headers map[Header]string

	// CreatedAt is the time that the event was produced and is generated by the event streamer.
	CreatedAt time.Time
}

type EventFilter 

type EventFilter func(e *Event) bool

EventFilter can be passed to the event streaming implementation to allow specific consumers to have an earlier on filtering process. True is returned when the event should be skipped.

type EventStreamer 

type EventStreamer interface {
	NewProducer(ctx context.Context, topic string) (Producer, error)
	NewConsumer(ctx context.Context, topic string, name string, opts ...ConsumerOption) (Consumer, error)
}

EventStreamer implementations should all be tested with adaptertest.TestEventStreamer

type FilterValue added in v0.1.2 

type FilterValue struct {
	Enabled bool
	Value   string
}
type Header string
const (
	HeaderWorkflowName Header = "workflow_name"
	HeaderForeignID    Header = "foreign_id"
	HeaderTopic        Header = "topic"
	HeaderRunID        Header = "run_id"
	HeaderRunState     Header = "run_state"
)

type Logger added in v0.2.0 

type Logger interface {
	// Debug will be used by workflow for debug logs when in debug mode.
	Debug(ctx context.Context, msg string, meta map[string]string)
	// Error is used when writing errors to the logs.
	Error(ctx context.Context, err error)
}

Logger interface allows the user of Workflow to provide a custom logger and not use the default which is provided in internal/logger. Workflow only writes two types of logs: Debug and Error. Error is only used at the highest level where an auto-retry process (consumers and pollers) errors and retries.

Error is used only when the error cannot be passed back to the caller and cannot be bubbled up any further.

Debug is used only when the Workflow is built with WithDebugMode.

type MermaidDirection 

type MermaidDirection string
const (
	UnknownDirection     MermaidDirection = ""
	TopToBottomDirection MermaidDirection = "TB"
	LeftToRightDirection MermaidDirection = "LR"
	RightToLeftDirection MermaidDirection = "RL"
	BottomToTopDirection MermaidDirection = "BT"
)

type MermaidFormat 

type MermaidFormat struct {
	WorkflowName   string
	Direction      MermaidDirection
	Nodes          []int
	StartingPoints []int
	TerminalPoints []int
	Transitions    []MermaidTransition
}

type MermaidTransition 

type MermaidTransition struct {
	From int
	To   []int
}

type Option added in v0.1.2 

type Option func(so *options)

func ConsumeLag added in v0.1.2 

func ConsumeLag(d time.Duration) Option

ConsumeLag defines the age of the event that the consumer will consume. The workflow consumer will not consume events newer than the time specified and will wait to consume them.

func ErrBackOff added in v0.1.2 

func ErrBackOff(d time.Duration) Option

ErrBackOff defines the time duration of the backoff of the workflow process when an error is encountered.

func LagAlert added in v0.1.2 

func LagAlert(d time.Duration) Option

LagAlert defines the time duration / threshold before the prometheus metric defined in /internal/metrics/metrics.go switches to true which means that the workflow consumer is struggling to consume events fast enough and might need to be converted to a parallel consumer.

func ParallelCount added in v0.1.2 

func ParallelCount(instances int) Option

ParallelCount defines the number of instances of the workflow process. The processes are shareded consistently and will be provided a name such as "consumer-1-of-5" to show the instance number and the total number of instances that the process is a part of.

func PauseAfterErrCount added in v0.1.2 

func PauseAfterErrCount(count int) Option

PauseAfterErrCount defines the number of times an error can occur until the record is updated to RunStatePaused which is similar to a Dead Letter Queue in the sense that the record will no longer be processed and won't block the workflow's consumers and can be investigated and retried later on.

func PollingFrequency added in v0.1.2 

func PollingFrequency(d time.Duration) Option

PollingFrequency defines the time duration of which the workflow process will poll for changes.

type OrderType added in v0.1.2 

type OrderType int
const (
	OrderTypeUnknown    OrderType = 0
	OrderTypeAscending  OrderType = 1
	OrderTypeDescending OrderType = 2
)

func (OrderType) String added in v0.1.2 

func (ot OrderType) String() string

type OutboxEvent added in v0.1.2 

type OutboxEvent struct {
	// ID is a unique ID for this specific OutboxEvent.
	ID string

	// WorkflowName refers to the name of the workflow that the OutboxEventData belongs to.
	WorkflowName string

	// Data represents a slice of bytes the OutboxEventDataMaker constructs via serialising event data
	// in an expected way for it to also be deserialized by the outbox consumer.
	Data []byte

	// CreatedAt is the time that this specific OutboxEvent was produced.
	CreatedAt time.Time
}

type OutboxEventData added in v0.1.2 

type OutboxEventData struct {
	ID string

	// WorkflowName refers to the name of the workflow that the OutboxEventData belongs to.
	WorkflowName string

	// Data represents a slice of bytes the OutboxEventDataMaker constructs via serialising event data
	// in an expected way for it to also be deserialized by the outbox consumer.
	Data []byte
}

func MakeOutboxEventData added in v0.2.0 

func MakeOutboxEventData(record Record) (OutboxEventData, error)

MakeOutboxEventData creates a OutboxEventData that houses all the information that must be stored and be retrievable from the outbox.

type Producer 

type Producer interface {
	Send(ctx context.Context, foreignID string, statusType int, headers map[Header]string) error
	Close() error
}

type Record 

type Record struct {
	WorkflowName string
	ForeignID    string
	RunID        string
	RunState     RunState
	Status       int
	Object       []byte
	CreatedAt    time.Time
	UpdatedAt    time.Time
}

Record is the cornerstone of Workflow. Record must always be wire compatible with no generics as it's intended purpose is to be the stored structure of a Run.

type RecordFilter added in v0.1.2 

type RecordFilter func(filters *recordFilters)

func FilterByForeignID added in v0.1.2 

func FilterByForeignID(val string) RecordFilter

func FilterByRunState added in v0.1.2 

func FilterByRunState(rs RunState) RecordFilter

func FilterByStatus added in v0.1.2 

func FilterByStatus(status int64) RecordFilter

type RecordStore 

type RecordStore interface {
	// Store should create or update a record depending on whether the underlying store is mutable or append only. Store
	// must implement transactions and a separate outbox store to store the outbox record (that should be
	// generated using MakeOutboxEventData) which can be retrieved when calling ListOutboxEvents and can be
	// deleted when DeleteOutboxEvent is called.
	Store(ctx context.Context, record *Record) error
	Lookup(ctx context.Context, runID string) (*Record, error)
	Latest(ctx context.Context, workflowName, foreignID string) (*Record, error)

	// List provides a slice of Record where the total items will be equal or less than the limit depending
	// on the offset provided and how many records remain after that ID.
	List(ctx context.Context, workflowName string, offsetID int64, limit int, order OrderType, filters ...RecordFilter) ([]Record, error)

	// ListOutboxEvents lists all events that are yet to be published to the event streamer. A requirement for
	// implementation of the RecordStore is to support a Transactional Outbox that has Event's written to it when
	// Store is called.
	ListOutboxEvents(ctx context.Context, workflowName string, limit int64) ([]OutboxEvent, error)
	// DeleteOutboxEvent will expect an Event's ID field and will remove the event from the outbox store when the
	// event has successfully been published to the event streamer.
	DeleteOutboxEvent(ctx context.Context, id string) error
}

RecordStore implementations should all be tested with adaptertest.TestRecordStore. The underlying implementation of store must support transactions or the ability to commit the record and an outbox event in a single call as well as being able to obtain an ID for the record before it is created.

type RoleScheduler 

type RoleScheduler interface {
	// Await must return a child context of the provided (parent) context. Await should block until the role is
	// assigned to the caller. Only one caller should be able to be assigned the role at any given time. The returned
	// context.CancelFunc is called after each process execution. Some process executions can be more long living and
	// others not but if any process errors the context.CancelFunc will be called after the specified error backoff
	// has finished.
	Await(ctx context.Context, role string) (context.Context, context.CancelFunc, error)
}

RoleScheduler implementations should all be tested with adaptertest.TestRoleScheduler

type Run added in v0.2.0 

type Run[Type any, Status StatusType] struct {
	TypedRecord[Type, Status]
	// contains filtered or unexported fields
}

Run is a representation of a workflow run. It incorporates all the fields from the Record as well as having defined types for the Status and Object fields along with access to the RunStateController which controls the state of the run aka "RunState".

func NewTestingRun added in v0.2.0 

func NewTestingRun[Type any, Status StatusType](
	t *testing.T,
	wr Record,
	object Type,
	opts ...TestingRunOption,
) Run[Type, Status]

NewTestingRun should be used when testing logic that defines a workflow.Run as a parameter. This is usually the case in unit tests and would not normally be found when doing an Acceptance test for the entire workflow.

func (*Run[Type, Status]) Cancel added in v0.2.0 

func (r *Run[Type, Status]) Cancel(ctx context.Context) (Status, error)

Cancel is intended to be used inside a workflow process where (Status, error) are the return signature. This allows the user to simply type "return r.Cancel(ctx)" to cancel a record from inside a workflow which results in the record being permanently left alone and will not be processed.

func (*Run[Type, Status]) Pause added in v0.2.0 

func (r *Run[Type, Status]) Pause(ctx context.Context) (Status, error)

Pause is intended to be used inside a workflow process where (Status, error) are the return signature. This allows the user to simply type "return r.Pause(ctx)" to pause a record from inside a workflow which results in the record being temporarily left alone and will not be processed until it is resumed.

func (*Run[Type, Status]) Skip added in v0.2.0 

func (r *Run[Type, Status]) Skip() (Status, error)

Skip is a util function to skip the update and move on to the next event (consumer) or execution (callback)

type RunState added in v0.1.2 

type RunState int
const (
	RunStateUnknown              RunState = 0
	RunStateInitiated            RunState = 1
	RunStateRunning              RunState = 2
	RunStatePaused               RunState = 3
	RunStateCancelled            RunState = 4
	RunStateCompleted            RunState = 5
	RunStateDataDeleted          RunState = 6
	RunStateRequestedDataDeleted RunState = 7
)

func (RunState) Finished added in v0.1.2 

func (rs RunState) Finished() bool

func (RunState) Stopped added in v0.1.2 

func (rs RunState) Stopped() bool

Stopped is the type of status that requires consumers to ignore the workflow run as it is in a stopped state. Only paused workflow runs can be resumed and must be done so via the workflow API or the Run methods. All cancelled workflow runs are cancelled permanently and cannot be undone whereas Pausing can be resumed.

func (RunState) String added in v0.1.2 

func (rs RunState) String() string

func (RunState) Valid added in v0.1.2 

func (rs RunState) Valid() bool

type RunStateChangeHookFunc added in v0.2.0 

type RunStateChangeHookFunc[Type any, Status StatusType] func(ctx context.Context, record *TypedRecord[Type, Status]) error

RunStateChangeHookFunc defines the function signature for all hooks associated to the run.

type RunStateController added in v0.1.2 

type RunStateController interface {
	// Pause will take the workflow run specified and move it into a temporary state where it will no longer be processed.
	// A paused workflow run can be resumed by calling Resume. ErrUnableToPause is returned when a workflow is not in a
	// state to be paused.
	Pause(ctx context.Context) error
	// Cancel can be called after Pause has been called. A paused run of the workflow can be indefinitely cancelled.
	// Once cancelled, DeleteData can be called and will move the run into an indefinite state of DataDeleted.
	// ErrUnableToCancel is returned when the workflow record is not in a state to be cancelled.
	Cancel(ctx context.Context) error
	// Resume can be called on a workflow run that has been paused. ErrUnableToResume is returned when the workflow
	// run is not in a state to be resumed.
	Resume(ctx context.Context) error
	// DeleteData can be called after a workflow run has been completed or cancelled. DeleteData should be used to
	// comply with the right to be forgotten such as complying with GDPR. ErrUnableToDelete is returned when the
	// workflow run is not in a state to be deleted.
	DeleteData(ctx context.Context) error
}

RunStateController allows the interaction with a specific workflow record.

func NewRunStateController added in v0.1.2 

func NewRunStateController(store storeFunc, wr *Record) RunStateController

type ScheduleOption added in v0.1.2 

type ScheduleOption[Type any, Status StatusType] func(o *scheduleOpts[Type, Status])

func WithScheduleFilter added in v0.1.2 

func WithScheduleFilter[Type any, Status StatusType](
	fn func(ctx context.Context) (bool, error),
) ScheduleOption[Type, Status]

func WithScheduleInitialValue added in v0.1.2 

func WithScheduleInitialValue[Type any, Status StatusType](t *Type) ScheduleOption[Type, Status]

type SkipType added in v0.1.2 

type SkipType int
var (
	SkipTypeDefault        SkipType = 0
	SkipTypeRunStateUpdate SkipType = -1
)

type State 

type State int
const (
	StateUnknown  State = 0
	StateShutdown State = 1
	StateRunning  State = 2
	StateIdle     State = 3
)

func (State) String added in v0.1.2 

func (s State) String() string

type StatusType 

type StatusType interface {
	~int | ~int32 | ~int64

	String() string
}

type TestingRecordStore 

type TestingRecordStore interface {
	RecordStore

	Snapshots(workflowName, foreignID, runID string) []*Record
}

type TestingRunOption added in v0.2.0 

type TestingRunOption func(*testingRunOpts)

func WithCancelFn added in v0.2.0 

func WithCancelFn(cancel func(ctx context.Context) error) TestingRunOption

func WithDeleteDataFn added in v0.2.0 

func WithDeleteDataFn(deleteData func(ctx context.Context) error) TestingRunOption

func WithPauseFn added in v0.2.0 

func WithPauseFn(pause func(ctx context.Context) error) TestingRunOption

func WithResumeFn added in v0.2.0 

func WithResumeFn(resume func(ctx context.Context) error) TestingRunOption

type TimeoutFunc 

type TimeoutFunc[Type any, Status StatusType] func(ctx context.Context, r *Run[Type, Status], now time.Time) (Status, error)

TimeoutFunc runs once the timeout has expired which is set by TimerFunc. If false is returned with a nil error then the timeout is skipped and not retried at a later date. If a non-nil error is returned the TimeoutFunc will be called again until a nil error is returned. If true is returned with a nil error then the provided record and any modifications made to it will be stored and the status updated - continuing the workflow.

type TimeoutRecord added in v0.1.2 

type TimeoutRecord struct {
	ID           int64
	WorkflowName string
	ForeignID    string
	RunID        string
	Status       int
	Completed    bool
	ExpireAt     time.Time
	CreatedAt    time.Time
}

type TimeoutStore 

type TimeoutStore interface {
	Create(ctx context.Context, workflowName, foreignID, runID string, status int, expireAt time.Time) error
	Complete(ctx context.Context, id int64) error
	Cancel(ctx context.Context, id int64) error
	List(ctx context.Context, workflowName string) ([]TimeoutRecord, error)
	ListValid(ctx context.Context, workflowName string, status int, now time.Time) ([]TimeoutRecord, error)
}

TimeoutStore implementations should all be tested with adaptertest.TestTimeoutStore

type TimerFunc 

type TimerFunc[Type any, Status StatusType] func(ctx context.Context, r *Run[Type, Status], now time.Time) (time.Time, error)

TimerFunc exists to allow the specification of when the timeout should expire dynamically. If not time is set then a timeout will not be created and the event will be skipped. If the time is set then a timeout will be created and once expired TimeoutFunc will be called. Any non-nil error will be retried with backoff.

func DurationTimerFunc 

func DurationTimerFunc[Type any, Status StatusType](duration time.Duration) TimerFunc[Type, Status]

func TimeTimerFunc 

func TimeTimerFunc[Type any, Status StatusType](t time.Time) TimerFunc[Type, Status]

type TriggerOption 

type TriggerOption[Type any, Status StatusType] func(o *triggerOpts[Type, Status])

func WithInitialValue 

func WithInitialValue[Type any, Status StatusType](t *Type) TriggerOption[Type, Status]

type TypedRecord added in v0.2.0 

type TypedRecord[Type any, Status StatusType] struct {
	Record
	Status Status
	Object *Type
}

TypedRecord differs from Record in that it contains a Typed Object and Typed Status

type Workflow 

type Workflow[Type any, Status StatusType] struct {
	// contains filtered or unexported fields
}

func (*Workflow[Type, Status]) Await 

func (w *Workflow[Type, Status]) Await(
	ctx context.Context,
	foreignID, runID string,
	status Status,
	opts ...AwaitOption,
) (*Run[Type, Status], error)

func (*Workflow[Type, Status]) Callback 

func (w *Workflow[Type, Status]) Callback(
	ctx context.Context,
	foreignID string,
	status Status,
	payload io.Reader,
) error

func (*Workflow[Type, Status]) Name 

func (w *Workflow[Type, Status]) Name() string

func (*Workflow[Type, Status]) Run 

func (w *Workflow[Type, Status]) Run(ctx context.Context)

func (*Workflow[Type, Status]) Schedule added in v0.1.2 

func (w *Workflow[Type, Status]) Schedule(
	foreignID string,
	startingStatus Status,
	spec string,
	opts ...ScheduleOption[Type, Status],
) error

func (*Workflow[Type, Status]) States 

func (w *Workflow[Type, Status]) States() map[string]State

func (*Workflow[Type, Status]) Stop 

func (w *Workflow[Type, Status]) Stop()

Stop cancels the context provided to all the background processes that the workflow launched and waits for all of them to shut down gracefully.

func (*Workflow[Type, Status]) Trigger 

func (w *Workflow[Type, Status]) Trigger(
	ctx context.Context,
	foreignID string,
	startingStatus Status,
	opts ...TriggerOption[Type, Status],
) (runID string, err error)

Source Files

View all Source files

Directories

Path Synopsis
_examples
callback module
connector module
gettingstarted module
schedule module
timeout module
webui module
adapters
adaptertest
memrecordstore
memrolescheduler
memstreamer
memtimeoutstore
jlog module
kafkastreamer module
reflexstreamer module
rinkrolescheduler module
sqlstore module
sqltimeout module
webui module
examples module
internal
errorcounter
graph
logger
metrics
outboxpb
util

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.