protofsm: add logging
This commit is contained in:
2 files changed
+78
-4
No files matched your search
@@ -8,8 +8,10 @@ import (
|
|||||||
"github.com/btcsuite/btcd/btcec/v2"
|
"github.com/btcsuite/btcd/btcec/v2"
|
||||||
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
||||||
"github.com/btcsuite/btcd/wire"
|
"github.com/btcsuite/btcd/wire"
|
||||||
|
"github.com/davecgh/go-spew/spew"
|
||||||
"github.com/lightningnetwork/lnd/chainntnfs"
|
"github.com/lightningnetwork/lnd/chainntnfs"
|
||||||
"github.com/lightningnetwork/lnd/fn"
|
"github.com/lightningnetwork/lnd/fn"
|
||||||
|
"github.com/lightningnetwork/lnd/lnutils"
|
||||||
"github.com/lightningnetwork/lnd/lnwire"
|
"github.com/lightningnetwork/lnd/lnwire"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -204,6 +206,10 @@ func (s *StateMachine[Event, Env]) Stop() {
|
|||||||
//
|
//
|
||||||
// TODO(roasbeef): bool if processed?
|
// TODO(roasbeef): bool if processed?
|
||||||
func (s *StateMachine[Event, Env]) SendEvent(event Event) {
|
func (s *StateMachine[Event, Env]) SendEvent(event Event) {
|
||||||
|
log.Debugf("FSM(%v): sending event: %v", s.cfg.Env.Name(),
|
||||||
|
lnutils.SpewLogClosure(event),
|
||||||
|
)
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case s.events <- event:
|
case s.events <- event:
|
||||||
case <-s.quit:
|
case <-s.quit:
|
||||||
@@ -236,6 +242,10 @@ func (s *StateMachine[Event, Env]) SendMessage(msg lnwire.Message) bool {
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
|
log.Debugf("FSM(%v): sending msg: %v", s.cfg.Env.Name(),
|
||||||
|
lnutils.SpewLogClosure(msg),
|
||||||
|
)
|
||||||
|
|
||||||
// Otherwise, try to map the message using the default message mapper.
|
// Otherwise, try to map the message using the default message mapper.
|
||||||
// If we can't extract an event, then we'll return false to indicate
|
// If we can't extract an event, then we'll return false to indicate
|
||||||
// that the message wasn't processed.
|
// that the message wasn't processed.
|
||||||
@@ -295,12 +305,20 @@ func (s *StateMachine[Event, Env]) RemoveStateSub(sub StateSubscriber[
|
|||||||
// executeDaemonEvent executes a daemon event, which is a special type of event
|
// executeDaemonEvent executes a daemon event, which is a special type of event
|
||||||
// that can be emitted as part of the state transition function of the state
|
// that can be emitted as part of the state transition function of the state
|
||||||
// machine. An error is returned if the type of event is unknown.
|
// machine. An error is returned if the type of event is unknown.
|
||||||
func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
func (s *StateMachine[Event, Env]) executeDaemonEvent( //nolint:funlen
|
||||||
|
event DaemonEvent) error {
|
||||||
|
|
||||||
switch daemonEvent := event.(type) {
|
switch daemonEvent := event.(type) {
|
||||||
// This is a send message event, so we'll send the event, and also mind
|
// This is a send message event, so we'll send the event, and also mind
|
||||||
// any preconditions as well as post-send events.
|
// any preconditions as well as post-send events.
|
||||||
case *SendMsgEvent[Event]:
|
case *SendMsgEvent[Event]:
|
||||||
sendAndCleanUp := func() error {
|
sendAndCleanUp := func() error {
|
||||||
|
log.Debugf("FSM(%v): sending message to target(%v): "+
|
||||||
|
"%v", s.cfg.Env.Name(),
|
||||||
|
daemonEvent.TargetPeer.SerializeCompressed(),
|
||||||
|
lnutils.SpewLogClosure(daemonEvent.Msgs),
|
||||||
|
)
|
||||||
|
|
||||||
err := s.cfg.Daemon.SendMessages(
|
err := s.cfg.Daemon.SendMessages(
|
||||||
daemonEvent.TargetPeer, daemonEvent.Msgs,
|
daemonEvent.TargetPeer, daemonEvent.Msgs,
|
||||||
)
|
)
|
||||||
@@ -317,6 +335,12 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
|||||||
go func() {
|
go func() {
|
||||||
defer s.wg.Done()
|
defer s.wg.Done()
|
||||||
|
|
||||||
|
log.Debugf("FSM(%v): sending "+
|
||||||
|
"post-send event: %v",
|
||||||
|
s.cfg.Env.Name(),
|
||||||
|
lnutils.SpewLogClosure(event),
|
||||||
|
)
|
||||||
|
|
||||||
s.SendEvent(event)
|
s.SendEvent(event)
|
||||||
}()
|
}()
|
||||||
})
|
})
|
||||||
@@ -340,6 +364,9 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
|||||||
predicateTicker := time.NewTicker(pollInterval)
|
predicateTicker := time.NewTicker(pollInterval)
|
||||||
defer predicateTicker.Stop()
|
defer predicateTicker.Stop()
|
||||||
|
|
||||||
|
log.Infof("FSM(%v): waiting for send predicate to "+
|
||||||
|
"be true", s.cfg.Env.Name())
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-predicateTicker.C:
|
case <-predicateTicker.C:
|
||||||
@@ -351,6 +378,10 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
|||||||
)
|
)
|
||||||
|
|
||||||
if canSend {
|
if canSend {
|
||||||
|
log.Infof("FSM(%v): send "+
|
||||||
|
"active predicate",
|
||||||
|
s.cfg.Env.Name())
|
||||||
|
|
||||||
err := sendAndCleanUp()
|
err := sendAndCleanUp()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
//nolint:lll
|
//nolint:lll
|
||||||
@@ -371,6 +402,9 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
|||||||
// If this is a broadcast transaction event, then we'll broadcast with
|
// If this is a broadcast transaction event, then we'll broadcast with
|
||||||
// the label attached.
|
// the label attached.
|
||||||
case *BroadcastTxn:
|
case *BroadcastTxn:
|
||||||
|
log.Debugf("FSM(%v): broadcasting txn, txid=%v",
|
||||||
|
s.cfg.Env.Name(), daemonEvent.Tx.TxHash())
|
||||||
|
|
||||||
err := s.cfg.Daemon.BroadcastTransaction(
|
err := s.cfg.Daemon.BroadcastTransaction(
|
||||||
daemonEvent.Tx, daemonEvent.Label,
|
daemonEvent.Tx, daemonEvent.Label,
|
||||||
)
|
)
|
||||||
@@ -383,6 +417,9 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
|||||||
// The state machine has requested a new event to be sent once a
|
// The state machine has requested a new event to be sent once a
|
||||||
// transaction spending a specified outpoint has confirmed.
|
// transaction spending a specified outpoint has confirmed.
|
||||||
case *RegisterSpend[Event]:
|
case *RegisterSpend[Event]:
|
||||||
|
log.Debugf("FSM(%v): registering spend: %v", s.cfg.Env.Name(),
|
||||||
|
daemonEvent.OutPoint)
|
||||||
|
|
||||||
spendEvent, err := s.cfg.Daemon.RegisterSpendNtfn(
|
spendEvent, err := s.cfg.Daemon.RegisterSpendNtfn(
|
||||||
&daemonEvent.OutPoint, daemonEvent.PkScript,
|
&daemonEvent.OutPoint, daemonEvent.PkScript,
|
||||||
daemonEvent.HeightHint,
|
daemonEvent.HeightHint,
|
||||||
@@ -418,6 +455,9 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
|||||||
// The state machine has requested a new event to be sent once a
|
// The state machine has requested a new event to be sent once a
|
||||||
// specified txid+pkScript pair has confirmed.
|
// specified txid+pkScript pair has confirmed.
|
||||||
case *RegisterConf[Event]:
|
case *RegisterConf[Event]:
|
||||||
|
log.Debugf("FSM(%v): registering conf: %v", s.cfg.Env.Name(),
|
||||||
|
daemonEvent.Txid)
|
||||||
|
|
||||||
numConfs := daemonEvent.NumConfs.UnwrapOr(1)
|
numConfs := daemonEvent.NumConfs.UnwrapOr(1)
|
||||||
confEvent, err := s.cfg.Daemon.RegisterConfirmationsNtfn(
|
confEvent, err := s.cfg.Daemon.RegisterConfirmationsNtfn(
|
||||||
&daemonEvent.Txid, daemonEvent.PkScript,
|
&daemonEvent.Txid, daemonEvent.PkScript,
|
||||||
@@ -462,6 +502,12 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(event DaemonEvent) error {
|
|||||||
func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
|
func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
|
||||||
newEvent Event) (State[Event, Env], error) {
|
newEvent Event) (State[Event, Env], error) {
|
||||||
|
|
||||||
|
log.Debugf("FSM(%v): applying new event", s.cfg.Env.Name(),
|
||||||
|
lnutils.NewLogClosure(func() string {
|
||||||
|
return spew.Sdump(newEvent)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
eventQueue := fn.NewQueue(newEvent)
|
eventQueue := fn.NewQueue(newEvent)
|
||||||
|
|
||||||
// Given the next event to handle, we'll process the event, then add
|
// Given the next event to handle, we'll process the event, then add
|
||||||
@@ -472,6 +518,13 @@ func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
|
|||||||
//nolint:lll
|
//nolint:lll
|
||||||
for nextEvent := eventQueue.Dequeue(); nextEvent.IsSome(); nextEvent = eventQueue.Dequeue() {
|
for nextEvent := eventQueue.Dequeue(); nextEvent.IsSome(); nextEvent = eventQueue.Dequeue() {
|
||||||
err := fn.MapOptionZ(nextEvent, func(event Event) error {
|
err := fn.MapOptionZ(nextEvent, func(event Event) error {
|
||||||
|
log.Debugf("FSM(%v): processing event: %v",
|
||||||
|
s.cfg.Env.Name(),
|
||||||
|
lnutils.NewLogClosure(func() string {
|
||||||
|
return spew.Sdump(event)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
// Apply the state transition function of the current
|
// Apply the state transition function of the current
|
||||||
// state given this new event and our existing env.
|
// state given this new event and our existing env.
|
||||||
transition, err := currentState.ProcessEvent(
|
transition, err := currentState.ProcessEvent(
|
||||||
@@ -489,6 +542,10 @@ func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
|
|||||||
//
|
//
|
||||||
//nolint:lll
|
//nolint:lll
|
||||||
err := fn.MapOptionZ(events.ExternalEvents, func(dEvents DaemonEventSet) error {
|
err := fn.MapOptionZ(events.ExternalEvents, func(dEvents DaemonEventSet) error {
|
||||||
|
log.Debugf("FSM(%v): processing "+
|
||||||
|
"daemon %v daemon events",
|
||||||
|
s.cfg.Env.Name(), len(dEvents))
|
||||||
|
|
||||||
for _, dEvent := range dEvents {
|
for _, dEvent := range dEvents {
|
||||||
err := s.executeDaemonEvent(
|
err := s.executeDaemonEvent(
|
||||||
dEvent,
|
dEvent,
|
||||||
@@ -508,8 +565,19 @@ func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
|
|||||||
// our event queue.
|
// our event queue.
|
||||||
//
|
//
|
||||||
//nolint:lll
|
//nolint:lll
|
||||||
events.InternalEvent.WhenSome(func(inEvent Event) {
|
events.InternalEvent.WhenSome(func(es Event) {
|
||||||
eventQueue.Enqueue(inEvent)
|
log.Debugf("FSM(%v): adding "+
|
||||||
|
"new internal event "+
|
||||||
|
"to queue: %v",
|
||||||
|
s.cfg.Env.Name(),
|
||||||
|
lnutils.NewLogClosure(func() string { //nolint:lll
|
||||||
|
return spew.Sdump( //nolint:lll
|
||||||
|
es,
|
||||||
|
)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
|
eventQueue.Enqueue(es)
|
||||||
})
|
})
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
@@ -544,6 +612,8 @@ func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
|
|||||||
func (s *StateMachine[Event, Env]) driveMachine() {
|
func (s *StateMachine[Event, Env]) driveMachine() {
|
||||||
defer s.wg.Done()
|
defer s.wg.Done()
|
||||||
|
|
||||||
|
log.Debugf("FSM(%v): starting state machine", s.cfg.Env.Name())
|
||||||
|
|
||||||
currentState := s.cfg.InitialState
|
currentState := s.cfg.InitialState
|
||||||
|
|
||||||
// Before we start, if we have an init daemon event specified, then
|
// Before we start, if we have an init daemon event specified, then
|
||||||
|
|||||||
@@ -113,7 +113,11 @@ func (d *dummyStateStart) ProcessEvent(event dummyEvents, env *dummyEnv,
|
|||||||
},
|
},
|
||||||
NewEvents: fn.Some(EmittedEvent[dummyEvents]{
|
NewEvents: fn.Some(EmittedEvent[dummyEvents]{
|
||||||
ExternalEvents: fn.Some(DaemonEventSet{
|
ExternalEvents: fn.Some(DaemonEventSet{
|
||||||
sendEvent, &BroadcastTxn{}, sendEvent2,
|
sendEvent, sendEvent2,
|
||||||
|
&BroadcastTxn{
|
||||||
|
Tx: &wire.MsgTx{},
|
||||||
|
Label: "test",
|
||||||
|
},
|
||||||
}),
|
}),
|
||||||
}),
|
}),
|
||||||
}, nil
|
}, nil
|
||||||
|
|||||||
Reference in new issue
Block a user