21485572389

Committed 29 Jan 2026 04:09PM UTC coverage: 65.247% (+0.2%) from 65.074%

Build # 21485572389

Build Type

Pull #10089

github

Committed by

web-flow

Commit Message

Merge 22d34d15e into 19b2ad797

Pull Request Pull Request #10089: Onion message forwarding

Run Details

1152 of 1448 new or added lines in 23 files covered. (79.56%)

4109 existing lines in 29 files now uncovered.

139515 of 213825 relevant lines covered (65.25%)

20529.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.0

/actor/router.go

package actor

import (
        "context"
        "errors"
        "sync/atomic"

        "github.com/lightningnetwork/lnd/fn/v2"
)

// ErrNoActorsAvailable is returned when a router cannot find any actors
// registered for its service key to forward a message to.
var ErrNoActorsAvailable = errors.New("no actors available for service key")

// RoutingStrategy defines the interface for selecting an actor from a list of
// available actors.
// The M (Message) and R (Response) type parameters ensure that the strategy
// is compatible with the types of actors it will be selecting.
type RoutingStrategy[M Message, R any] interface {
        // Select chooses an ActorRef from the provided slice. It returns the
        // selected actor or an error if no actor can be selected (e.g., if the
        // list is empty or another strategy-specific issue occurs).
        Select(refs []ActorRef[M, R]) (ActorRef[M, R], error)
}

// RoundRobinStrategy implements a round-robin selection strategy. It is generic
// over M and R to match the RoutingStrategy interface, though its logic doesn't
// depend on these types directly for the selection mechanism itself.
type RoundRobinStrategy[M Message, R any] struct {
        // index is used to pick the next actor in a round-robin fashion. It
        // must be accessed atomically to ensure thread-safety if multiple
        // goroutines use the same strategy instance (which they will via the
        // router).
        index uint64
}

// NewRoundRobinStrategy creates a new RoundRobinStrategy, initialized for
// round-robin selection.
func NewRoundRobinStrategy[M Message, R any]() *RoundRobinStrategy[M, R] {
        return &RoundRobinStrategy[M, R]{}
}

// Select picks an actor from the list using a round-robin algorithm.
func (s *RoundRobinStrategy[M, R]) Select(refs []ActorRef[M, R]) (ActorRef[M, R], error) {
        if len(refs) == 0 {
                return nil, ErrNoActorsAvailable
        }

        // Atomically increment and get the current index for selection.
        // We subtract 1 because AddUint64 returns the new value (which is
        // 1-based for the first call after initialization to 0), and slice
        // indexing is 0-based.
        idx := atomic.AddUint64(&s.index, 1) - 1
        selectedRef := refs[idx%uint64(len(refs))]

        return selectedRef, nil
}

// Router is a message-dispatching component that fronts multiple actors
// registered under a specific ServiceKey. It uses a RoutingStrategy to
// distribute messages to one of the available actors. It is generic over M
// (Message type) and R (Response type) to match the actors it routes to.
type Router[M Message, R any] struct {
        receptionist *Receptionist
        serviceKey   ServiceKey[M, R]
        strategy     RoutingStrategy[M, R]
        dlo          ActorRef[Message, any] // Dead Letter Office reference.
}

// NewRouter creates a new Router for a given service key and strategy. The
// receptionist is used to discover actors registered with the service key.
// The router itself is not an actor but a message dispatcher that behaves like
// an ActorRef from the sender's perspective.
func NewRouter[M Message, R any](receptionist *Receptionist,
        key ServiceKey[M, R], strategy RoutingStrategy[M, R],
        dlo ActorRef[Message, any]) *Router[M, R] {

        return &Router[M, R]{
                receptionist: receptionist,
                serviceKey:   key,
                strategy:     strategy,
                dlo:          dlo,
        }
}

// getActor dynamically finds available actors for the service key and selects
// one using the configured strategy. This method is called internally by Tell
// and Ask on each invocation to ensure up-to-date actor discovery.
func (r *Router[M, R]) getActor() (ActorRef[M, R], error) {
        // Discover available actors from the receptionist.
        availableActors := FindInReceptionist(r.receptionist, r.serviceKey)
        if len(availableActors) == 0 {
                return nil, ErrNoActorsAvailable
        }

        // Select one actor using the strategy.
        return r.strategy.Select(availableActors)
}

// Tell sends a message to one of the actors managed by the router, selected by
// the routing strategy. If no actors are available or the send context is
// cancelled before the message can be enqueued in the target actor's mailbox,
// the message may be dropped. Errors during actor selection (e.g.,
// ErrNoActorsAvailable) are currently not propagated from Tell, aligning with
// its fire-and-forget nature. Such errors could be logged internally if needed.
func (r *Router[M, R]) Tell(ctx context.Context, msg M) {
        selectedActor, err := r.getActor()
        if err != nil {
                // If no actors are available for the service, and a DLO is
                // configured, forward the message there.
                if errors.Is(err, ErrNoActorsAvailable) && r.dlo != nil {
                        r.dlo.Tell(context.Background(), msg)
                }
                return
        }

        selectedActor.Tell(ctx, msg)
}

// Ask sends a message to one of the actors managed by the router, selected by
// the routing strategy, and returns a Future for the response. If no actors are
// available (ErrNoActorsAvailable), the Future will be completed with this
// error. If the send context is cancelled before the message can be enqueued in
// the chosen actor's mailbox, the Future will be completed with the context's error.
func (r *Router[M, R]) Ask(ctx context.Context, msg M) Future[R] {
        selectedActor, err := r.getActor()
        if err != nil {
                // If no actor could be selected (e.g., none available),
                // complete the promise immediately with the selection error.
                promise := NewPromise[R]()
                promise.Complete(fn.Err[R](err))
                return promise.Future()
        }

        return selectedActor.Ask(ctx, msg)
}

// ID provides an identifier for the router. Since a router isn't an actor
// itself but a dispatcher for a service, its ID can be based on the service
// key.
func (r *Router[M, R]) ID() string {
        return "router(" + r.serviceKey.name + ")"
}

1	package actor
2
3	import (
4	"context"
5	"errors"
6	"sync/atomic"
7
8	"github.com/lightningnetwork/lnd/fn/v2"
9	)
10
11	// ErrNoActorsAvailable is returned when a router cannot find any actors
12	// registered for its service key to forward a message to.
13	var ErrNoActorsAvailable = errors.New("no actors available for service key")
14
15	// RoutingStrategy defines the interface for selecting an actor from a list of
16	// available actors.
17	// The M (Message) and R (Response) type parameters ensure that the strategy
18	// is compatible with the types of actors it will be selecting.
19	type RoutingStrategy[M Message, R any] interface {
20	// Select chooses an ActorRef from the provided slice. It returns the
21	// selected actor or an error if no actor can be selected (e.g., if the
22	// list is empty or another strategy-specific issue occurs).
23	Select(refs []ActorRef[M, R]) (ActorRef[M, R], error)
24	}
25
26	// RoundRobinStrategy implements a round-robin selection strategy. It is generic
27	// over M and R to match the RoutingStrategy interface, though its logic doesn't
28	// depend on these types directly for the selection mechanism itself.
29	type RoundRobinStrategy[M Message, R any] struct {
30	// index is used to pick the next actor in a round-robin fashion. It
31	// must be accessed atomically to ensure thread-safety if multiple
32	// goroutines use the same strategy instance (which they will via the
33	// router).
34	index uint64
35	}
36
37	// NewRoundRobinStrategy creates a new RoundRobinStrategy, initialized for
38	// round-robin selection.
39	func NewRoundRobinStrategy[M Message, R any]() *RoundRobinStrategy[M, R] {	6✔
40	return &RoundRobinStrategy[M, R]{}	6✔
41	}	6✔
42
43	// Select picks an actor from the list using a round-robin algorithm.
44	func (s *RoundRobinStrategy[M, R]) Select(refs []ActorRef[M, R]) (ActorRef[M, R], error) {	17✔
45	if len(refs) == 0 {	17✔
NEW 46	return nil, ErrNoActorsAvailable	×
NEW 47	}	×
48
49	// Atomically increment and get the current index for selection.
50	// We subtract 1 because AddUint64 returns the new value (which is
51	// 1-based for the first call after initialization to 0), and slice
52	// indexing is 0-based.
53	idx := atomic.AddUint64(&s.index, 1) - 1	17✔
54	selectedRef := refs[idx%uint64(len(refs))]	17✔
55		17✔
56	return selectedRef, nil	17✔
57	}
58
59	// Router is a message-dispatching component that fronts multiple actors
60	// registered under a specific ServiceKey. It uses a RoutingStrategy to
61	// distribute messages to one of the available actors. It is generic over M
62	// (Message type) and R (Response type) to match the actors it routes to.
63	type Router[M Message, R any] struct {
64	receptionist *Receptionist
65	serviceKey ServiceKey[M, R]
66	strategy RoutingStrategy[M, R]
67	dlo ActorRef[Message, any] // Dead Letter Office reference.
68	}
69
70	// NewRouter creates a new Router for a given service key and strategy. The
71	// receptionist is used to discover actors registered with the service key.
72	// The router itself is not an actor but a message dispatcher that behaves like
73	// an ActorRef from the sender's perspective.
74	func NewRouter[M Message, R any](receptionist *Receptionist,
75	key ServiceKey[M, R], strategy RoutingStrategy[M, R],
76	dlo ActorRef[Message, any]) *Router[M, R] {	6✔
77		6✔
78	return &Router[M, R]{	6✔
79	receptionist: receptionist,	6✔
80	serviceKey: key,	6✔
81	strategy: strategy,	6✔
82	dlo: dlo,	6✔
83	}	6✔
84	}	6✔
85
86	// getActor dynamically finds available actors for the service key and selects
87	// one using the configured strategy. This method is called internally by Tell
88	// and Ask on each invocation to ensure up-to-date actor discovery.
89	func (r *Router[M, R]) getActor() (ActorRef[M, R], error) {	21✔
90	// Discover available actors from the receptionist.	21✔
91	availableActors := FindInReceptionist(r.receptionist, r.serviceKey)	21✔
92	if len(availableActors) == 0 {	25✔
93	return nil, ErrNoActorsAvailable	4✔
94	}	4✔
95
96	// Select one actor using the strategy.
97	return r.strategy.Select(availableActors)	17✔
98	}
99
100	// Tell sends a message to one of the actors managed by the router, selected by
101	// the routing strategy. If no actors are available or the send context is
102	// cancelled before the message can be enqueued in the target actor's mailbox,
103	// the message may be dropped. Errors during actor selection (e.g.,
104	// ErrNoActorsAvailable) are currently not propagated from Tell, aligning with
105	// its fire-and-forget nature. Such errors could be logged internally if needed.
106	func (r *Router[M, R]) Tell(ctx context.Context, msg M) {	6✔
107	selectedActor, err := r.getActor()	6✔
108	if err != nil {	8✔
109	// If no actors are available for the service, and a DLO is	2✔
110	// configured, forward the message there.	2✔
111	if errors.Is(err, ErrNoActorsAvailable) && r.dlo != nil {	4✔
112	r.dlo.Tell(context.Background(), msg)	2✔
113	}	2✔
114	return	2✔
115	}
116
117	selectedActor.Tell(ctx, msg)	4✔
118	}
119
120	// Ask sends a message to one of the actors managed by the router, selected by
121	// the routing strategy, and returns a Future for the response. If no actors are
122	// available (ErrNoActorsAvailable), the Future will be completed with this
123	// error. If the send context is cancelled before the message can be enqueued in
124	// the chosen actor's mailbox, the Future will be completed with the context's error.
125	func (r *Router[M, R]) Ask(ctx context.Context, msg M) Future[R] {	15✔
126	selectedActor, err := r.getActor()	15✔
127	if err != nil {	17✔
128	// If no actor could be selected (e.g., none available),	2✔
129	// complete the promise immediately with the selection error.	2✔
130	promise := NewPromise[R]()	2✔
131	promise.Complete(fn.Err[R](err))	2✔
132	return promise.Future()	2✔
133	}	2✔
134
135	return selectedActor.Ask(ctx, msg)	13✔
136	}
137
138	// ID provides an identifier for the router. Since a router isn't an actor
139	// itself but a dispatcher for a service, its ID can be based on the service
140	// key.
141	func (r *Router[M, R]) ID() string {	2✔
142	return "router(" + r.serviceKey.name + ")"	2✔
143	}	2✔

lightningnetwork / lnd / 21485572389

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous