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) }) }