Merge pull request #9820 from lightningnetwork/actor

actor: add new package for structured concurrency based on the Actor model
This commit is contained in:
Olaoluwa Osuntokun authored and GitHub committed 2026-02-09 17:02:02 -08:00
commit 3807c6c563
17 files changed
+4141

No files matched your search

+478
View File
@@ -0,0 +1,478 @@
# Actor Package
## Introduction to Actors
The actor model is a conceptual model for concurrent computation that treats
"actors" as the universal primitives of concurrent computation. Originating from
Carl Hewitt's work in the 1970s and popularized by languages like Erlang and
frameworks like Akka, actors provide a high-level abstraction for building
robust, concurrent, and distributed systems.
At its core, an actor is an independent unit of computation that encapsulates:
- **State**: An actor can maintain private state that it alone can modify.
- **Behavior**: An actor defines how it reacts to messages it receives.
- **Mailbox**: Each actor has a mailbox to queue incoming messages.
Actors communicate exclusively through asynchronous message passing. When an
actor receives a message, it can:
1. Send a finite number of messages to other actors.
2. Create a finite number of new actors.
3. Designate the behavior to be used for the next message it receives (which
can be the same behavior).
Concurrency is managed by the actor system, allowing many actors to execute
concurrently without explicit lock management by the developer for actor state.
This model inherently promotes loose coupling, as actors do not share state and
interact only through messages.
## Motivation for this Package
In large, long-lived systems like `lnd`, managing complexity, concurrency, and
component lifecycles becomes increasingly challenging. This `actor` package is
introduced to address several key motivations:
### Structured Message Passing
To move away from direct, synchronous method calls between major components,
especially where concurrency or complex state interactions are involved. Message
passing encourages clearer, more auditable interactions and helps manage
concurrent access to component state.
### Eliminating "God Structs"
Over time, systems can develop large "god structs" that hold references to
numerous sub-systems. This leads to tight coupling, makes dependency management
difficult, and can obscure the flow of control and data. Actors, by
encapsulating state and behavior and interacting via messages, help break down
these monolithic structures into more manageable, independent units.
### Decoupled Lifecycles
Often, the lifecycle of a sub-system is unnecessarily tied to a parent system,
or access to a sub-system requires traversing through a central "manager"
object. Actors can have independent lifecycles managed by an actor system,
allowing for more granular control over starting, stopping, and restarting
components.
An example of such interaction is when an RPC call needs to go through several
other structs to obtain a reference to a given sub-system, in order to make a
direct method call on that sub-system.
With the model described in this document, the RPC server just needs to know
about what is effectively an _abstract address_ of that sub-system. It can then
use that to obtain something similar to a mailbox to do the method call.
This allows for a more decoupled architecture, as the RPC server doesn't need to
know the exact "shape" of the method to call, just which message to send.
Refactors of the sub-system won't break the RPC server, as long as the message
(which can be constructed via a dedicated constructor) is the same.
---
This package provides a foundational actor framework tailored for Go, enabling
developers to build components that are easier to reason about, test, and
maintain in a concurrent environment.
## Core Concepts
Let's explore the fundamental building blocks provided by this package.
### Messages
Actors communicate by sending and receiving messages. Any type that an actor
needs to process must implement the `actor.Message` interface. A simple way to
do this is by embedding `actor.BaseMessage`:
```go
package mymodule
import "github.com/lightningnetwork/lnd/actor"
// MyRequest is a custom message type.
type MyRequest struct {
// Embed BaseMessage to satisfy the Message interface.
actor.BaseMessage
Data string
}
// MessageType returns a string identifier for this message type.
func (m *MyRequest) MessageType() string {
return "MyRequest"
}
// MyResponse might be a corresponding response type.
type MyResponse struct {
actor.BaseMessage
Reply string
}
func (m *MyResponse) MessageType() string {
return "MyResponse"
}
```
The `MessageType()` method provides a string representation of the message type,
which can be useful for debugging or routing.
### Actor Behavior
The logic of an actor (how it responds to messages) is defined by its
`ActorBehavior`. This is an interface that you implement:
```go
package actor
// ActorBehavior defines the logic for how an actor processes incoming messages.
type ActorBehavior[M Message, R any] interface {
Receive(actorCtx context.Context, msg M) fn.Result[R]
}
```
The `Receive` method passes in a caller context (useful for shutdown detection)
and the incoming message. It returns an `fn.Result[R]`, which can encapsulate
either a successful response of type `R` or an error.
For simple cases, you can use `actor.FunctionBehavior` to adapt a Go function
into an `ActorBehavior`:
```go
import (
"context"
"fmt"
"github.com/lightningnetwork/lnd/actor"
"github.com/lightningnetwork/lnd/fn/v2"
)
// myActorLogic defines the processing for MyRequest messages.
func myActorLogic(ctx context.Context, msg *MyRequest) fn.Result[*MyResponse] {
// In a real actor, you might interact with state or other services.
// The actor's context (ctx) can be checked for shutdown signals.
select {
case <-ctx.Done():
return fn.Err[*MyResponse](errors.New("actor shutting down"))
default:
}
response := &MyResponse{Reply: fmt.Sprintf("Processed: %s", msg.Data)}
return fn.Ok(response)
}
// Create a behavior from the function.
behavior := actor.NewFunctionBehavior(myActorLogic)
```
For more complex cases, you can implement the `Receive` method on a new struct,
and pass that around directly.
### Service Keys and Actor References: The Interaction Layer
Direct interaction with an actor's internal state or its concrete struct is
discouraged. Instead, communication and discovery are managed through two key
abstractions: `ServiceKey` and `ActorRef`. These provide a layer of indirection,
promoting loose coupling and location transparency (though the current
implementation is in-process).
#### `ServiceKey[M Message, R any]`
A `ServiceKey` is a type-safe identifier used for registering actors that
provide a particular service and for discovering them later. The generic type
parameters `M` (the type of message the actor handles) and `R` (the type of
response the actor produces for `Ask` operations) ensure that you discover
actors compatible with the interactions you intend to perform.
```go
// Define a service key for actors that handle MyRequest and produce MyResponse.
myServiceKey := actor.NewServiceKey[*MyRequest, *MyResponse]("my-custom-service")
// Later, this key would be used with a Receptionist (part of an ActorSystem)
// to find ActorRefs for actors offering this service.
```
#### `ActorRef[M Message, R any]`
An `ActorRef` is a lightweight, shareable reference to an actor. It's the
primary means by which you send messages to an actor. It is also generic over
the message type `M` and response type `R` that the target actor handles.
You typically obtain an `ActorRef` by looking it up in a `Receptionist` using a
`ServiceKey` (covered later when discussing the `ActorSystem`), or directly from
an actor instance via its `.Ref()` method (e.g., `sampleActor.Ref()` if you have
the `Actor` instance).
There are two main ways to send messages using an `ActorRef`:
1. **Tell (Fire-and-Forget)**: Used for sending messages when you don't need a
direct reply. The call returns immediately after attempting to enqueue the
message.
```go
// Assuming 'actorRef' is an ActorRef[*MyRequest, *MyResponse] obtained for an actor.
requestMsg := &MyRequest{Data: "A fire-and-forget message"}
actorRef.Tell(context.Background(), requestMsg)
// The message is now in the actor's mailbox (or will be shortly).
```
The `context.Context` passed to `Tell` can be used to cancel the send
operation if, for example, the actor's mailbox is full and the send would
block for too long.
2. **Ask (Request-Response)**: Used when you need a response from the actor.
This returns a `Future[R]`, which represents the eventual reply.
```go
// Assuming 'actorRef' is an ActorRef[*MyRequest, *MyResponse].
askMsg := &MyRequest{Data: "A request needing a response"}
futureResponse := actorRef.Ask(context.Background(), askMsg)
```
A `Future[R]` represents a result that will be available at some point. You
can block until it's ready using `Await`:
```go
// Await the result. It's good practice to use a context with a timeout.
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
result := futureResponse.Await(ctx)
response, err := result.Unpack()
if err != nil {
fmt.Printf("Ask failed: %v\n", err)
// return or handle error
} else {
fmt.Printf("Received reply: %s\n", response.Reply)
}
```
The `Future` interface also offers non-blocking ways to handle results, like
`OnComplete` (for callbacks) and `ThenApply` (for chaining transformations).
A more restricted `TellOnlyRef[M]` is also available if only fire-and-forget
semantics are required (obtained via an actor's `TellRef()` method).
### Actors
An `Actor` is the concrete entity that runs a behavior, manages a mailbox, and
has a lifecycle. You create an actor using `actor.NewActor` with an
`ActorConfig`:
```go
cfg := actor.ActorConfig[*MyRequest, *MyResponse]{
ID: "my-sample-actor",
Behavior: behavior,
MailboxSize: 10,
// Dead Letter Office (covered later)
DLO: nil,
}
sampleActor, err := actor.NewActor(cfg)
if err != nil {
// Handle invalid config (empty ID, nil behavior).
return err
}
```
An actor doesn't start processing messages until its `Start()` method is called.
This launches a dedicated goroutine for the actor.
```go
sampleActor.Start()
```
To stop an actor, you call its `Stop()` method. This cancels the actor's
internal context, causing its goroutine to clean up and exit.
```go
// Sometime later...
sampleActor.Stop()
```
## Visualizing Actor Relationships
The following diagram illustrates the primary components of the actor package
and their relationships. It provides a high-level overview of how actors are
managed, discovered, and interacted with.
```mermaid
classDiagram
direction TB
class ActorSystem {
+Receptionist
+DeadLetters
+Shutdown()
}
class Receptionist {
+Find(ServiceKey) ActorRef[]
+Register(ServiceKey, ActorRef)
}
class DeadLetterOffice {
+Receive(undeliverable Message)
}
class ServiceKey {
+Spawn(ActorSystem, Behavior) ActorRef
}
class Actor {
-mailbox
-behavior
+Ref() ActorRef
+Start()
+Stop()
}
class ActorRef {
<<Interface>>
+Tell(Message)
+Ask(Message) Future
}
class Message {
<<Interface>>
}
class Future {
+Await() Result
}
class Router {
+Tell(Message)
+Ask(Message) Future
}
%% Core system relationships
ActorSystem *-- Receptionist : has
ActorSystem *-- DeadLetterOffice : provides
ActorSystem o-- "manages" Actor
%% Actor and communication
Actor --> ActorRef : provides
Actor ..> Message : processes
ActorRef ..> Message : sends
ActorRef ..> Future : returns for Ask
%% Service discovery and routing
Receptionist o-- ServiceKey : uses for lookup
ServiceKey ..> Actor : creates
Router --> ActorRef : routes to
Router --> Receptionist : discovers actors via
note for ActorSystem "Central manager for actor lifecycle and service discovery"
note for Actor "Independent unit with encapsulated state and behavior"
note for ActorRef "Location-transparent handle for sending messages"
note for Message "Data exchanged between actors"
note for ServiceKey "Type-safe identifier for actor registration and discovery"
note for Router "Distributes messages among multiple actors"
note for DeadLetterOffice "Handles messages that cannot be delivered"
```
## The Actor System
While individual actors are useful, they often need to be managed and
coordinated. The `ActorSystem` serves this purpose.
```go
system := actor.NewActorSystem()
// Ensures all actors in the system are stopped.
defer system.Shutdown()
```
### Actor Lifecycle and Registration
The `ActorSystem` can manage the lifecycle of actors. You can register actors
with the system:
```go
// Using 'behavior' from earlier and 'myServiceKey' defined in the
// "Service Keys and Actor References" section.
// RegisterWithSystem creates, starts, and registers the actor.
actorRefFromSystem := actor.RegisterWithSystem(
system, "system-managed-actor", myServiceKey, behavior,
)
```
Alternatively, a `ServiceKey` itself provides a `Spawn` method for convenience:
```go
actorRefSpawned := myServiceKey.Spawn(system, "spawned-actor", behavior)
```
Actors registered with the system are automatically stopped when
`system.Shutdown()` is called. You can also stop and remove individual actors
using `system.StopAndRemoveActor(actorID)`.
A `ServiceKey` is essentially the mailbox address of an actor.
### Receptionist: Service Discovery
Actors often need to find other actors to communicate with. The `Receptionist`
facilitates this. Actors are registered with the receptionist using a
`ServiceKey`, which is type-safe.
```go
// Get the system's receptionist.
receptionist := system.Receptionist()
// Find actors registered for a specific service key.
foundRefs := actor.FindInReceptionist(receptionist, myServiceKey)
if len(foundRefs) > 0 {
targetActor := foundRefs[0]
targetActor.Tell(context.Background(), &MyRequest{Data: "Hello from a discoverer!"})
} else {
fmt.Println("No actors found for service key:", myServiceKey)
}
```
When an actor is stopped (e.g., via `ServiceKey.Unregister` or system shutdown),
it should also be unregistered from the receptionist.
### Dead Letter Office (DLO)
What happens to messages that cannot be delivered? For example, if an actor is
stopped while messages are still in its mailbox, or if a message is sent to an
actor that doesn't exist (though the current `ActorRef` design makes the latter
less likely for direct sends).
The `ActorSystem` provides a default `DeadLetterActor`. When an actor is
configured (via `ActorConfig.DLO`), undeliverable messages (e.g., those drained
from its mailbox upon shutdown) can be routed to this DLO. This allows for
logging, auditing, or potential manual intervention for "lost" messages.
```go
// Actors created via RegisterWithSystem or ServiceKey.Spawn
// are automatically configured to use the system's DLO.
// system.DeadLetters() returns an ActorRef to the system's DLO.
```
## Routers: Distributing Work
Sometimes, you might have multiple actors performing the same kind of task, and
you want to distribute messages among them. A `Router` can do this. It's not an
actor itself but acts as a dispatcher.
A `Router` uses a `RoutingStrategy` to pick one actor from a group registered
under a `ServiceKey`.
```go
// Assume 'system' and 'myServiceKey' are set up, and multiple actors
// are registered with 'myServiceKey'.
// Create a round-robin routing strategy.
roundRobinStrategy := actor.NewRoundRobinStrategy[*MyRequest, *MyResponse]()
// Create a router for 'myServiceKey' using this strategy.
// Messages sent to this router will be forwarded to one of the actors
// registered under 'myServiceKey'.
// The router also needs a DLO for messages it can't route (e.g., if no actors are available).
serviceRouter := actor.NewRouter(
system.Receptionist(),
myServiceKey,
roundRobinStrategy,
system.DeadLetters(),
)
// Now, interact with the router as if it were an ActorRef:
serviceRouter.Tell(context.Background(), &MyRequest{Data: "Message via router"})
futureReplyFromRouter := serviceRouter.Ask(context.Background(), &MyRequest{Data: "Ask via router"})
// ... await futureReplyFromRouter ...
```
If the router cannot find any available actors for the `ServiceKey` (e.g., none
are registered or running), `Tell` operations will typically send the message to
the router's configured DLO, and `Ask` operations will return a `Future`
completed with `ErrNoActorsAvailable`.
+324
View File
@@ -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
}
+446
View File
@@ -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)
})
}
+101
View File
@@ -0,0 +1,101 @@
package actor_test
import (
"context"
"fmt"
"time"
"github.com/lightningnetwork/lnd/actor"
"github.com/lightningnetwork/lnd/fn/v2"
)
// BasicGreetingMsg is a simple message type for the basic actor example.
type BasicGreetingMsg struct {
actor.BaseMessage
Name string
}
// MessageType implements actor.Message.
func (m BasicGreetingMsg) MessageType() string { return "BasicGreetingMsg" }
// BasicGreetingResponse is a simple response type.
type BasicGreetingResponse struct {
Greeting string
}
// ExampleActor demonstrates creating a single actor, sending it a message
// directly using Ask, and then unregistering and stopping it.
func ExampleActor() {
system := actor.NewActorSystem()
defer system.Shutdown()
//nolint:ll
greeterKey := actor.NewServiceKey[BasicGreetingMsg, BasicGreetingResponse](
"basic-greeter",
)
actorID := "my-greeter"
greeterBehavior := actor.NewFunctionBehavior(
func(ctx context.Context,
msg BasicGreetingMsg) fn.Result[BasicGreetingResponse] {
return fn.Ok(BasicGreetingResponse{
Greeting: "Hello, " + msg.Name + " from " +
actorID,
})
},
)
// Spawn the actor. This registers it with the system and receptionist,
// and starts it. It returns an ActorRef.
greeterRef, err := greeterKey.Spawn(system, actorID, greeterBehavior)
if err != nil {
fmt.Printf("Failed to spawn actor: %v\n", err)
return
}
fmt.Printf("Actor %s spawned.\n", greeterRef.ID())
// Send a message directly to the actor's reference.
askCtx, askCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
defer askCancel()
futureResponse := greeterRef.Ask(
askCtx, BasicGreetingMsg{Name: "World"},
)
awaitCtx, awaitCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
defer awaitCancel()
result := futureResponse.Await(awaitCtx)
result.WhenErr(func(err error) {
fmt.Printf("Error awaiting response: %v\n", err)
})
result.WhenOk(func(response BasicGreetingResponse) {
fmt.Printf("Received: %s\n", response.Greeting)
})
// Unregister the actor. This also stops the actor.
unregistered := greeterKey.Unregister(system, greeterRef)
if unregistered {
fmt.Printf("Actor %s unregistered and stopped.\n",
greeterRef.ID())
} else {
fmt.Printf("Failed to unregister actor %s.\n", greeterRef.ID())
}
// Verify it's no longer in the receptionist.
refsAfterUnregister := actor.FindInReceptionist(
system.Receptionist(), greeterKey,
)
fmt.Printf("Actors for key '%s' after unregister: %d\n",
"basic-greeter", len(refsAfterUnregister))
// Output:
// Actor my-greeter spawned.
// Received: Hello, World from my-greeter
// Actor my-greeter unregistered and stopped.
// Actors for key 'basic-greeter' after unregister: 0
}
+120
View File
@@ -0,0 +1,120 @@
package actor_test
import (
"context"
"fmt"
"time"
"github.com/lightningnetwork/lnd/actor"
"github.com/lightningnetwork/lnd/fn/v2"
)
// RouterGreetingMsg is a message type for the router example.
type RouterGreetingMsg struct {
actor.BaseMessage
Name string
}
// MessageType implements actor.Message.
func (m RouterGreetingMsg) MessageType() string { return "RouterGreetingMsg" }
// RouterGreetingResponse is a response type for the router example.
type RouterGreetingResponse struct {
Greeting string
HandlerID string
}
// ExampleRouter demonstrates creating multiple actors under the same service
// key and using a router to dispatch messages to them.
func ExampleRouter() {
system := actor.NewActorSystem()
defer system.Shutdown()
//nolint:ll
routerGreeterKey := actor.NewServiceKey[RouterGreetingMsg, RouterGreetingResponse](
"router-greeter-service",
)
// Behavior for the first greeter actor.
actorID1 := "router-greeter-1"
greeterBehavior1 := actor.NewFunctionBehavior(
func(ctx context.Context,
msg RouterGreetingMsg) fn.Result[RouterGreetingResponse] {
return fn.Ok(RouterGreetingResponse{
Greeting: "Greetings, " + msg.Name + "!",
HandlerID: actorID1,
})
},
)
_, err := routerGreeterKey.Spawn(system, actorID1, greeterBehavior1)
if err != nil {
fmt.Printf("Failed to spawn actor: %v\n", err)
return
}
fmt.Printf("Actor %s spawned.\n", actorID1)
// Behavior for the second greeter actor.
actorID2 := "router-greeter-2"
greeterBehavior2 := actor.NewFunctionBehavior(
func(ctx context.Context,
msg RouterGreetingMsg) fn.Result[RouterGreetingResponse] {
return fn.Ok(RouterGreetingResponse{
Greeting: "Salutations, " + msg.Name + "!",
HandlerID: actorID2,
})
},
)
_, err = routerGreeterKey.Spawn(system, actorID2, greeterBehavior2)
if err != nil {
fmt.Printf("Failed to spawn actor: %v\n", err)
return
}
fmt.Printf("Actor %s spawned.\n", actorID2)
// Create a router for the "router-greeter-service".
greeterRouter := actor.NewRouter(
system.Receptionist(), routerGreeterKey,
actor.NewRoundRobinStrategy[RouterGreetingMsg,
RouterGreetingResponse](),
system.DeadLetters(),
)
fmt.Printf("Router %s created for service key '%s'.\n",
greeterRouter.ID(), "router-greeter-service")
// Send messages through the router.
names := []string{"Alice", "Bob", "Charlie", "David"}
for _, name := range names {
askCtx, askCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
futureResponse := greeterRouter.Ask(
askCtx, RouterGreetingMsg{Name: name},
)
awaitCtx, awaitCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
result := futureResponse.Await(awaitCtx)
result.WhenErr(func(err error) {
fmt.Printf("For %s: Error - %v\n", name, err)
})
result.WhenOk(func(response RouterGreetingResponse) {
fmt.Printf("For %s: Received '%s' from %s\n",
name, response.Greeting, response.HandlerID)
})
awaitCancel()
askCancel()
}
// Output:
// Actor router-greeter-1 spawned.
// Actor router-greeter-2 spawned.
// Router router(router-greeter-service) created for service key 'router-greeter-service'.
// For Alice: Received 'Greetings, Alice!' from router-greeter-1
// For Bob: Received 'Salutations, Bob!' from router-greeter-2
// For Charlie: Received 'Greetings, Charlie!' from router-greeter-1
// For David: Received 'Salutations, David!' from router-greeter-2
}
+153
View File
@@ -0,0 +1,153 @@
package actor_test
import (
"context"
"fmt"
"time"
"github.com/lightningnetwork/lnd/actor"
"github.com/lightningnetwork/lnd/fn/v2"
)
// CounterMsg is a message type for the stateful counter actor.
// It can be used to increment the counter or get its current value.
type CounterMsg struct {
actor.BaseMessage
Increment int
GetValue bool
Who string
}
// MessageType implements actor.Message.
func (m CounterMsg) MessageType() string { return "CounterMsg" }
// CounterResponse is a response type for the counter actor.
type CounterResponse struct {
Value int
Responder string
}
// StatefulCounterActor demonstrates an actor that maintains internal state (a
// counter) and processes messages to modify or query that state.
type StatefulCounterActor struct {
counter int
actorID string
}
// NewStatefulCounterActor creates a new counter actor.
func NewStatefulCounterActor(id string) *StatefulCounterActor {
return &StatefulCounterActor{
actorID: id,
}
}
// Receive is the message handler for the StatefulCounterActor.
// It implements the actor.ActorBehavior interface implicitly when wrapped.
func (s *StatefulCounterActor) Receive(ctx context.Context,
msg CounterMsg) fn.Result[CounterResponse] {
if msg.Increment > 0 {
// For increment, we can just acknowledge or return the new
// value. Messages are sent serially, so we don't need to worry
// about a mutex here.
s.counter += msg.Increment
return fn.Ok(CounterResponse{
Value: s.counter,
Responder: s.actorID,
})
}
if msg.GetValue {
return fn.Ok(CounterResponse{
Value: s.counter,
Responder: s.actorID,
})
}
return fn.Err[CounterResponse](fmt.Errorf("invalid CounterMsg"))
}
// ExampleActor_stateful demonstrates creating an actor whose behavior is defined
// by a struct with methods, allowing it to maintain internal state.
func ExampleActor_stateful() {
system := actor.NewActorSystem()
defer system.Shutdown()
counterServiceKey := actor.NewServiceKey[CounterMsg, CounterResponse](
"struct-counter-service",
)
// Create an instance of our stateful actor logic.
actorID := "counter-actor-1"
counterLogic := NewStatefulCounterActor(actorID)
// Spawn the actor.
// The counterLogic instance itself satisfies the ActorBehavior
// interface because its Receive method matches the required signature.
counterRef, err := counterServiceKey.Spawn(
system, actorID, counterLogic,
)
if err != nil {
fmt.Printf("Failed to spawn actor: %v\n", err)
return
}
fmt.Printf("Actor %s spawned.\n", counterRef.ID())
// Send messages to increment the counter.
for i := 1; i <= 3; i++ {
askCtx, askCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
futureResp := counterRef.Ask(askCtx,
CounterMsg{
Increment: i,
Who: fmt.Sprintf("Incrementer-%d", i),
},
)
awaitCtx, awaitCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
resp := futureResp.Await(awaitCtx)
resp.WhenOk(func(r CounterResponse) {
fmt.Printf("Incremented by %d, new value: %d "+
"(from %s)\n", i, r.Value, r.Responder)
})
resp.WhenErr(func(e error) {
fmt.Printf("Error incrementing: %v\n", e)
})
awaitCancel()
askCancel()
}
// Send a message to get the current value.
askCtx, askCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
futureResp := counterRef.Ask(
askCtx, CounterMsg{GetValue: true, Who: "Getter"},
)
awaitCtx, awaitCancel := context.WithTimeout(
context.Background(), 1*time.Second,
)
finalValueResp := futureResp.Await(awaitCtx)
finalValueResp.WhenOk(func(r CounterResponse) {
fmt.Printf("Final counter value: %d (from %s)\n",
r.Value, r.Responder)
})
finalValueResp.WhenErr(func(e error) {
fmt.Printf("Error getting value: %v\n", e)
})
awaitCancel()
askCancel()
// Output:
// Actor counter-actor-1 spawned.
// Incremented by 1, new value: 1 (from counter-actor-1)
// Incremented by 2, new value: 3 (from counter-actor-1)
// Incremented by 3, new value: 6 (from counter-actor-1)
// Final counter value: 6 (from counter-actor-1)
}
+137
View File
@@ -0,0 +1,137 @@
package actor_test
import (
"context"
"fmt"
"strings"
"sync"
"time"
"github.com/lightningnetwork/lnd/actor"
"github.com/lightningnetwork/lnd/fn/v2"
)
// LogMsg is a message type for the TellOnly example.
type LogMsg struct {
actor.BaseMessage
Text string
}
// MessageType implements actor.Message.
func (m LogMsg) MessageType() string { return "LogMsg" }
// LoggerActorBehavior is a simple actor behavior that logs messages. It doesn't
// produce a meaningful response for Ask, so it's a good candidate for TellOnly
// interactions.
type LoggerActorBehavior struct {
mu sync.Mutex
logs []string
actorID string
}
func NewLoggerActorBehavior(id string) *LoggerActorBehavior {
return &LoggerActorBehavior{actorID: id}
}
// Receive processes LogMsg messages by appending them to an internal log. The
// response type is 'any' as it's not typically used with Ask.
func (l *LoggerActorBehavior) Receive(ctx context.Context,
msg actor.Message) fn.Result[any] {
logMessage, ok := msg.(LogMsg)
if !ok {
return fn.Err[any](fmt.Errorf("unexpected message "+
"type: %s", msg.MessageType()))
}
l.mu.Lock()
defer l.mu.Unlock()
entry := fmt.Sprintf("[%s from %s]: %s", time.Now().Format("15:04:05"),
l.actorID, logMessage.Text)
l.logs = append(l.logs, entry)
// For Tell, the result is often ignored, but we must return something.
return fn.Ok[any](nil)
}
func (l *LoggerActorBehavior) GetLogs() []string {
l.mu.Lock()
defer l.mu.Unlock()
copiedLogs := make([]string, len(l.logs))
copy(copiedLogs, l.logs)
return copiedLogs
}
// ExampleTellOnlyRef demonstrates using a TellOnlyRef for fire-and-forget
// messaging with an actor.
func ExampleTellOnlyRef() {
system := actor.NewActorSystem()
defer system.Shutdown()
// The logger actor doesn't really have a response type for Ask, so we
// use 'any'.
loggerServiceKey := actor.NewServiceKey[actor.Message, any](
"tell-only-logger-service",
)
actorID := "my-logger"
loggerLogic := NewLoggerActorBehavior(actorID)
// Spawn the actor.
fullRef, err := loggerServiceKey.Spawn(system, actorID, loggerLogic)
if err != nil {
fmt.Printf("Failed to spawn actor: %v\n", err)
return
}
fmt.Printf("Actor %s spawned.\n", fullRef.ID())
// Get a TellOnlyRef for the actor. We can get this from the Actor
// instance itself if we had it, or by type assertion if we know the
// underlying ref supports it. Since fullRef is ActorRef[actor.Message,
// any], it already satisfies TellOnlyRef[actor.Message].
//
// Or, if we had the *Actor instance: tellOnlyLogger =
// actorInstance.TellRef()
var tellOnlyLogger actor.TellOnlyRef[actor.Message] = fullRef
fmt.Printf("Obtained TellOnlyRef for %s.\n", tellOnlyLogger.ID())
// Send messages using Tell.
tellOnlyLogger.Tell(
context.Background(), LogMsg{Text: "First log entry."},
)
tellOnlyLogger.Tell(
context.Background(), LogMsg{Text: "Second log entry."},
)
// Allow some time for messages to be processed.
time.Sleep(10 * time.Millisecond)
// Retrieve logs directly from the behavior for verification in this
// example. In a real scenario, this might not be possible or desired.
logs := loggerLogic.GetLogs()
fmt.Println("Logged entries:")
for _, entry := range logs {
// Strip the timestamp and actor ID for consistent example
// output. Example entry: "[15:04:05 from my-logger]: Actual log
// text"
parts := strings.SplitN(entry, "]: ", 2)
if len(parts) == 2 {
fmt.Println(parts[1])
}
}
// Attempting to Ask using tellOnlyLogger would be a compile-time error:
// tellOnlyLogger.Ask(context.Background(), LogMsg{Text: "This would
// fail"})
// Output:
// Actor my-logger spawned.
// Obtained TellOnlyRef for my-logger.
// Logged entries:
// First log entry.
// Second log entry.
}
+45
View File
@@ -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)
},
)
}
+158
View File
@@ -0,0 +1,158 @@
package actor
import (
"context"
"sync"
"sync/atomic"
"github.com/lightningnetwork/lnd/fn/v2"
)
// promiseImpl is a structure that can be used to complete a Future. It provides
// methods to set the result of an asynchronous operation and to obtain the
// Future interface for consumers.
// The promiseImpl itself is not typically exposed directly to consumers of the
// future's result; they interact with the Future interface.
type promiseImpl[T any] struct {
fut *futureImpl[T]
}
// NewPromise creates a new Promise. The associated Future, which consumers can
// use to await the result, can be obtained via the Future() method. The Future
// is completed by calling the Complete() method on this Promise.
func NewPromise[T any]() Promise[T] {
return &promiseImpl[T]{
fut: &futureImpl[T]{
// done is a channel that will be closed when the future
// is completed.
done: make(chan struct{}),
},
}
}
// Future returns the Future interface associated with this Promise. Consumers
// can use this to Await the result or register callbacks.
func (p *promiseImpl[T]) Future() Future[T] {
return p.fut
}
// Complete attempts to set the result of the future. It returns true if this
// call successfully set the result (i.e., it was the first to complete it),
// and false if the future had already been completed. This ensures that a
// future can only be completed once. The completion involves storing the result
// and signaling any goroutines waiting on the future's done channel.
func (p *promiseImpl[T]) Complete(result fn.Result[T]) bool {
var success bool
p.fut.completeOnce.Do(func() {
p.fut.resultCache.Store(&result)
close(p.fut.done)
success = true
})
return success
}
// futureImpl is the concrete implementation of the Future interface. It manages
// the state of an asynchronous computation's result.
type futureImpl[T any] struct {
// resultCache stores the fn.Result[T] after the future is completed.
// It's of type atomic.Pointer to allow lock-free reads after completion
// with improved type safety over atomic.Value.
resultCache atomic.Pointer[fn.Result[T]]
// done is closed once the future is completed, signaling any waiting
// Await calls.
done chan struct{}
// completeOnce ensures that the logic to set the result and close the
// done channel is executed only once.
completeOnce sync.Once
}
// Await blocks until the result is available or the passed context is
// cancelled. If the future is already completed, it returns the result
// immediately. Otherwise, it waits for either the future's completion or the
// context's cancellation.
func (f *futureImpl[T]) Await(ctx context.Context) fn.Result[T] {
// First, try a non-blocking load from the cache. If the future is
// already completed, this will return the result directly.
if resPtr := f.resultCache.Load(); resPtr != nil {
return *resPtr
}
// Wait for either the future to be done or the context to be cancelled.
select {
case <-f.done:
// The future has been completed. Load the result from the
// cache. It must be present now. Load and dereference.
// This load is safe because the 'done' channel is closed only
// after the resultCache is written (ensured by completeOnce).
resPtr := f.resultCache.Load()
// resPtr should not be nil here as <-f.done was signaled.
return *resPtr
case <-ctx.Done():
// The waiting context was cancelled before the future completed.
return fn.Err[T](ctx.Err())
}
}
// ThenApply registers a function to transform the result of a future. The
// original future is not modified; a new Future instance representing the
// transformed result is returned. Once the original future completes
// successfully, the provided transformation function (fApply) is called with
// the result. The transformation is applied asynchronously in a new goroutine.
// If the passed context is cancelled while waiting for the
// original future to complete, the returned future will yield the context's
// error.
func (f *futureImpl[T]) ThenApply(ctx context.Context,
fApply func(T) T) Future[T] {
// Create a new promise for the transformed result.
transformedPromise := NewPromise[T]()
go func() {
// Await the original future's result, respecting the passed
// context for cancellation.
originalResult := f.Await(ctx)
// If the original future completed with an error (or Await was
// cancelled by its context), complete the transformed future
// with the same error.
// This also handles the case where originalResult.Await(ctx)
// itself returned ctx.Err().
if originalResult.IsErr() {
transformedPromise.Complete(originalResult)
return
}
// Otherwise, the original future completed successfully. Apply the
// transformation function to its result.
originalResult.WhenOk(func(res T) {
newValue := fApply(res)
transformedPromise.Complete(fn.Ok(newValue))
})
}()
return transformedPromise.Future()
}
// OnComplete registers a function to be called when the result is ready. If the
// passed context is cancelled before the future completes, the callback
// function (cFunc) will be invoked with the context's error. The callback is
// executed in a new goroutine, so it does not block the completion path of the
// original future.
func (f *futureImpl[T]) OnComplete(ctx context.Context,
cFunc func(fn.Result[T])) {
go func() {
// Await the original future's result, respecting the passed
// context for cancellation.
result := f.Await(ctx)
// Call the callback function with the result.
cFunc(result)
}()
}
+467
View File
@@ -0,0 +1,467 @@
package actor
import (
"context"
"fmt"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/stretchr/testify/require"
"pgregory.net/rapid"
)
// TestFutureAwaitContextCancellation tests that Await respects context
// cancellation if the context is cancelled before the future resolves.
func TestFutureAwaitContextCancellation(t *testing.T) {
t.Parallel()
rapid.Check(t, func(t *rapid.T) {
// Test cancellation when the Await context is cancelled via
// context.Cancel. The underlying future will not be completed, allowing
// us to test the cancellation path of Await.
prom1 := NewPromise[int]()
fut1 := prom1.Future()
ctx1, cancel1 := context.WithCancel(context.Background())
// We'll cancel the future immediately after creating it.
cancel1()
result1 := fut1.Await(ctx1)
require.True(t, result1.IsErr())
require.ErrorIs(
t, result1.Err(), context.Canceled,
"await with immediate cancel",
)
// Test cancellation when the Await context times out. The
// underlying future will also not be completed.
prom2 := NewPromise[int]()
fut2 := prom2.Future()
// Use a very short timeout that will trigger.
ctx2, cancel2 := context.WithTimeout(
context.Background(), 1*time.Nanosecond,
)
defer cancel2()
// Await the future; it should fall through to the timeout
// because the future itself is not completed.
result2 := fut2.Await(ctx2)
require.True(t, result2.IsErr())
require.ErrorIs(
t, result2.Err(), context.DeadlineExceeded,
"await with timeout",
)
})
}
// TestFutureAwaitFutureCompletes tests that Await returns the future's
// result if the context is not cancelled before the future resolves.
func TestFutureAwaitFutureCompletes(t *testing.T) {
t.Parallel()
rapid.Check(t, func(t *rapid.T) {
valToSet := rapid.Int().Draw(t, "valToSet")
// With a 50% chance, configure the test to complete the future
// with an error instead of a successful value.
var errToSet error
if rapid.Bool().Draw(t, "have_error") {
errToSet = fmt.Errorf("err")
}
promise := NewPromise[int]()
fut := promise.Future()
// Use a background context for Await, as we expect the future
// to complete normally.
ctx := context.Background()
// Complete the future in a separate goroutine to simulate an
// asynchronous operation.
go func() {
if errToSet != nil {
promise.Complete(fn.Err[int](errToSet))
} else {
promise.Complete(fn.Ok(valToSet))
}
}()
// Now we'll wait for the future to complete, then verify below
// that the result (value or error) is as expected.
result := fut.Await(ctx)
if errToSet != nil {
// If an error was set, verify that Await returns that
// specific error.
require.True(t, result.IsErr())
require.ErrorIs(
t, result.Err(), errToSet,
"await with error",
)
} else {
// If no error was set, verify that Await returns the
// correct value.
require.False(t, result.IsErr(), "await with value")
result.WhenOk(func(val int) {
require.Equal(
t, valToSet, val, "await with value",
)
})
}
})
}
// TestFutureThenApplyContextCancellation tests that ThenApply respects its
// context, yielding a context error if cancelled before the original future
// completes.
func TestFutureThenApplyContextCancellation(t *testing.T) {
t.Parallel()
rapid.Check(t, func(t *rapid.T) {
// The original future will not be completed in this test case,
// allowing us to specifically test the cancellation behavior of
// the context passed to ThenApply.
originalPromise := NewPromise[int]()
originalFut := originalPromise.Future()
// Create a context for ThenApply and cancel it immediately.
ctxApply, cancelApply := context.WithCancel(
context.Background(),
)
cancelApply()
var transformCalled atomic.Bool
transform := func(i int) int {
transformCalled.Store(true)
return i * 2
}
// Register the transformation. The ThenApply operation itself
// will start a goroutine to await the originalFut.
newFut := originalFut.ThenApply(ctxApply, transform)
// Await the new (transformed) future. Use a background context
// for this Await to isolate the test to the cancellation of
// ctxApply.
result := newFut.Await(context.Background())
require.True(t, result.IsErr())
require.ErrorIs(
t, result.Err(), context.Canceled,
"ThenApply with cancelled context",
)
require.False(
t, transformCalled.Load(),
"ThenApply transform function called despite "+
"context cancellation",
)
})
}
// TestFutureThenApplyOriginalFutureCompletes tests ThenApply's behavior when
// the original future completes (with a value or error) before ThenApply's
// context is cancelled.
func TestFutureThenApplyOriginalFutureCompletes(t *testing.T) {
t.Parallel()
rapid.Check(t, func(t *rapid.T) {
initialVal := rapid.Int().Draw(t, "initialVal")
// Configure whether the original future completes with an error
// or a successful value.
var originalErr error
if rapid.Bool().Draw(t, "have_error") {
originalErr = fmt.Errorf("original error")
}
originalPromise := NewPromise[int]()
originalFut := originalPromise.Future()
// Create a context for ThenApply that should not cancel before
// the original future completes.
ctxApply, cancelApply := context.WithTimeout(
context.Background(), 50*time.Millisecond,
)
defer cancelApply()
var transformCalled atomic.Bool
transform := func(i int) int {
transformCalled.Store(true)
return i * 2
}
newFut := originalFut.ThenApply(ctxApply, transform)
// Complete the original future in a separate goroutine to
// simulate asynchrony.
go func() {
if originalErr != nil {
originalPromise.Complete(
fn.Err[int](originalErr),
)
} else {
originalPromise.Complete(fn.Ok(initialVal))
}
}()
// Await our new future which transforms the original future's
// result. Use a background context for this Await.
result := newFut.Await(context.Background())
if originalErr != nil {
// If the original future had an error, the transformed
// future should also yield that same error.
require.True(t, result.IsErr())
require.ErrorIs(
t, result.Err(), originalErr,
"ThenApply with original error",
)
require.False(
t, transformCalled.Load(),
"ThenApply transform function called despite "+
"original future having an error",
)
} else {
// If the original future completed successfully, the
// transformed future should contain the transformed value.
require.False(
t, result.IsErr(),
"ThenApply with original value",
)
require.True(
t, transformCalled.Load(),
"ThenApply transform function not called for "+
"successful original future",
)
result.WhenOk(func(val int) {
expectedTransformedVal := initialVal * 2
require.Equal(
t, expectedTransformedVal, val,
"ThenApply with original value",
)
})
}
})
}
// TestFutureOnCompleteContextCancellation tests that OnComplete's callback
// receives a context error if its context is cancelled before the future
// completes.
func TestFutureOnCompleteContextCancellation(t *testing.T) {
t.Parallel()
rapid.Check(t, func(t *rapid.T) {
// The original future will not complete in this test, allowing
// us to focus on the cancellation of OnComplete's context.
originalPromise := NewPromise[int]()
originalFut := originalPromise.Future()
// Create a context for OnComplete and cancel it immediately to
// simulate a premature cancellation.
ctxComplete, cancelComplete := context.WithCancel(
context.Background(),
)
cancelComplete()
var wg sync.WaitGroup
wg.Add(1)
var (
callbackInvoked atomic.Bool
callbackResultValue fn.Result[int]
// mu is a mutex to protect callbackResultValue as it's
// written by the callback goroutine and read by the
// test goroutine.
mu sync.Mutex
)
// Register an OnComplete callback. The callback itself runs in
// a new goroutine started by OnComplete.
originalFut.OnComplete(ctxComplete, func(res fn.Result[int]) {
mu.Lock()
callbackResultValue = res
mu.Unlock()
callbackInvoked.Store(true)
wg.Done()
})
// Use a wait group and a channel to wait for the callback to
// be invoked.
waitChan := make(chan struct{})
go func() {
wg.Wait()
close(waitChan)
}()
select {
// The callback should be invoked, even if with a context error.
case <-waitChan:
case <-time.After(50 * time.Millisecond):
require.Fail(
t, "OnComplete callback timed out waiting "+
"for execution after context cancel",
)
}
require.True(
t, callbackInvoked.Load(),
"OnComplete callback not invoked",
)
mu.Lock()
defer mu.Unlock()
// Verify that the callback received a context.Canceled error
// because its context (ctxComplete) was cancelled.
require.True(t, callbackResultValue.IsErr())
require.ErrorIs(
t, callbackResultValue.Err(), context.Canceled,
"OnComplete with cancelled context",
)
})
}
// TestFutureOnCompleteFutureCompletes tests OnComplete's behavior when the
// future completes (with value or error) before its context is cancelled.
func TestFutureOnCompleteFutureCompletes(t *testing.T) {
t.Parallel()
rapid.Check(t, func(t *rapid.T) {
valToSet := rapid.Int().Draw(t, "valToSet")
// Configure whether the original future completes with an error
// or a successful value.
var originalErr error
if rapid.Bool().Draw(t, "have_error") {
originalErr = fmt.Errorf("original error")
}
originalPromise := NewPromise[int]()
originalFut := originalPromise.Future()
// Use a background context for OnComplete, as we expect the
// future to complete normally.
ctxComplete := context.Background()
var wg sync.WaitGroup
wg.Add(1)
var (
callbackInvoked atomic.Bool
callbackResultValue fn.Result[int]
mu sync.Mutex
)
// Register an OnComplete callback. This callback will execute
// once the originalFut completes.
originalFut.OnComplete(ctxComplete, func(res fn.Result[int]) {
mu.Lock()
callbackResultValue = res
mu.Unlock()
callbackInvoked.Store(true)
wg.Done()
})
// Complete the original future in a separate goroutine to
// simulate an asynchronous operation.
go func() {
if originalErr != nil {
originalPromise.Complete(
fn.Err[int](originalErr),
)
} else {
originalPromise.Complete(fn.Ok(valToSet))
}
}()
// Use a wait group and a channel to wait for the callback's
// execution.
waitChan := make(chan struct{})
go func() {
wg.Wait()
close(waitChan)
}()
select {
// The callback should be invoked as the future completes.
case <-waitChan:
case <-time.After(50 * time.Millisecond):
require.Fail(
t, "OnComplete callback timed out waiting "+
"for execution",
)
}
require.True(t, callbackInvoked.Load())
mu.Lock()
defer mu.Unlock()
// Verify that the callback received the correct result (either
// the error or the value from the completed future).
if originalErr != nil {
require.True(t, callbackResultValue.IsErr())
require.ErrorIs(
t, callbackResultValue.Err(), originalErr,
"OnComplete with error",
)
} else {
require.False(
t, callbackResultValue.IsErr(),
"OnComplete with value",
)
callbackResultValue.WhenOk(func(val int) {
require.Equal(
t, valToSet, val,
"OnComplete with value",
)
})
}
})
}
// TestPromiseCompleteIdempotency verifies that calling Complete on a Promise
// multiple times is safe and only the first completion takes effect. Subsequent
// calls should return false and not alter the future's result.
func TestPromiseCompleteIdempotency(t *testing.T) {
t.Parallel()
promise := NewPromise[string]()
future := promise.Future()
// First completion should succeed.
firstResult := fn.Ok("first-value")
ok := promise.Complete(firstResult)
require.True(t, ok, "first Complete should return true")
// Second completion with a different value should be ignored.
secondResult := fn.Ok("second-value")
ok = promise.Complete(secondResult)
require.False(t, ok, "second Complete should return false")
// Third completion with an error should also be ignored.
thirdResult := fn.Err[string](fmt.Errorf("should be ignored"))
ok = promise.Complete(thirdResult)
require.False(t, ok, "third Complete should return false")
// The future should contain the first value.
result := future.Await(context.Background())
require.False(t, result.IsErr(), "future should not be an error")
result.WhenOk(func(val string) {
require.Equal(
t, "first-value", val,
"future should contain the first completion value",
)
})
}
+19
View File
@@ -0,0 +1,19 @@
module github.com/lightningnetwork/lnd/actor
go 1.24.9
require (
github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084
github.com/lightningnetwork/lnd/fn/v2 v2.0.8
github.com/stretchr/testify v1.8.1
pgregory.net/rapid v1.2.0
)
require (
github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
golang.org/x/exp v0.0.0-20231226003508-02704c960a9b // indirect
golang.org/x/sync v0.7.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
+29
View File
@@ -0,0 +1,29 @@
github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c h1:4HxD1lBUGUddhzgaNgrCPsFWd7cGYNpeFUgd9ZIgyM0=
github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c/go.mod h1:w7xnGOhwT3lmrS4H3b/D1XAXxvh+tbhUm8xeHN2y3TQ=
github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084 h1:y3bvkt8ki0KX35eUEU8XShRHusz1S+55QwXUTmxn888=
github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084/go.mod h1:XItGUfVOxotJL8kkuk2Hj3EVow5KCugXl3wWfQ6K0AE=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/lightningnetwork/lnd/fn/v2 v2.0.8 h1:r2SLz7gZYQPVc3IZhU82M66guz3Zk2oY+Rlj9QN5S3g=
github.com/lightningnetwork/lnd/fn/v2 v2.0.8/go.mod h1:TOzwrhjB/Azw1V7aa8t21ufcQmdsQOQMDtxVOQWNl8s=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
golang.org/x/exp v0.0.0-20231226003508-02704c960a9b h1:kLiC65FbiHWFAOu+lxwNPujcsl8VYyTYYEZnsOO1WK4=
golang.org/x/exp v0.0.0-20231226003508-02704c960a9b/go.mod h1:iRJReGqOEeBhDZGkGbynYwcHlctCvnjTYIamk7uXpHI=
golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk=
pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=
+119
View File
@@ -0,0 +1,119 @@
package actor
import (
"context"
"fmt"
"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
// result is available (OnComplete).
type Future[T any] interface {
// Await blocks until the result is available or the context is
// cancelled, then returns it.
Await(ctx context.Context) fn.Result[T]
// ThenApply registers a function to transform the result of a future.
// The original future is not modified, a new instance of the future is
// returned. If the passed context is cancelled while waiting for the
// original future to complete, the new future will complete with the
// context's error.
ThenApply(ctx context.Context, fn func(T) T) Future[T]
// OnComplete registers a function to be called when the result of the
// future is ready. If the passed context is cancelled before the future
// completes, the callback function will be invoked with the context's
// error.
OnComplete(ctx context.Context, fn func(fn.Result[T]))
}
// Promise is an interface that allows for the completion of an associated
// Future. It provides a way to set the result of an asynchronous operation.
// The producer of an asynchronous result uses a Promise to set the outcome,
// while consumers use the associated Future to retrieve it.
type Promise[T any] interface {
// Future returns the Future interface associated with this Promise.
// Consumers can use this to Await the result or register callbacks.
Future() Future[T]
// Complete attempts to set the result of the future. It returns true if
// this call successfully set the result (i.e., it was the first to
// 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]
}
+12
View File
@@ -0,0 +1,12 @@
package actor
import "github.com/btcsuite/btclog/v2"
// log is a logger that is initialized as disabled. This means the package will
// not perform any logging by default until a logger is set.
var log = btclog.Disabled
// UseLogger uses a specified Logger to output package logging info.
func UseLogger(logger btclog.Logger) {
log = logger
}
+154
View File
@@ -0,0 +1,154 @@
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")
// Compile-time assertion that Router satisfies the ActorRef interface.
var _ ActorRef[Message, any] = (*Router[Message, any])(nil)
// 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)
} else {
log.Warnf("Router(%s): message %s dropped "+
"(no actors available, no DLO configured)",
r.serviceKey.name, msg.MessageType())
}
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 + ")"
}
+421
View File
@@ -0,0 +1,421 @@
package actor
import (
"context"
"errors"
"fmt"
"sync"
"github.com/lightningnetwork/lnd/fn/v2"
)
// stoppable defines an interface for components that can be stopped.
// This is unexported as it's an internal detail of ActorSystem for managing
// actors that need to be shut down.
type stoppable interface {
Stop()
}
// SystemConfig holds configuration parameters for the ActorSystem.
type SystemConfig struct {
// MailboxCapacity is the default capacity for actor mailboxes.
MailboxCapacity int
}
// DefaultConfig returns a default configuration for the ActorSystem.
// The default mailbox capacity of 100 means each actor can buffer up to 100
// pending messages (envelopes). Each envelope holds a message and an optional
// promise pointer, so the memory overhead per actor is roughly proportional to
// the size of the messages being sent multiplied by this capacity.
func DefaultConfig() SystemConfig {
return SystemConfig{
MailboxCapacity: 100,
}
}
// ActorSystem manages the lifecycle of actors and provides coordination
// services such as a receptionist for actor discovery and a dead letter office
// for undeliverable messages. It also handles the graceful shutdown of all
// managed actors.
type ActorSystem struct {
// receptionist is used for actor discovery.
receptionist *Receptionist
// actors stores all actors managed by the system, keyed by their ID.
// This includes the deadLetterActor.
actors map[string]stoppable
// deadLetterActor handles undeliverable messages.
deadLetterActor ActorRef[Message, any]
// config holds the system-wide configuration.
config SystemConfig
// mu protects the 'actors' map.
mu sync.RWMutex
// ctx is the main context for the actor system.
ctx context.Context
// cancel cancels the main system context.
cancel context.CancelFunc
}
// NewActorSystem creates a new actor system using the default configuration.
func NewActorSystem() *ActorSystem {
return NewActorSystemWithConfig(DefaultConfig())
}
// NewActorSystemWithConfig creates a new actor system with custom configuration
func NewActorSystemWithConfig(config SystemConfig) *ActorSystem {
ctx, cancel := context.WithCancel(context.Background())
// Initialize the core ActorSystem components.
system := &ActorSystem{
receptionist: newReceptionist(),
config: config,
actors: make(map[string]stoppable),
ctx: ctx,
cancel: cancel,
}
// Define the behavior for the dead letter actor. It logs undeliverable
// messages and returns an error.
deadLetterBehavior := NewFunctionBehavior(
func(ctx context.Context, msg Message) fn.Result[any] {
log.Warnf("Dead letter received: message type=%s",
msg.MessageType())
return fn.Err[any](errors.New(
"message undeliverable: " + msg.MessageType(),
))
},
)
// Create the raw dead letter actor (*Actor instance). The DLO's own DLO
// reference is nil to prevent loops if messages to the DLO itself fail.
deadLetterActorCfg := ActorConfig[Message, any]{
ID: "dead-letters",
Behavior: deadLetterBehavior,
DLO: nil,
MailboxSize: config.MailboxCapacity,
}
deadLetterRawActor, err := NewActor[Message, any](deadLetterActorCfg)
if err != nil {
// This should never happen since we control the DLO config.
panic("failed to create dead letter actor: " + err.Error())
}
deadLetterRawActor.Start()
system.deadLetterActor = deadLetterRawActor.Ref()
// Add the raw actor to the map of stoppable actors. No lock needed here
// as 'system' is not yet accessible concurrently.
system.actors[deadLetterRawActor.id] = deadLetterRawActor
// The system is now fully initialized and ready.
return system
}
// RegisterWithSystem creates an actor with the given ID, service key, and
// behavior within the specified ActorSystem. It starts the actor, adds it to
// the system's management, registers it with the receptionist using the
// provided key, and returns its ActorRef.
func RegisterWithSystem[M Message, R any](as *ActorSystem, id string,
key ServiceKey[M, R],
behavior ActorBehavior[M, R]) (ActorRef[M, R], error) {
actorCfg := ActorConfig[M, R]{
ID: id,
Behavior: behavior,
DLO: as.deadLetterActor,
MailboxSize: as.config.MailboxCapacity,
}
// Check for duplicate actor ID before creating the actor.
as.mu.Lock()
if _, exists := as.actors[id]; exists {
as.mu.Unlock()
return nil, fmt.Errorf("%w: %s", ErrDuplicateActorID, id)
}
actorInstance, err := NewActor(actorCfg)
if err != nil {
as.mu.Unlock()
return nil, err
}
actorInstance.Start()
// Add the actor instance to the system's list of stoppable actors.
as.actors[actorInstance.id] = actorInstance
as.mu.Unlock()
log.Infof("ActorSystem: registered actor %s with service key %s",
id, key.name)
// Register the actor's reference with the receptionist under the given
// service key, making it discoverable by other parts of the system.
RegisterWithReceptionist(as.receptionist, key, actorInstance.Ref())
return actorInstance.Ref(), nil
}
// Receptionist returns the system's receptionist, which can be used for
// actor service discovery (finding actors by ServiceKey).
func (as *ActorSystem) Receptionist() *Receptionist {
return as.receptionist
}
// DeadLetters returns a reference to the system's dead letter actor. Messages
// that cannot be delivered to their intended recipient (e.g., if an Ask
// context is cancelled before enqueuing) may be routed here if not otherwise
// handled.
func (as *ActorSystem) DeadLetters() ActorRef[Message, any] {
return as.deadLetterActor
}
// Shutdown gracefully stops the actor system. It iterates through all managed
// actors, including the dead letter actor, and calls their Stop method.
// After initiating the stop for all actors, it cancels the main system context.
// This method is safe for concurrent use.
func (as *ActorSystem) Shutdown() error {
log.Infof("ActorSystem: initiating shutdown")
// Create a slice of actors to stop. This avoids holding the lock while
// calling Stop() on each actor, and includes the dead letter actor.
var actorsToStop []stoppable
as.mu.RLock()
for _, actor := range as.actors {
actorsToStop = append(actorsToStop, actor)
}
as.mu.RUnlock()
// Notify all managed actors to stop. Actor.Stop() is non-blocking.
// Each actor's Stop method will cancel its internal context, leading
// to the termination of its processing goroutine.
for _, actor := range actorsToStop {
actor.Stop()
}
// Clear the actors map after initiating their shutdown.
as.mu.Lock()
as.actors = nil
as.mu.Unlock()
// Finally cancel the main context
// This signals to any other components observing the system's context
// that shutdown has been initiated.
as.cancel()
return nil
}
// StopAndRemoveActor stops a specific actor by its ID and removes it from the
// ActorSystem's management. It returns true if the actor was found and stopped,
// false otherwise.
func (as *ActorSystem) StopAndRemoveActor(id string) bool {
as.mu.Lock()
defer as.mu.Unlock()
actorToStop, exists := as.actors[id]
if !exists {
return false
}
// Stop the actor. This is non-blocking.
actorToStop.Stop()
// Remove from the system's management.
delete(as.actors, id)
return true
}
// UnregisterFromReceptionist removes an actor reference from a service key in
// the given receptionist. It returns true if the reference was found and
// removed, and false otherwise. This is a package-level generic function
// because methods cannot have their own type parameters in Go.
func UnregisterFromReceptionist[M Message, R any](r *Receptionist,
key ServiceKey[M, R], refToRemove ActorRef[M, R]) bool {
r.mu.Lock()
defer r.mu.Unlock()
refs, exists := r.registrations[key.name]
if !exists {
return false
}
found := false
// Build a new slice containing only the references that are not the one
// to be removed.
newRefs := make([]any, 0, max(0, len(refs)-1))
for _, itemInSlice := range refs {
// Try to assert the item from the slice to the specific
// ActorRef[M,R] type we are trying to remove.
if specificActorRef, ok := itemInSlice.(ActorRef[M, R]); ok {
// If the type assertion is successful and it's the one
// we want to remove, mark as found and skip adding it
// to newRefs.
if specificActorRef == refToRemove {
found = true
continue
}
}
newRefs = append(newRefs, itemInSlice)
}
if !found {
return false
}
// If the new list of references is empty, remove the key from the map.
// Otherwise, update the map with the new slice.
if len(newRefs) == 0 {
delete(r.registrations, key.name)
} else {
r.registrations[key.name] = newRefs
}
return true
}
// ServiceKey is a type-safe identifier used for registering and discovering
// actors via the Receptionist. The generic type parameters M (Message) and R
// (Response) ensure that only actors handling compatible message/response types
// are associated with and retrieved for this key.
type ServiceKey[M Message, R any] struct {
name string
}
// NewServiceKey creates a new service key with the given name. The name is used
// as the lookup key within the Receptionist.
func NewServiceKey[M Message, R any](name string) ServiceKey[M, R] {
return ServiceKey[M, R]{name: name}
}
// Spawn registers an actor for this service key within the given ActorSystem.
// It's a convenience method that calls RegisterWithSystem, starting the actor
// and registering it with the receptionist.
func (sk ServiceKey[M, R]) Spawn(as *ActorSystem, id string,
behavior ActorBehavior[M, R]) (ActorRef[M, R], error) {
return RegisterWithSystem(as, id, sk, behavior)
}
// Unregister removes an actor reference associated with this service key from
// the ActorSystem's receptionist and also stops the actor.
// It returns true if the actor was successfully unregistered from the
// receptionist AND successfully stopped and removed from the system's
// management. Otherwise, it returns false.
func (sk ServiceKey[M, R]) Unregister(as *ActorSystem,
refToRemove ActorRef[M, R]) bool {
unregisteredFromReceptionist := UnregisterFromReceptionist(
as.Receptionist(), sk, refToRemove,
)
// If not found in receptionist, no need to try stopping.
if !unregisteredFromReceptionist {
return false
}
// Attempt to stop and remove the actor from the system.
stoppedAndRemoved := as.StopAndRemoveActor(refToRemove.ID())
return unregisteredFromReceptionist && stoppedAndRemoved
}
// UnregisterAll finds all actor references associated with this service key in
// the ActorSystem's receptionist. For each found actor, it attempts to stop it
// and remove it from system management, and also unregisters it from the
// receptionist.
func (sk ServiceKey[M, R]) UnregisterAll(as *ActorSystem) int {
// First find all the refs that match this service key.
refsFound := FindInReceptionist(as.Receptionist(), sk)
actorsStoppedCount := 0
for _, ref := range refsFound {
// Attempt to stop and remove the actor from the system's active
// management. This is the primary action to deactivate the
// actor. If StopAndRemoveActor returns true, it means an active
// actor was found in the system's `actors` map and was stopped.
if as.StopAndRemoveActor(ref.ID()) {
actorsStoppedCount++
}
// Regardless of whether the actor was actively managed by the
// system (i.e., found in as.actors), attempt to unregister its
// reference from the receptionist. This helps clean up any
// potentially stale entries in the receptionist if an actor was
// removed from the system's management without also being
// unregistered from the receptionist.
UnregisterFromReceptionist(as.Receptionist(), sk, ref)
}
return actorsStoppedCount
}
// Receptionist provides service discovery for actors. Actors can be registered
// under a ServiceKey and later discovered by other actors or system components.
type Receptionist struct {
// registrations stores ActorRef instances, keyed by ServiceKey.name.
registrations map[string][]any
// mu protects access to registrations.
mu sync.RWMutex
}
// newReceptionist creates a new Receptionist instance.
func newReceptionist() *Receptionist {
return &Receptionist{
registrations: make(map[string][]any),
}
}
// RegisterWithReceptionist registers an actor with a service key in the given
// receptionist. This is a package-level generic function because methods
// cannot have their own type parameters in Go (as of the current version).
// It appends the actor reference to the list associated with the key's name.
func RegisterWithReceptionist[M Message, R any](r *Receptionist,
key ServiceKey[M, R], ref ActorRef[M, R]) {
r.mu.Lock()
defer r.mu.Unlock()
// Initialize the slice for this key if it's the first registration.
if _, exists := r.registrations[key.name]; !exists {
r.registrations[key.name] = make([]any, 0)
}
r.registrations[key.name] = append(r.registrations[key.name], ref)
}
// FindInReceptionist returns all actors registered with a service key in the
// given receptionist. This is a package-level generic function because methods
// cannot have their own type parameters. It performs a type assertion to ensure
// that only ActorRefs matching the ServiceKey's generic types (M, R) are
// returned, providing type safety.
func FindInReceptionist[M Message, R any](r *Receptionist,
key ServiceKey[M, R]) []ActorRef[M, R] {
r.mu.RLock()
defer r.mu.RUnlock()
if refs, exists := r.registrations[key.name]; exists {
typedRefs := make([]ActorRef[M, R], 0, len(refs))
for _, ref := range refs {
// Make sure that the reference is of the correct type.
// This type assertion is crucial for type safety, ensuring
// that the returned ActorRefs match the expected M and R.
if typedRef, ok := ref.(ActorRef[M, R]); ok {
typedRefs = append(typedRefs, typedRef)
}
}
return typedRefs
}
return nil
}
+958
View File
@@ -0,0 +1,958 @@
package actor
import (
"context"
"errors"
"fmt"
"sync/atomic"
"testing"
"time"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/stretchr/testify/require"
)
// TestActorSystemNewActorSystem verifies the basic initialization of an
// ActorSystem, including its default DLO.
func TestActorSystemNewActorSystem(t *testing.T) {
t.Parallel()
as := NewActorSystem()
require.NotNil(t, as, "newActorSystem should not return nil")
require.NotNil(t, as.Receptionist(), "receptionist should not be nil")
require.NotNil(t, as.DeadLetters(), "deadLetters should not be nil")
require.Equal(t, "dead-letters", as.DeadLetters().ID(), "dLO ID mismatch")
// Test the DLO's behavior (it should return an error for Ask).
testDLOMsg := newTestMsg("to-dlo")
future := as.DeadLetters().Ask(context.Background(), testDLOMsg)
result := future.Await(context.Background())
// We should get back an error for asks.
require.True(
t, result.IsErr(), "system DLO should return an error on Ask",
)
expectedErrStr := "message undeliverable: " + testDLOMsg.MessageType()
require.EqualError(
t, result.Err(), expectedErrStr, "dLO error message mismatch",
)
// Shutdown the system to clean up resources.
err := as.Shutdown()
require.NoError(t, err, "actorSystem shutdown failed")
}
// TestActorSystemRegisterWithSystem verifies actor registration, lifecycle
// management within the system.
func TestActorSystemRegisterWithSystem(t *testing.T) {
t.Parallel()
as := NewActorSystem()
defer func() {
err := as.Shutdown()
require.NoError(t, err)
}()
actorID := "test-actor-sys-reg"
serviceKey := NewServiceKey[*testMsg, string]("test-service")
// Using echoBehavior from actor_test.go (implicitly available)
beh := newEchoBehavior(t, 0)
// We'll start off by registering the actor.
actorRef, err := RegisterWithSystem(as, actorID, serviceKey, beh)
require.NoError(t, err)
require.NotNil(t, actorRef, "registerWithSystem should return a valid ActorRef")
require.Equal(t, actorID, actorRef.ID(), "registered actor ID mismatch")
// The actor should be found in the receptionist.
foundActors := FindInReceptionist(as.Receptionist(), serviceKey)
require.Len(t, foundActors, 1, "actor not found in receptionist")
require.Equal(t, actorRef, foundActors[0], "incorrect actor in receptionist")
// Next, we'll send out a simple tell, using our reply channel to make
// sure it's actually processed.
msgData := "hello-system-actor"
replyChan := make(chan string, 1)
actorRef.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")
// Stop the actor through the system.
stopped := as.StopAndRemoveActor(actorID)
require.True(t, stopped, "StopAndRemoveActor failed")
// Wait for actor to fully stop.
time.Sleep(50 * time.Millisecond)
// Send a message to the now-stopped actor's ref. This should go to the
// system's DLO.
afterStopMsg := newTestMsg("after-stop-to-dlo")
require.NotPanics(t, func() {
actorRef.Tell(context.Background(), afterStopMsg)
}, "tell to stopped actor should not panic")
}
// TestActorSystemShutdown verifies that all actors are stopped and the system
// context is cancelled upon shutdown.
func TestActorSystemShutdown(t *testing.T) {
t.Parallel()
as := NewActorSystem()
// We'll start by making 3 new actors, each with a unique ID.
numActors := 3
actorRefs := make([]ActorRef[*testMsg, string], numActors)
for i := 0; i < numActors; i++ {
actorID := fmt.Sprintf("shutdown-test-actor-%d", i)
key := NewServiceKey[*testMsg, string](
fmt.Sprintf("service-%d", i),
)
beh := newEchoBehavior(t, 0)
ref, regErr := RegisterWithSystem(as, actorID, key, beh)
require.NoError(t, regErr)
actorRefs[i] = ref
}
// We'll now send a message to each actor to ensure that they're
// running.
for i, ref := range actorRefs {
future := ref.Ask(
context.Background(),
newTestMsg(fmt.Sprintf("ping-%d", i)),
)
ctxAwait, cancelAwait := context.WithTimeout(
context.Background(), time.Second,
)
res := future.Await(ctxAwait)
cancelAwait()
require.False(
t, res.IsErr(),
"actor %d failed to respond before shutdown: %v",
i, res.Err(),
)
}
// Next, trigger a shutdown, and assert that the done channel gets
// closed.
err := as.Shutdown()
require.NoError(t, err, "actorSystem shutdown failed")
// Check if the system context is done using RecvOrTimeout with a zero
// timeout for a non-blocking check.
_, err = fn.RecvOrTimeout(as.ctx.Done(), time.Millisecond*100)
require.NoError(t, err, "actorSystem context not cancelled after shutdown")
// We'll now try to send a message to each of the actors, this should
// result in an error.
for i, ref := range actorRefs {
future := ref.Ask(
context.Background(),
newTestMsg(fmt.Sprintf("ping-after-shutdown-%d", i)),
)
res := future.Await(context.Background())
require.True(
t, res.IsErr(),
"actor %d Ask should fail after shutdown", i,
)
require.ErrorIs(t, res.Err(), ErrActorTerminated)
}
as.mu.RLock()
require.Nil(t, as.actors, "actors map should be nil after shutdown")
as.mu.RUnlock()
// Once shutdown, we shouldn't be able to send to the DLO either.
dloRef := as.DeadLetters()
futureDLO := dloRef.Ask(
context.Background(), newTestMsg("ping-dlo-after-shutdown"),
)
resDLO := futureDLO.Await(context.Background())
require.True(
t, resDLO.IsErr(), "DLO Ask should fail after system shutdown",
)
require.ErrorIs(
t, resDLO.Err(), ErrActorTerminated,
)
}
// TestActorSystemStopAndRemoveActor verifies specific actor stopping and
// removal.
func TestActorSystemStopAndRemoveActor(t *testing.T) {
t.Parallel()
as := NewActorSystem()
defer func() {
err := as.Shutdown()
require.NoError(t, err)
}()
// Make some actor IDs, then unique service keys, then use that to
// register two actors.
actor1ID := "actor-to-stop"
actor2ID := "actor-to-keep"
key1 := NewServiceKey[*testMsg, string]("service1")
key2 := NewServiceKey[*testMsg, string]("service2")
beh := newEchoBehavior(t, 0)
ref1, err := RegisterWithSystem(as, actor1ID, key1, beh)
require.NoError(t, err)
ref2, err := RegisterWithSystem(as, actor2ID, key2, beh)
require.NoError(t, err)
// If we remove one actor, then try to send to it, we should get an
// error.
stopped := as.StopAndRemoveActor(actor1ID)
require.True(t, stopped, "failed to stop and remove actor1")
future1 := ref1.Ask(context.Background(), newTestMsg("ping-actor1"))
res1 := future1.Await(context.Background())
require.True(t, res1.IsErr(), "actor1 should be stopped")
require.ErrorIs(t, res1.Err(), ErrActorTerminated)
as.mu.RLock()
_, exists := as.actors[actor1ID]
as.mu.RUnlock()
// The actor should no longer be found.
require.False(t, exists, "actor1 still in system's actor map")
// Make sure that we can still send messages to the existing actor.
future2 := ref2.Ask(
context.Background(), newTestMsg("ping-actor2"),
)
ctxAwait2, cancelAwait2 := context.WithTimeout(
context.Background(), time.Second,
)
res2 := future2.Await(ctxAwait2)
cancelAwait2()
require.False(
t, res2.IsErr(), "actor2 should still be running: %v",
res2.Err(),
)
res2.WhenOk(func(s string) {
require.Equal(t, "echo: ping-actor2", s)
})
stoppedNonExistent := as.StopAndRemoveActor("non-existent-actor")
require.False(
t, stoppedNonExistent, "stopping non-existent actor should "+
"return false",
)
}
// TestReceptionist covers basic registration, finding, and unregistration.
func TestReceptionist(t *testing.T) {
t.Parallel()
as := NewActorSystem()
defer func() {
err := as.Shutdown()
require.NoError(t, err)
}()
receptionist := as.Receptionist()
key1 := NewServiceKey[*testMsg, string]("key1")
key2 := NewServiceKey[*testMsg, string]("key2")
key1Again := NewServiceKey[*testMsg, string]("key1")
// Register 3 actor instance using the service keys we created above.
beh := newEchoBehavior(t, 0)
actor1Ref, err := RegisterWithSystem(as, "actor1-rec", key1, beh)
require.NoError(t, err)
actor2Ref, err := RegisterWithSystem(as, "actor2-rec", key1, beh)
require.NoError(t, err)
actor3Ref, err := RegisterWithSystem(as, "actor3-rec", key2, beh)
require.NoError(t, err)
// We should be able to find the actors we registered.
foundForKey1 := FindInReceptionist(receptionist, key1)
require.Len(t, foundForKey1, 2, "should find 2 actors for key1")
require.Contains(t, foundForKey1, actor1Ref)
require.Contains(t, foundForKey1, actor2Ref)
foundForKey1Again := FindInReceptionist(receptionist, key1Again)
require.ElementsMatch(t, foundForKey1, foundForKey1Again)
// Same goes for the second key we added.
foundForKey2 := FindInReceptionist(receptionist, key2)
require.Len(t, foundForKey2, 1, "should find 1 actor for key2")
require.Equal(t, actor3Ref, foundForKey2[0])
// We shouldn't be able to find a key we didn't add.
nonExistentKey := NewServiceKey[*testMsg, string]("non-existent")
foundForNonExistent := FindInReceptionist(receptionist, nonExistentKey)
require.Empty(t, foundForNonExistent)
// We should be able to unregister the actors we added.
unregistered := UnregisterFromReceptionist(
receptionist, key1, actor1Ref,
)
require.True(t, unregistered, "failed to unregister actor1Ref")
foundForKey1AfterUnreg := FindInReceptionist(receptionist, key1)
require.Len(t, foundForKey1AfterUnreg, 1)
require.Equal(t, actor2Ref, foundForKey1AfterUnreg[0])
// If we try to unregister the same actor again, it should fail.
unregisteredAgain := UnregisterFromReceptionist(receptionist, key1, actor1Ref)
require.False(t, unregisteredAgain)
unregisteredLast := UnregisterFromReceptionist(receptionist, key1, actor2Ref)
require.True(t, unregisteredLast)
foundForKey1AfterAllUnreg := FindInReceptionist(receptionist, key1)
require.Empty(t, foundForKey1AfterAllUnreg)
receptionist.mu.RLock()
_, exists := receptionist.registrations[key1.name]
receptionist.mu.RUnlock()
require.False(t, exists, "key1 should be removed from registrations map")
// Finally, if we use the wrong key, or one that doesn't exist, that
// should also fail.
unregisteredWrongKey := UnregisterFromReceptionist(receptionist, key1, actor3Ref)
require.False(t, unregisteredWrongKey)
unregisteredNonExistentKey := UnregisterFromReceptionist(receptionist, nonExistentKey, actor1Ref)
require.False(t, unregisteredNonExistentKey)
}
// TestServiceKeyMethods tests Spawn and Unregister methods on ServiceKey.
func TestServiceKeyMethods(t *testing.T) {
t.Parallel()
as := NewActorSystem()
defer func() {
err := as.Shutdown()
require.NoError(t, err)
}()
key := NewServiceKey[*testMsg, string]("sk-service")
beh := newEchoBehavior(t, 0)
// Attempt to spawn a new actor using the service key and desired
// behavior.
actorRef, err := key.Spawn(as, "actor-sk-spawn", beh)
require.NoError(t, err)
require.NotNil(t, actorRef)
require.Equal(t, "actor-sk-spawn", actorRef.ID())
// We should be able to find the actor in the receptionist.
found := FindInReceptionist(as.Receptionist(), key)
require.Len(t, found, 1)
require.Equal(t, actorRef, found[0])
as.mu.RLock()
_, sysExists := as.actors[actorRef.ID()]
as.mu.RUnlock()
require.True(t, sysExists)
// Next, try to unregister the actor using the service key.
success := key.Unregister(as, actorRef)
require.True(t, success, "serviceKey.Unregister failed")
// The actor should no longer be found in the receptionist.
foundAfter := FindInReceptionist(as.Receptionist(), key)
require.Empty(t, foundAfter)
as.mu.RLock()
_, sysExistsAfter := as.actors[actorRef.ID()]
as.mu.RUnlock()
require.False(t, sysExistsAfter)
// If we try to send a message to the actor after unregistering it, then
// we should get an error.
future := actorRef.Ask(context.Background(), newTestMsg("ping"))
res := future.Await(context.Background())
require.True(t, res.IsErr() && errors.Is(res.Err(), ErrActorTerminated))
successAgain := key.Unregister(as, actorRef)
require.False(t, successAgain)
otherSys := NewActorSystem() // Create a different actor system
defer func() {
err := otherSys.Shutdown()
require.NoError(t, err)
}()
// Create a dummy actor in otherSys of the correct generic type for the
// key. This actor won't be found in 'as', so Unregister should fail.
dummyBehOther := newEchoBehavior(t, 0)
dummyKeyOther := NewServiceKey[*testMsg, string]("dummy-other")
dummyActorRefOtherSys, err := RegisterWithSystem(
otherSys, "dummy-other-actor", dummyKeyOther, dummyBehOther,
)
require.NoError(t, err)
successNonMember := key.Unregister(as, dummyActorRefOtherSys)
require.False(t, successNonMember)
}
// TestServiceKeyUnregisterAll tests the UnregisterAll method on ServiceKey.
// It covers scenarios including basic unregistration of multiple actors,
// attempting to unregister with no actors present, unregistering actors for
// one key while leaving others intact, and the idempotency of the operation.
func TestServiceKeyUnregisterAll(t *testing.T) {
t.Parallel()
// Common setup for all sub-tests.
as := NewActorSystem()
defer func() {
err := as.Shutdown()
require.NoError(t, err, "ActorSystem shutdown failed.")
}()
// Common behavior for test actors used across sub-tests.
beh := newEchoBehavior(t, 0)
t.Run("unregister all multiple actors", func(st *testing.T) {
key1 := NewServiceKey[*testMsg, string]("sk-ua-key1")
actor1Key1, err := key1.Spawn(as, "actor1-k1-ua", beh)
require.NoError(st, err)
actor2Key1, err := key1.Spawn(as, "actor2-k1-ua", beh)
require.NoError(st, err)
// Verify they are registered in the receptionist.
foundActorsForKey1 := FindInReceptionist(
as.Receptionist(), key1,
)
require.Len(
st, foundActorsForKey1, 2,
"actors for key1 not in receptionist initially.",
)
// Verify they are in the system's actor map.
as.mu.RLock()
_, actor1Key1Exists := as.actors[actor1Key1.ID()]
_, actor2Key1Exists := as.actors[actor2Key1.ID()]
as.mu.RUnlock()
require.True(
st, actor1Key1Exists,
"actor1 for key1 not in system actors map initially.",
)
require.True(
st, actor2Key1Exists,
"actor2 for key1 not in system actors map initially.",
)
// Unregister all for key1.
stoppedCountKey1 := key1.UnregisterAll(as)
require.Equal(
st, 2, stoppedCountKey1,
"UnregisterAll for key1 returned incorrect count.",
)
// Verify they are unregistered from the receptionist.
foundActorsForKey1After := FindInReceptionist(
as.Receptionist(), key1,
)
require.Empty(
st, foundActorsForKey1After,
"actors for key1 still in receptionist after "+
"UnregisterAll.",
)
// Verify they are removed from system actors map.
as.mu.RLock()
_, actor1Key1ExistsAfter := as.actors[actor1Key1.ID()]
_, actor2Key1ExistsAfter := as.actors[actor2Key1.ID()]
as.mu.RUnlock()
require.False(
st, actor1Key1ExistsAfter,
"Actor1 for key1 still in system actors "+
"map after UnregisterAll.",
)
require.False(
st, actor2Key1ExistsAfter,
"Actor2 for key1 still in system actors "+
"map after UnregisterAll.",
)
// Verify actors are stopped.
resultActor1Key1 := actor1Key1.Ask(
context.Background(), newTestMsg("ping-k1-a1"),
).Await(context.Background())
require.True(
st, resultActor1Key1.IsErr(),
"Actor1 key1 Ask should fail after UnregisterAll.",
)
require.ErrorIs(
st, resultActor1Key1.Err(), ErrActorTerminated,
"Actor1 key1 not terminated with correct error.",
)
resultActor2Key1 := actor2Key1.Ask(
context.Background(), newTestMsg("ping-k1-a2"),
).Await(context.Background())
require.True(
st, resultActor2Key1.IsErr(),
"Actor2 key1 Ask should fail after UnregisterAll.",
)
require.ErrorIs(
st, resultActor2Key1.Err(), ErrActorTerminated,
"Actor2 key1 not terminated with correct error.",
)
})
t.Run("unregister all with no actors for the key", func(st *testing.T) {
keyEmpty := NewServiceKey[*testMsg, string]("sk-ua-key-empty")
stoppedCountEmptyKey := keyEmpty.UnregisterAll(as)
require.Equal(
st, 0, stoppedCountEmptyKey,
"UnregisterAll for empty key returned non-zero count.",
)
foundActorsForKeyEmpty := FindInReceptionist(
as.Receptionist(), keyEmpty,
)
require.Empty(
st, foundActorsForKeyEmpty,
"Receptionist not empty for keyEmpty "+
"after UnregisterAll.",
)
})
t.Run("unregister all with mixed keys", func(st *testing.T) {
keyA := NewServiceKey[*testMsg, string]("sk-ua-keyA")
keyB := NewServiceKey[*testMsg, string]("sk-ua-keyB")
// Spawn 3 actors, two of them will share the same service key.
actorA1, err := keyA.Spawn(as, "actorA1-ua-mixed", beh)
require.NoError(st, err)
actorA2, err := keyA.Spawn(as, "actorA2-ua-mixed", beh)
require.NoError(st, err)
actorB1, err := keyB.Spawn(as, "actorB1-ua-mixed", beh)
require.NoError(st, err)
// Make sure we're able to find them in the receptionist.
require.Len(
st, FindInReceptionist(as.Receptionist(), keyA), 2,
"KeyA initial registration count mismatch.",
)
require.Len(
st, FindInReceptionist(as.Receptionist(), keyB), 1,
"KeyB initial registration count mismatch.",
)
// We'll start by unregistering all actors for keyA.
stoppedCountKeyA := keyA.UnregisterAll(as)
require.Equal(
st, 2, stoppedCountKeyA,
"UnregisterAll for keyA returned incorrect count.",
)
// Verify keyA actors are gone from receptionist, keyB actor
// remains.
require.Empty(
st, FindInReceptionist(as.Receptionist(), keyA),
"actors for keyA still in receptionist after "+
"UnregisterAll.",
)
foundActorsForKeyBAfterA := FindInReceptionist(
as.Receptionist(), keyB,
)
require.Len(
st, foundActorsForKeyBAfterA, 1,
"Actor for keyB affected by UnregisterAll on keyA.",
)
require.Equal(
st, actorB1, foundActorsForKeyBAfterA[0],
"Wrong actor found for keyB.",
)
// Verify keyA actors are removed from system map, keyB actor
// remains.
as.mu.RLock()
_, actorA1ExistsAfterMixed := as.actors[actorA1.ID()]
_, actorA2ExistsAfterMixed := as.actors[actorA2.ID()]
_, actorB1ExistsAfterMixed := as.actors[actorB1.ID()]
as.mu.RUnlock()
require.False(
st, actorA1ExistsAfterMixed,
"ActorA1 still in system actors map after "+
"mixed UnregisterAll.",
)
require.False(
st, actorA2ExistsAfterMixed,
"ActorA2 still in system actors map after "+
"mixed UnregisterAll.",
)
require.True(
st, actorB1ExistsAfterMixed,
"ActorB1 removed from system actors map incorrectly.",
)
// Verify keyA actors are stopped, keyB actor is running.
resultActorA1Mixed := actorA1.Ask(
context.Background(), newTestMsg("ping-kA-a1"),
).Await(context.Background())
require.True(st, resultActorA1Mixed.IsErr())
require.ErrorIs(
st, resultActorA1Mixed.Err(), ErrActorTerminated,
)
resultActorB1Mixed := actorB1.Ask(
context.Background(), newTestMsg("ping-kB-a1"),
).Await(context.Background())
require.False(
st, resultActorB1Mixed.IsErr(),
"ActorB1 terminated incorrectly (mixed test): %v",
resultActorB1Mixed.Err(),
)
resultActorB1Mixed.WhenOk(func(s string) {
require.Equal(st, "echo: ping-kB-a1", s)
})
})
t.Run("idempotency of UnregisterAll", func(st *testing.T) {
keyIdempotent := NewServiceKey[*testMsg, string](
"sk-ua-key-idem",
)
actorIdem, err := keyIdempotent.Spawn(as, "actor-idem-ua", beh)
require.NoError(st, err)
// First call should unregister and stop.
stoppedCountFirstCall := keyIdempotent.UnregisterAll(as)
require.Equal(
st, 1, stoppedCountFirstCall,
"UnregisterAll (first call) incorrect count.",
)
// Second call should do nothing and return 0.
stoppedCountSecondCall := keyIdempotent.UnregisterAll(as)
require.Equal(
st, 0, stoppedCountSecondCall,
"UnregisterAll (second call) incorrect count, not "+
"idempotent.",
)
// Verify actor is gone from receptionist and system map, and is
// stopped.
require.Empty(
st, FindInReceptionist(as.Receptionist(), keyIdempotent),
"Actors for keyIdempotent still in receptionist "+
"after calls.",
)
as.mu.RLock()
_, actorIdemExistsAfter := as.actors[actorIdem.ID()]
as.mu.RUnlock()
require.False(
st, actorIdemExistsAfter,
"ActorIdem still in system actors map after calls.",
)
resultActorIdem := actorIdem.Ask(
context.Background(), newTestMsg("ping-kidem-a1"),
).Await(context.Background())
require.True(st, resultActorIdem.IsErr())
require.ErrorIs(st, resultActorIdem.Err(), ErrActorTerminated)
})
}
// routerTestHarness helps set up routers and their associated actors for testing.
// It uses an actorTestHarness internally for DLO observation for the router.
type routerTestHarness struct {
*actorTestHarness
as *ActorSystem
receptionist *Receptionist
}
// newRouterTestHarness sets up a new harness for router testing.
// It creates an ActorSystem for actors that the router will route to,
// and uses the embedded actorTestHarness for the router's own DLO.
func newRouterTestHarness(t *testing.T) *routerTestHarness {
t.Helper()
system := NewActorSystem()
t.Cleanup(func() {
err := system.Shutdown()
require.NoError(t, err, "router test actor system shutdown failed")
})
// The DLO for the router itself will come from actorTestHarness.
// Actors managed by `system` (router targets) will use `system.DeadLetters()`.
return &routerTestHarness{
actorTestHarness: newActorTestHarness(t),
as: system,
receptionist: system.Receptionist(),
}
}
// newRouterTargetActor creates an actor, registers it with the harness's
// ActorSystem (h.as) and Receptionist under the given service key. This actor
// is intended to be a target for the router.
func (h *routerTestHarness) newRouterTargetActor(id string,
key ServiceKey[*testMsg, string],
beh ActorBehavior[*testMsg, string]) ActorRef[*testMsg, string] {
h.t.Helper()
ref, err := RegisterWithSystem(h.as, id, key, beh)
require.NoError(h.t, err)
return ref
}
// TestRouterNewRouter verifies that a new router can be created as expected.
func TestRouterNewRouter(t *testing.T) {
t.Parallel()
h := newRouterTestHarness(t)
key := NewServiceKey[*testMsg, string]("router-service")
strategy := NewRoundRobinStrategy[*testMsg, string]()
router := NewRouter(h.receptionist, key, strategy, h.dlo.Ref())
require.NotNil(t, router, "newRouter should not return nil")
require.Equal(t, "router(router-service)", router.ID(), "router ID mismatch")
}
// countingEchoBehavior is an echo behavior that also counts how many messages
// it has processed.
type countingEchoBehavior struct {
*echoBehavior
id string
processedMsgs int64
}
func newCountingEchoBehavior(t *testing.T, id string) *countingEchoBehavior {
return &countingEchoBehavior{
echoBehavior: newEchoBehavior(t, 0),
id: id,
}
}
func (b *countingEchoBehavior) Receive(ctx context.Context,
msg *testMsg) fn.Result[string] {
atomic.AddInt64(&b.processedMsgs, 1)
// Include actor ID in reply for easier verification.
res := b.echoBehavior.Receive(ctx, msg)
val, err := res.Unpack()
if err == nil {
return fn.Ok(fmt.Sprintf("%s:%s", b.id, val))
}
return res
}
// TestRouterTellAndAskRoundRobin verifies that the router distributes messages
// in a round robin properly.
func TestRouterTellAndAskRoundRobin(t *testing.T) {
t.Parallel()
h := newRouterTestHarness(t)
// Make a new router for the given service key and round robin strategy.
serviceKey := NewServiceKey[*testMsg, string]("rr-service")
strategy := NewRoundRobinStrategy[*testMsg, string]()
router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref())
// We'll now register two actors with the router, each with a different
// service key.
actor1Beh := newCountingEchoBehavior(t, "actor1")
actor2Beh := newCountingEchoBehavior(t, "actor2")
_ = h.newRouterTargetActor("actor1-rr", serviceKey, actor1Beh)
_ = h.newRouterTargetActor("actor2-rr", serviceKey, actor2Beh)
// Nxet, we'll send a mix of Tell and Ask messages to the router.
numMessages := 6
for i := 0; i < numMessages; i++ {
msgData := fmt.Sprintf("message-%d", i)
if i%2 == 0 {
router.Tell(context.Background(), newTestMsg(msgData))
} else {
future := router.Ask(
context.Background(), newTestMsg(msgData),
)
ctxAwait, cancelAwait := context.WithTimeout(
context.Background(), time.Second,
)
result := future.Await(ctxAwait)
cancelAwait()
require.False(
t, result.IsErr(), "ask failed: %v", result.Err(),
)
}
}
// Wait a bit for Tell messages to be processed.
time.Sleep(100 * time.Millisecond)
// Each actor should have processed numMessages / 2 messages.
require.EqualValues(
t, numMessages/2, atomic.LoadInt64(&actor1Beh.processedMsgs),
"actor1 processed message count mismatch",
)
require.EqualValues(
t, numMessages/2, atomic.LoadInt64(&actor2Beh.processedMsgs),
"actor2 processed message count mismatch",
)
// Router's DLO should be empty.
h.assertNoDLOMessages()
}
// TestRouterNoActorsAvailable verifies that if no actors are available for the
// message, then an error is returned.
func TestRouterNoActorsAvailable(t *testing.T) {
t.Parallel()
h := newRouterTestHarness(t)
serviceKey := NewServiceKey[*testMsg, string]("no-actor-service")
strategy := NewRoundRobinStrategy[*testMsg, string]()
router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref())
// We'll send a message, then assert that it goes to the DLO.
tellMsg := newTestMsg("tell-no-actor")
router.Tell(context.Background(), tellMsg)
h.assertDLOMessage(tellMsg)
// If we use an ask instead, then we should get an error.
askMsg := newTestMsg("ask-no-actor")
future := router.Ask(context.Background(), askMsg)
result := future.Await(context.Background())
require.True(
t, result.IsErr(), "ask should fail when no actors are available",
)
require.ErrorIs(t, result.Err(), ErrNoActorsAvailable, "error mismatch")
}
// TestRouterTellAskContextCancellation verifies that if the context is
// canceled, then sending aborts.
func TestRouterTellAskContextCancellation(t *testing.T) {
t.Parallel()
h := newRouterTestHarness(t)
serviceKey := NewServiceKey[*testMsg, string]("ctx-cancel-service")
strategy := NewRoundRobinStrategy[*testMsg, string]()
router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref())
// Use a regular echo actor, but we'll control context for Tell/Ask.
targetActorBeh := newEchoBehavior(t, 50*time.Millisecond)
_ = h.newRouterTargetActor("target-ctx", serviceKey, targetActorBeh)
// Next, we'll send a Tell message with a context that will be cancelled
// before we even send.
ctxTell, cancelTell := context.WithCancel(context.Background())
cancelTell()
router.Tell(ctxTell, newTestMsg("tell-ctx-cancelled"))
// The Message should be dropped by actorRefImpl.Tell if ctx is
// cancelled. Router's DLO should not receive it from this path.
h.assertNoDLOMessages()
// Next, we'll do the same for Ask. This time, we should get an error.
ctxAsk, cancelAsk := context.WithCancel(context.Background())
cancelAsk()
futureAsk := router.Ask(ctxAsk, newTestMsg("ask-ctx-cancelled"))
resultAsk := futureAsk.Await(context.Background())
require.True(
t, resultAsk.IsErr(), "ask with cancelled context should fail",
)
require.ErrorIs(
t, resultAsk.Err(), context.Canceled,
"error should be context.Canceled",
)
}
// TestRouterDynamicActorRegistration tests that we're able to dynamically add
// and remove actors from the router.
func TestRouterDynamicActorRegistration(t *testing.T) {
t.Parallel()
h := newRouterTestHarness(t)
serviceKey := NewServiceKey[*testMsg, string]("dynamic-service")
strategy := NewRoundRobinStrategy[*testMsg, string]()
router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref())
// If we try to send a mesasge to the router before any actors are
// added, we should get an error.
futureNoActor := router.Ask(context.Background(), newTestMsg("ping-no-actors"))
resNoActor := futureNoActor.Await(context.Background())
require.ErrorIs(t, resNoActor.Err(), ErrNoActorsAvailable)
actor1Beh := newCountingEchoBehavior(t, "actor1")
actor1Ref := h.newRouterTargetActor("actor1-dynamic", serviceKey, actor1Beh)
// At this point, we have a new actor added, but we'll try to send a
// message to a different actor ID. This should go to the router's DLO.
futureActor1 := router.Ask(context.Background(), newTestMsg("ping-actor1"))
ctxAwaitA1, cancelAwaitA1 := context.WithTimeout(context.Background(), time.Second)
resActor1 := futureActor1.Await(ctxAwaitA1)
cancelAwaitA1()
require.False(t, resActor1.IsErr(), "ask to actor1 failed: %v", resActor1.Err())
resActor1.WhenOk(func(s string) {
require.Equal(t, "actor1:echo: ping-actor1", s)
})
actor2Beh := newCountingEchoBehavior(t, "actor2")
actor2Ref := h.newRouterTargetActor(
"actor2-dynamic", serviceKey, actor2Beh,
)
// Now that we've added two actors above, we should round robin between
// them when sending.
ctxAwaitDA1, cancelAwaitDA1 := context.WithTimeout(
context.Background(), time.Second,
)
router.Ask(context.Background(), newTestMsg("dynamic-ask1")).Await(
ctxAwaitDA1,
)
cancelAwaitDA1()
ctxAwaitDA2, cancelAwaitDA2 := context.WithTimeout(context.Background(), time.Second)
router.Ask(context.Background(), newTestMsg("dynamic-ask2")).Await(ctxAwaitDA2)
cancelAwaitDA2()
time.Sleep(50 * time.Millisecond)
// actor1 should have processed 2 messages (ping-actor1, dynamic-ask1),
require.EqualValues(t, 2, atomic.LoadInt64(&actor1Beh.processedMsgs))
require.EqualValues(t, 1, atomic.LoadInt64(&actor2Beh.processedMsgs))
// Next, we'll unregister the first actor ref.
unregistered := UnregisterFromReceptionist(
h.receptionist, serviceKey, actor1Ref,
)
require.True(t, unregistered)
// All the messages should now go to the second actor.
for i := 0; i < 2; i++ {
msgData := fmt.Sprintf("to-actor2-%d", i)
future := router.Ask(context.Background(), newTestMsg(msgData))
ctxAwaitLoop, cancelAwaitLoop := context.WithTimeout(
context.Background(), time.Second,
)
res := future.Await(ctxAwaitLoop)
cancelAwaitLoop()
require.False(
t, res.IsErr(), "ask to actor2 failed: %v", res.Err(),
)
res.WhenOk(func(s string) {
require.Equal(t, "actor2:echo: "+msgData, s)
})
}
// Actor 1 shouldn't have got any of the messages, they should go to
// actor 2.
require.EqualValues(t, 2, atomic.LoadInt64(&actor1Beh.processedMsgs))
require.EqualValues(t, 1+2, atomic.LoadInt64(&actor2Beh.processedMsgs))
// Next, we'll unregister the second actor ref.
unregistered2 := UnregisterFromReceptionist(
h.receptionist, serviceKey, actor2Ref,
)
require.True(t, unregistered2)
// If we try to send another message, it should go to the DL.
tellMsg := newTestMsg("dynamic-tell-no-actors")
router.Tell(context.Background(), tellMsg)
h.assertDLOMessage(tellMsg)
}