From 921700750eeccc07c0e152f6c71b92cbe5070129 Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Wed, 18 Mar 2026 19:47:48 -0700 Subject: [PATCH 1/5] fix(statemachine): drop stale state timeout messages --- statemachine/statemachine.go | 57 ++++++++++++++++++++++++++++++------ 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/statemachine/statemachine.go b/statemachine/statemachine.go index a7ea6cd..8f43bc4 100644 --- a/statemachine/statemachine.go +++ b/statemachine/statemachine.go @@ -5,6 +5,7 @@ import ( "reflect" "runtime" "strings" + "sync/atomic" "time" "ergo.services/ergo/gen" @@ -79,6 +80,10 @@ type StateMachine[D any] struct { // Pointer to the most recently configured state timeout. stateTimeout *ActiveStateTimeout + // Monotonic generation for state timeouts so stale timeout messages can be + // recognized and dropped after a state transition. + stateTimeoutGeneration uint64 + // Pointer to the most recently configured message timeout. messageTimeout *ActiveMessageTimeout @@ -98,10 +103,17 @@ type StateTimeout struct { func (StateTimeout) isAction() {} type ActiveStateTimeout struct { - state gen.Atom - timeout StateTimeout - cancel func() - cancelled bool + state gen.Atom + timeout StateTimeout + generation uint64 + cancel func() + cancelled bool +} + +type stateTimeoutMessage struct { + state gen.Atom + generation uint64 + payload any } type GenericTimeout struct { @@ -228,7 +240,7 @@ func (s *StateMachine[D]) SetCurrentState(state gen.Atom) error { // touch it. Otherwise we should cancel the active state timeout if there // is one. if s.hasActiveStateTimeout() && s.stateTimeout.state != state { - s.Log().Debug("StateMachine: canceling state timeout for state %s", state) + s.Log().Warning("StateMachine: canceling state timeout due to state transition", "fromState", oldState, "toState", state, "timeoutState", s.stateTimeout.state, "timeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String()) s.stateTimeout.cancel() s.stateTimeout.cancelled = true } @@ -383,6 +395,17 @@ func (s *StateMachine[D]) ProcessRun() (rr error) { switch message.Type { case gen.MailboxMessageTypeRegular: + if timeoutMsg, ok := message.Message.(stateTimeoutMessage); ok { + if !s.hasActiveStateTimeout() { + s.Log().Warning("StateMachine: dropping stale state timeout without active timeout", "messageState", timeoutMsg.state, "messageGeneration", timeoutMsg.generation, "currentState", s.currentState) + return nil + } + if s.stateTimeout.generation != timeoutMsg.generation || s.stateTimeout.state != timeoutMsg.state { + s.Log().Warning("StateMachine: dropping stale state timeout after state transition", "messageState", timeoutMsg.state, "messageGeneration", timeoutMsg.generation, "currentState", s.currentState, "activeTimeoutState", s.stateTimeout.state, "activeTimeoutGeneration", s.stateTimeout.generation, "activeTimeoutCancelled", s.stateTimeout.cancelled) + return nil + } + message.Message = timeoutMsg.payload + } switch message.Message.(type) { case startMonitoringEvents: // start monitoring @@ -399,6 +422,17 @@ func (s *StateMachine[D]) ProcessRun() (rr error) { messageType := reflect.TypeOf(message.Message).String() handler, ok := s.lookupMessageHandler(messageType) if ok == false { + activeTimeoutState := gen.Atom("") + activeTimeoutMessage := "" + activeTimeoutCancelled := false + if s.stateTimeout != nil { + activeTimeoutState = s.stateTimeout.state + activeTimeoutCancelled = s.stateTimeout.cancelled + if s.stateTimeout.timeout.Message != nil { + activeTimeoutMessage = reflect.TypeOf(s.stateTimeout.timeout.Message).String() + } + } + s.Log().Error("StateMachine: no handler for message", "messageType", messageType, "currentState", s.currentState, "activeTimeoutState", activeTimeoutState, "activeTimeoutMessage", activeTimeoutMessage, "activeTimeoutCancelled", activeTimeoutCancelled) return fmt.Errorf("No handler for message %s in state %s", messageType, s.currentState) } return s.invokeMessageHandler(handler, message) @@ -499,6 +533,7 @@ func (s *StateMachine[D]) ProcessActions(actions []Action, state gen.Atom) { switch action := action.(type) { case StateTimeout: if s.hasActiveStateTimeout() { + s.Log().Warning("StateMachine: replacing active state timeout", "currentState", s.currentState, "oldTimeoutState", s.stateTimeout.state, "oldTimeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String(), "newTimeoutMessage", reflect.TypeOf(action.Message).String()) s.stateTimeout.cancel() s.stateTimeout.cancelled = true } @@ -506,16 +541,20 @@ func (s *StateMachine[D]) ProcessActions(actions []Action, state gen.Atom) { // In Ergo v3.2.0+, proc.Send() returns "not allowed" when called from // a goroutine while the process is in Sleep state. SendAfter properly // handles this by using the node's routing methods directly. - cancelFunc, err := s.SendAfter(s.PID(), action.Message, action.Duration) + generation := atomic.AddUint64(&s.stateTimeoutGeneration, 1) + wrapped := stateTimeoutMessage{state: state, generation: generation, payload: action.Message} + cancelFunc, err := s.SendAfter(s.PID(), wrapped, action.Duration) if err != nil { s.Log().Error("StateMachine: failed to schedule state timeout: %v", err) continue } s.stateTimeout = &ActiveStateTimeout{ - state: state, - timeout: action, - cancel: func() { cancelFunc() }, + state: state, + timeout: action, + generation: generation, + cancel: func() { cancelFunc() }, } + s.Log().Warning("StateMachine: scheduled state timeout", "state", state, "message", reflect.TypeOf(action.Message).String(), "duration", action.Duration) case GenericTimeout: if s.hasActiveGenericTimeout(action.Name) { s.genericTimeouts[action.Name].cancel() From 1572cd04589a0f77800632063b9348d2efc27f07 Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Tue, 14 Apr 2026 10:13:40 -0700 Subject: [PATCH 2/5] fix(statemachine): address review feedback on timeout staleness protection - Replace atomic.AddUint64 with plain increment (single-threaded ProcessRun) - Extend generation-based stale message detection to GenericTimeout and MessageTimeout (previously only StateTimeout was protected) - Rename cancelled -> canceled (Go stdlib convention, per context.Canceled) --- statemachine/statemachine.go | 103 +++++++++++++++++++++++++---------- 1 file changed, 74 insertions(+), 29 deletions(-) diff --git a/statemachine/statemachine.go b/statemachine/statemachine.go index 8f43bc4..e5cdde3 100644 --- a/statemachine/statemachine.go +++ b/statemachine/statemachine.go @@ -5,7 +5,6 @@ import ( "reflect" "runtime" "strings" - "sync/atomic" "time" "ergo.services/ergo/gen" @@ -80,9 +79,10 @@ type StateMachine[D any] struct { // Pointer to the most recently configured state timeout. stateTimeout *ActiveStateTimeout - // Monotonic generation for state timeouts so stale timeout messages can be - // recognized and dropped after a state transition. - stateTimeoutGeneration uint64 + // Monotonic generation counter for all timeout types so stale timeout + // messages can be recognized and dropped after cancellation or state + // transition. Only accessed from the single-threaded ProcessRun loop. + timeoutGeneration uint64 // Pointer to the most recently configured message timeout. messageTimeout *ActiveMessageTimeout @@ -107,7 +107,7 @@ type ActiveStateTimeout struct { timeout StateTimeout generation uint64 cancel func() - cancelled bool + canceled bool } type stateTimeoutMessage struct { @@ -125,9 +125,16 @@ type GenericTimeout struct { func (GenericTimeout) isAction() {} type ActiveGenericTimeout struct { - timeout GenericTimeout - cancel func() - cancelled bool + timeout GenericTimeout + generation uint64 + cancel func() + canceled bool +} + +type genericTimeoutMessage struct { + name gen.Atom + generation uint64 + payload any } type MessageTimeout struct { @@ -138,9 +145,15 @@ type MessageTimeout struct { func (MessageTimeout) isAction() {} type ActiveMessageTimeout struct { - timeout MessageTimeout - cancel func() - cancelled bool + timeout MessageTimeout + generation uint64 + cancel func() + canceled bool +} + +type messageTimeoutMessage struct { + generation uint64 + payload any } // Type alias for MessageHandler callbacks. @@ -242,7 +255,7 @@ func (s *StateMachine[D]) SetCurrentState(state gen.Atom) error { if s.hasActiveStateTimeout() && s.stateTimeout.state != state { s.Log().Warning("StateMachine: canceling state timeout due to state transition", "fromState", oldState, "toState", state, "timeoutState", s.stateTimeout.state, "timeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String()) s.stateTimeout.cancel() - s.stateTimeout.cancelled = true + s.stateTimeout.canceled = true } // Execute state enter callback until no new transition is triggered. if s.stateEnterCallback != nil { @@ -266,16 +279,16 @@ func (s *StateMachine[D]) SetData(data D) { } func (s *StateMachine[D]) hasActiveStateTimeout() bool { - return s.stateTimeout != nil && !s.stateTimeout.cancelled + return s.stateTimeout != nil && !s.stateTimeout.canceled } func (s *StateMachine[D]) hasActiveMessageTimeout() bool { - return s.messageTimeout != nil && !s.messageTimeout.cancelled + return s.messageTimeout != nil && !s.messageTimeout.canceled } func (s *StateMachine[D]) hasActiveGenericTimeout(name gen.Atom) bool { if timeout, exists := s.genericTimeouts[name]; exists { - return !timeout.cancelled + return !timeout.canceled } return false } @@ -390,7 +403,7 @@ func (s *StateMachine[D]) ProcessRun() (rr error) { // Any message should cancel the active message timeout if s.hasActiveMessageTimeout() { s.messageTimeout.cancel() - s.messageTimeout.cancelled = true + s.messageTimeout.canceled = true } switch message.Type { @@ -401,7 +414,30 @@ func (s *StateMachine[D]) ProcessRun() (rr error) { return nil } if s.stateTimeout.generation != timeoutMsg.generation || s.stateTimeout.state != timeoutMsg.state { - s.Log().Warning("StateMachine: dropping stale state timeout after state transition", "messageState", timeoutMsg.state, "messageGeneration", timeoutMsg.generation, "currentState", s.currentState, "activeTimeoutState", s.stateTimeout.state, "activeTimeoutGeneration", s.stateTimeout.generation, "activeTimeoutCancelled", s.stateTimeout.cancelled) + s.Log().Warning("StateMachine: dropping stale state timeout after state transition", "messageState", timeoutMsg.state, "messageGeneration", timeoutMsg.generation, "currentState", s.currentState, "activeTimeoutState", s.stateTimeout.state, "activeTimeoutGeneration", s.stateTimeout.generation, "activeTimeoutCanceled", s.stateTimeout.canceled) + return nil + } + message.Message = timeoutMsg.payload + } + if timeoutMsg, ok := message.Message.(genericTimeoutMessage); ok { + if !s.hasActiveGenericTimeout(timeoutMsg.name) { + s.Log().Warning("StateMachine: dropping stale generic timeout", "name", timeoutMsg.name, "messageGeneration", timeoutMsg.generation, "currentState", s.currentState) + return nil + } + active := s.genericTimeouts[timeoutMsg.name] + if active.generation != timeoutMsg.generation { + s.Log().Warning("StateMachine: dropping stale generic timeout after replacement", "name", timeoutMsg.name, "messageGeneration", timeoutMsg.generation, "activeGeneration", active.generation, "currentState", s.currentState) + return nil + } + message.Message = timeoutMsg.payload + } + if timeoutMsg, ok := message.Message.(messageTimeoutMessage); ok { + if !s.hasActiveMessageTimeout() { + s.Log().Warning("StateMachine: dropping stale message timeout", "messageGeneration", timeoutMsg.generation, "currentState", s.currentState) + return nil + } + if s.messageTimeout.generation != timeoutMsg.generation { + s.Log().Warning("StateMachine: dropping stale message timeout after replacement", "messageGeneration", timeoutMsg.generation, "activeGeneration", s.messageTimeout.generation, "currentState", s.currentState) return nil } message.Message = timeoutMsg.payload @@ -424,15 +460,15 @@ func (s *StateMachine[D]) ProcessRun() (rr error) { if ok == false { activeTimeoutState := gen.Atom("") activeTimeoutMessage := "" - activeTimeoutCancelled := false + activeTimeoutCanceled := false if s.stateTimeout != nil { activeTimeoutState = s.stateTimeout.state - activeTimeoutCancelled = s.stateTimeout.cancelled + activeTimeoutCanceled = s.stateTimeout.canceled if s.stateTimeout.timeout.Message != nil { activeTimeoutMessage = reflect.TypeOf(s.stateTimeout.timeout.Message).String() } } - s.Log().Error("StateMachine: no handler for message", "messageType", messageType, "currentState", s.currentState, "activeTimeoutState", activeTimeoutState, "activeTimeoutMessage", activeTimeoutMessage, "activeTimeoutCancelled", activeTimeoutCancelled) + s.Log().Error("StateMachine: no handler for message", "messageType", messageType, "currentState", s.currentState, "activeTimeoutState", activeTimeoutState, "activeTimeoutMessage", activeTimeoutMessage, "activeTimeoutCanceled", activeTimeoutCanceled) return fmt.Errorf("No handler for message %s in state %s", messageType, s.currentState) } return s.invokeMessageHandler(handler, message) @@ -535,13 +571,14 @@ func (s *StateMachine[D]) ProcessActions(actions []Action, state gen.Atom) { if s.hasActiveStateTimeout() { s.Log().Warning("StateMachine: replacing active state timeout", "currentState", s.currentState, "oldTimeoutState", s.stateTimeout.state, "oldTimeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String(), "newTimeoutMessage", reflect.TypeOf(action.Message).String()) s.stateTimeout.cancel() - s.stateTimeout.cancelled = true + s.stateTimeout.canceled = true } // Use SendAfter instead of manual goroutine + Send. // In Ergo v3.2.0+, proc.Send() returns "not allowed" when called from // a goroutine while the process is in Sleep state. SendAfter properly // handles this by using the node's routing methods directly. - generation := atomic.AddUint64(&s.stateTimeoutGeneration, 1) + s.timeoutGeneration++ + generation := s.timeoutGeneration wrapped := stateTimeoutMessage{state: state, generation: generation, payload: action.Message} cancelFunc, err := s.SendAfter(s.PID(), wrapped, action.Duration) if err != nil { @@ -558,28 +595,36 @@ func (s *StateMachine[D]) ProcessActions(actions []Action, state gen.Atom) { case GenericTimeout: if s.hasActiveGenericTimeout(action.Name) { s.genericTimeouts[action.Name].cancel() - s.genericTimeouts[action.Name].cancelled = true + s.genericTimeouts[action.Name].canceled = true } // Use SendAfter instead of manual goroutine + Send. - cancelFunc, err := s.SendAfter(s.PID(), action.Message, action.Duration) + s.timeoutGeneration++ + generation := s.timeoutGeneration + wrapped := genericTimeoutMessage{name: action.Name, generation: generation, payload: action.Message} + cancelFunc, err := s.SendAfter(s.PID(), wrapped, action.Duration) if err != nil { s.Log().Error("StateMachine: failed to schedule generic timeout %s: %v", action.Name, err) continue } s.genericTimeouts[action.Name] = &ActiveGenericTimeout{ - timeout: action, - cancel: func() { cancelFunc() }, + timeout: action, + generation: generation, + cancel: func() { cancelFunc() }, } case MessageTimeout: // Use SendAfter instead of manual goroutine + Send. - cancelFunc, err := s.SendAfter(s.PID(), action.Message, action.Duration) + s.timeoutGeneration++ + generation := s.timeoutGeneration + wrapped := messageTimeoutMessage{generation: generation, payload: action.Message} + cancelFunc, err := s.SendAfter(s.PID(), wrapped, action.Duration) if err != nil { s.Log().Error("StateMachine: failed to schedule message timeout: %v", err) continue } s.messageTimeout = &ActiveMessageTimeout{ - timeout: action, - cancel: func() { cancelFunc() }, + timeout: action, + generation: generation, + cancel: func() { cancelFunc() }, } default: panic("unsupported action") From 1ef660e4a3bcf9f29e8f07021a3b5d4fbb5f91b8 Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Tue, 14 Apr 2026 10:13:45 -0700 Subject: [PATCH 3/5] fix(statemachine): update test handler signatures with from argument Add missing gen.PID first argument to changeState and changeStateSync to match the current StateMessageHandler/StateCallHandler type signatures. --- statemachine/statemachine_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/statemachine/statemachine_test.go b/statemachine/statemachine_test.go index 9170e19..ee981aa 100644 --- a/statemachine/statemachine_test.go +++ b/statemachine/statemachine_test.go @@ -30,11 +30,11 @@ func (b *BasicStatemachine) Init(args ...any) (StateMachineSpec[Data], error) { return spec, nil } -func changeState(state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, []Action, error) { +func changeState(from gen.PID, state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, []Action, error) { return gen.Atom("StateB"), data, nil, nil } -func changeStateSync(state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, gen.Atom, []Action, error) { +func changeStateSync(from gen.PID, state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, gen.Atom, []Action, error) { return gen.Atom("StateB"), data, gen.Atom("StateB"), nil, nil } From 97867d619df36151fcd05081cb4bdd326254d721 Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Wed, 15 Apr 2026 14:37:36 -0700 Subject: [PATCH 4/5] fix(statemachine): demote state timeout lifecycle logs from Warning to Debug --- statemachine/statemachine.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/statemachine/statemachine.go b/statemachine/statemachine.go index e5cdde3..3d06af3 100644 --- a/statemachine/statemachine.go +++ b/statemachine/statemachine.go @@ -253,7 +253,7 @@ func (s *StateMachine[D]) SetCurrentState(state gen.Atom) error { // touch it. Otherwise we should cancel the active state timeout if there // is one. if s.hasActiveStateTimeout() && s.stateTimeout.state != state { - s.Log().Warning("StateMachine: canceling state timeout due to state transition", "fromState", oldState, "toState", state, "timeoutState", s.stateTimeout.state, "timeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String()) + s.Log().Debug("StateMachine: canceling state timeout due to state transition", "fromState", oldState, "toState", state, "timeoutState", s.stateTimeout.state, "timeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String()) s.stateTimeout.cancel() s.stateTimeout.canceled = true } @@ -569,7 +569,7 @@ func (s *StateMachine[D]) ProcessActions(actions []Action, state gen.Atom) { switch action := action.(type) { case StateTimeout: if s.hasActiveStateTimeout() { - s.Log().Warning("StateMachine: replacing active state timeout", "currentState", s.currentState, "oldTimeoutState", s.stateTimeout.state, "oldTimeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String(), "newTimeoutMessage", reflect.TypeOf(action.Message).String()) + s.Log().Debug("StateMachine: replacing active state timeout", "currentState", s.currentState, "oldTimeoutState", s.stateTimeout.state, "oldTimeoutMessage", reflect.TypeOf(s.stateTimeout.timeout.Message).String(), "newTimeoutMessage", reflect.TypeOf(action.Message).String()) s.stateTimeout.cancel() s.stateTimeout.canceled = true } @@ -591,7 +591,7 @@ func (s *StateMachine[D]) ProcessActions(actions []Action, state gen.Atom) { generation: generation, cancel: func() { cancelFunc() }, } - s.Log().Warning("StateMachine: scheduled state timeout", "state", state, "message", reflect.TypeOf(action.Message).String(), "duration", action.Duration) + s.Log().Debug("StateMachine: scheduled state timeout", "state", state, "message", reflect.TypeOf(action.Message).String(), "duration", action.Duration) case GenericTimeout: if s.hasActiveGenericTimeout(action.Name) { s.genericTimeouts[action.Name].cancel() From c8a88e0fba07edaddf677320b1d7423c7cc584dc Mon Sep 17 00:00:00 2001 From: "clanker-pel[bot]" <295400756+clanker-pel[bot]@users.noreply.github.com> Date: Mon, 22 Jun 2026 22:12:00 +0000 Subject: [PATCH 5/5] chore: demote stale-timeout generation-race logs from warning to debug Demote the two "dropping stale generic timeout" state-machine logs (the plain and the "after replacement" variant) from Warning to Debug. They report a benign generation race during timeout replacement, which is expected and self-correcting, so they should not surface as warnings. --- statemachine/statemachine.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/statemachine/statemachine.go b/statemachine/statemachine.go index 3d06af3..19fafba 100644 --- a/statemachine/statemachine.go +++ b/statemachine/statemachine.go @@ -421,12 +421,12 @@ func (s *StateMachine[D]) ProcessRun() (rr error) { } if timeoutMsg, ok := message.Message.(genericTimeoutMessage); ok { if !s.hasActiveGenericTimeout(timeoutMsg.name) { - s.Log().Warning("StateMachine: dropping stale generic timeout", "name", timeoutMsg.name, "messageGeneration", timeoutMsg.generation, "currentState", s.currentState) + s.Log().Debug("StateMachine: dropping stale generic timeout", "name", timeoutMsg.name, "messageGeneration", timeoutMsg.generation, "currentState", s.currentState) return nil } active := s.genericTimeouts[timeoutMsg.name] if active.generation != timeoutMsg.generation { - s.Log().Warning("StateMachine: dropping stale generic timeout after replacement", "name", timeoutMsg.name, "messageGeneration", timeoutMsg.generation, "activeGeneration", active.generation, "currentState", s.currentState) + s.Log().Debug("StateMachine: dropping stale generic timeout after replacement", "name", timeoutMsg.name, "messageGeneration", timeoutMsg.generation, "activeGeneration", active.generation, "currentState", s.currentState) return nil } message.Message = timeoutMsg.payload