actor: add fundamental interfaces and concrete Actor impl
What changed, and why it matters
This commit introduces a brand-new internal actor-model concurrency package for the LND codebase. It adds interfaces for actors, references, behaviors, futures/promises, and a concrete actor implementation with a mailbox, goroutine lifecycle, and a dead-letter office. It is purely foundational code with no production consumers in this commit and no security-relevant behavior beyond normal concurrency hygiene.
No security action required. Treat as normal code-review item for a new internal concurrency primitive; monitor future commits that wire this package into production subsystems for lifecycle, mailbox sizing, and DLO handling issues.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The patch creates the actor package: interface.go defines Message, Future, Promise, TellOnlyRef, ActorRef, ActorBehavior, and sentinel errors; actor.go implements Actor[M,R] with a buffered mailbox, Start/Stop lifecycle using context.Context and sync.Once, a process loop handling Tell/Ask envelopes, graceful draining on shutdown, and DLO routing; func_actor.go provides function-based behavior adapters; and actor_test.go adds unit tests. No existing production code is modified or consumed by this package in the diff.
Changed components
actor/actor.goactor/actor_test.goactor/func_actor.goactor/interface.goInspect captured patch +888 / −0
diff --git a/actor/actor.go b/actor/actor.go
new file mode 100644
index 0000000..373898b
--- /dev/null
+++ b/actor/actor.go
@@ -0,0 +1,324 @@
+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 chan envelope[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],
+ error) {
+
+ if cfg.ID == "" {
+ return nil, ErrEmptyActorID
+ }
+
+ if cfg.Behavior == nil {
+ return nil, ErrNilBehavior
+ }
+
+ 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
+ }
+
+ actor := &Actor[M, R]{
+ id: cfg.ID,
+ behavior: cfg.Behavior,
+ mailbox: make(chan envelope[M, R], mailboxCapacity),
+ ctx: ctx,
+ cancel: cancel,
+ dlo: cfg.DLO,
+ }
+
+ // Create and cache the actor's own reference.
+ actor.ref = &actorRefImpl[M, R]{
+ actor: actor,
+ }
+
+ return actor, nil
+}
+
+// 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() {
+ log.Infof("Actor %s: starting", a.id)
+
+ 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() {
+ for {
+ select {
+ case env := <-a.mailbox:
+ 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)
+ }
+
+ // The actor's context has been cancelled, signaling a stop
+ // request. Exit the processing loop to terminate the actor's
+ // goroutine. Before exiting, drain any remaining messages from
+ // the mailbox.
+ //
+ // NOTE: We intentionally do NOT close the mailbox channel here.
+ // Closing it would create a TOCTOU race with Tell/Ask, which
+ // check ctx.Err() before sending to the mailbox. Between that
+ // check and the actual send, the channel could be closed,
+ // causing a panic. Instead, we drain using a non-blocking
+ // select loop. Messages sent after this point will remain in
+ // the buffered channel and be garbage collected.
+ case <-a.ctx.Done():
+ log.Debugf("Actor %s: context cancelled, draining "+
+ "mailbox", a.id)
+
+ // Drain any remaining messages from the mailbox
+ // without closing the channel.
+ drained := 0
+ for {
+ select {
+ case env := <-a.mailbox:
+ drained++
+
+ // 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,
+ ),
+ )
+ }
+
+ default:
+ if drained > 0 {
+ log.Debugf("Actor %s: "+
+ "drained %d "+
+ "message(s) during "+
+ "shutdown",
+ a.id, drained)
+ }
+
+ // No more messages in the mailbox.
+ return
+ }
+ }
+ }
+ }
+}
+
+// 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() {
+ log.Infof("Actor %s: stopping", a.id)
+
+ 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:ll
+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
+ }
+
+ select {
+ // Message successfully enqueued in the actor's mailbox.
+ case ref.actor.mailbox <- envelope[M, R]{message: msg, promise: nil}:
+
+ // The context for the Tell operation was cancelled before the message
+ // could be enqueued. The message is dropped.
+ case <-ctx.Done():
+ log.Warnf("Tell to actor %s: message %s dropped "+
+ "(caller context cancelled)", ref.actor.id,
+ msg.MessageType())
+
+ // The actor itself has been stopped/terminated.
+ case <-ref.actor.ctx.Done():
+ // If the actor is terminated and has a DLO, send the message
+ // there. Otherwise, it's dropped.
+ ref.trySendToDLO(msg)
+ }
+}
+
+// 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:ll
+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()
+ }
+
+ // Check if the context is already done before attempting to send. This
+ // ensures deterministic behavior and prevents a race where the message
+ // could be enqueued even though the context was already cancelled.
+ if ctx.Err() != nil {
+ promise.Complete(fn.Err[R](ctx.Err()))
+ return promise.Future()
+ }
+
+ select {
+ // Attempt to send the message along with its promise to the actor's
+ // mailbox.
+ case ref.actor.mailbox <- envelope[M, R]{message: msg, promise: promise}:
+
+ // The context for the Ask operation was cancelled before the message
+ // could be enqueued. Complete the promise with the context's error to
+ // unblock the caller.
+ case <-ctx.Done():
+ promise.Complete(fn.Err[R](ctx.Err()))
+
+ // The actor's context was cancelled (e.g., actor stopped) while this
+ // Ask operation was attempting to send (e.g., mailbox was full).
+ case <-ref.actor.ctx.Done():
+ promise.Complete(fn.Err[R](ErrActorTerminated))
+ }
+
+ // 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
+}
diff --git a/actor/actor_test.go b/actor/actor_test.go
new file mode 100644
index 0000000..3d49aa6
--- /dev/null
+++ b/actor/actor_test.go
@@ -0,0 +1,446 @@
+package actor
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "reflect"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/lightningnetwork/lnd/fn/v2"
+ "github.com/stretchr/testify/require"
+)
+
+// testMsg is a simple message type for testing. It embeds BaseMessage to
+// satisfy the actor.Message interface.
+type testMsg struct {
+ BaseMessage
+ data string
+
+ replyChan chan string
+}
+
+// MessageType returns the type name of the message.
+func (m *testMsg) MessageType() string {
+ return "testMsg"
+}
+
+// newTestMsg creates a new test message.
+func newTestMsg(data string) *testMsg {
+ return &testMsg{data: data}
+}
+
+// newTestMsgWithReply creates a new test message that includes a reply channel.
+// This can be used by test behaviors to send data back to the test
+// synchronously, especially for Tell operations.
+func newTestMsgWithReply(data string, replyChan chan string) *testMsg {
+ return &testMsg{data: data, replyChan: replyChan}
+}
+
+// echoBehavior is a simple actor behavior that processes *testMsg messages. It
+// stores the last message's data and, for Ask, echoes it back. For Tell, if
+// replyChan is set in testMsg, it sends data back on it.
+type echoBehavior struct {
+ lastMsgData atomic.Value
+ processingDelay time.Duration
+ t *testing.T
+}
+
+// newEchoBehavior creates a new echoBehavior.
+func newEchoBehavior(t *testing.T, delay time.Duration) *echoBehavior {
+ return &echoBehavior{t: t, processingDelay: delay}
+}
+
+// Receive handles incoming messages. It simulates work if processingDelay is
+// set, stores the message data, and responds for Ask operations or via
+// replyChan for Tell.
+func (b *echoBehavior) Receive(_ context.Context,
+ msg *testMsg) fn.Result[string] {
+
+ if b.processingDelay > 0 {
+ time.Sleep(b.processingDelay)
+ }
+
+ b.lastMsgData.Store(msg.data)
+
+ if msg.replyChan != nil {
+ // Attempt to send the data on the reply channel, but quit if
+ // it takes longer than 1 second (e.g., channel unbuffered
+ // and no receiver).
+ select {
+ case msg.replyChan <- msg.data:
+ case <-time.After(time.Second):
+ b.t.Logf("warning: replyChan send timed out")
+ }
+ }
+
+ return fn.Ok(fmt.Sprintf("echo: %s", msg.data))
+}
+
+// GetLastMsgData retrieves the data from the last message processed.
+func (b *echoBehavior) GetLastMsgData() (string, bool) {
+ val := b.lastMsgData.Load()
+ if val == nil {
+ return "", false
+ }
+ data, ok := val.(string)
+ return data, ok
+}
+
+// errorBehavior is an actor behavior that always returns a predefined error
+// upon receiving a message.
+type errorBehavior struct {
+ err error
+}
+
+// newErrorBehavior creates a new errorBehavior.
+func newErrorBehavior(err error) *errorBehavior {
+ return &errorBehavior{err: err}
+}
+
+// Receive always returns the configured error.
+func (b *errorBehavior) Receive(_ context.Context,
+ _ *testMsg) fn.Result[string] {
+
+ return fn.Err[string](b.err)
+}
+
+// blockingBehavior is an actor behavior that blocks until its actorCtx is done.
+type blockingBehavior struct{}
+
+// Receive blocks until the actor's context is cancelled, then returns the
+// context's error.
+func (b *blockingBehavior) Receive(actorCtx context.Context,
+ _ *testMsg) fn.Result[string] {
+
+ <-actorCtx.Done()
+ return fn.Err[string](actorCtx.Err())
+}
+
+// deadLetterTestMsg is a distinct message type used for testing DLO
+// interactions.
+type deadLetterTestMsg struct {
+ BaseMessage
+ id string
+}
+
+// MessageType returns the type name of the message.
+func (m *deadLetterTestMsg) MessageType() string {
+ return "deadLetterTestMsg"
+}
+
+// deadLetterObserverBehavior is a behavior for a test Dead Letter Office actor.
+// It records all messages sent to it, allowing tests to verify DLO
+// interactions.
+type deadLetterObserverBehavior struct {
+ mu sync.Mutex
+ receivedMsgs []Message
+}
+
+// newDeadLetterObserverBehavior creates a new deadLetterObserverBehavior.
+func newDeadLetterObserverBehavior() *deadLetterObserverBehavior {
+ return &deadLetterObserverBehavior{
+ receivedMsgs: make([]Message, 0),
+ }
+}
+
+// Receive records the incoming message and returns a successful result.
+func (b *deadLetterObserverBehavior) Receive(_ context.Context,
+ msg Message) fn.Result[any] {
+
+ b.mu.Lock()
+ b.receivedMsgs = append(b.receivedMsgs, msg)
+ b.mu.Unlock()
+
+ return fn.Ok[any](nil)
+}
+
+// GetReceivedMsgs returns a copy of all messages received by this DLO.
+func (b *deadLetterObserverBehavior) GetReceivedMsgs() []Message {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ msgs := make([]Message, len(b.receivedMsgs))
+ copy(msgs, b.receivedMsgs)
+
+ return msgs
+}
+
+// actorTestHarness provides helper methods for setting up actors in tests. It
+// manages a dedicated DLO for actors created through it.
+type actorTestHarness struct {
+ t *testing.T
+ dlo *Actor[Message, any]
+ dloBeh *deadLetterObserverBehavior
+}
+
+// newActorTestHarness sets up a test harness with a dedicated DLO. The DLO is
+// automatically stopped when the test cleans up.
+func newActorTestHarness(t *testing.T) *actorTestHarness {
+ t.Helper()
+
+ dloBeh := newDeadLetterObserverBehavior()
+ dloCfg := ActorConfig[Message, any]{
+ ID: "test-dlo-" + t.Name(),
+ Behavior: dloBeh,
+ DLO: nil,
+ MailboxSize: 10,
+ }
+ dloActor, err := NewActor[Message, any](dloCfg)
+ require.NoError(t, err)
+ dloActor.Start()
+
+ t.Cleanup(dloActor.Stop)
+
+ return &actorTestHarness{
+ t: t,
+ dlo: dloActor,
+ dloBeh: dloBeh,
+ }
+}
+
+// newActor creates, starts, and registers a new actor for cleanup. The actor
+// will use the harness's DLO.
+func (h *actorTestHarness) newActor(id string,
+ beh ActorBehavior[*testMsg, string],
+ mailboxSize int) *Actor[*testMsg, string] {
+
+ h.t.Helper()
+
+ cfg := ActorConfig[*testMsg, string]{
+ ID: id,
+ Behavior: beh,
+ DLO: h.dlo.Ref(),
+ MailboxSize: mailboxSize,
+ }
+ actor, err := NewActor(cfg)
+ require.NoError(h.t, err)
+ actor.Start()
+
+ h.t.Cleanup(actor.Stop)
+
+ return actor
+}
+
+// assertDLOMessage checks that the DLO eventually receives a specific message.
+func (h *actorTestHarness) assertDLOMessage(expectedMsg Message) {
+ h.t.Helper()
+ require.Eventually(h.t, func() bool {
+ msgs := h.dloBeh.GetReceivedMsgs()
+ for _, m := range msgs {
+ if reflect.DeepEqual(m, expectedMsg) {
+ return true
+ }
+ }
+ return false
+ }, time.Second, 10*time.Millisecond,
+ "dLO did not receive expected message: %v", expectedMsg,
+ )
+}
+
+// assertNoDLOMessages checks that the DLO has not received any messages.
+func (h *actorTestHarness) assertNoDLOMessages() {
+ h.t.Helper()
+
+ // Allow a very brief moment for any async DLO sends to occur.
+ time.Sleep(20 * time.Millisecond)
+
+ msgs := h.dloBeh.GetReceivedMsgs()
+
+ require.Empty(h.t, msgs, "dLO received unexpected messages")
+}
+
+// TestActorNewActorIDAndRefs verifies that NewActor correctly initializes an
+// actor's ID and provides functional ActorRef and TellOnlyRef instances.
+func TestActorNewActorIDAndRefs(t *testing.T) {
+ t.Parallel()
+
+ h := newActorTestHarness(t)
+ actorID := "test-actor-1"
+ beh := newEchoBehavior(t, 0)
+ actor := h.newActor(actorID, beh, 1)
+
+ require.Equal(t, actorID, actor.Ref().ID(), "actorRef ID mismatch")
+ require.Equal(
+ t, actorID, actor.TellRef().ID(), "tellOnlyRef ID mismatch",
+ )
+ require.NotNil(t, actor.Ref(), "actorRef should not be nil")
+ require.NotNil(t, actor.TellRef(), "tellOnlyRef should not be nil")
+}
+
+// TestActorStartStop verifies the basic lifecycle of an actor: starting,
+// processing messages, and stopping.
+func TestActorStartStop(t *testing.T) {
+ t.Parallel()
+
+ h := newActorTestHarness(t)
+ beh := newEchoBehavior(t, 0)
+ actor := h.newActor("test-actor-lifecycle", beh, 1)
+
+ // Actor should be running and process a message.
+ msgData := "hello"
+ replyChan := make(chan string, 1)
+ actor.Ref().Tell(
+ context.Background(), newTestMsgWithReply(msgData, replyChan),
+ )
+
+ received, err := fn.RecvOrTimeout(replyChan, 100*time.Millisecond)
+ require.NoError(t, err, "timed out waiting for actor to process message")
+ require.Equal(
+ t, msgData, received, "actor did not process message before stop",
+ )
+
+ actor.Stop()
+ time.Sleep(50 * time.Millisecond)
+
+ // Try sending another message; it should ideally not be processed or go
+ // to DLO.
+ msgDataAfterStop := "message-after-stop"
+ replyChanAfterStop := make(chan string, 1)
+ actor.Ref().Tell(
+ context.Background(),
+ newTestMsgWithReply(msgDataAfterStop, replyChanAfterStop),
+ )
+
+ // We expect a timeout here, meaning the message was not processed by
+ // the echoBehavior's replyChan.
+ _, err = fn.RecvOrTimeout(replyChanAfterStop, 100*time.Millisecond)
+ // err == nil would mean a message was received, meaning the actor
+ // processed it after Stop().
+ require.Error(t, err, "actor processed message after Stop()")
+ require.ErrorContains(t, err, "timeout hit")
+
+ h.assertDLOMessage(
+ &testMsg{data: msgDataAfterStop, replyChan: replyChanAfterStop},
+ )
+}
+
+// TestActorTellBasic verifies that a message sent via Tell is processed by the
+// actor's behavior.
+func TestActorTellBasic(t *testing.T) {
+ t.Parallel()
+
+ h := newActorTestHarness(t)
+ beh := newEchoBehavior(t, 0)
+ actor := h.newActor("test-actor-tell", beh, 1)
+
+ msgData := "tell-message"
+ replyChan := make(chan string, 1)
+ actor.Ref().Tell(
+ context.Background(), newTestMsgWithReply(msgData, replyChan),
+ )
+
+ receivedTell, errTell := fn.RecvOrTimeout(replyChan, 100*time.Millisecond)
+ require.NoError(t, errTell, "timed out waiting for Tell message processing")
+ require.Equal(
+ t, msgData, receivedTell, "behavior did not receive Tell message data",
+ )
+
+ lastData, ok := beh.GetLastMsgData()
+ require.True(t, ok, "last message data not set in behavior")
+ require.Equal(t, msgData, lastData, "last message data mismatch")
+ h.assertNoDLOMessages()
+}
+
+// TestActorAskSuccess verifies that a message sent via Ask is processed, and
+// the returned Future is completed with the behavior's successful result.
+func TestActorAskSuccess(t *testing.T) {
+ t.Parallel()
+
+ h := newActorTestHarness(t)
+ beh := newEchoBehavior(t, 0)
+ actor := h.newActor("test-actor-ask-success", beh, 1)
+
+ msgData := "ask-message"
+ future := actor.Ref().Ask(context.Background(), newTestMsg(msgData))
+
+ result := future.Await(context.Background())
+ require.False(t, result.IsErr(), "ask returned an error: %v", result.Err())
+
+ result.WhenOk(func(val string) {
+ expectedReply := fmt.Sprintf("echo: %s", msgData)
+ require.Equal(t, expectedReply, val, "ask response mismatch")
+ })
+
+ lastData, ok := beh.GetLastMsgData()
+ require.True(t, ok, "last message data not set in behavior")
+ require.Equal(t, msgData, lastData, "last message data mismatch")
+ h.assertNoDLOMessages()
+}
+
+// TestActorAskErrorBehavior verifies that if an actor's behavior returns an
+// error, the Future from an Ask call is completed with that error.
+func TestActorAskErrorBehavior(t *testing.T) {
+ t.Parallel()
+
+ h := newActorTestHarness(t)
+ expectedErr := errors.New("behavior error")
+ beh := newErrorBehavior(expectedErr)
+ actor := h.newActor("test-actor-ask-error", beh, 1)
+
+ future := actor.Ref().Ask(
+ context.Background(), newTestMsg("ask-error-test"),
+ )
+
+ result := future.Await(context.Background())
+ require.True(t, result.IsErr(), "ask should have returned an error")
+ require.ErrorIs(t, result.Err(), expectedErr, "ask error mismatch")
+
+ h.assertNoDLOMessages()
+}
+
+// TestFunctionBehaviorFromSimple verifies that FunctionBehaviorFromSimple
+// correctly adapts a simple (msg) -> (result, error) function into an
+// ActorBehavior, handling both success and error cases.
+func TestFunctionBehaviorFromSimple(t *testing.T) {
+ t.Parallel()
+
+ t.Run("success", func(t *testing.T) {
+ t.Parallel()
+
+ h := newActorTestHarness(t)
+
+ beh := FunctionBehaviorFromSimple(
+ func(msg *testMsg) (string, error) {
+ return "simple: " + msg.data, nil
+ },
+ )
+ actor := h.newActor("test-simple-success", beh, 1)
+
+ future := actor.Ref().Ask(
+ context.Background(), newTestMsg("hello"),
+ )
+ result := future.Await(context.Background())
+ require.False(
+ t, result.IsErr(),
+ "expected success, got: %v", result.Err(),
+ )
+ result.WhenOk(func(val string) {
+ require.Equal(t, "simple: hello", val)
+ })
+ })
+
+ t.Run("error", func(t *testing.T) {
+ t.Parallel()
+
+ h := newActorTestHarness(t)
+
+ expectedErr := errors.New("simple behavior error")
+ beh := FunctionBehaviorFromSimple(
+ func(msg *testMsg) (string, error) {
+ return "", expectedErr
+ },
+ )
+ actor := h.newActor("test-simple-error", beh, 1)
+
+ future := actor.Ref().Ask(
+ context.Background(), newTestMsg("hello"),
+ )
+ result := future.Await(context.Background())
+ require.True(t, result.IsErr())
+ require.ErrorIs(t, result.Err(), expectedErr)
+ })
+}
diff --git a/actor/func_actor.go b/actor/func_actor.go
new file mode 100644
index 0000000..f1580f0
--- /dev/null
+++ b/actor/func_actor.go
@@ -0,0 +1,45 @@
+package actor
+
+import (
+ "context"
+
+ "github.com/lightningnetwork/lnd/fn/v2"
+)
+
+// ActorFunc is a function type that represents an actor which functions purely
+// based on a simple function processor.
+type ActorFunc[M Message, R any] func(context.Context, M) fn.Result[R]
+
+// FunctionBehavior adapts a function to the ActorBehavior interface.
+type FunctionBehavior[M Message, R any] struct {
+ fn ActorFunc[M, R]
+}
+
+// NewFunctionBehavior creates a behavior from a function.
+func NewFunctionBehavior[M Message, R any](
+ fn ActorFunc[M, R]) *FunctionBehavior[M, R] {
+
+ return &FunctionBehavior[M, R]{fn: fn}
+}
+
+// Receive implements ActorBehavior interface for the function.
+//
+// TODO(roasbeef): just base it off the function direct instead?
+func (b *FunctionBehavior[M, R]) Receive(ctx context.Context,
+ msg M) fn.Result[R] {
+
+ return b.fn(ctx, msg)
+}
+
+// FunctionBehaviorFromSimple adapts a simpler function to the ActorBehavior
+// interface.
+func FunctionBehaviorFromSimple[M Message, R any](
+ sFunc func(M) (R, error)) *FunctionBehavior[M, R] {
+
+ return NewFunctionBehavior(
+ func(ctx context.Context, msg M) fn.Result[R] {
+ val, err := sFunc(msg)
+ return fn.NewResult(val, err)
+ },
+ )
+}
diff --git a/actor/interface.go b/actor/interface.go
index 8a59509..6ddbe8f 100644
--- a/actor/interface.go
+++ b/actor/interface.go
@@ -7,6 +7,44 @@ import (
"github.com/lightningnetwork/lnd/fn/v2"
)
+// ErrActorTerminated indicates that an operation failed because the target
+// actor was terminated or in the process of shutting down.
+var ErrActorTerminated = fmt.Errorf("actor terminated")
+
+// ErrEmptyActorID is returned when an actor is created with an empty ID.
+var ErrEmptyActorID = fmt.Errorf("actor ID must not be empty")
+
+// ErrNilBehavior is returned when an actor is created with a nil behavior.
+var ErrNilBehavior = fmt.Errorf("actor behavior must not be nil")
+
+// ErrDuplicateActorID is returned when attempting to register an actor with an
+// ID that is already in use within the actor system.
+var ErrDuplicateActorID = fmt.Errorf("actor ID already registered")
+
+// BaseMessage is a helper struct that can be embedded in message types defined
+// outside the actor package to satisfy the Message interface's unexported
+// messageMarker method.
+type BaseMessage struct{}
+
+// messageMarker implements the unexported method for the Message interface,
+// allowing types that embed BaseMessage to satisfy the Message interface.
+func (BaseMessage) messageMarker() {}
+
+// Message is a sealed interface for actor messages. Actors will receive
+// messages conforming to this interface. The interface is "sealed" by the
+// unexported messageMarker method, meaning only types that can satisfy it
+// (e.g., by embedding BaseMessage or being in the same package) can be
+// Messages.
+type Message interface {
+ // messageMarker is a private method that makes this a sealed interface
+ // (see BaseMessage for embedding).
+ messageMarker()
+
+ // MessageType returns the type name of the message for
+ // routing/filtering.
+ MessageType() string
+}
+
// Future represents the result of an asynchronous computation. It allows
// consumers to wait for the result (Await), apply transformations upon
// completion (ThenApply), or register a callback to be executed when the
@@ -44,3 +82,38 @@ type Promise[T any] interface {
// complete it), and false if the future had already been completed.
Complete(result fn.Result[T]) bool
}
+
+// TellOnlyRef is a reference to an actor that only supports "tell" operations.
+// This is useful for scenarios where only fire-and-forget message passing is
+// needed, or to restrict capabilities.
+type TellOnlyRef[M Message] interface {
+ // 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.
+ Tell(ctx context.Context, msg M)
+
+ // ID returns the unique identifier for this actor.
+ ID() string
+}
+
+// ActorRef is a reference to an actor that supports both "tell" and "ask"
+// operations. It embeds TellOnlyRef and adds the Ask method for
+// request-response interactions.
+type ActorRef[M Message, R any] interface {
+ TellOnlyRef[M]
+
+ // 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).
+ Ask(ctx context.Context, msg M) Future[R]
+}
+
+// ActorBehavior defines the logic for how an actor processes incoming messages.
+// It is a strategy interface that encapsulates the actor's reaction to
+// messages.
+type ActorBehavior[M Message, R any] interface {
+ // Receive processes a message and returns a Result. The provided
+ // context is the actor's internal context, which can be used to
+ // detect actor shutdown requests.
+ Receive(actorCtx context.Context, msg M) fn.Result[R]
+}
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.