history

package
v0.19.1 Latest Latest
Warning

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

 Go to latest Published: Jun 23, 2026 License: MIT Imports: 9 Imported by: 3

README

/pkg/history

cd /

[!NOTE] asyncmachine-go is a pathless control-flow graph with a consensus (AOP, actor model, state-machine).

/pkg/history provides mutation history tracking and traversal, which plays an essential role in making informed decision about state flow. It contains rich machine time information including subsets for tracked states, various diffs, sums, and also binds mutations to human time. Because of storage constraints, additional info about transitions is optional. Each history backend has it's own query mechanism, but all implement the common Query interface.

Another use case for pkg/history is resurrecting state machines after a restart, based on ticks and (optionally) a journal - it happens manually inside the MachineRestored state, which is added by the Machine.Import() method. During the import, MachineTick increments by +1. MachineRecord can optionally store also the full schema and ordered list of state names, to restore dynamic state-machines.

This layer works with both local and network machines via the Tracer API. All times are in UTC, methods thread-safe, and IDs deterministic.

SQL view

History Backends
TODO
  • conditions for extra transition info
  • proper time and space benchmarks
  • more backends (sqlc)
  • store time distances

Below are the key APIs, the rest can be found in the godoc.

type TimeRecord struct {
    // TransitionId is an optional ID of the related [TransitionRecord].
    TransitionId string
    // MutType is a mutation type.
    MutType am.MutationType
    // MTimeSum is a machine time sum after this transition.
    MTimeSum uint64
    // MTimeSum is a machine time sum after this transition for tracked states
    // only.
    MTimeTrackedSum uint64
    // MTimeDiff is a machine time difference for this transition.
    MTimeDiff uint64
    // MTimeDiff is a machine time difference for this transition for tracked
    // states only.
    MTimeTrackedDiff uint64
    // MTimeRecordDiff is a machine time difference since the previous
    // [TimeRecord].
    MTimeRecordDiff uint64
    // HTime is a human time in UTC.
    HTime time.Time
    // MTime is a machine time after this mutation.
    MTimeTracked am.Time
    // MachTick is the machine tick at the time of this transition.
    MachTick uint32
}


type MemoryApi interface {
    // predefined queries

    ActivatedBetween(ctx context.Context, state string, start, end time.Time) bool
    ActiveBetween(ctx context.Context, state string, start, end time.Time) bool
    DeactivatedBetween(ctx context.Context, state string, start, end time.Time) bool
    InactiveBetween(ctx context.Context, state string, start, end time.Time) bool

    // DB queries

    Find(ctx context.Context, inclTx bool, cond Query) (*MemoryRecord, error)

    // converters

    ToTimeRecord(format any) (*TimeRecord, error)
    ToMachineRecord(format any) (*MachineRecord, error)
    ToTransitionRecord(format any) (*TransitionRecord, error)

    // misc

    Machine() am.Api
    Config() Config
    Context() context.Context
    MachineRecord() *MachineRecord
    Dispose() error
}

In-Process History

This is the default backend and uses a simple Go slice.

Pros:

  • lightweight
  • concurrent reads
  • no encode/decode overhead
  • works in WASM

Cons:

  • no persistence
  • manual pattern queries via slice indexes
  • eats memory

Example:

import amhist "github.com/pancsta/asyncmachine-go/pkg/history"

// ...

// var mach *am.Machine
// var ctx context.Context

// start tracking mutations of states A and C, with a 1k limit
cfg := Config{
    TrackedStates: am.S{"A", "C"},
    MaxRecords: 1_000,
}
mem, err := NewMemory(ctx, nil, mach, cfg, onErr)

// mutate A
mach.Add1("A", nil)

// run a query
now := time.Now().UTC()
mem.Sync()
time.Sleep(100 * time.Millisecond)
mem.ActivatedBetween(ctx, "A", now.Add(-time.Second), now) // true

Benchmark:

=== RUN   TestTrackMany
    test_hist.go:27: rounds: 50000
    test_hist.go:34: mach: 147.189692ms
    test_hist.go:38: db: 147.432228ms
    test_hist.go:65: query: 147.689852ms
--- PASS: TestTrackMany (0.15s)
PASS

SQL History

The SQL backend uses GORM, and ships with a WASM-based SQLite (WAL enabled), although it can be used with any SQL database.

Pros:

  • StarTrek-ready, great tooling
  • easy pattern queries via WHERE and JOIN
  • can offload data over the network
  • concurrent reads
  • multiple DB connections

Cons:

  • slow startup or provisioning required
  • SQLite adds 1-3MBs to the binary size
  • slow writes
SQL Schema Diagram

Example:

import (
    amhist "github.com/pancsta/asyncmachine-go/pkg/history"
    amhistg "github.com/pancsta/asyncmachine-go/pkg/history/gorm"
)

// ...

// var mach *am.Machine
// var ctx context.Context

// injected err handler
onErr := func(err error) {
    log.Print(err.Error())
}

// backend and base configs
cfg := Config{
    BaseConfig: amhist.Config{
        MaxRecords: 10 ^ 6,
        TrackedStates: am.S{"A", "C"},
    },
    EncJson: true,
}

// create amhist.sqlite
db, err := amhistg.NewSqlite("")
defer db.Close()

mem, err := amhistg.NewMemory(ctx, db, mach, cfg, onErr)

// mutate and query
mach.Add1("A", nil)
now := time.Now().UTC()
mem.Sync()
time.Sleep(100 * time.Millisecond)
mem.ActivatedBetween(ctx, "A", now.Add(-time.Second), now) // true

Benchmark:

=== RUN   TestGormTrackMany
    test_hist.go:27: rounds: 50000
    test_hist.go:34: mach: 189.432252ms
    test_hist.go:38: db: 998.199453ms
    test_hist.go:65: query: 998.957906ms
--- PASS: TestGormTrackMany (1.02s)
PASS

Key-Value History - BoltDB

The Key-Value store backend uses etcd-io/bbolt with vmihailenco/msgpack and writes to a single file. For debugging there's also JSON encoding, with 2x the size.

Pros:

  • instant startup
  • small binary size
  • concurrent reads

Cons:

  • abysmal tooling [0][1]
  • manual pattern queries via cursor scanning
  • single DB connection only

Schema:

  • _machines
  • MyMachId1
    • Times
    • Transitions

Example:

import (
    amhist "github.com/pancsta/asyncmachine-go/pkg/history"
    amhistbb "github.com/pancsta/asyncmachine-go/pkg/history/bbolt"
)

// ...

// var mach *am.Machine
// var ctx context.Context

// injected err handler
onErr := func(err error) {
    log.Print(err.Error())
}

// backend and base configs
cfg := Config{
    BaseConfig: amhist.Config{
        MaxRecords: 10 ^ 6,
        TrackedStates: am.S{"A", "C"},
    },
    EncJson: true,
}

// create amhist.db
db, err := amhistbb.NewDb("")
defer db.Close()

mem, err := amhistbb.NewMemory(ctx, db, mach, cfg, onErr)

// mutate and query
mach.Add1("A", nil)
now := time.Now().UTC()
mem.Sync()
time.Sleep(100 * time.Millisecond)
mem.ActivatedBetween(ctx, "A", now.Add(-time.Second), now) // true

Benchmark:

=== RUN   TestBboltTrackMany
    test_hist.go:27: rounds: 50000
    test_hist.go:34: mach: 140.18248ms
    test_hist.go:38: db: 154.976653ms
    test_hist.go:65: query: 155.065849ms
    bbolt_test.go:121: write time: 88.05461ms
--- PASS: TestBboltTrackMany (0.17s)
PASS

Key-Value History - BadgerDB

The Key-Value store backend uses dgraph-io/badger with vmihailenco/msgpack and writes to a single file. For debugging there's also JSON encoding, with 2x the size.

Pros:

  • instant startup
  • small binary size
  • concurrent reads
  • works in WASM

Cons:

  • abysmal tooling [0]
  • manual pattern queries via cursor scanning
  • single DB connection only

Schema:

  • _machines
  • MyMachId1
    • Times
    • Transitions

Example:

import (
    amhist "github.com/pancsta/asyncmachine-go/pkg/history"
    amhistb "github.com/pancsta/asyncmachine-go/pkg/history/badger"
)

// ...

// var mach *am.Machine
// var ctx context.Context

// injected err handler
onErr := func(err error) {
    log.Print(err.Error())
}

// backend and base configs
cfg := Config{
    BaseConfig: amhist.Config{
        MaxRecords: 10 ^ 6,
        TrackedStates: am.S{"A", "C"},
    },
    EncJson: true,
}

// create amhist.db
db, err := amhistb.NewDb("")
defer db.Close()

mem, err := amhistb.NewMemory(ctx, db, mach, cfg, onErr)

// mutate and query
mach.Add1("A", nil)
now := time.Now().UTC()
mem.Sync()
time.Sleep(100 * time.Millisecond)
mem.ActivatedBetween(ctx, "A", now.Add(-time.Second), now) // true

Benchmark:

=== RUN   TestBadgerTrackMany
    test_hist.go:27: rounds: 50000
    test_hist.go:34: mach: 123.825858ms
    test_hist.go:38: db: 124.52048ms
    test_hist.go:65: query: 124.607522ms
--- PASS: TestBadgerTrackMany (0.17s)
PASS

Columnar History

There's an experimental Columnar backend based on FrostDB and Parquet in /pkg/x/history/frostdb.

Documentation

Status

Testing, not semantically versioned.

monorepo

Go back to the monorepo root to continue reading.

Documentation

Overview

Package history provides machine history tracking and traversal using the process' memory and structs.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrIncompatibleType = fmt.Errorf("incompatible type")
	ErrCondition        = fmt.Errorf("incorrect condition")
)
View Source 
var (
	BackendMemory = Backend{"memory"}
	BackendSqlite = Backend{"sqlite"}
	BackendBbolt  = Backend{"bbolt"}
	BackendBadger = Backend{"badger"}

	BackendEnum = enum.New(BackendMemory, BackendSqlite, BackendBbolt,
		BackendBadger)
)
View Source 
var ErrConfig = fmt.Errorf("incorrect config")

TODO bind ErrConfig

Functions

This section is empty.

Types

type Backend added in v0.17.1 

type Backend enum.Member[string]

Backend enumerates all available asyncmachine history backends.

type BaseConfig added in v0.16.0 

type BaseConfig struct {
	Log bool

	// Called is a list of mutation states (called ones) required to track a
	// transition. See also CalledExclude. Optional.
	Called am.S
	// CalledExclude flips Called to be a blocklist.
	CalledExclude bool
	// Changed is a list of transition states which had clock changes, required to
	// track a transition. See also ChangedExclude. A state can be called, but not
	// changed. Optional.
	Changed am.S
	// ChangedExclude flips Changed to be a blocklist.
	ChangedExclude bool
	// TrackRejected is a flag to track rejected transitions.
	TrackRejected bool

	// TrackedStates is a list of states to store clock values of.
	TrackedStates am.S
	// StoreTransitions is a flag to store TransitionRecord, in addition to
	// TimeRecord, for each tracked transition.
	StoreTransitions bool
	// TODO StoreSchema keeps the latest machine schema and state names within
	// [MachineRecord]. Useful for dynamic machines.
	StoreSchema bool

	// MaxRecords is the maximum number of records to keep in the history before
	// a rotation begins.
	MaxRecords int
}

BaseConfig describes the tracking configuration - conditions, list of tracked states and data expiration TTLs.

type BaseMemory added in v0.16.0 

type BaseMemory struct {
	Ctx  context.Context
	Mach am.Api
	// read-only config for this history
	Cfg *BaseConfig
	// contains filtered or unexported fields
}

BaseMemory are the common methods for all memory implementations, operating on common models, like TimeRecord or Query.

func NewBaseMemory added in v0.16.0 

func NewBaseMemory(
	ctx context.Context, mach am.Api, config BaseConfig, memImpl MemoryApi,
) *BaseMemory

func (*BaseMemory) ActivatedBetween added in v0.16.0 

func (m *BaseMemory) ActivatedBetween(
	ctx context.Context, state string, start, end time.Time,
) bool

ActivatedBetween returns true if the state was activated within the passed human time range.

func (*BaseMemory) ActiveBetween added in v0.16.0 

func (m *BaseMemory) ActiveBetween(
	ctx context.Context, state string, start, end time.Time,
) bool

ActiveBetween returns true if the state was activated all the time within the passed human time range.

func (*BaseMemory) Config added in v0.16.0 

func (m *BaseMemory) Config() BaseConfig

func (*BaseMemory) Context added in v0.16.0 

func (m *BaseMemory) Context() context.Context

func (*BaseMemory) DeactivatedBetween added in v0.16.0 

func (m *BaseMemory) DeactivatedBetween(
	ctx context.Context, state string, start, end time.Time,
) bool

func (*BaseMemory) Dispose added in v0.16.0 

func (m *BaseMemory) Dispose() error

func (*BaseMemory) FindLatest added in v0.16.0 

func (m *BaseMemory) FindLatest(
	ctx context.Context, retTx bool, limit int, query Query,
) ([]*MemoryRecord, error)

FindLatest returns the latest records matching the given conditions, in order from the newest to the oldest. If the limit is 0, all records are returned.

func (*BaseMemory) InactiveBetween added in v0.16.0 

func (m *BaseMemory) InactiveBetween(
	ctx context.Context, state string, start, end time.Time,
) bool

func (*BaseMemory) Index added in v0.16.0 

func (m *BaseMemory) Index(states am.S) []int

Index returns the indexes of the given states in the history records, or -1 if the state is not being tracked.

func (*BaseMemory) Index1 added in v0.16.0 

func (m *BaseMemory) Index1(state string) int

Index1 is BaseMemory.Index for a single state.

func (*BaseMemory) IsTracked added in v0.16.0 

func (m *BaseMemory) IsTracked(states am.S) bool

IsTracked returns true if the given states are all being tracked by this memory instance.

func (*BaseMemory) IsTracked1 added in v0.16.0 

func (m *BaseMemory) IsTracked1(state string) bool

IsTracked1 is IsTracked for a single state.

func (*BaseMemory) Machine added in v0.16.0 

func (m *BaseMemory) Machine() am.Api

func (*BaseMemory) MachineRecord added in v0.16.0 

func (m *BaseMemory) MachineRecord() *MachineRecord

MachineRecord returns a copy of the history record for the tracked machine.

func (*BaseMemory) Sync added in v0.16.0 

func (m *BaseMemory) Sync() error

func (*BaseMemory) ToMachineRecord added in v0.16.0 

func (m *BaseMemory) ToMachineRecord(format any) (*MachineRecord, error)

func (*BaseMemory) ToMemoryRecord added in v0.16.0 

func (m *BaseMemory) ToMemoryRecord(format any) (*MemoryRecord, error)

func (*BaseMemory) ToTimeRecord added in v0.16.0 

func (m *BaseMemory) ToTimeRecord(format any) (*TimeRecord, error)

func (*BaseMemory) ToTransitionRecord added in v0.16.0 

func (m *BaseMemory) ToTransitionRecord(format any) (*TransitionRecord, error)

func (*BaseMemory) ValidateQuery added in v0.16.0 

func (m *BaseMemory) ValidateQuery(query Query) error

type ConditionTime added in v0.16.0 

type ConditionTime struct {
	// MTimeStates is a set of states for MTime.
	MTimeStates am.S
	// MTime is the machine time for a mutation to match.
	MTime am.Time
	// HTime is the human time for a mutation to match.
	HTime time.Time
	// MTimeSum is a machine time sum after this transition.
	MTimeSum uint64
	// MTimeSum is a machine time sum after this transition for tracked states
	// only.
	MTimeTrackedSum uint64
	// MTimeDiff is a machine time difference for this transition, compared to the
	// previous transition.
	MTimeDiff uint64
	// MTimeDiff is a machine time difference for this transition for tracked
	// states only.
	MTimeTrackedDiff uint64
	// MTimeRecordDiff is a machine time difference since the previous
	// [TimeRecord].
	MTimeRecordDiff uint64
	// MachTick is the machine tick at the time of this transition.
	MachTick uint32
}

type ConditionTx added in v0.16.0 

type ConditionTx struct {
	Query

	// Called is a set of states that were called in the mutation.
	Called am.S

	SourceTx   string
	SourceMach string
	IsAuto     bool
	IsAccepted bool
	IsCheck    bool
	IsBroken   bool
	QueueLen   uint16

	QueuedAt   uint64
	ExecutedAt uint64
}

ConditionTx represents a condition for a single transition. Requires BaseConfig.StoreTransitions to be true.

type Config added in v0.16.0 

type Config = BaseConfig

Config for the in-process memory.

type MachineRecord added in v0.16.0 

type MachineRecord struct {
	// ID of the tracked machine
	MachId     string    `msgpack:"mi"`
	StateNames am.S      `msgpack:"sn"`
	Schema     am.Schema `msgpack:"s"`

	// first time the machine has been tracked
	FirstTracking time.Time `msgpack:"ft"`
	// last time a tracking of this machine has started
	LastTracking time.Time `msgpack:"lt"`
	// last time a sync has been performed
	LastSync time.Time `msgpack:"ls"`

	// current (total) machine time
	MTime am.Time `msgpack:"mt"`
	// sum of the current machine time
	MTimeSum uint64 `msgpack:"mts"`
	// current machine start tick
	MachTick uint32 `msgpack:"mt2"`
	// next ID for time records
	NextId uint64 `msgpack:"ni"`
}

type MatcherFn added in v0.16.0 

type MatcherFn func(now *am.TimeIndex, db []*MemoryRecord) []*MemoryRecord

type Memory added in v0.16.0 

type Memory struct {
	*BaseMemory
	// contains filtered or unexported fields
}

func NewMemory added in v0.16.0 

func NewMemory(
	ctx context.Context, machRecord *MachineRecord, mach am.Api,
	config BaseConfig, onErr func(err error),
) (*Memory, error)

NewMemory returns a new memory instance that tracks the given machine according to the given tracking configuration. All states are tracked by default, which often is not desired. Keeps 1000 records by default.

Example 
package main

import (
	"context"

	amhist "github.com/pancsta/asyncmachine-go/pkg/history"
	am "github.com/pancsta/asyncmachine-go/pkg/machine"
	amss "github.com/pancsta/asyncmachine-go/pkg/states"
	ssdbg "github.com/pancsta/asyncmachine-go/tools/debugger/states"
)

var ss = ssdbg.DebuggerStates

func main() {
	ctx := context.Background()

	// configs
	onErr := func(err error) {
		panic(err)
	}
	cfg := amhist.Config{
		TrackedStates: am.S{ss.Start},
	}

	// init machine and history
	mach := am.New(ctx, amss.BasicSchema, &am.Opts{Id: "MyMach1"})
	mem, err := amhist.NewMemory(ctx, nil, mach, cfg, onErr)
	if err != nil {
		panic(err)
	}
	// query etc
	_ = mem.Dispose()
}

func (*Memory) Config added in v0.16.0 

func (m *Memory) Config() BaseConfig

func (*Memory) Dispose added in v0.16.0 

func (m *Memory) Dispose() error

func (*Memory) Export added in v0.16.0 

func (m *Memory) Export() []*MemoryRecord

func (*Memory) FindLatest added in v0.16.0 

func (m *Memory) FindLatest(
	ctx context.Context, _ bool, limit int, query Query,
) ([]*MemoryRecord, error)

FindLatest is BaseMemory.FindLatest for in-process memory.

func (*Memory) Machine added in v0.16.0 

func (m *Memory) Machine() am.Api

func (*Memory) MachineRecord added in v0.16.0 

func (m *Memory) MachineRecord() *MachineRecord

func (*Memory) Match added in v0.16.0 

func (m *Memory) Match(
	ctx context.Context, matcherFn MatcherFn,
) ([]*MemoryRecord, error)

Match returns the first record that matches the MatcherFn.

type MemoryApi added in v0.16.0 

type MemoryApi interface {

	// ActivatedBetween returns true if the state become active withing the
	// passed conditions.
	ActivatedBetween(ctx context.Context, state string, start, end time.Time) bool
	// ActiveBetween returns true if the state was active at least once
	// within the passed conditions. Always true is ActivatedBetween is true.
	ActiveBetween(ctx context.Context, state string, start, end time.Time) bool
	DeactivatedBetween(
		ctx context.Context, state string, start, end time.Time,
	) bool
	InactiveBetween(ctx context.Context, state string, start, end time.Time) bool

	FindLatest(
		ctx context.Context, retTx bool, limit int, query Query,
	) ([]*MemoryRecord, error)
	// Sync synchronizes the batch buffer with the underlying backend, making
	// those new records appear in queries. Doesn't guarantee persistence.
	// Useful when queries very fresh records.
	Sync() error

	ToTimeRecord(format any) (*TimeRecord, error)
	ToMachineRecord(format any) (*MachineRecord, error)
	ToTransitionRecord(format any) (*TransitionRecord, error)

	IsTracked(states am.S) bool
	IsTracked1(state string) bool
	Index(states am.S) []int
	Index1(state string) int

	Machine() am.Api
	Config() BaseConfig
	Context() context.Context
	MachineRecord() *MachineRecord
	Dispose() error
}

type MemoryRecord added in v0.16.0 

type MemoryRecord struct {
	Time       *TimeRecord
	Transition *TransitionRecord
}

type Query added in v0.16.0 

type Query struct {

	// Active is a set of states that were active AFTER the mutation
	Active am.S
	// Activated is a set of states that were activated during the transition.
	Activated am.S
	// Inactive is a set of states that were inactive AFTER the mutation
	Inactive am.S
	// Deactivated is a set of states that were deactivated during the
	// transition.
	Deactivated am.S

	// Start is the beginning of a scalar time condition and requires an
	// equivalent in [Query.End].
	Start ConditionTime
	// End is the end of a scalar time condition and requires an equivalent in
	// [Query.Start].
	End ConditionTime
}

Query represents various conditions for a single mutation. All are optional, but at least one must be set.

type TimeRecord added in v0.16.0 

type TimeRecord struct {
	// MutType is a mutation type.
	MutType am.MutationType `msgpack:"mt"`
	// MTimeSum is a machine time sum after this transition.
	MTimeSum uint64 `msgpack:"mts"`
	// MTimeSum is a machine time sum after this transition for tracked states
	// only.
	MTimeTrackedSum uint64 `msgpack:"mtts"`
	// MTimeDiffSum is a machine time difference for this transition.
	MTimeDiffSum uint64 `msgpack:"mtds"`
	// MTimeDiffSum is a machine time difference for this transition for tracked
	// states only.
	MTimeTrackedDiffSum uint64 `msgpack:"mttds"`
	// MTimeRecordDiffSum is a machine time difference since the previous
	// [TimeRecord].
	MTimeRecordDiffSum uint64 `msgpack:"mtrds"`
	// HTime is a human time in UTC.
	HTime time.Time `msgpack:"ht"`
	// MTime is a machine time after this mutation.
	MTimeTracked am.Time `msgpack:"mtt"`
	// MTimeTrackedDiff is a machine time diff compared to the previous mutation
	// (not a record).
	MTimeTrackedDiff am.Time `msgpack:"mttd"`
	// MachTick is the machine tick at the time of this transition.
	MachTick uint32 `msgpack:"mt2"`
}

type TransitionRecord added in v0.16.0 

type TransitionRecord struct {
	TransitionId string `msgpack:"ti"`
	TimeRecordId uint64 `msgpack:"tri"`
	SourceTx     string `msgpack:"st"`
	SourceMach   string `msgpack:"sm"`
	IsAuto       bool   `msgpack:"ia"`
	IsAccepted   bool   `msgpack:"ia2"`
	IsCheck      bool   `msgpack:"ic"`
	IsBroken     bool   `msgpack:"ib"`
	QueueLen     uint16 `msgpack:"ql"`

	QueuedAt   uint64 `msgpack:"qa"`
	ExecutedAt uint64 `msgpack:"ea"`

	// extra
	Called    []int             `msgpack:"c"`
	Arguments map[string]string `msgpack:"a"`
}

Source Files

View all Source files

Directories

Path Synopsis
Package badger provides machine history tracking and traversal using the Badger K/V database.
 Package badger provides machine history tracking and traversal using the Badger K/V database.
Package bbolt provides machine history tracking and traversal using the bbolt K/V database.
 Package bbolt provides machine history tracking and traversal using the bbolt K/V database.

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.