From b055af6420736cbb04f9a2cc245859ee53402335 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:55:48 -0700 Subject: [PATCH] actor: add fundamental interfaces and concrete Actor impl In this commit, we add the actual Actor implementation. We define a series of types and interfaces, that in concert, describe our actor. An actor has some ID, a reference (used to send messages to it), and also a set of defined messages that it'll accept. An actor can be implemented using a simple function if it's stateless. Otherwise, a struct can implement the Receive method, and handle its internal message passing and state that way. --- actor/actor.go | 324 ++++++++++++++++++++++++++++++++ actor/actor_test.go | 446 ++++++++++++++++++++++++++++++++++++++++++++ actor/func_actor.go | 45 +++++ actor/interface.go | 73 ++++++++ 4 files changed, 888 insertions(+) create mode 100644 actor/actor.go create mode 100644 actor/actor_test.go create mode 100644 actor/func_actor.go diff --git a/actor/actor.go b/actor/actor.go new file mode 100644 index 000000000..373898b87 --- /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 000000000..3d49aa6eb --- /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 000000000..f1580f03f --- /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 8a59509db..6ddbe8fcc 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] +}