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

78.69

/actor/actor.go

package actor

import (
        "context"
        "sync"

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

// ActorConfig holds the configuration parameters for creating a new Actor.
// It is generic over M (Message type) and R (Response type) to accommodate
// the actor's specific behavior.
type ActorConfig[M Message, R any] struct {
        // ID is the unique identifier for the actor.
        ID string

        // Behavior defines how the actor responds to messages.
        Behavior ActorBehavior[M, R]

        // DLO is a reference to the dead letter office for this actor system.
        // If nil, undeliverable messages during shutdown or due to a full
        // mailbox (if such logic were added) might be dropped.
        DLO ActorRef[Message, any]

        // MailboxSize defines the buffer capacity of the actor's mailbox.
        MailboxSize int
}

// envelope wraps a message with its associated promise. This allows the sender
// of an "ask" message to await a response. If the promise is nil, it
// signifies a "tell" operation (fire-and-forget).
type envelope[M Message, R any] struct {
        message M
        promise Promise[R]
}

// Actor represents a concrete actor implementation. It encapsulates a behavior,
// manages its internal state implicitly through that behavior, and processes
// messages from its mailbox sequentially in its own goroutine.
type Actor[M Message, R any] struct {
        // id is the unique identifier for the actor.
        id string

        // behavior defines how the actor responds to messages.
        behavior ActorBehavior[M, R]

        // mailbox is the incoming message queue for the actor.
        mailbox Mailbox[M, R]

        // ctx is the context governing the actor's lifecycle.
        ctx context.Context

        // cancel is the function to cancel the actor's context.
        cancel context.CancelFunc

        // dlo is a reference to the dead letter office for this actor system.
        dlo ActorRef[Message, any]

        // startOnce ensures the actor's processing loop is started only once.
        startOnce sync.Once

        // stopOnce ensures the actor's processing loop is stopped only once.
        stopOnce sync.Once

        // ref is the cached ActorRef for this actor.
        ref ActorRef[M, R]
}

// NewActor creates a new actor instance with the given ID and behavior.
// It initializes the actor's internal structures but does not start its
// message processing goroutine. The Start() method must be called to begin
// processing messages.
func NewActor[M Message, R any](cfg ActorConfig[M, R]) *Actor[M, R] {
        ctx, cancel := context.WithCancel(context.Background())

        // Ensure MailboxSize has a sane default if not specified or zero. A
        // capacity of 0 would make the channel unbuffered, which is generally
        // not desired for actor mailboxes.
        mailboxCapacity := cfg.MailboxSize
        if mailboxCapacity <= 0 {
                // Default to a small capacity if an invalid one is given. This
                // could also come from a global constant.
                mailboxCapacity = 1
        }

        // Create mailbox - could be injected via config in the future.
        mailbox := NewChannelMailbox[M, R](ctx, mailboxCapacity)

        actor := &Actor[M, R]{
                id:       cfg.ID,
                behavior: cfg.Behavior,
                mailbox:  mailbox,
                ctx:      ctx,
                cancel:   cancel,
                dlo:      cfg.DLO,
        }

        // Create and cache the actor's own reference.
        actor.ref = &actorRefImpl[M, R]{
                actor: actor,
        }

        return actor
}

// Start initiates the actor's message processing loop in a new goroutine. This
// method should be called once after the actor is created.
func (a *Actor[M, R]) Start() {
        a.startOnce.Do(func() {
                go a.process()
        })
}

// process is the main event loop for the actor. It continuously monitors its
// mailbox for incoming messages and its context for cancellation signals.
func (a *Actor[M, R]) process() {
        // Use the new iterator pattern for receiving messages.
        for env := range a.mailbox.Receive(a.ctx) {
                result := a.behavior.Receive(a.ctx, env.message)

                // If a promise was provided (i.e., it was an "ask"
                // operation), complete the promise with the result from
                // the behavior.
                if env.promise != nil {
                        env.promise.Complete(result)
                }
        }

        // Context was cancelled or mailbox closed, drain remaining messages.
        a.mailbox.Close()

        for env := range a.mailbox.Drain() {
                // If a DLO is configured, send the original message there
                // for auditing or potential manual reprocessing.
                if a.dlo != nil {
                        a.dlo.Tell(context.Background(), env.message)
                }

                // If it was an Ask, complete the promise with an error
                // indicating the actor terminated.
                if env.promise != nil {
                        env.promise.Complete(fn.Err[R](ErrActorTerminated))
                }
        }
}

// Stop signals the actor to terminate its processing loop and shut down.
// This is achieved by cancelling the actor's internal context. The actor's
// goroutine will exit once it detects the context cancellation.
func (a *Actor[M, R]) Stop() {
        a.stopOnce.Do(func() {
                a.cancel()
        })
}

// actorRefImpl provides a concrete implementation of the ActorRef interface. It
// holds a reference to the target Actor instance, enabling message sending.
type actorRefImpl[M Message, R any] struct {
        actor *Actor[M, R]
}

// Tell sends a message without waiting for a response. If the context is
// cancelled before the message can be sent to the actor's mailbox, the message
// may be dropped.
//
//nolint:lll
func (ref *actorRefImpl[M, R]) Tell(ctx context.Context, msg M) {
        // If the actor's own context is already done, don't try to send.
        // Route to DLO if available.
        if ref.actor.ctx.Err() != nil {
                ref.trySendToDLO(msg)
                return
        }

        env := envelope[M, R]{message: msg, promise: nil}

        // Use mailbox Send method which internally checks both contexts.
        if !ref.actor.mailbox.Send(ctx, env) {
                // Failed to send - check if actor terminated.
                if ref.actor.ctx.Err() != nil {
                        ref.trySendToDLO(msg)
                }
                // Otherwise it was the caller's context that cancelled.
        }
}

// Ask sends a message and returns a Future for the response. The Future will be
// completed with the actor's reply or an error if the operation fails (e.g.,
// context cancellation before send).
//
//nolint:lll
func (ref *actorRefImpl[M, R]) Ask(ctx context.Context, msg M) Future[R] {
        // Create a new promise that will be fulfilled with the actor's response.
        promise := NewPromise[R]()

        // If the actor's own context is already done, complete the promise with
        // ErrActorTerminated and return immediately. This is the primary guard
        // against trying to send to a stopped actor.
        if ref.actor.ctx.Err() != nil {
                promise.Complete(fn.Err[R](ErrActorTerminated))
                return promise.Future()
        }

        env := envelope[M, R]{message: msg, promise: promise}

        // Use mailbox Send method which internally checks both contexts.
        if !ref.actor.mailbox.Send(ctx, env) {
                // Determine the error based on what failed.
                if ref.actor.ctx.Err() != nil {
                        promise.Complete(fn.Err[R](ErrActorTerminated))
                } else {
                        promise.Complete(fn.Err[R](ctx.Err()))
                }
        }

        // Return the future associated with the promise, allowing the caller to
        // await the response.
        return promise.Future()
}

// trySendToDLO attempts to send the message to the actor's DLO if configured.
func (ref *actorRefImpl[M, R]) trySendToDLO(msg M) {
        if ref.actor.dlo != nil {
                // Use context.Background() for sending to DLO as the
                // original context might be done or the operation
                // should not be bound by it.
                // This Tell to DLO is fire-and-forget.
                ref.actor.dlo.Tell(context.Background(), msg)
        }
}

// ID returns the unique identifier for this actor.
func (ref *actorRefImpl[M, R]) ID() string {
        return ref.actor.id
}

// Ref returns an ActorRef for this actor. This allows clients to interact with
// the actor (send messages) without having direct access to the Actor struct
// itself, promoting encapsulation and location transparency.
func (a *Actor[M, R]) Ref() ActorRef[M, R] {
        return a.ref
}

// TellRef returns a TellOnlyRef for this actor. This allows clients to send
// messages to the actor using only the "tell" pattern (fire-and-forget),
// without having access to "ask" capabilities.
func (a *Actor[M, R]) TellRef() TellOnlyRef[M] {
        return a.ref
}

1	package actor
2
3	import (
4	"context"
5	"sync"
6
7	"github.com/lightningnetwork/lnd/fn/v2"
8	)
9
10	// ActorConfig holds the configuration parameters for creating a new Actor.
11	// It is generic over M (Message type) and R (Response type) to accommodate
12	// the actor's specific behavior.
13	type ActorConfig[M Message, R any] struct {
14	// ID is the unique identifier for the actor.
15	ID string
16
17	// Behavior defines how the actor responds to messages.
18	Behavior ActorBehavior[M, R]
19
20	// DLO is a reference to the dead letter office for this actor system.
21	// If nil, undeliverable messages during shutdown or due to a full
22	// mailbox (if such logic were added) might be dropped.
23	DLO ActorRef[Message, any]
24
25	// MailboxSize defines the buffer capacity of the actor's mailbox.
26	MailboxSize int
27	}
28
29	// envelope wraps a message with its associated promise. This allows the sender
30	// of an "ask" message to await a response. If the promise is nil, it
31	// signifies a "tell" operation (fire-and-forget).
32	type envelope[M Message, R any] struct {
33	message M
34	promise Promise[R]
35	}
36
37	// Actor represents a concrete actor implementation. It encapsulates a behavior,
38	// manages its internal state implicitly through that behavior, and processes
39	// messages from its mailbox sequentially in its own goroutine.
40	type Actor[M Message, R any] struct {
41	// id is the unique identifier for the actor.
42	id string
43
44	// behavior defines how the actor responds to messages.
45	behavior ActorBehavior[M, R]
46
47	// mailbox is the incoming message queue for the actor.
48	mailbox Mailbox[M, R]
49
50	// ctx is the context governing the actor's lifecycle.
51	ctx context.Context
52
53	// cancel is the function to cancel the actor's context.
54	cancel context.CancelFunc
55
56	// dlo is a reference to the dead letter office for this actor system.
57	dlo ActorRef[Message, any]
58
59	// startOnce ensures the actor's processing loop is started only once.
60	startOnce sync.Once
61
62	// stopOnce ensures the actor's processing loop is stopped only once.
63	stopOnce sync.Once
64
65	// ref is the cached ActorRef for this actor.
66	ref ActorRef[M, R]
67	}
68
69	// NewActor creates a new actor instance with the given ID and behavior.
70	// It initializes the actor's internal structures but does not start its
71	// message processing goroutine. The Start() method must be called to begin
72	// processing messages.
73	func NewActor[M Message, R any](cfg ActorConfig[M, R]) *Actor[M, R] {	97✔
74	ctx, cancel := context.WithCancel(context.Background())	97✔
75		97✔
76	// Ensure MailboxSize has a sane default if not specified or zero. A	97✔
77	// capacity of 0 would make the channel unbuffered, which is generally	97✔
78	// not desired for actor mailboxes.	97✔
79	mailboxCapacity := cfg.MailboxSize	97✔
80	if mailboxCapacity <= 0 {	97✔
NEW 81	// Default to a small capacity if an invalid one is given. This	×
NEW 82	// could also come from a global constant.	×
NEW 83	mailboxCapacity = 1	×
NEW 84	}	×
85
86	// Create mailbox - could be injected via config in the future.	97✔
87	mailbox := NewChannelMailbox[M, R](ctx, mailboxCapacity)	97✔
88		97✔
89	actor := &Actor[M, R]{	97✔
90	id: cfg.ID,	97✔
91	behavior: cfg.Behavior,	97✔
92	mailbox: mailbox,	97✔
93	ctx: ctx,	97✔
94	cancel: cancel,	97✔
95	dlo: cfg.DLO,	97✔
96	}	97✔
97		97✔
98	// Create and cache the actor's own reference.	97✔
99	actor.ref = &actorRefImpl[M, R]{	97✔
100	actor: actor,	97✔
101	}
102
103	return actor
104	}
105		97✔
106	// Start initiates the actor's message processing loop in a new goroutine. This	194✔
107	// method should be called once after the actor is created.	97✔
108	func (a *Actor[M, R]) Start() {	97✔
109	a.startOnce.Do(func() {
110	go a.process()
111	})
112	}
113		97✔
114	// process is the main event loop for the actor. It continuously monitors its	235✔
115	// mailbox for incoming messages and its context for cancellation signals.	138✔
116	func (a *Actor[M, R]) process() {	41✔
117	// Use the new iterator pattern for receiving messages.	41✔
118	for env := range a.mailbox.Receive(a.ctx) {	41✔
119	result := a.behavior.Receive(a.ctx, env.message)	41✔
120		41✔
121	// If a promise was provided (i.e., it was an "ask"	41✔
122	// operation), complete the promise with the result from	66✔
123	// the behavior.	25✔
124	if env.promise != nil {	25✔
125	env.promise.Complete(result)
126	}
127	}
128
129	// Context was cancelled or mailbox closed, drain remaining messages.
130	a.mailbox.Close()	97✔
131		97✔
132	for env := range a.mailbox.Drain() {	97✔
133	// If a DLO is configured, send the original message there	97✔
134	// for auditing or potential manual reprocessing.	97✔
135	if a.dlo != nil {	97✔
136	a.dlo.Tell(context.Background(), env.message)	97✔
NEW 137	}	×
NEW 138		×
NEW 139	// If it was an Ask, complete the promise with an error	×
NEW 140	// indicating the actor terminated.	×
NEW 141	if env.promise != nil {	×
NEW 142	env.promise.Complete(fn.Err[R](ErrActorTerminated))	×
NEW 143	}	×
NEW 144	}	×
NEW 145	}	×
146
147	// Stop signals the actor to terminate its processing loop and shut down.
148	// This is achieved by cancelling the actor's internal context. The actor's
NEW 149	// goroutine will exit once it detects the context cancellation.	×
NEW 150	func (a *Actor[M, R]) Stop() {	×
NEW 151	a.stopOnce.Do(func() {	×
NEW 152	a.cancel()	×
NEW 153	})	×
154	}
155
156	// actorRefImpl provides a concrete implementation of the ActorRef interface. It	97✔
157	// holds a reference to the target Actor instance, enabling message sending.
158	type actorRefImpl[M Message, R any] struct {
159	actor *Actor[M, R]
160	}
161
162	// Tell sends a message without waiting for a response. If the context is
163	// cancelled before the message can be sent to the actor's mailbox, the message
164	// may be dropped.	98✔
165	//	195✔
166	//nolint:lll	97✔
167	func (ref *actorRefImpl[M, R]) Tell(ctx context.Context, msg M) {	97✔
168	// If the actor's own context is already done, don't try to send.
169	// Route to DLO if available.
170	if ref.actor.ctx.Err() != nil {
171	ref.trySendToDLO(msg)
172	return
173	}
174
175	env := envelope[M, R]{message: msg, promise: nil}
176
177	// Use mailbox Send method which internally checks both contexts.
178	if !ref.actor.mailbox.Send(ctx, env) {
179	// Failed to send - check if actor terminated.
180	if ref.actor.ctx.Err() != nil {
181	ref.trySendToDLO(msg)	19✔
182	}	19✔
183	// Otherwise it was the caller's context that cancelled.	19✔
184	}	21✔
185	}	2✔
186		2✔
187	// Ask sends a message and returns a Future for the response. The Future will be	2✔
188	// completed with the actor's reply or an error if the operation fails (e.g.,
189	// context cancellation before send).	17✔
190	//
191	//nolint:lll	16✔
192	func (ref *actorRefImpl[M, R]) Ask(ctx context.Context, msg M) Future[R] {
193	// Create a new promise that will be fulfilled with the actor's response.
194	promise := NewPromise[R]()
195		1✔
196	// If the actor's own context is already done, complete the promise with
197	// ErrActorTerminated and return immediately. This is the primary guard
NEW 198	// against trying to send to a stopped actor.	×
NEW 199	if ref.actor.ctx.Err() != nil {	×
NEW 200	promise.Complete(fn.Err[R](ErrActorTerminated))	×
NEW 201	return promise.Future()	×
202	}
203
204	env := envelope[M, R]{message: msg, promise: promise}
205
206	// Use mailbox Send method which internally checks both contexts.
207	if !ref.actor.mailbox.Send(ctx, env) {
208	// Determine the error based on what failed.
209	if ref.actor.ctx.Err() != nil {
210	promise.Complete(fn.Err[R](ErrActorTerminated))	36✔
211	} else {	36✔
212	promise.Complete(fn.Err[R](ctx.Err()))	36✔
213	}	36✔
214	}	36✔
215		36✔
216	// Return the future associated with the promise, allowing the caller to	36✔
217	// await the response.	46✔
218	return promise.Future()	10✔
219	}	10✔
220		10✔
221	// trySendToDLO attempts to send the message to the actor's DLO if configured.
222	func (ref *actorRefImpl[M, R]) trySendToDLO(msg M) {
223	if ref.actor.dlo != nil {
224	// Use context.Background() for sending to DLO as the
225	// original context might be done or the operation	27✔
226	// should not be bound by it.	1✔
227	// This Tell to DLO is fire-and-forget.	1✔
228	ref.actor.dlo.Tell(context.Background(), msg)	1✔
229	}
230	}	25✔
231
232	// ID returns the unique identifier for this actor.
233	func (ref *actorRefImpl[M, R]) ID() string {	25✔
234	return ref.actor.id
235	}
236
237	// Ref returns an ActorRef for this actor. This allows clients to interact with
NEW 238	// the actor (send messages) without having direct access to the Actor struct	×
NEW 239	// itself, promoting encapsulation and location transparency.	×
240	func (a *Actor[M, R]) Ref() ActorRef[M, R] {
241	return a.ref
242	}
NEW 243		×
NEW 244	// TellRef returns a TellOnlyRef for this actor. This allows clients to send	×
245	// messages to the actor using only the "tell" pattern (fire-and-forget),
246	// without having access to "ask" capabilities.
247	func (a *Actor[M, R]) TellRef() TellOnlyRef[M] {
248	return a.ref
249	}	25✔

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