diff --git a/actor/router.go b/actor/router.go new file mode 100644 index 000000000..87e2dae44 --- /dev/null +++ b/actor/router.go @@ -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 + ")" +} diff --git a/actor/system.go b/actor/system.go new file mode 100644 index 000000000..5e765de6e --- /dev/null +++ b/actor/system.go @@ -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 +} diff --git a/actor/system_test.go b/actor/system_test.go new file mode 100644 index 000000000..cc1510ac6 --- /dev/null +++ b/actor/system_test.go @@ -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) +}