wip: cleanup

Author: gijswijs

onionmessage/actor.go

@@ -9,11 +9,17 @@ import (
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
-// OMRequest is a message sent to an Onion Peer Actor to send an onion message
-// to the peer for which the actor is responsible.
+// OMRequest is a message sent to an Onion Peer Actor to request sending an
+// onion message to a specific peer. Each actor is responsible for a single
+// peer connection, and this message is used to queue onion messages for
+// delivery to that peer.
 type OMRequest struct {
 	// Embed BaseMessage to satisfy the actor package Message interface.
 	actor.BaseMessage
+
+	// msg is the onion message to send. This field is unexported as it's an
+	// implementation detail of the actor system and should not be accessed
+	// directly by external code.
 	msg lnwire.OnionMessage
 }
 
@@ -37,9 +43,18 @@ func (m *OMResponse) MessageType() string {
 type OnionPeerActorRef actor.ActorRef[*OMRequest, *OMResponse]
 
 // SpawnOnionPeerActor spawns a new Onion Peer Actor responsible for sending
-// onion messages to a single peer identified by pubKey. The sender function
-// is used to send the actual onion messages. The function returns a reference
-// to the spawned actor.
+// onion messages to a single peer identified by pubKey.
+//
+// Actor Lifecycle:
+// - One actor per connected peer
+// - Registered with receptionist using peer's public key as service key
+// - Automatically cleaned up when peer disconnects
+//
+// The sender function is used to send the actual onion messages to the peer.
+// This function should be provided by the brontide peer connection logic.
+//
+// Returns a reference to the spawned actor that can be used to send messages
+// via Tell() or Ask() operations.
 func SpawnOnionPeerActor(system *actor.ActorSystem,
 	sender func(msg *lnwire.OnionMessage),
 	pubKey [33]byte) OnionPeerActorRef {
@@ -73,3 +88,12 @@ func SpawnOnionPeerActor(system *actor.ActorSystem,
 
 	return actorRef
 }
+
+func findPeerActor(receptionist *actor.Receptionist, pubKey [33]byte,
+) fn.Option[actor.ActorRef[*OMRequest, *OMResponse]] {
+
+	pubKeyHex := hex.EncodeToString(pubKey[:])
+	serviceKey := actor.NewServiceKey[*OMRequest, *OMResponse](pubKeyHex)
+	refs := actor.FindInReceptionist(receptionist, serviceKey)
+	return fn.Head(refs)
+}

onionmessage/endpoint.go

@@ -34,10 +34,17 @@ type OnionMessageUpdate struct {
 	// blinded routes by default.
 	OnionBlob []byte
 
+	// CustomRecords contains any custom TLV records included in the
+	// payload.
 	CustomRecords record.CustomSet
 
+	// ReplyPath contains the reply path information for the onion message.
 	ReplyPath *lnwire.ReplyPath
 
+	// EncryptedRecipientData contains the encrypted recipient data for the
+	// onion message, created by the creator of the blinded route. This is
+	// the receiver for the last leg of the route, and the sender for the
+	// first leg uyp to the introduction point.
 	EncryptedRecipientData []byte
 }
 
@@ -72,7 +79,14 @@ var _ msgmux.Endpoint = (*OnionEndpoint)(nil)
 
 // NewOnionEndpoint creates a new OnionEndpoint with the given options.
 func NewOnionEndpoint(receptionist *actor.Receptionist, router *sphinx.Router,
-	opts ...OnionEndpointOption) *OnionEndpoint {
+	opts ...OnionEndpointOption) (*OnionEndpoint, error) {
+
+	if receptionist == nil {
+		return nil, ErrNilReceptionist
+	}
+	if router == nil {
+		return nil, ErrNilRouter
+	}
 
 	o := &OnionEndpoint{
 		receptionist: receptionist,
@@ -82,7 +96,7 @@ func NewOnionEndpoint(receptionist *actor.Receptionist, router *sphinx.Router,
 		opt(o)
 	}
 
-	return o
+	return o, nil
 }
 
 // Name returns the unique name of the endpoint.
@@ -118,49 +132,62 @@ func (o *OnionEndpoint) SendMessage(ctx context.Context,
 		btclog.HexN("onion_blob", onionMsg.OnionBlob, 10),
 		slog.Int("blob_length", len(onionMsg.OnionBlob)))
 
-	routingDecisionResult := processOnionMessage(o.router, onionMsg)
+	routingActionResult := processOnionMessage(o.router, onionMsg)
 
-	routingDecision, err := routingDecisionResult.Unpack()
+	routingAction, err := routingActionResult.Unpack()
 	if err != nil {
 		log.ErrorS(logCtx, "Failed to handle onion message", err)
 		return false
 	}
 
-	var payload *lnwire.OnionMessagePayload
-
-	routingDecision.WhenLeft(func(forwardAction ForwardAction) {
-		err = o.forwardMessage(
-			ctx, forwardAction.NextNodeID,
-			forwardAction.NextPathKey,
-			forwardAction.NextPacket,
-		)
-		payload = forwardAction.Payload
-	})
-
-	if err != nil {
-		log.ErrorS(logCtx, "Failed to forward onion message", err)
-		return false
-	}
-
-	routingDecision.WhenRight(func(deliverAction DeliverAction) {
-		payload = deliverAction.Payload
-	})
-
+	var isProcessedSuccessful bool = false
+
+	// Handle the routing action.
+	payload := fn.ElimEither(routingAction,
+		func(forwardAction forwardAction) *lnwire.OnionMessagePayload {
+			log.DebugS(logCtx, "Forwarding onion message",
+				lnutils.LogPubKey("next_node_id",
+					forwardAction.nextNodeID),
+			)
+
+			err := o.forwardMessage(
+				ctx, forwardAction.nextNodeID,
+				forwardAction.nextPathKey,
+				forwardAction.nextPacket,
+			)
+
+			if err != nil {
+				log.ErrorS(logCtx, "Failed to forward onion "+
+					"message", err)
+				isProcessedSuccessful = false
+			}
+
+			isProcessedSuccessful = true
+
+			return forwardAction.payload
+		},
+		func(deliverAction deliverAction) *lnwire.OnionMessagePayload {
+			log.DebugS(logCtx, "Delivering onion message to self")
+			isProcessedSuccessful = true
+			return deliverAction.payload
+		})
+
+	// Convert peer []byte to [33]byte.
 	var peerArr [33]byte
 	copy(peerArr[:], peer)
 
 	// Convert path key []byte to [33]byte.
-	pathKey := onionMsg.PathKey.SerializeCompressed()
 	var pathKeyArr [33]byte
-	copy(pathKeyArr[:], pathKey)
+	copy(pathKeyArr[:], onionMsg.PathKey.SerializeCompressed())
 
+	// Create the onion message update to send to subscribers.
 	update := &OnionMessageUpdate{
 		Peer:      peerArr,
 		PathKey:   pathKeyArr,
 		OnionBlob: onionMsg.OnionBlob,
 	}
 
-	// If we have a payload (no error), add its contents to our update.
+	// If we have a payload, add its contents to our update.
 	if payload != nil {
 		customRecords := make(record.CustomSet)
 		for _, v := range payload.FinalHopPayloads {
@@ -173,18 +200,17 @@ func (o *OnionEndpoint) SendMessage(ctx context.Context,
 
 	// Send the update to any subscribers.
 	if sendErr := o.onionMessageServer.SendUpdate(update); sendErr != nil {
-		log.Errorf("Failed to send onion message update: %v", sendErr)
-		return false
-	}
+		log.ErrorS(logCtx, "Failed to send onion message update",
+			sendErr)
 
-	// If we failed to handle the onion message, we return false.
-	if err != nil {
 		return false
 	}
 
-	return true
+	return isProcessedSuccessful
 }
 
+// forwardMessage forwards the onion message to the next node using the peer
+// actor that was spawned for that node.
 func (o *OnionEndpoint) forwardMessage(ctx context.Context,
 	nextNodeID *btcec.PublicKey, nextBlindingPoint *btcec.PublicKey,
 	nextPacket []byte) error {
@@ -195,25 +221,24 @@ func (o *OnionEndpoint) forwardMessage(ctx context.Context,
 	// Find the onion peer actor for the next node ID.
 	actorRefOpt := findPeerActor(o.receptionist, nextNodeIDBytes)
 
+	// If no actor found, return an error.
 	if actorRefOpt.IsNone() {
-		return fmt.Errorf("no actors found for nextNodeId: %v",
+		return fmt.Errorf("failed to forward onion message: no peer "+
+			"actor found for node %v (peer may be disconnected)",
 			nextNodeID)
 	}
 
-	onionMsg := lnwire.NewOnionMessage(
-		nextBlindingPoint, nextPacket,
-	)
-	req := &OMRequest{msg: *onionMsg}
-	actorRefOpt.UnsafeFromSome().Tell(ctx, req)
+	// Send the onion message to the peer actor.
+	actorRefOpt.WhenSome(func(actorRef actor.ActorRef[*OMRequest,
+		*OMResponse]) {
 
-	return nil
-}
-
-func findPeerActor(receptionist *actor.Receptionist, pubKey [33]byte,
-) fn.Option[actor.ActorRef[*OMRequest, *OMResponse]] {
+		onionMsg := lnwire.NewOnionMessage(
+			nextBlindingPoint, nextPacket,
+		)
+		req := &OMRequest{msg: *onionMsg}
+		actorRef.Tell(ctx, req)
+	})
 
-	pubKeyHex := hex.EncodeToString(pubKey[:])
-	serviceKey := actor.NewServiceKey[*OMRequest, *OMResponse](pubKeyHex)
-	refs := actor.FindInReceptionist(receptionist, serviceKey)
-	return fn.Head(refs)
+	// Successfully forwarded the message.
+	return nil
 }

onionmessage/errors.go

@@ -3,38 +3,17 @@ package onionmessage
 import "errors"
 
 var (
-	// ErrBadMessage is returned when we can't process an onion message.
-	ErrBadMessage = errors.New("onion message processing failed")
-
-	// ErrBadOnionMsg is returned when we receive a bad onion message.
-	ErrBadOnionMsg = errors.New("invalid onion message")
-
-	// ErrBadOnionBlob is returned when we receive a bad onion blob within
-	// our onion message.
-	ErrBadOnionBlob = errors.New("invalid onion blob")
-
-	// ErrNoForwardingOnion is returned when we try to forward an onion
-	// message but no next onion is provided.
-	ErrNoForwardingOnion = errors.New("no next onion provided to forward")
-
-	// ErrNoEncryptedData is returned when the encrypted data TLV is not
-	// present when it is required.
-	ErrNoEncryptedData = errors.New("encrypted data blob required")
-
-	// ErrNoForwardingPayload is returned when no onion message payload
-	// is provided to allow forwarding messages.
-	ErrNoForwardingPayload = errors.New("no payload provided for " +
-		"forwarding")
-
-	// ErrNoNextNodeID is returned when we require a next node id in our
-	// encrypted data blob and one was not provided.
-	ErrNoNextNodeID = errors.New("next node ID required")
-
 	// ErrActortShuttingDown is returned by the actor logic when its context
 	// is cancelled.
 	ErrActorShuttingDown = errors.New("actor shutting down")
 
 	// ErrNextNodeIdEmpty is returned when the next node ID is missing from
 	// the route data.
 	ErrNextNodeIdEmpty = errors.New("next node ID empty")
+
+	// ErrNilReceptionist is returned when a nil receptionist is provided.
+	ErrNilReceptionist = errors.New("receptionist cannot be nil")
+
+	// ErrNilRouter is returned when a nil router is provided.
+	ErrNilRouter = errors.New("router cannot be nil")
 )