asyncmachine-go

README

Samples . Getting Started . Packages . Devtools . Apps . Docs . FAQ . Changes

asyncmachine-go

[!NOTE] State machines communicate through states.

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.

[!NOTE] git clone https://github.com/pancsta/asyncmachine-go.git

Each state represents

• 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.

[!NOTE] Flow is state, and state is flow, in a graph.

Samples

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

[!NOTE] Clock-based navigation in time.

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.

Getting Started

• 🦾 /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

Packages

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.

Devtools

• /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.

[!NOTE] Inspecting cause-and-effect in distributed systems.

Apps

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

Documentation

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

Goals

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

Community

• GitHub discussions
• Matrix chat
• Author's RSS

[!NOTE] Hundreds of clones.

Status

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

[!NOTE] Managing distributed concurrency.

Development

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

Roadmap

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

[!NOTE] Step by step.

FAQ

How does asyncmachine work?

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.

What is a "state" in asyncmachine?

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

What does "clock-based" mean?

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.

What's the difference between states and events?

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.

Changes

• Changelog
• Breaking changes
• Repo traffic
• Release feed

[!NOTE] Don't lose your sync.