Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

protofsm: update GR Manager usage and start using structured logging #9342

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+142 −125
Open

protofsm: update GR Manager usage and start using structured logging #9342

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4a5a362
go.mod: update btclog dep
ellemouton Dec 10, 2024
bc3cd32
protofsm: use updated GoroutineManager API
ellemouton Dec 9, 2024
64e2e94
protofsm: use prefixed logger for StateMachine
ellemouton Dec 12, 2024
aa8bb29
protofsm: use structured logging
ellemouton Dec 12, 2024
d0144f2
protofsm: use pointer to GoroutineManager
ellemouton Dec 11, 2024
2625e95
docs: add release notes entry
ellemouton Dec 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion protofsm/log.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
package protofsm

import (
"github.com/btcsuite/btclog"
"github.com/btcsuite/btclog/v2"
"github.com/lightningnetwork/lnd/build"
)

Expand Down
75 changes: 33 additions & 42 deletions protofsm/state_machine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btclog/v2"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/lnutils"
Expand Down Expand Up @@ -130,6 +131,8 @@ type stateQuery[Event any, Env Environment] struct {
type StateMachine[Event any, Env Environment] struct {
cfg StateMachineCfg[Event, Env]

log btclog.Logger

// events is the channel that will be used to send new events to the
// FSM.
events chan Event
Expand Down Expand Up @@ -197,7 +200,10 @@ func NewStateMachine[Event any, Env Environment](
cfg StateMachineCfg[Event, Env]) StateMachine[Event, Env] {

return StateMachine[Event, Env]{
cfg: cfg,
cfg: cfg,
log: log.WithPrefix(
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice!

fmt.Sprintf("FSM(%v)", cfg.Env.Name()),
),
events: make(chan Event, 1),
stateQuery: make(chan stateQuery[Event, Env]),
wg: *fn.NewGoroutineManager(),
Expand Down Expand Up @@ -229,9 +235,7 @@ func (s *StateMachine[Event, Env]) Stop() {
//
// TODO(roasbeef): bool if processed?
func (s *StateMachine[Event, Env]) SendEvent(ctx context.Context, event Event) {
log.Debugf("FSM(%v): sending event: %v", s.cfg.Env.Name(),
lnutils.SpewLogClosure(event),
)
s.log.Debugf("Sending event: %v", lnutils.SpewLogClosure(event))

select {
case s.events <- event:
Expand Down Expand Up @@ -269,9 +273,7 @@ func (s *StateMachine[Event, Env]) SendMessage(ctx context.Context,
return false
}

log.Debugf("FSM(%v): sending msg: %v", s.cfg.Env.Name(),
lnutils.SpewLogClosure(msg),
)
s.log.Debugf("Sending msg: %v", lnutils.SpewLogClosure(msg))

// 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
Expand Down Expand Up @@ -342,11 +344,10 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
// any preconditions as well as post-send events.
case *SendMsgEvent[Event]:
sendAndCleanUp := func() error {
log.Debugf("FSM(%v): sending message to target(%x): "+
"%v", s.cfg.Env.Name(),
s.log.Debugf("Sending message to target(%x): "+
"%v",
daemonEvent.TargetPeer.SerializeCompressed(),
lnutils.SpewLogClosure(daemonEvent.Msgs),
)
lnutils.SpewLogClosure(daemonEvent.Msgs))

err := s.cfg.Daemon.SendMessages(
daemonEvent.TargetPeer, daemonEvent.Msgs,
Expand All @@ -361,9 +362,8 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
return fn.MapOptionZ(daemonEvent.PostSendEvent, func(event Event) error { //nolint:ll
launched := s.wg.Go(
ctx, func(ctx context.Context) {
log.Debugf("FSM(%v): sending "+
s.log.Debugf("Sending "+
"post-send event: %v",
s.cfg.Env.Name(),
lnutils.SpewLogClosure(event))

s.SendEvent(ctx, event)
Expand Down Expand Up @@ -393,8 +393,7 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
)
defer predicateTicker.Stop()

log.Infof("FSM(%v): waiting for send predicate to "+
"be true", s.cfg.Env.Name())
s.log.Infof("Waiting for send predicate to be true")

for {
select {
Expand All @@ -407,14 +406,13 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
)

if canSend {
log.Infof("FSM(%v): send "+
"active predicate",
s.cfg.Env.Name())
s.log.Infof("Send active " +
"predicate")

err := sendAndCleanUp()
if err != nil {
//nolint:ll
log.Errorf("FSM(%v): unable to send message: %v", err)
s.log.Errorf("Unable to send message: %v", err)
}

return
Expand All @@ -435,8 +433,8 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
// If this is a broadcast transaction event, then we'll broadcast with
// the label attached.
case *BroadcastTxn:
log.Debugf("FSM(%v): broadcasting txn, txid=%v",
s.cfg.Env.Name(), daemonEvent.Tx.TxHash())
s.log.Debugf("Broadcasting txn, txid=%v",
daemonEvent.Tx.TxHash())

err := s.cfg.Daemon.BroadcastTransaction(
daemonEvent.Tx, daemonEvent.Label,
Expand All @@ -450,8 +448,7 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
// The state machine has requested a new event to be sent once a
// transaction spending a specified outpoint has confirmed.
case *RegisterSpend[Event]:
log.Debugf("FSM(%v): registering spend: %v", s.cfg.Env.Name(),
daemonEvent.OutPoint)
s.log.Debugf("Registering spend: %v", daemonEvent.OutPoint)

spendEvent, err := s.cfg.Daemon.RegisterSpendNtfn(
&daemonEvent.OutPoint, daemonEvent.PkScript,
Expand Down Expand Up @@ -495,8 +492,7 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
// The state machine has requested a new event to be sent once a
// specified txid+pkScript pair has confirmed.
case *RegisterConf[Event]:
log.Debugf("FSM(%v): registering conf: %v", s.cfg.Env.Name(),
daemonEvent.Txid)
s.log.Debugf("Registering conf: %v", daemonEvent.Txid)

numConfs := daemonEvent.NumConfs.UnwrapOr(1)
confEvent, err := s.cfg.Daemon.RegisterConfirmationsNtfn(
Expand Down Expand Up @@ -547,9 +543,9 @@ func (s *StateMachine[Event, Env]) applyEvents(ctx context.Context,
currentState State[Event, Env], newEvent Event) (State[Event, Env],
error) {

log.Debugf("FSM(%v): applying new event", s.cfg.Env.Name(),
lnutils.SpewLogClosure(newEvent),
)
s.log.Debugf("Applying new event: %v",
lnutils.SpewLogClosure(newEvent))

eventQueue := fn.NewQueue(newEvent)

// Given the next event to handle, we'll process the event, then add
Expand All @@ -560,10 +556,8 @@ func (s *StateMachine[Event, Env]) applyEvents(ctx context.Context,
//nolint:ll
for nextEvent := eventQueue.Dequeue(); nextEvent.IsSome(); nextEvent = eventQueue.Dequeue() {
err := fn.MapOptionZ(nextEvent, func(event Event) error {
log.Debugf("FSM(%v): processing event: %v",
s.cfg.Env.Name(),
lnutils.SpewLogClosure(event),
)
s.log.Debugf("Processing event: %v",
lnutils.SpewLogClosure(event))

// Apply the state transition function of the current
// state given this new event and our existing env.
Expand Down Expand Up @@ -593,14 +587,12 @@ func (s *StateMachine[Event, Env]) applyEvents(ctx context.Context,
//
//nolint:ll
for _, inEvent := range events.InternalEvent {
log.Debugf("FSM(%v): adding "+
s.log.Debugf("Adding "+
"new internal event "+
"to queue: %v",
s.cfg.Env.Name(),
lnutils.SpewLogClosure(
inEvent,
),
)
))

eventQueue.Enqueue(inEvent)
}
Expand All @@ -611,9 +603,8 @@ func (s *StateMachine[Event, Env]) applyEvents(ctx context.Context,
return err
}

log.Infof("FSM(%v): state transition: from_state=%T, "+
"to_state=%T",
s.cfg.Env.Name(), currentState,
s.log.Infof("State transition: from_state=%T, "+
"to_state=%T", currentState,
transition.NextState)

// With our events processed, we'll now update our
Expand All @@ -640,7 +631,7 @@ func (s *StateMachine[Event, Env]) applyEvents(ctx context.Context,
// incoming events, and then drives the state machine forward until it reaches
// a terminal state.
func (s *StateMachine[Event, Env]) driveMachine(ctx context.Context) {
log.Debugf("FSM(%v): starting state machine", s.cfg.Env.Name())
s.log.Debugf("Starting state machine")

currentState := s.cfg.InitialState

Expand All @@ -650,7 +641,7 @@ func (s *StateMachine[Event, Env]) driveMachine(ctx context.Context) {
return s.executeDaemonEvent(ctx, event)
})
if err != nil {
log.Errorf("unable to execute init event: %w", err)
s.log.Errorf("Unable to execute init event: %v", err)
return
}

Expand All @@ -670,7 +661,7 @@ func (s *StateMachine[Event, Env]) driveMachine(ctx context.Context) {
if err != nil {
s.cfg.ErrorReporter.ReportError(err)

log.Errorf("unable to apply event: %v", err)
s.log.Errorf("Unable to apply event: %v", err)

// An error occurred, so we'll tear down the
// entire state machine as we can't proceed.
Expand Down