fix: post v0.13 fixes

pancsta · pancsta · commit d1cb5ddea280 · 2025-07-09T14:13:35.000+02:00
- loop race
- CI
- linter
diff --git a/.golangci.yml b/.golangci.yml
@@ -1,7 +1,8 @@
-run:
-  exclude-files:
-    - "examples/.*"
-    - "scripts/.*"
+issues:
+  exclude-dirs:
+    - "vendor"
+    - "third_party"
+    - "pkg/pubsub/uds"
 
 linters:
   enable:
diff --git a/pkg/helpers/help.go b/pkg/helpers/help.go
@@ -20,9 +20,10 @@ import (
 
 	"github.com/failsafe-go/failsafe-go"
 	"github.com/failsafe-go/failsafe-go/retrypolicy"
+	"golang.org/x/sync/errgroup"
+
 	"github.com/pancsta/asyncmachine-go/internal/utils"
 	"github.com/pancsta/asyncmachine-go/pkg/states/pipes"
-	"golang.org/x/sync/errgroup"
 
 	am "github.com/pancsta/asyncmachine-go/pkg/machine"
 	ss "github.com/pancsta/asyncmachine-go/pkg/states"
diff --git a/pkg/machine/machine.go b/pkg/machine/machine.go
@@ -97,6 +97,7 @@ type Machine struct {
 	stateNames         S
 	stateNamesExport   S
 	handlersLock       sync.Mutex
+	loopLock           sync.Mutex
 	handlers           []*handler
 	clock              Clock
 	cancel             context.CancelFunc
@@ -396,6 +397,9 @@ func (m *Machine) doDispose(force bool) {
 	m.tracers = nil
 	go func() {
 		time.Sleep(100 * time.Millisecond)
+		m.loopLock.Lock()
+		defer m.loopLock.Unlock()
+
 		closeSafe(m.handlerEnd)
 		closeSafe(m.handlerPanic)
 		closeSafe(m.handlerStart)
@@ -1620,21 +1624,25 @@ func (m *Machine) MustParseStates(states S) S {
 	}
 
 	// check if all states are defined in m.Struct
-	// TODO optimize?
-	ret := make(S, len(states))
 	seen := make(map[string]struct{})
+	dups := false
 	for i := range states {
 		if _, ok := m.schema[states[i]]; !ok {
 			panic(fmt.Errorf(
 				"%w: %s not defined in schema for %s", ErrStateMissing,
 				states[i], m.id))
 		}
 		if _, ok := seen[states[i]]; !ok {
-			ret = append(ret, states[i])
 			seen[states[i]] = struct{}{}
+		} else {
+			// mark as duplicated
+			dups = true
 		}
 	}
 
+	if dups {
+		return slicesUniq(states)
+	}
 	return states
 }
 
@@ -2546,14 +2554,17 @@ func (m *Machine) handlerLoop() {
 				return
 			}
 
+			m.loopLock.Lock()
+
 			// pass the result to handlerLoop
 			select {
 			case <-m.ctx.Done():
 				m.handlerLoopDone()
+				m.loopLock.Unlock()
 				return
 
 			case m.handlerEnd <- ret:
-				// pass
+				m.loopLock.Unlock()
 			}
 		}
 	}
@@ -2675,8 +2686,8 @@ func (m *Machine) IsQueued(mutationType MutationType, states S,
 }
 
 // IsQueuedAbove allows for rate-limiting of mutations for a specific state.
-func (m *Machine) IsQueuedAbove(threshold int, mutationType MutationType, states S,
-	withoutArgsOnly bool, statesStrictEqual bool, startIndex int,
+func (m *Machine) IsQueuedAbove(threshold int, mutationType MutationType,
+	states S, withoutArgsOnly bool, statesStrictEqual bool, startIndex int,
 ) bool {
 	if m.disposed.Load() {
 		return false
diff --git a/pkg/machine/misc_mach.go b/pkg/machine/misc_mach.go
@@ -483,9 +483,6 @@ type Event struct {
 	// Machine is the machine that the event belongs to. It can be used to access
 	// the current Transition and Mutation.
 	machine *Machine
-	// internal events always have a nil step
-	step     *Step
-	internal bool
 }
 
 // Mutation returns the Mutation of an Event.
diff --git a/pkg/machine/transition.go b/pkg/machine/transition.go
@@ -192,7 +192,6 @@ func (t *Transition) CleanCache() {
 
 // StatesBefore is a list of states before the transition.
 func (t *Transition) StatesBefore() S {
-
 	// cache
 	if v := t.cacheStatesBefore.Load(); v != nil {
 		return *v
@@ -216,7 +215,6 @@ func (t *Transition) StatesBefore() S {
 
 // TargetStates is a list of states after parsing the relations.
 func (t *Transition) TargetStates() S {
-
 	// cache
 	if v := t.cacheTargetStates.Load(); v != nil {
 		return *v
@@ -455,7 +453,6 @@ func (t *Transition) emitExitEvents() Result {
 func (t *Transition) emitHandler(
 	from, to string, isFinal, isEnter bool, event string, args A,
 ) Result {
-
 	t.latestStepToState = to
 	ret, handlerCalled := t.Machine.handle(event, args, isFinal, isEnter, false)
 
diff --git a/pkg/machine/types.go b/pkg/machine/types.go
@@ -355,19 +355,19 @@ type Mutation struct {
 	cacheCalled atomic.Pointer[S]
 }
 
-func (m Mutation) String() string {
+func (m *Mutation) String() string {
 	return fmt.Sprintf("[%s] %v", m.Type, m.Called)
 }
 
-func (m Mutation) IsCalled(idx int) bool {
+func (m *Mutation) IsCalled(idx int) bool {
 	return slices.Contains(m.Called, idx)
 }
 
-func (m Mutation) CalledIndex(index S) *TimeIndex {
+func (m *Mutation) CalledIndex(index S) *TimeIndex {
 	return NewTimeIndex(index, m.Called)
 }
 
-func (m Mutation) StringFromIndex(index S) string {
+func (m *Mutation) StringFromIndex(index S) string {
 	called := NewTimeIndex(index, m.Called)
 	return "[" + m.Type.String() + "] " + j(called.ActiveStates())
 }
diff --git a/pkg/pubsub/pubsub_test.go b/pkg/pubsub/pubsub_test.go
@@ -14,10 +14,11 @@ import (
 
 	"github.com/joho/godotenv"
 	ma "github.com/multiformats/go-multiaddr"
-	amtele "github.com/pancsta/asyncmachine-go/pkg/telemetry"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/sync/errgroup"
 
+	amtele "github.com/pancsta/asyncmachine-go/pkg/telemetry"
+
 	amhelp "github.com/pancsta/asyncmachine-go/pkg/helpers"
 	amhelpt "github.com/pancsta/asyncmachine-go/pkg/helpers/testing"
 	am "github.com/pancsta/asyncmachine-go/pkg/machine"
@@ -66,7 +67,8 @@ type mirrorHandlers struct {
 	ProcessMsgsState       am.HandlerFinal
 }
 
-var mirrorStates = am.S{am.Exception,
+var mirrorStates = am.S{
+	am.Exception,
 	ss.Joined, ss.Ready, ss.MissPeersByUpdates, ss.MissPeersByGossip,
 	ss.MissUpdatesByGossip, ss.SendGossips, ss.SendUpdates, ss.SendInfo,
 	ss.MsgReqUpdates, ss.MsgReqInfo, ss.ProcessMsgs, ss.ReqMissingUpdates,
@@ -534,7 +536,7 @@ func TestExposingMany(t *testing.T) {
 		wg.Add(1)
 		go func() {
 			ps.Dispose()
-			amhelp.WaitForAll(ctx, 100*time.Millisecond,
+			_ = amhelp.WaitForAll(ctx, 100*time.Millisecond,
 				ps.Mach.When1(ss.Disposed, nil))
 			ps.Mach.Dispose()
 			wg.Done()
diff --git a/pkg/pubsub/topic.go b/pkg/pubsub/topic.go
@@ -214,7 +214,6 @@ func (t *Topic) ExceptionEnter(e *am.Event) bool {
 }
 
 func (t *Topic) ExceptionState(e *am.Event) {
-
 	// super
 	t.ExceptionHandler.ExceptionState(e)
 	args := am.ParseArgs(e.Args)
@@ -1023,7 +1022,6 @@ func (t *Topic) PeerJoinedEnter(e *am.Event) bool {
 }
 
 func (t *Topic) PeerJoinedState(e *am.Event) {
-
 	t.Mach.EvRemove1(e, ss.PeerJoined, nil)
 	args := ParseArgs(e.Args)
 	peerId := args.PeerId
@@ -1928,7 +1926,6 @@ func (t *Topic) metric(msg, host string) bool {
 	metric := t.MachMetrics.Load()
 	if metric == nil {
 		return false
-
 	}
 
 	key := msg + host
@@ -1994,5 +1991,4 @@ func (t *Topic) syncMetrics() {
 			add("Peer")
 		}
 	}, nil)
-
 }
diff --git a/tools/debugger/debugger.go b/tools/debugger/debugger.go
@@ -1746,7 +1746,6 @@ func (d *Debugger) initGraphGen(
 	snapshot *graph.Graph, id string, details, clients int, outDir,
 	svgName string,
 ) []*amvis.Visualizer {
-
 	var vizs []*amvis.Visualizer
 
 	// render single (current one)
@@ -1850,8 +1849,8 @@ func (d *Debugger) syncOptsTimelines() {
 
 func (d *Debugger) diagramsRender(
 	e *am.Event, shot *amgraph.Graph, id string, details, clients int,
-	outDir string, svgName string) {
-
+	outDir string, svgName string,
+) {
 	ctx := d.Mach.NewStateCtx(ss.DiagramsRendering)
 	if ctx.Err() != nil {
 		return // expired
@@ -1907,7 +1906,6 @@ func (d *Debugger) diagramsMemCache(
 	e *am.Event, id string, cache *goquery.Document, tx *telemetry.DbgMsgTx,
 	outDir, svgName string,
 ) {
-
 	svgPath := path.Join(outDir, svgName+".svg")
 	ctx := d.Mach.NewStateCtx(ss.DiagramsRendering)
 	if ctx.Err() != nil {
@@ -1946,7 +1944,6 @@ func (d *Debugger) diagramsFileCache(
 	e *am.Event, id string, tx *telemetry.DbgMsgTx, lvl int, outDir,
 	svgName string,
 ) {
-
 	svgPath := path.Join(outDir, svgName+".svg")
 	ctx := d.Mach.NewStateCtx(ss.DiagramsRendering)
 	if ctx.Err() != nil {
diff --git a/tools/debugger/handlers.go b/tools/debugger/handlers.go
@@ -1583,7 +1583,6 @@ func (d *Debugger) UpdateFocusState(e *am.Event) {
 
 	// unblock bc of locks
 	go func() {
-
 		// change focus (or not) when changing view types
 		switch d.Mach.Switch(ss.GroupFocused) {
 		case ss.ClientListFocused:
diff --git a/tools/generator/grafana.go b/tools/generator/grafana.go
@@ -219,14 +219,14 @@ func MachDashboardEnv(mach *am.Machine) error {
 		return nil
 	}
 
-	p := cli.GrafanaParams{}
-	// TODO named vars for env vars
-	p.GrafanaUrl = os.Getenv("AM_GRAFANA_URL")
-	p.Token = os.Getenv("AM_GRAFANA_TOKEN")
-	p.Folder = "asyncmachine"
-	p.Ids = mach.Id()
-	p.Name = mach.Id()
-	p.Source = os.Getenv("AM_SERVICE")
+	p := cli.GrafanaParams{
+		Ids:        mach.Id(),
+		Name:       mach.Id(),
+		Folder:     "asyncmachine",
+		GrafanaUrl: os.Getenv("AM_GRAFANA_URL"),
+		Token:      os.Getenv("AM_GRAFANA_TOKEN"),
+		Source:     os.Getenv("AM_SERVICE"),
+	}
 
 	if p.GrafanaUrl == "" || p.Token == "" || p.Source == "" {
 		return nil
diff --git a/tools/visualizer/visualizer.go b/tools/visualizer/visualizer.go
@@ -702,8 +702,8 @@ type Selection struct {
 
 func UpdateCache(
 	ctx context.Context, filepath string, dom *goquery.Document,
-	selections ...*Selection) error {
-
+	selections ...*Selection,
+) error {
 	for _, sel := range selections {
 		for _, state := range sel.States {
 			if ctx.Err() != nil {
@@ -767,5 +767,5 @@ func UpdateCache(
 		return err
 	}
 
-	return os.WriteFile(filepath, []byte(html), 0644)
+	return os.WriteFile(filepath, []byte(html), 0o644)
 }
diff --git a/tools/visualizer/visualizer_test.go b/tools/visualizer/visualizer_test.go
@@ -3,7 +3,9 @@ package visualizer
 // func TestUpdates(t *testing.T) {
 // 	sel := &Selection{
 // 		MachId: "cook",
-// 		States: am.S{"Exception", "StoryCookingStarted", "StoryJoke", "Ready", "Requesting", "StoryStartAgain", "Start", "BaseDBStarting", "CheckStories"},
+// 		States: am.S{"Exception", "StoryCookingStarted", "StoryJoke", "Ready",
+// 			"Requesting", "StoryStartAgain", "Start", "BaseDBStarting",
+// 			"CheckStories"},
 // 		// Active: am.S{"StoryJoke"},
 // 		Active: am.S{"Ready", "StoryStartAgain", "Start", "CheckStories"},
 // 	}

Original file line number	Diff line number	Diff line change
`@@ -483,9 +483,6 @@ type Event struct {`
`483`	`483`	`// Machine is the machine that the event belongs to. It can be used to access`
`484`	`484`	`// the current Transition and Mutation.`
`485`	`485`	`machine *Machine`
`486`		`- // internal events always have a nil step`
`487`		`- step *Step`
`488`		`- internal bool`
`489`	`486`	`}`
`490`	`487`
`491`	`488`	`// Mutation returns the Mutation of an Event.`
Original file line number	Diff line number	Diff line change
`@@ -355,19 +355,19 @@ type Mutation struct {`
`355`	`355`	`cacheCalled atomic.Pointer[S]`
`356`	`356`	`}`
`357`	`357`
`358`		`-func (m Mutation) String() string {`
	`358`	`+func (m *Mutation) String() string {`
`359`	`359`	`return fmt.Sprintf("[%s] %v", m.Type, m.Called)`
`360`	`360`	`}`
`361`	`361`
`362`		`-func (m Mutation) IsCalled(idx int) bool {`
	`362`	`+func (m *Mutation) IsCalled(idx int) bool {`
`363`	`363`	`return slices.Contains(m.Called, idx)`
`364`	`364`	`}`
`365`	`365`
`366`		`-func (m Mutation) CalledIndex(index S) *TimeIndex {`
	`366`	`+func (m Mutation) CalledIndex(index S) TimeIndex {`
`367`	`367`	`return NewTimeIndex(index, m.Called)`
`368`	`368`	`}`
`369`	`369`
`370`		`-func (m Mutation) StringFromIndex(index S) string {`
	`370`	`+func (m *Mutation) StringFromIndex(index S) string {`
`371`	`371`	`called := NewTimeIndex(index, m.Called)`
`372`	`372`	`return "[" + m.Type.String() + "] " + j(called.ActiveStates())`
`373`	`373`	`}`
Original file line number	Diff line number	Diff line change
`@@ -214,7 +214,6 @@ func (t Topic) ExceptionEnter(e am.Event) bool {`
`214`	`214`	`}`
`215`	`215`
`216`	`216`	`func (t Topic) ExceptionState(e am.Event) {`
`217`		`-`
`218`	`217`	`// super`
`219`	`218`	`t.ExceptionHandler.ExceptionState(e)`
`220`	`219`	`args := am.ParseArgs(e.Args)`
`@@ -1023,7 +1022,6 @@ func (t Topic) PeerJoinedEnter(e am.Event) bool {`
`1023`	`1022`	`}`
`1024`	`1023`
`1025`	`1024`	`func (t Topic) PeerJoinedState(e am.Event) {`
`1026`		`-`
`1027`	`1025`	`t.Mach.EvRemove1(e, ss.PeerJoined, nil)`
`1028`	`1026`	`args := ParseArgs(e.Args)`
`1029`	`1027`	`peerId := args.PeerId`
`@@ -1928,7 +1926,6 @@ func (t *Topic) metric(msg, host string) bool {`
`1928`	`1926`	`metric := t.MachMetrics.Load()`
`1929`	`1927`	`if metric == nil {`
`1930`	`1928`	`return false`
`1931`		`-`
`1932`	`1929`	`}`
`1933`	`1930`
`1934`	`1931`	`key := msg + host`
`@@ -1994,5 +1991,4 @@ func (t *Topic) syncMetrics() {`
`1994`	`1991`	`add("Peer")`
`1995`	`1992`	`}`
`1996`	`1993`	`}, nil)`
`1997`		`-`
`1998`	`1994`	`}`