protofsm: extend EmittedEvents with new Outbox field

go.mod

@@ -64,6 +64,8 @@ require (
 	pgregory.net/rapid v1.2.0
 )
 
+require github.com/lightningnetwork/lnd/actor v0.0.1-alpha
+
 require (
 	dario.cat/mergo v1.0.1 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
@@ -213,6 +215,9 @@ replace github.com/gogo/protobuf => github.com/gogo/protobuf v1.3.2
 // allows us to specify that as an option.
 replace google.golang.org/protobuf => github.com/lightninglabs/protobuf-go-hex-display v1.33.0-hex-display
 
+// Use the local actor module for development.
+replace github.com/lightningnetwork/lnd/actor => ./actor
+
 // If you change this please also update docs/INSTALL.md and GO_VERSION in
 // Makefile (then run `make lint` to see where else it needs to be updated as
 // well).

protofsm/state_machine.go

@@ -11,6 +11,7 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/actor"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/lnutils"
@@ -42,6 +43,12 @@ type EmittedEvent[Event any] struct {
 	// ExternalEvent is an optional external event that is to be sent to
 	// the daemon for dispatch. Usually, this is some form of I/O.
 	ExternalEvents DaemonEventSet
+
+	// Outbox is an optional set of events that are accumulated during event
+	// processing and returned to the caller for processing into the main
+	// state machine. This enables nested state machines to emit events that
+	// bubble up to their parent.
+	Outbox []Event
 }
 
 // StateTransition is a state transition type. It denotes the next state to go
@@ -124,6 +131,18 @@ type stateQuery[Event any, Env Environment] struct {
 	CurrentState chan State[Event, Env]
 }
 
+// syncEventRequest is used to send an event to the state machine synchronously,
+// waiting for the event processing to complete and returning the accumulated
+// outbox events.
+type syncEventRequest[Event any] struct {
+	// event is the event to process.
+	event Event
+
+	// promise is used to signal completion and return the accumulated
+	// outbox events or an error.
+	promise actor.Promise[[]Event]
+}
+
 // StateMachine represents an abstract FSM that is able to process new incoming
 // events and drive a state machine to termination. This implementation uses
 // type params to abstract over the types of events and environment. Events
@@ -140,6 +159,10 @@ type StateMachine[Event any, Env Environment] struct {
 	// FSM.
 	events chan Event
 
+	// syncEvents is the channel that will be used to send synchronous event
+	// requests to the FSM, returning the accumulated outbox events.
+	syncEvents chan syncEventRequest[Event]
+
 	// newStateEvents is an EventDistributor that will be used to notify
 	// any relevant callers of new state transitions that occur.
 	newStateEvents *fn.EventDistributor[State[Event, Env]]
@@ -214,6 +237,7 @@ func NewStateMachine[Event any, Env Environment](
 			fmt.Sprintf("FSM(%v):", cfg.Env.Name()),
 		),
 		events:         make(chan Event, 1),
+		syncEvents:     make(chan syncEventRequest[Event], 1),
 		stateQuery:     make(chan stateQuery[Event, Env]),
 		gm:             *fn.NewGoroutineManager(),
 		newStateEvents: fn.NewEventDistributor[State[Event, Env]](),
@@ -259,6 +283,63 @@ func (s *StateMachine[Event, Env]) SendEvent(ctx context.Context, event Event) {
 	}
 }
 
+// AskEvent sends a new event to the state machine using the Ask pattern
+// (request-response), waiting for the event to be fully processed. It
+// returns a Future that will be resolved with the accumulated outbox events
+// from all state transitions triggered by this event, including nested
+// internal events. The Future's Await method will return fn.Result[[]Event]
+// containing either the accumulated outbox events or an error if processing
+// failed.
+func (s *StateMachine[Event, Env]) AskEvent(ctx context.Context,
+	event Event) actor.Future[[]Event] {
+
+	s.log.Debugf("Asking event %T", event)
+
+	// Create a promise to signal completion and return results.
+	promise := actor.NewPromise[[]Event]()
+
+	req := syncEventRequest[Event]{
+		event:   event,
+		promise: promise,
+	}
+
+	// Check for context cancellation or shutdown first to avoid races.
+	select {
+	case <-ctx.Done():
+		promise.Complete(
+			fn.Errf[[]Event]("context cancelled: %w",
+				ctx.Err()),
+		)
+
+		return promise.Future()
+
+	case <-s.quit:
+		promise.Complete(fn.Err[[]Event](ErrStateMachineShutdown))
+
+		return promise.Future()
+
+	default:
+	}
+
+	// Send the request to the state machine. If we can't send it due to
+	// context cancellation or shutdown, complete the promise with an error.
+	select {
+	// Successfully sent, the promise will be completed by driveMachine.
+	case s.syncEvents <- req:
+
+	case <-ctx.Done():
+		promise.Complete(
+			fn.Errf[[]Event]("context cancelled: %w",
+				ctx.Err()),
+		)
+
+	case <-s.quit:
+		promise.Complete(fn.Err[[]Event](ErrStateMachineShutdown))
+	}
+
+	return promise.Future()
+}
+
 // CanHandle returns true if the target message can be routed to the state
 // machine.
 func (s *StateMachine[Event, Env]) CanHandle(msg msgmux.PeerMsg) bool {