asyncmachine-go

Declarative workflows with relations (state machine)

Usage »
Examples »
Documentation »
Tools »
Integrations »
Case Studies »

asyncmachine-go is a minimal implementation of AsyncMachine in Golang using channels and context. It aims at simplicity and speed.

It can be used as a lightweight in-memory Temporal alternative, worker for Asynq, or to create simple consensus engines, stateful firewalls, telemetry, bots, etc.

asyncmachine-go is a general purpose state machine for managing complex asynchronous workflows in a safe and structured way

Comparison

Common differences from other state machines:

• many states can be active at the same time
• transitions between all the states are allowed
• states are connected by relations
• every transition can be rejected
• error is a state

Buzzwords

AM technically is: event emitter, queue, dependency graph, AOP, logical clocks, ~2.5k LoC, no deps

AM gives you: states, events, thread-safety, logging, metrics, traces, debugger, history, flow constraints, scheduler

Flow constraints are: state mutations, negotiation, relations, "when" methods, state contexts, external contexts

Usage

Basics

// ProcessingFile -> FileProcessed (1 async and 1 sync state)
package main

import am "github.com/pancsta/asyncmachine-go/pkg/machine"

func main() {
    // init the machine
    mach := am.New(nil, am.Struct{
        "ProcessingFile": {
            Add: am.S{"InProgress"},
            Remove: am.S{"FileProcessed"},
        },
        "FileProcessed": {
            Remove: am.S{"ProcessingFile", "InProgress"},
        },
        "InProgress": {},
    }, nil)
    mach.BindHandlers(&Handlers{
        Filename: "README.md",
    })
    // change the state
    mach.Add1("ProcessingFile", nil)
    // wait for completed
    select {
    case <-time.After(5 * time.Second):
        println("timeout")
    case <-mach.WhenErr(nil):
        println("err:", mach.Err)
    case <-mach.When1("FileProcessed", nil):
        println("done")
    }
}

type Handlers struct {
    Filename string
}

// negotiation handler
func (h *Handlers) ProcessingFileEnter(e *am.Event) bool {
    // read-only ops
    // decide if moving fwd is ok
    // no blocking
    // lock-free critical zone
    return true
}

// final handler
func (h *Handlers) ProcessingFileState(e *am.Event) {
    // read & write ops
    // no blocking
    // lock-free critical zone
    mach := e.Machine
    // tick-based context
    stateCtx := mach.NewStateCtx("ProcessingFile")
    go func() {
        // block in the background, locks needed
        if stateCtx.Err() != nil {
            return // expired
        }
        // blocking call
        err := processFile(h.Filename, stateCtx)
        if err != nil {
            mach.AddErr(err)
            return
        }
        // re-check the tick ctx after a blocking call
        if stateCtx.Err() != nil {
            return // expired
        }
        // move to the next state in the flow
        mach.Add1("FileProcessed", nil)
    }()
}

Waiting

// wait until FileDownloaded becomes active
<-mach.When1("FileDownloaded", nil)

// wait until FileDownloaded becomes inactive
<-mach.WhenNot1("DownloadingFile", args, nil)

// wait for EventConnected to be activated with an arg ID=123
<-mach.WhenArgs("EventConnected", am.A{"ID": 123}, nil)

// wait for Foo to have a tick >= 6 and Bar tick >= 10
<-mach.WhenTime(am.S{"Foo", "Bar"}, am.T{6, 10}, nil)

// wait for DownloadingFile to have a tick increased by 2 since now
<-mach.WhenTick("DownloadingFile", 2, nil)

// wait for an error
<-mach.WhenErr()

See docs/cookbook.md for more snippets.

Examples

FSM - Finite State Machine

• origin

var (
    states = am.Struct{
        // input states
        InputPush: {},
        InputCoin: {},

        // "state" states
        Locked: {
            Auto:   true,
            Remove: groupUnlocked,
        },
        Unlocked: {Remove: groupUnlocked},
    }
)

NFA - Nondeterministic Finite Automaton

• origin

var (
    states = am.Struct{
        // input states
        Input: {Multi: true},

        // action states
        Start: {Add: am.S{StepX}},

        // "state" states
        StepX: {Remove: groupSteps},
        Step0: {Remove: groupSteps},
        Step1: {Remove: groupSteps},
        Step2: {Remove: groupSteps},
        Step3: {Remove: groupSteps},
    }
)

PathWatcher

• origin

// States map defines relations and properties of states (for files).
var States = am.Struct{
    Init: {Add: S{Watching}},

    Watching: {
        Add:   S{Init},
        After: S{Init},
    },
    ChangeEvent: {
        Multi:   true,
        Require: S{Watching},
    },

    Refreshing: {
        Multi:  true,
        Remove: S{AllRefreshed},
    },
    Refreshed:    {Multi: true},
    AllRefreshed: {},
}

// StatesDir map defines relations and properties of states (for directories).
var StatesDir = am.Struct{
    Refreshing:   {Remove: groupRefreshed},
    Refreshed:    {Remove: groupRefreshed},
    DirDebounced: {Remove: groupRefreshed},
    DirCached:    {},
}

Temporal Expense Workflow

• origin

// States map defines relations and properties of states.
var States = am.Struct{
    CreatingExpense: {Remove: GroupExpense},
    ExpenseCreated:  {Remove: GroupExpense},
    WaitingForApproval: {
        Auto:   true,
        Remove: GroupApproval,
    },
    ApprovalGranted: {Remove: GroupApproval},
    PaymentInProgress: {
        Auto:   true,
        Remove: GroupPayment,
    },
    PaymentCompleted: {Remove: GroupPayment},
}

Temporal FileProcessing Workflow

• origin
• Asynq worker version

// States map defines relations and properties of states.
var States = am.Struct{
    DownloadingFile: {Remove: GroupFileDownloaded},
    FileDownloaded:  {Remove: GroupFileDownloaded},
    ProcessingFile: {
        Auto:    true,
        Require: S{FileDownloaded},
        Remove:  GroupFileProcessed,
    },
    FileProcessed: {Remove: GroupFileProcessed},
    UploadingFile: {
        Auto:    true,
        Require: S{FileProcessed},
        Remove:  GroupFileUploaded,
    },
    FileUploaded: {Remove: GroupFileUploaded},
}

Documentation

• godoc
• cookbook
• discussions
• manual.md | manual.pdf
  • Machine and States
    • State Clocks and Context
    • Auto States
    • Categories of States
    • ...
  • Changing State
    • State Mutations
    • Transition Lifecycle
    • Calculating Target States
    • Negotiation Handlers
    • Final Handlers
    • ...
  • Advanced Topics
    • State's Relations
    • Queue and History
    • Typesafe States
    • ...
  • Cheatsheet

Tools

Generator

am-gen will quickly bootstrap a typesafe states file for you.

$ am-gen states-file Foo,Bar

package states

import am "github.com/pancsta/asyncmachine-go/pkg/machine"

// S is a type alias for a list of state names.
type S = am.S

// States map defines relations and properties of states.
var States = am.Struct{
    Foo: {},
    Bar: {},
}

// Groups of mutually exclusive states.

//var (
//      GroupPlaying = S{Playing, Paused}
//)

//#region boilerplate defs

// Names of all the states (pkg enum).

const (
    Foo = "Foo"
    Bar = "Bar"
)

// Names is an ordered list of all the state names.
var Names = S{
    Foo,
    Bar,
    am.Exception,
}

//#endregion

See tools/cmd/am-gen for more info.

Debugger

am-dbg is a lightweight, multi-client debugger for AM. It easily handles >100 client machines simultaneously streaming telemetry data (and potentially many more). Some features include:

• states tree
• log view
• time travel
• transition steps
• import / export
• filters
• matrix view

See tools/cmd/am-dbg for more info.

Integrations

Open Telemetry

pkg/telemetry provides Open Telemetry integration which exposes machine's states and transitions as Otel traces, compatible with Jaeger.

See pkg/telemetry for more info or import a sample asset.

Prometheus

pkg/telemetry/prometheus binds to machine's transactions and averages the values withing an interval window and exposes various metrics. Combined with Grafana, it can be used to monitor the metrics of you machines.

See pkg/telemetry/prometheus for more info.

Case Studies

Several case studies are available to show how to implement various types of machines, measure performance and produce a lot of inspectable data.

libp2p-pubsub benchmark

• pubsub host - eg ps-17 (20 states)
  PubSub machine is a simple event loop with Multi states which get responses via arg channels. Heavy use of Machine.Eval().
• discovery - eg ps-17-disc (10 states)
  Discovery machine is a simple event loop with Multi states and a periodic refresh state.
• discovery bootstrap - eg ps-17-disc-bf3 (5 states)
  BootstrapFlow is a non-linear flow for topic bootstrapping with some retry logic.

See github.com/pancsta/go-libp2p-pubsub-benchmark or the pdf results for more info.

libp2p-pubsub simulator

• simulator sim (14 states)
  Root simulator machine, initializes the network and manipulates it during heartbeats according to frequency definitions. Heavily dependent on state negotiation.
• simulator's peer - eg sim-p17 (17 states)
  Handles peer's connections, topics and messages. This machine has a decent amount of relations. Each sim peer has its own pubsub host.
• topics - eg sim-t-se7ev (5 states)
  State-only machine (no handlers, no goroutine). States represent correlations with peer machines.

See github.com/pancsta/go-libp2p-pubsub-benchmark for more info.

am-dbg

am-dbg is a cview TUI app with a single machine consisting of:

• input events (7 states)
• external state (11 states)
• actions (14 states)

This machine features a decent amount of relations within a large number of states and 4 state groups. It's also a good example to see how easily an AM-based program can be controller with a script in tools/cmd/am-dbg-demo.

// States map defines relations and properties of states.
var States = am.Struct{
    ///// Input events

    ClientMsg:       {Multi: true},
    ConnectEvent:    {Multi: true},
    DisconnectEvent: {Multi: true},

    // user scrolling tx / steps
    UserFwd: {
        Add:    S{Fwd},
        Remove: GroupPlaying,
    },
    UserBack: {
        Add:    S{Back},
        Remove: GroupPlaying,
    },
    UserFwdStep: {
        Add:     S{FwdStep},
        Require: S{ClientSelected},
        Remove:  am.SMerge(GroupPlaying, S{LogUserScrolled}),
    },
    UserBackStep: {
        Add:     S{BackStep},
        Require: S{ClientSelected},
        Remove:  am.SMerge(GroupPlaying, S{LogUserScrolled}),
    },

    ///// External state (eg UI)

    // focus group

    TreeFocused:          {Remove: GroupFocused},
    LogFocused:           {Remove: GroupFocused},
    SidebarFocused:       {Remove: GroupFocused},
    TimelineTxsFocused:   {Remove: GroupFocused},
    TimelineStepsFocused: {Remove: GroupFocused},
    MatrixFocused:        {Remove: GroupFocused},
    DialogFocused:        {Remove: GroupFocused},

    StateNameSelected:    {Require: S{ClientSelected}},
    HelpDialog:           {Remove: GroupDialog},
    ExportDialog: {
        Require: S{ClientSelected},
        Remove:  GroupDialog,
    },
    LogUserScrolled: {},
    Ready:           {Require: S{Start}},

    ///// Actions

    Start: {},
    TreeLogView: {
        Auto:   true,
        Remove: GroupViews,
    },
    MatrixView:     {Remove: GroupViews},
    TreeMatrixView: {Remove: GroupViews},
    TailMode: {
        Require: S{ClientSelected},
        Remove:  GroupPlaying,
    },
    Playing: {
        Require: S{ClientSelected},
        Remove:  am.SMerge(GroupPlaying, S{LogUserScrolled}),
    },
    Paused: {
        Auto:    true,
        Require: S{ClientSelected},
        Remove:  GroupPlaying,
    },

    // tx / steps back / fwd

    Fwd: {
        Require: S{ClientSelected},
        Remove:  S{Playing},
    },
    Back: {
        Require: S{ClientSelected},
        Remove:  S{Playing},
    },
    FwdStep: {
        Require: S{ClientSelected},
        Remove:  S{Playing},
    },
    BackStep: {
        Require: S{ClientSelected},
        Remove:  S{Playing},
    },

    ScrollToTx: {Require: S{ClientSelected}},

    // client selection

    SelectingClient: {Remove: S{ClientSelected}},
    ClientSelected: {
        Remove: S{SelectingClient, LogUserScrolled},
    },
    RemoveClient: {Require: S{ClientSelected}},
}

See tools/debugger/states for more info.

Roadmap

• negotiation testers (eg CanAdd)
• helpers for composing networks of machines
• helpers for queue and history traversal
• "state-trace" navbar in am-dbg (via AddFromEv)
• go1.22 traces
• inference
• optimizations
• manual updated to a spec

See also issues.

Development

• all PRs welcome
• before
  • ./scripts/dep-taskfile.sh
  • task install-deps
• after
  • task test
  • task format
  • task lint

Changelog

Latest release: v0.6.1

• fix(am-dbg): fix tail mode with filters #85 (@pancsta)
• fix: address misc issues #84 (@pancsta)
• docs: add pdf manual #83 (@pancsta)
• docs: minor manual updates #82 (@pancsta)
• refac(am-dbg): split and reorg files #81 (@pancsta)
• feat(telemetry): include log level in msgs #80 (@pancsta)
• feat(am-dbg): add tx and log filtering #79 (@pancsta)
• feat(machine): add global AnyAny negotiation handler #78 (@pancsta)
• refac(machine): extract When* ctx disposal #77 (@pancsta)
• refac(machine): refac to directional channs #76 (@pancsta)
• refac(machine): reorder Eval params #75 (@pancsta)
• feat(machine): add NoOpTracer for embedding #74 (@pancsta)
• feat(machine): add Tracer.QueueEnd #73 (@pancsta)
• test(machine): increase coverage to 85% #72 (@pancsta)
• refac(machine): make WhenDisposed a method #71 (@pancsta)

Maintenance release: v0.5.1

• fix(machine): fix Dispose() deadlock #70 (@pancsta)
• fix(machine): allow for nil ctx #69 (@pancsta)
• fix(am-dbg): fix tail mode delay #68 (@pancsta)
• fix(am-dbg): fix crash for machs with log level 0 #67 (@pancsta)
• feat(machine): add WhenQueueEnds #65 (@pancsta)
• docs: update manual to v0.5.0 #64 (@pancsta)

See CHANELOG.md for the full list.