asyncmachine-go is a distributed workflow engine (technically a pathless control-flow graph with a consensus), which implements AOP and actor model through a clock-based state-machine. It features atomic transitions, relations, transparent RPC, TUI debugger, telemetry, REPL, selective distribution, diagrams, and WASM support.

As a control flow library, it decides about running of predefined bits of code (transition handlers) - their order and which ones to run, according to currently active states (flags). Thanks to a novel state machine, the number of handlers can be minimized while maximizing scenario coverage. It's lightweight, fault-tolerant by design, has rule-based mutations, and can target virtually any step-in-time, in any workflow. It's a low-level tool with acceptable performance.

asyncmachine-go takes care of context, select, and panic, while allowing for graph-structured concurrency with goroutine cancelation. The history log and relations have vector formats. It aims to create autonomous workflows with organic control flow and stateful APIs.

• binary flag
• node with relations
• AOP aspect
• logical clock
• subscription topic
• multiple methods
• metric
• trace
• lock
• breakpoint

Besides the main use-case of workflows, it can be used for stateful applications of any size - daemons, UIs, configs, bots, firewalls, synchronization consensus, games, smart graphs, microservice orchestration, robots, contracts, streams, DI containers, message broking, test scenarios, simulators, as well as "real-time" systems which rely on instant cancelation.

Minimal - an untyped definition of 2 states and 1 relation, then 1 mutation and a check.

import am "github.com/pancsta/asyncmachine-go/pkg/machine"
// ...
mach := am.New(nil, am.Schema{
    "Foo": {Require: am.S{"Bar"}},
    "Bar": {},
}, nil)
mach.Add1("Foo", nil)
mach.Is1("Foo") // false

Complicated - wait on a multi state (event) and the Ready state with a 1s timeout, then mutate with typed args, on top of a state context.

// state ctx is an expiration ctx
ctx := client.Mach.NewStateCtx(ssC.WorkerReady)
// clock-based subscription
whenPayload := client.Mach.WhenTicks(ssC.WorkerPayload, 1, ctx)
// state mutation
client.RpcWorker.NetMach.Add1(ssW.WorkRequested, Pass(&A{
    Input: 2}))
// WaitFor* wraps select statements
err := amhelp.WaitForAll(ctx, time.Second,
    // post-mutation subscription
    mach.When1(ss.BasicStatesDef.Ready, nil),
    // pre-mutation subscription
    whenPayload)
// check cancelation
if ctx.Err() != nil {
    return // state ctx expired
}
// check error
if err != nil {
    // error state mutation
    client.Mach.AddErr(err, nil)
    return // no err required
}
// client/WorkerPayload and mach/Ready activated

Handlers - Aspect Oriented transition handlers.

// can Foo activate?
func (h *Handlers) FooEnter(e *am.Event) bool {
    return true
}
// with Foo active, can Bar activate?
func (h *Handlers) FooBar(e *am.Event) bool {
    return true
}
// Foo activates
func (h *Handlers) FooState(e *am.Event) {
    h.foo = NewConn()
}
// Foo de-activates
func (h *Handlers) FooEnd(e *am.Event) {
    h.foo.Close()
}

Schemas - relational schemas (aRPC server).

var ServerSchema = am.Schema{
    ssF.ClientConnected: {Require: S{ssS.RpcReady}},
    ssF.ErrDelivery:     {Require: S{ssS.Exception}},
    ssF.ErrHandlerTimeout: {
        Add:     S{ssS.Exception},
        Multi:   true,
        Require: S{ssS.Exception},
    },
    ssF.ErrNetwork: {
        Remove:  S{ssS.ClientConnected},
        Require: S{ssS.Exception},
    },
    ssF.ErrNetworkTimeout: {Require: S{ssS.Exception}},
    ssF.ErrOnClient:       {Require: S{ssS.Exception}},
    ssF.ErrProviding:      {Require: S{ssS.Exception}},
    ssF.ErrRpc:            {Require: S{ssS.Exception}},
    ssF.ErrSendPayload:    {Require: S{ssS.Exception}},
    ssF.Exception:         {Multi: true},
    ssF.HandshakeDone: {
        Remove:  S{ssS.Handshaking, ssS.HandshakeDone, ssS.Exception},
        Require: S{ssS.Start, ssS.ClientConnected},
    },
    ssF.Handshaking: {
        Remove:  S{ssS.Handshaking, ssS.HandshakeDone},
        Require: S{ssS.Start},
    },
    ssF.Healthcheck: {Multi: true},
    ssF.Heartbeat:   {},
    ssF.MetricSync:  {Multi: true},
    ssF.Ready: {
        Auto:    true,
        Require: S{ssS.HandshakeDone, ssS.RpcReady},
    },
    ssF.RpcAccepting: {
        Remove:  S{ssS.RpcStarting, ssS.RpcAccepting, ssS.RpcReady},
        Require: S{ssS.Start},
    },
    ssF.RpcReady: {
        Remove:  S{ssS.RpcStarting, ssS.RpcAccepting, ssS.RpcReady},
        Require: S{ssS.Start},
    },
    ssF.RpcStarting: {
        Remove:  S{ssS.RpcStarting, ssS.RpcAccepting, ssS.RpcReady},
        Require: S{ssS.Start},
    },
    ssF.SendPayload:     {Multi: true},
    ssF.Start:           {Add: S{ssS.RpcStarting}},
    ssF.WebSocketTunnel: {},
}

All examples and benchmarks can be found in /examples.

• 🦾 /pkg/machine is the main package
• /docs/manual.md is the go-to
• /docs/diagrams.md try to explain things visually
• /examples show use cases and integrations
  • with /examples/mach_template being ready for copy-paste
  • also Basic, CLI Daemon, aRPC, WASM, WASM Workflow, TUI
• /tools/cmd/am-gen will bootstrap
• /tools/cmd/am-dbg will record every detail
• code in /pkg/node shows a high-level usage
• and reading tests is always a good idea

This monorepo offers the following importable packages, especially:

• 🦾 /pkg/machine State machine, dependency free, semver compatible.
• /pkg/states Reusable state schemas, handlers, and piping.
• /pkg/helpers Useful functions when working with async state machines.
• /pkg/telemetry Telemetry exporters for dbg, metrics, traces, and logs.

Other packages:

• /pkg/rpc Remote state machines, with the same API as local ones.
• /pkg/history History tracking and traversal, including Key-Value and SQL.
• /pkg/integrations Integrations with NATS and JSON.
• /pkg/graph Directional multigraph of connected state machines.
• /pkg/node Distributed worker pools with supervisors.
• /pkg/pubsub Decentralized PubSub based on libp2p gossipsub.

• /tools/cmd/am-dbg Multi-client TUI debugger.
• /tools/cmd/am-gen Generates schema files and Grafana dashboards.
• /tools/cmd/arpc Network-native REPL and CLI.
• /tools/cmd/am-vis Generates D2 diagrams.
• /tools/cmd/am-relay Rotates logs and relays WASM.

asyncmachine-go synchronizes state for the following projects:

• secai - AI Workflows framework
• secai Web UI - WebAssembly go-app PWA
• Self-hosting of pkg/rpc, pkg/node, pkg/pubsub
• arpc REPL - Cobra-based REPL
• am-dbg TUI Debugger - Single state-machine TUI app
• libp2p PubSub Simulator - Sandbox simulator for libp2p-pubsub
• libp2p PubSub Benchmark - Benchmark of libp2p-pubsub ported to asyncmachine-go

• API: pkg.go.dev / code.asyncmachine.dev
• diagrams / cookbook
• manual MD / manual PDF
  • Machine and States
  • Changing State
  • Advanced Topics
  • Cheatsheet

• scale up, not down
• defaults work by default
• everything can be traced and debugged
• automation is evolution
• state != data

• GitHub discussions
• Matrix chat
• Author's RSS

Under development, status depends on each package. The bottom layers seem prod grade, the top ones are alpha or testing.

• good first issues
• before
  • ./scripts/dep-taskfile.sh
  • task install-deps
• after
  • task test
  • task format
  • task lint
  • task precommit

• more tooling and diagrams
• bug fixes, optimizations
• network security, ACLs
• ROADMAP.md

It calls struct methods according to conventions, a schema, and currently active states (eg BarEnter, FooFoo, FooBar, BarState). It tackles nondeterminism by embracing it - like an UDP event stream with structure.

State is a binary ID as in status / switch / flag, eg "process RUNNING" or "car BROKEN".

Each state has a counter of activations & deactivations, and all state counters create "machine time". These are logical clocks, and the queue is also (partially) counted.

The same event happening many times will cause only 1 state activation, until the state becomes inactive.

The complete FAQ is available at FAQ.md.

• Changelog
• Breaking changes
• Repo traffic
• Release feed