Repository: Tochemey/goakt
Pull request #431 (2024-08-19 23:20:28)

perf: refactor processing error handling

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/431?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 86.11%. Comparing base [(`4183b60`)](https://app.codecov.io/gh/Tochemey/goakt/commit/4183b600fc9269700a455ee5b24fb3127cc8d7ef?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) to head [(`f9efd3c`)](https://app.codecov.io/gh/Tochemey/goakt/commit/f9efd3c0a4756c9eb86143a188d466de74dc9a29?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #431      +/-   ##
==========================================
+ Coverage   86.07%   86.11%   +0.04%     
==========================================
  Files          46       46              
  Lines        3518     3529      +11     
==========================================
+ Hits         3028     3039      +11     
  Misses        345      345              
  Partials      145      145              
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/431?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).


Patch:
diff --git a/actors/api.go b/actors/api.go
index 13d839b..fe0b6e6 100644
--- a/actors/api.go
+++ b/actors/api.go
@@ -72,7 +72,7 @@ func Ask(ctx context.Context, to PID, message proto.Message, timeout time.Durati
 		case <-await:
 			to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 			err = ErrRequestTimeout
-			to.handleError(messageContext, err)
+			to.onError(messageContext, err)
 			return
 		}
 	}
@@ -148,7 +148,7 @@ func BatchAsk(ctx context.Context, to PID, timeout time.Duration, messages ...pr
 			case <-await:
 				to.setLastProcessingDuration(time.Since(to.getLastProcessingTime()))
 				err = ErrRequestTimeout
-				to.handleError(messageContext, err)
+				to.onError(messageContext, err)
 				return
 			}
 		}
diff --git a/actors/context.go b/actors/context.go
index 0bc50b6..d5ebd90 100644
--- a/actors/context.go
+++ b/actors/context.go
@@ -86,7 +86,7 @@ type ReceiveContext interface {
 	// When TellStream encounter a single message it will fall back to a Tell call.
 	BatchTell(to PID, messages ...proto.Message)
 	// Ask sends a synchronous message to another actor and expect a response. This method is good when interacting with a child actor.
-	// Ask has a timeout which can cause the sender to panic. When ask times out, the receiving actor does not know and may still process the message.
+	// Ask has a timeout which can cause the sender to return an error. When ask times out, the receiving actor does not know and may still process the message.
 	// It is recommended to set a good timeout to quickly receive response and try to avoid false positives
 	Ask(to PID, message proto.Message) (response proto.Message)
 	// BatchAsk sends a synchronous bunch of messages to the given PID and expect responses in the same order as the messages.
@@ -117,7 +117,7 @@ type ReceiveContext interface {
 	// All current messages in the mailbox will be processed before the actor shutdown after a period of time
 	// that can be configured. All child actors will be gracefully shutdown.
 	Shutdown()
-	// Spawn creates a child actor or panic
+	// Spawn creates a child actor or sets the context error
 	Spawn(name string, actor Actor) PID
 	// Children returns the list of all the children of the given actor
 	Children() []PID
@@ -132,13 +132,14 @@ type ReceiveContext interface {
 	// RemoteReSpawn restarts an actor on a remote node.
 	RemoteReSpawn(host string, port int, name string)
 	// Err is used instead of panicking within a message handler.
-	// One can also call panic which is not the recommended way
 	Err(err error)
 	// PipeTo processes a long-running task and pipes the result to the provided actor.
 	// The successful result of the task will be put onto the provided actor mailbox.
 	// This is useful when interacting with external services.
 	// It’s common that you would like to use the value of the response in the actor when the long-running task is completed
 	PipeTo(to PID, task future.Task)
+
+	getError() error
 }
 
 type receiveContext struct {
@@ -150,6 +151,7 @@ type receiveContext struct {
 	recipient    PID
 	async        bool
 	sendTime     atomic.Time
+	err          error
 }
 
 // force compilation error
@@ -186,11 +188,9 @@ func (c *receiveContext) Self() PID {
 }
 
 // Err is used instead of panicking within a message handler.
-// One can also call panic which is not the recommended way
 func (c *receiveContext) Err(err error) {
 	if err != nil {
-		// this will be recovered
-		panic(err)
+		c.err = err
 	}
 }
 
@@ -198,8 +198,8 @@ func (c *receiveContext) Err(err error) {
 func (c *receiveContext) Response(resp proto.Message) {
 	// only set a response when the message is sync message
 	if !c.async {
-		defer close(c.response)
 		c.response <- resp
+		close(c.response)
 	}
 }
 
@@ -282,7 +282,7 @@ func (c *receiveContext) Tell(to PID, message proto.Message) {
 	recipient := c.recipient
 	ctx := context.WithoutCancel(c.ctx)
 	if err := recipient.Tell(ctx, to, message); err != nil {
-		panic(err)
+		c.Err(err)
 	}
 }
 
@@ -299,7 +299,7 @@ func (c *receiveContext) BatchTell(to PID, messages ...proto.Message) {
 }
 
 // Ask sends a synchronous message to another actor and expect a response. This method is good when interacting with a child actor.
-// Ask has a timeout which can cause the sender to panic. When ask times out, the receiving actor does not know and may still process the message.
+// Ask has a timeout which can cause the sender to set the context error. When ask times out, the receiving actor does not know and may still process the message.
 // It is recommended to set a good timeout to quickly receive response and try to avoid false positives
 func (c *receiveContext) Ask(to PID, message proto.Message) (response proto.Message) {
 	recipient := c.recipient
@@ -391,7 +391,7 @@ func (c *receiveContext) Shutdown() {
 	}
 }
 
-// Spawn creates a child actor or panic
+// Spawn creates a child actor or return error
 func (c *receiveContext) Spawn(name string, actor Actor) PID {
 	recipient := c.recipient
 	ctx := context.WithoutCancel(c.ctx)
@@ -418,7 +418,7 @@ func (c *receiveContext) Child(name string) PID {
 }
 
 // Stop forces the child Actor under the given name to terminate after it finishes processing its current message.
-// Nothing happens if child is already stopped. However, it panics when the child cannot be stopped.
+// Nothing happens if child is already stopped. However, it returns an error when the child cannot be stopped.
 func (c *receiveContext) Stop(child PID) {
 	recipient := c.recipient
 	ctx := context.WithoutCancel(c.ctx)
@@ -451,7 +451,7 @@ func (c *receiveContext) Forward(to PID) {
 // Unhandled is used to handle unhandled messages instead of throwing error
 func (c *receiveContext) Unhandled() {
 	me := c.recipient
-	me.handleError(c, ErrUnhandled)
+	me.onError(c, ErrUnhandled)
 }
 
 // RemoteReSpawn restarts an actor on a remote node.
@@ -474,3 +474,8 @@ func (c *receiveContext) PipeTo(to PID, task future.Task) {
 		c.Err(err)
 	}
 }
+
+// getError returns any error during message processing
+func (c *receiveContext) getError() error {
+	return c.err
+}
diff --git a/actors/pid.go b/actors/pid.go
index 9a4d3df..7b49f36 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -29,7 +29,6 @@ import (
 	"errors"
 	"fmt"
 	stdhttp "net/http"
-	"runtime"
 	"strings"
 	"sync"
 	"time"
@@ -51,7 +50,7 @@ import (
 	"github.com/tochemey/goakt/v2/internal/internalpb/internalpbconnect"
 	"github.com/tochemey/goakt/v2/internal/metric"
 	"github.com/tochemey/goakt/v2/internal/queue"
-	"github.com/tochemey/goakt/v2/internal/slices"
+	"github.com/tochemey/goakt/v2/internal/slice"
 	"github.com/tochemey/goakt/v2/internal/types"
 	"github.com/tochemey/goakt/v2/log"
 	"github.com/tochemey/goakt/v2/telemetry"
@@ -159,7 +158,7 @@ type PID interface {
 	// push a message to the actor's receiveContextBuffer
 	doReceive(ctx ReceiveContext)
 	// watchers returns the list of actors watching this actor
-	watchers() *slices.Safe[*watcher]
+	watchers() *slice.Slice[*watcher]
 	// watchees returns the list of actors watched by this actor
 	watchees() *pidMap
 	// setBehavior is a utility function that helps set the actor behavior
@@ -179,7 +178,7 @@ type PID interface {
 	// unstash unstashes the oldest message in the stash and prepends to the mailbox
 	unstash() error
 	// toDeadletters add the given message to the deadletters queue
-	handleError(receiveCtx ReceiveContext, err error)
+	onError(receiveCtx ReceiveContext, err error)
 	// removeChild is a utility function to remove child actor
 	removeChild(pid PID)
 
@@ -227,7 +226,7 @@ type pid struct {
 	haltPassivationLnr chan types.Unit
 
 	// hold the watchersList watching the given actor
-	watchersList *slices.Safe[*watcher]
+	watchersList *slice.Slice[*watcher]
 
 	// hold the list of the children
 	children *pidMap
@@ -300,7 +299,7 @@ func newPID(ctx context.Context, actorPath *Path, actor Actor, opts ...pidOption
 		logger:               log.DefaultLogger,
 		children:             newPIDMap(10),
 		supervisorDirective:  DefaultSupervisoryStrategy,
-		watchersList:         slices.NewSafe[*watcher](),
+		watchersList:         slice.New[*watcher](),
 		watchedList:          newPIDMap(10),
 		telemetry:            telemetry.New(),
 		actorPath:            actorPath,
@@ -403,7 +402,7 @@ func (x *pid) Parents() []PID {
 	x.fieldsLocker.Unlock()
 	var parents []PID
 	if watchers.Len() > 0 {
-		for item := range watchers.Iter() {
+		for _, item := range watchers.Items() {
 			watcher := item.Value
 			if watcher != nil {
 				pid := watcher.WatcherID
@@ -646,7 +645,7 @@ func (x *pid) Ask(ctx context.Context, to PID, message proto.Message) (response
 		case <-await:
 			x.recordLastReceivedDurationMetric(ctx)
 			err = ErrRequestTimeout
-			x.handleError(messageContext, err)
+			x.onError(messageContext, err)
 			return nil, err
 		}
 	}
@@ -714,7 +713,7 @@ func (x *pid) BatchAsk(ctx context.Context, to PID, messages ...proto.Message) (
 			case <-await:
 				err = ErrRequestTimeout
 				x.recordLastReceivedDurationMetric(ctx)
-				x.handleError(messageContext, err)
+				x.onError(messageContext, err)
 				return nil, err
 			}
 		}
@@ -1125,7 +1124,7 @@ func (x *pid) Watch(cid PID) {
 
 // UnWatch stops watching a given actor
 func (x *pid) UnWatch(pid PID) {
-	for item := range pid.watchers().Iter() {
+	for _, item := range pid.watchers().Items() {
 		w := item.Value
 		if w.WatcherID.Equals(x) {
 			w.Done <- types.Unit{}
@@ -1137,7 +1136,7 @@ func (x *pid) UnWatch(pid PID) {
 }
 
 // watchers return the list of watchersList
-func (x *pid) watchers() *slices.Safe[*watcher] {
+func (x *pid) watchers() *slice.Slice[*watcher] {
 	return x.watchersList
 }
 
@@ -1194,7 +1193,7 @@ func (x *pid) reset() {
 	x.lastReceivedDuration.Store(0)
 	x.initTimeout.Store(DefaultInitTimeout)
 	x.children = newPIDMap(10)
-	x.watchersList = slices.NewSafe[*watcher]()
+	x.watchersList = slice.New[*watcher]()
 	x.telemetry = telemetry.New()
 	x.resetBehavior()
 	if x.metricEnabled.Load() {
@@ -1210,7 +1209,7 @@ func (x *pid) freeWatchers(ctx context.Context) {
 	x.logger.Debug("freeing all watcher actors...")
 	watchers := x.watchers()
 	if watchers.Len() > 0 {
-		for item := range watchers.Iter() {
+		for _, item := range watchers.Items() {
 			watcher := item.Value
 			terminated := &goaktpb.Terminated{
 				ActorId: x.ID(),
@@ -1267,7 +1266,6 @@ func (x *pid) receive() {
 		// fetch the data and continue the loop when there are no records yet
 		received, ok := x.mailbox.Pop()
 		if !ok {
-			runtime.Gosched()
 			continue
 		}
 
@@ -1281,26 +1279,41 @@ func (x *pid) receive() {
 	}
 }
 
+// handleReceived picks the right behavior and processes the message
 func (x *pid) handleReceived(received ReceiveContext) {
-	// handle panic when the processing of the message fails
-	defer func() {
-		if r := recover(); r != nil {
-			err := fmt.Errorf("%s", r)
-			for item := range x.watchersList.Iter() {
-				item.Value.ErrChan <- err
-			}
-		}
-		// set the total messages processed
-		x.processedCount.Inc()
-	}()
-
+	// handle panic or any error during processing
+	defer x.recovery(received)
+	// pick the current behvior from the stack
 	behaviorStack := x.behaviorStack
 	// send the message to the current actor behavior
 	if behavior, ok := behaviorStack.Peek(); ok {
+		// set the total messages processed
+		x.processedCount.Inc()
+		// call the behavior to handle the message
 		behavior(received)
 	}
 }
 
+// recovery is called upon after message is processed
+func (x *pid) recovery(received ReceiveContext) {
+	// track some processing panic error
+	var err error
+	if r := recover(); r != nil {
+		err = fmt.Errorf("%s", r)
+	}
+
+	// no panic
+	if err == nil {
+		// check whether the proper way to return error is used
+		err = received.getError()
+	}
+
+	// only handle error when defined
+	if err != nil {
+		x.onError(received, err)
+	}
+}
+
 // passivationLoop checks whether the actor is processing public or not.
 // when the actor is idle, it automatically shuts down to free resources
 func (x *pid) passivationLoop() {
@@ -1450,8 +1463,25 @@ func (x *pid) removeChild(pid PID) {
 	}
 }
 
-// handleError handles the error during a message handling
-func (x *pid) handleError(receiveCtx ReceiveContext, err error) {
+// onError handles the error during a message handling
+func (x *pid) onError(receiveCtx ReceiveContext, err error) {
+	// all unhandled error goes to deadletter
+	if !errors.Is(err, ErrUnhandled) {
+		// check whether there are supervisors and push the error message to them
+		if x.watchers().Len() != 0 {
+			// TODO: roundrobin pick or all of them
+			for _, item := range x.watchers().Items() {
+				item.Value.ErrChan <- err
+			}
+			return
+		}
+	}
+	x.toDeadletterQueue(receiveCtx, err)
+}
+
+// toDeadletterQueue sends message to deadletter queue
+func (x *pid) toDeadletterQueue(receiveCtx ReceiveContext, err error) {
+	// the message is lost
 	if x.eventsStream == nil {
 		return
 	}
@@ -1470,16 +1500,13 @@ func (x *pid) handleError(receiveCtx ReceiveContext, err error) {
 		senderAddr = receiveCtx.Sender().ActorPath().RemoteAddress()
 	}
 
-	receiver := x.actorPath.RemoteAddress()
-	deadletter := &goaktpb.Deadletter{
+	x.eventsStream.Publish(eventsTopic, &goaktpb.Deadletter{
 		Sender:   senderAddr,
-		Receiver: receiver,
+		Receiver: x.actorPath.RemoteAddress(),
 		Message:  msg,
 		SendTime: timestamppb.Now(),
 		Reason:   err.Error(),
-	}
-
-	x.eventsStream.Publish(eventsTopic, deadletter)
+	})
 }
 
 // registerMetrics register the PID metrics with OTel instrumentation.
@@ -1500,8 +1527,8 @@ func (x *pid) registerMetrics() error {
 	return err
 }
 
-// grpcClientOptions returns the gRPC client connections options
-func (x *pid) grpcClientOptions() ([]connect.ClientOption, error) {
+// clientOptions returns the gRPC client connections options
+func (x *pid) clientOptions() ([]connect.ClientOption, error) {
 	var interceptor *otelconnect.Interceptor
 	var err error
 	if x.metricEnabled.Load() {
@@ -1522,7 +1549,7 @@ func (x *pid) grpcClientOptions() ([]connect.ClientOption, error) {
 
 // remotingClient returns an instance of the Remote Service client
 func (x *pid) remotingClient(host string, port int) (internalpbconnect.RemotingServiceClient, error) {
-	clientConnectionOptions, err := x.grpcClientOptions()
+	clientOptions, err := x.clientOptions()
 	if err != nil {
 		return nil, err
 	}
@@ -1530,7 +1557,7 @@ func (x *pid) remotingClient(host string, port int) (internalpbconnect.RemotingS
 	return internalpbconnect.NewRemotingServiceClient(
 		x.httpClient,
 		http.URL(host, port),
-		clientConnectionOptions...,
+		clientOptions...,
 	), nil
 }
 
diff --git a/actors/redistribution.go b/actors/redistribution.go
index 84f18bc..2694410 100644
--- a/actors/redistribution.go
+++ b/actors/redistribution.go
@@ -35,7 +35,7 @@ import (
 	"github.com/tochemey/goakt/v2/goaktpb"
 	"github.com/tochemey/goakt/v2/internal/cluster"
 	"github.com/tochemey/goakt/v2/internal/internalpb"
-	"github.com/tochemey/goakt/v2/internal/slices"
+	"github.com/tochemey/goakt/v2/internal/slice"
 )
 
 // redistribute is used when cluster topology changes particularly
@@ -92,7 +92,7 @@ func (x *actorSystem) redistribute(ctx context.Context, event *cluster.Event) er
 		quotient := actorsCount / totalPeers
 		remainder := actorsCount % totalPeers
 		leaderActors = peerState.GetActors()[:remainder]
-		chunks = slices.Chunk[*internalpb.WireActor](peerState.GetActors()[remainder:], quotient)
+		chunks = slice.Chunk[*internalpb.WireActor](peerState.GetActors()[remainder:], quotient)
 	}
 
 	if len(chunks) > 0 {
diff --git a/go.mod b/go.mod
index dcb305c..d492b12 100644
--- a/go.mod
+++ b/go.mod
@@ -37,8 +37,6 @@ require (
 	k8s.io/utils v0.0.0-20240711033017-18e509b52bc8
 )
 
-require google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect
-
 require (
 	github.com/RoaringBitmap/roaring v1.9.4 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
@@ -99,7 +97,6 @@ require (
 	golang.org/x/text v0.17.0 // indirect
 	golang.org/x/time v0.6.0 // indirect
 	golang.org/x/tools v0.24.0 // indirect
-	google.golang.org/grpc v1.65.0
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
diff --git a/go.sum b/go.sum
index 306e202..ad09a87 100644
--- a/go.sum
+++ b/go.sum
@@ -377,10 +377,6 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 h1:Zy9XzmMEflZ/MAaA7vNcoebnRAld7FsPW1EeBB7V0m8=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157/go.mod h1:EfXuqaE1J41VCDicxHzUDm+8rk+7ZdXzHV0IhO/I6s0=
-google.golang.org/grpc v1.65.0 h1:bs/cUb4lp1G5iImFFd3u5ixQzweKizoZJAwBNLR42lc=
-google.golang.org/grpc v1.65.0/go.mod h1:WgYC2ypjlB0EiQi6wdKixMqukr6lBc0Vo+oOgjrM5ZQ=
 google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
 google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
diff --git a/internal/eventstream/stream.go b/internal/eventstream/stream.go
index 4007780..fe44ce7 100644
--- a/internal/eventstream/stream.go
+++ b/internal/eventstream/stream.go
@@ -50,9 +50,9 @@ func New() *EventsStream {
 // AddSubscriber adds a subscriber
 func (b *EventsStream) AddSubscriber() Subscriber {
 	b.mu.Lock()
-	defer b.mu.Unlock()
 	c := newSubscriber()
 	b.subs[c.ID()] = c
+	b.mu.Unlock()
 	return c
 }
 
@@ -90,8 +90,9 @@ func (b *EventsStream) Broadcast(msg any, topics []string) {
 func (b *EventsStream) SubscribersCount(topic string) int {
 	// get total subscribers subscribed to given topic.
 	b.mu.Lock()
-	defer b.mu.Unlock()
-	return len(b.topics[topic])
+	count := len(b.topics[topic])
+	b.mu.Unlock()
+	return count
 }
 
 // Subscribe subscribes a subscriber to a topic
@@ -158,3 +159,11 @@ func (b *EventsStream) Shutdown() {
 	b.subs = Subscribers{}
 	b.topics = map[string]Subscribers{}
 }
+
+// Subscribers returns the list of subscribers
+func (b *EventsStream) Subscribers() Subscribers {
+	b.mu.Lock()
+	subs := b.subs
+	b.mu.Unlock()
+	return subs
+}
diff --git a/internal/slices/chunk.go b/internal/slice/chunk.go
similarity index 99%
rename from internal/slices/chunk.go
rename to internal/slice/chunk.go
index c47408e..7b54047 100644
--- a/internal/slices/chunk.go
+++ b/internal/slice/chunk.go
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package slices
+package slice
 
 // Chunk helps chunk a slice into equal size
 func Chunk[T any](slice []T, chunkSize int) [][]T {
diff --git a/internal/slices/slices.go b/internal/slice/slice.go
similarity index 70%
rename from internal/slices/slices.go
rename to internal/slice/slice.go
index ac5ea7b..210b861 100644
--- a/internal/slices/slices.go
+++ b/internal/slice/slice.go
@@ -22,12 +22,12 @@
  * SOFTWARE.
  */
 
-package slices
+package slice
 
 import "sync"
 
-// Safe type that can be safely shared between goroutines.
-type Safe[T any] struct {
+// Slice type that can be safely shared between goroutines.
+type Slice[T any] struct {
 	sync.RWMutex
 	items []T
 }
@@ -38,9 +38,9 @@ type Item[T any] struct {
 	Value T
 }
 
-// NewSafe creates a new synchronized slice.
-func NewSafe[T any]() *Safe[T] {
-	cs := &Safe[T]{
+// New creates a new synchronized slice.
+func New[T any]() *Slice[T] {
+	cs := &Slice[T]{
 		items: make([]T, 0),
 	}
 
@@ -48,21 +48,22 @@ func NewSafe[T any]() *Safe[T] {
 }
 
 // Len returns the number of items
-func (cs *Safe[T]) Len() int {
+func (cs *Slice[T]) Len() int {
 	cs.Lock()
-	defer cs.Unlock()
-	return len(cs.items)
+	length := len(cs.items)
+	cs.Unlock()
+	return length
 }
 
 // Append adds an item to the concurrent slice.
-func (cs *Safe[T]) Append(item T) {
+func (cs *Slice[T]) Append(item T) {
 	cs.Lock()
-	defer cs.Unlock()
 	cs.items = append(cs.items, item)
+	cs.Unlock()
 }
 
 // Get returns the slice item at the given index
-func (cs *Safe[T]) Get(index int) (item any) {
+func (cs *Slice[T]) Get(index int) (item any) {
 	cs.RLock()
 	defer cs.RUnlock()
 	if isSet(cs.items, index) {
@@ -72,36 +73,33 @@ func (cs *Safe[T]) Get(index int) (item any) {
 }
 
 // Delete an item from the slice
-func (cs *Safe[T]) Delete(index int) {
+func (cs *Slice[T]) Delete(index int) {
 	cs.RLock()
-	defer cs.RUnlock()
+	isSet := isSet(cs.items, index)
+	cs.RUnlock()
 	var nilState T
-	if isSet(cs.items, index) {
+	if isSet {
+		cs.Lock()
 		// Pop the element at index from the slice
 		cs.items[index] = cs.items[len(cs.items)-1] // Copy last element to index.
 		cs.items[len(cs.items)-1] = nilState        // Erase last element (write zero value).
 		cs.items = cs.items[:len(cs.items)-1]       // Truncate slice.
+		cs.Unlock()
 	}
 }
 
-func isSet[T any](arr []T, index int) bool {
-	return len(arr) > index
-}
-
-// Iter iterates the items in the concurrent slice.
-// Each item is sent over a channel, so that
-// we can iterate over the slice using the builtin range keyword.
-func (cs *Safe[T]) Iter() <-chan Item[T] {
-	c := make(chan Item[T])
-	f := func() {
-		cs.RLock()
-		defer cs.RUnlock()
-		for index, value := range cs.items {
-			c <- Item[T]{Index: index, Value: value}
-		}
-		close(c)
+// Items returns the list of items
+func (cs *Slice[T]) Items() []Item[T] {
+	cs.RLock()
+	items := cs.items
+	cs.RUnlock()
+	output := make([]Item[T], len(items))
+	for index, value := range items {
+		output[index] = Item[T]{Index: index, Value: value}
 	}
-	go f()
+	return output
+}
 
-	return c
+func isSet[T any](arr []T, index int) bool {
+	return len(arr) > index
 }


Test patch:
diff --git a/actors/context_test.go b/actors/context_test.go
index 9d31884..10c80d6 100644
--- a/actors/context_test.go
+++ b/actors/context_test.go
@@ -35,7 +35,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/travisjeffery/go-dynaport"
 	"google.golang.org/protobuf/proto"
-	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/tochemey/goakt/v2/goaktpb"
 	"github.com/tochemey/goakt/v2/internal/eventstream"
@@ -134,10 +133,8 @@ func TestReceiveContext(t *testing.T) {
 		require.NoError(t, err)
 		require.NotNil(t, pid2)
 
-		op := func() {
-			context.Tell(pid2, new(testpb.TestSend))
-		}
-		assert.NotPanics(t, op)
+		context.Tell(pid2, new(testpb.TestSend))
+		require.NoError(t, context.getError())
 
 		time.Sleep(time.Second)
 		assert.NoError(t, pid1.Shutdown(ctx))
@@ -177,11 +174,8 @@ func TestReceiveContext(t *testing.T) {
 		// wait a while and shutdown actor2
 		time.Sleep(time.Second)
 		assert.NoError(t, pid2.Shutdown(ctx))
-
-		op := func() {
-			context.Tell(pid2, new(testpb.TestSend))
-		}
-		assert.Panics(t, op)
+		context.Tell(pid2, new(testpb.TestSend))
+		require.Error(t, context.getError())
 
 		time.Sleep(time.Second)
 		assert.NoError(t, pid1.Shutdown(ctx))
@@ -261,10 +255,8 @@ func TestReceiveContext(t *testing.T) {
 		time.Sleep(time.Second)
 		assert.NoError(t, pid2.Shutdown(ctx))
 
-		op := func() {
-			context.Ask(pid2, new(testpb.TestReply))
-		}
-		assert.Panics(t, op)
+		context.Ask(pid2, new(testpb.TestReply))
+		require.Error(t, context.getError())
 
 		time.Sleep(time.Second)
 		assert.NoError(t, pid1.Shutdown(ctx))
@@ -380,16 +372,13 @@ func TestReceiveContext(t *testing.T) {
 			async:     true,
 		}
 
-		op := func() {
-			context.RemoteAsk(&goaktpb.Address{
-				Host: "127.0.0.1",
-				Port: int32(remotingPort),
-				Name: actorName2,
-				Id:   "",
-			}, new(testpb.TestReply))
-		}
-
-		assert.Panics(t, op)
+		context.RemoteAsk(&goaktpb.Address{
+			Host: "127.0.0.1",
+			Port: int32(remotingPort),
+			Name: actorName2,
+			Id:   "",
+		}, new(testpb.TestReply))
+		require.Error(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -451,10 +440,8 @@ func TestReceiveContext(t *testing.T) {
 		// get the address of the exchanger actor one
 		addr1 := context.RemoteLookup(host, remotingPort, actorName2)
 		// send the message to t exchanger actor one using remote messaging
-		assert.NotPanics(t, func() {
-			context.RemoteTell(addr1, new(testpb.TestRemoteSend))
-		})
-
+		context.RemoteTell(addr1, new(testpb.TestRemoteSend))
+		require.NoError(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -510,15 +497,13 @@ func TestReceiveContext(t *testing.T) {
 		}
 
 		// send the message to the exchanger actor one using remote messaging
-		assert.Panics(t, func() {
-			context.RemoteTell(&goaktpb.Address{
-				Host: "127.0.0.1",
-				Port: int32(remotingPort),
-				Name: actorName2,
-				Id:   "",
-			}, new(testpb.TestRemoteSend))
-		})
-
+		context.RemoteTell(&goaktpb.Address{
+			Host: "127.0.0.1",
+			Port: int32(remotingPort),
+			Name: actorName2,
+			Id:   "",
+		}, new(testpb.TestRemoteSend))
+		require.Error(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -627,11 +612,8 @@ func TestReceiveContext(t *testing.T) {
 			recipient: pid1,
 			async:     true,
 		}
-
-		assert.Panics(t, func() {
-			context.RemoteLookup(host, remotingPort, actorName2)
-		})
-
+		context.RemoteLookup(host, remotingPort, actorName2)
+		require.Error(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -664,9 +646,8 @@ func TestReceiveContext(t *testing.T) {
 			async:     true,
 		}
 
-		assert.NotPanics(t, func() {
-			context.Shutdown()
-		})
+		context.Shutdown()
+		require.NoError(t, context.getError())
 	})
 	t.Run("With successful SpawnChild", func(t *testing.T) {
 		ctx := context.TODO()
@@ -732,9 +713,8 @@ func TestReceiveContext(t *testing.T) {
 		context.Shutdown()
 
 		// create the child actor
-		assert.Panics(t, func() {
-			context.Spawn("SpawnChild", newTestSupervised())
-		})
+		context.Spawn("SpawnChild", newTestSupervised())
+		require.Error(t, context.getError())
 	})
 	t.Run("With not found Child", func(t *testing.T) {
 		ctx := context.TODO()
@@ -768,10 +748,8 @@ func TestReceiveContext(t *testing.T) {
 		// stop the child
 		require.NoError(t, child.Shutdown(ctx))
 
-		assert.Panics(t, func() {
-			context.Child(name)
-		})
-
+		context.Child(name)
+		require.Error(t, context.getError())
 		t.Cleanup(func() {
 			context.Shutdown()
 		})
@@ -808,10 +786,8 @@ func TestReceiveContext(t *testing.T) {
 		// stop the parent
 		context.Shutdown()
 
-		assert.Panics(t, func() {
-			context.Child(name)
-		})
-
+		context.Child(name)
+		require.Error(t, context.getError())
 		t.Cleanup(func() {
 			require.NoError(t, child.Shutdown(ctx))
 		})
@@ -846,10 +822,8 @@ func TestReceiveContext(t *testing.T) {
 		assert.Len(t, context.Children(), 1)
 
 		// stop the child actor
-		assert.NotPanics(t, func() {
-			context.Stop(child)
-		})
-
+		context.Stop(child)
+		require.NoError(t, context.getError())
 		time.Sleep(time.Second)
 		assert.Empty(t, context.Children())
 		t.Cleanup(func() {
@@ -923,10 +897,8 @@ func TestReceiveContext(t *testing.T) {
 
 		time.Sleep(time.Second)
 		// stop the child actor
-		assert.Panics(t, func() {
-			context.Stop(NoSender)
-		})
-
+		context.Stop(NoSender)
+		require.Error(t, context.getError())
 		t.Cleanup(func() {
 			context.Shutdown()
 		})
@@ -966,9 +938,8 @@ func TestReceiveContext(t *testing.T) {
 		time.Sleep(time.Second)
 
 		// stop the child actor
-		assert.Panics(t, func() {
-			context.Stop(child)
-		})
+		context.Stop(child)
+		require.Error(t, context.getError())
 	})
 	t.Run("With failed Stop: actor not found", func(t *testing.T) {
 		ctx := context.TODO()
@@ -1004,10 +975,8 @@ func TestReceiveContext(t *testing.T) {
 		require.NoError(t, err)
 
 		// stop the child actor
-		assert.Panics(t, func() {
-			context.Stop(child)
-		})
-
+		context.Stop(child)
+		require.Error(t, context.getError())
 		t.Cleanup(func() {
 			context.Shutdown()
 			assert.NoError(t, child.Shutdown(ctx))
@@ -1045,10 +1014,8 @@ func TestReceiveContext(t *testing.T) {
 		// stop the child
 		assert.NoError(t, child.Shutdown(ctx))
 		// stop the child actor
-		assert.Panics(t, func() {
-			context.Stop(child)
-		})
-
+		context.Stop(child)
+		require.Error(t, context.getError())
 		time.Sleep(time.Second)
 		assert.Empty(t, context.Children())
 		t.Cleanup(func() {
@@ -1079,9 +1046,8 @@ func TestReceiveContext(t *testing.T) {
 			async:     true,
 		}
 
-		assert.Panics(t, func() {
-			context.Shutdown()
-		})
+		context.Shutdown()
+		require.Error(t, context.getError())
 	})
 	t.Run("With successful Forward", func(t *testing.T) {
 		ctx := context.TODO()
@@ -1358,10 +1324,8 @@ func TestReceiveContext(t *testing.T) {
 		require.NoError(t, err)
 		require.NotNil(t, pid2)
 
-		op := func() {
-			context.BatchTell(pid2, new(testpb.TestSend), new(testpb.TestSend))
-		}
-		assert.NotPanics(t, op)
+		context.BatchTell(pid2, new(testpb.TestSend), new(testpb.TestSend))
+		require.NoError(t, context.getError())
 
 		time.Sleep(time.Second)
 		assert.NoError(t, pid1.Shutdown(ctx))
@@ -1398,10 +1362,8 @@ func TestReceiveContext(t *testing.T) {
 		require.NoError(t, err)
 		require.NotNil(t, pid2)
 
-		op := func() {
-			context.BatchTell(pid2, new(testpb.TestSend))
-		}
-		assert.NotPanics(t, op)
+		context.BatchTell(pid2, new(testpb.TestSend))
+		require.NoError(t, context.getError())
 
 		time.Sleep(time.Second)
 		assert.NoError(t, pid1.Shutdown(ctx))
@@ -1442,10 +1404,8 @@ func TestReceiveContext(t *testing.T) {
 		time.Sleep(time.Second)
 		assert.NoError(t, pid2.Shutdown(ctx))
 
-		op := func() {
-			context.BatchTell(pid2, new(testpb.TestSend), new(testpb.TestSend))
-		}
-		assert.Panics(t, op)
+		context.BatchTell(pid2, new(testpb.TestSend), new(testpb.TestSend))
+		require.Error(t, context.getError())
 
 		time.Sleep(time.Second)
 		assert.NoError(t, pid1.Shutdown(ctx))
@@ -1528,10 +1488,8 @@ func TestReceiveContext(t *testing.T) {
 		time.Sleep(time.Second)
 		assert.NoError(t, pid2.Shutdown(ctx))
 
-		op := func() {
-			context.BatchAsk(pid2, new(testpb.TestReply), new(testpb.TestReply))
-		}
-		assert.Panics(t, op)
+		context.BatchAsk(pid2, new(testpb.TestReply), new(testpb.TestReply))
+		require.Error(t, context.getError())
 
 		time.Sleep(time.Second)
 		assert.NoError(t, pid1.Shutdown(ctx))
@@ -1577,10 +1535,8 @@ func TestReceiveContext(t *testing.T) {
 		// get the address of the exchanger actor one
 		testerAddr := context.RemoteLookup(host, remotingPort, tester)
 		// send the message to t exchanger actor one using remote messaging
-		assert.NotPanics(t, func() {
-			context.RemoteBatchTell(testerAddr, new(testpb.TestSend), new(testpb.TestSend), new(testpb.TestSend))
-		})
-
+		context.RemoteBatchTell(testerAddr, new(testpb.TestSend), new(testpb.TestSend), new(testpb.TestSend))
+		require.NoError(t, context.getError())
 		// wait for processing to complete on the actor side
 		time.Sleep(500 * time.Millisecond)
 		require.EqualValues(t, 3, testActor.counter.Load())
@@ -1633,10 +1589,8 @@ func TestReceiveContext(t *testing.T) {
 		// get the address of the exchanger actor one
 		testerAddr := context.RemoteLookup(host, remotingPort, tester)
 		// send the message to t exchanger actor one using remote messaging
-		var replies []*anypb.Any
-		assert.NotPanics(t, func() {
-			replies = context.RemoteBatchAsk(testerAddr, new(testpb.TestReply), new(testpb.TestReply), new(testpb.TestReply))
-		})
+		replies := context.RemoteBatchAsk(testerAddr, new(testpb.TestReply), new(testpb.TestReply), new(testpb.TestReply))
+		require.NoError(t, context.getError())
 		require.Len(t, replies, 3)
 		time.Sleep(time.Second)
 
@@ -1693,15 +1647,13 @@ func TestReceiveContext(t *testing.T) {
 		}
 
 		// send the message to the exchanger actor one using remote messaging
-		assert.Panics(t, func() {
-			context.RemoteBatchTell(&goaktpb.Address{
-				Host: "127.0.0.1",
-				Port: int32(remotingPort),
-				Name: actorName2,
-				Id:   "",
-			}, new(testpb.TestRemoteSend))
-		})
-
+		context.RemoteBatchTell(&goaktpb.Address{
+			Host: "127.0.0.1",
+			Port: int32(remotingPort),
+			Name: actorName2,
+			Id:   "",
+		}, new(testpb.TestRemoteSend))
+		require.Error(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -1756,16 +1708,13 @@ func TestReceiveContext(t *testing.T) {
 			async:     true,
 		}
 
-		op := func() {
-			context.RemoteBatchAsk(&goaktpb.Address{
-				Host: "127.0.0.1",
-				Port: int32(remotingPort),
-				Name: actorName2,
-				Id:   "",
-			}, new(testpb.TestReply))
-		}
-
-		assert.Panics(t, op)
+		context.RemoteBatchAsk(&goaktpb.Address{
+			Host: "127.0.0.1",
+			Port: int32(remotingPort),
+			Name: actorName2,
+			Id:   "",
+		}, new(testpb.TestReply))
+		require.Error(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -1820,10 +1769,8 @@ func TestReceiveContext(t *testing.T) {
 			async:     true,
 		}
 
-		assert.NotPanics(t, func() {
-			context.RemoteReSpawn(host, remotingPort, actorName2)
-		})
-
+		context.RemoteReSpawn(host, remotingPort, actorName2)
+		require.NoError(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -1877,10 +1824,8 @@ func TestReceiveContext(t *testing.T) {
 			async:     true,
 		}
 
-		assert.Panics(t, func() {
-			context.RemoteReSpawn(host, remotingPort, actorName2)
-		})
-
+		context.RemoteReSpawn(host, remotingPort, actorName2)
+		require.Error(t, context.getError())
 		time.Sleep(time.Second)
 
 		t.Cleanup(func() {
@@ -1990,10 +1935,8 @@ func TestReceiveContext(t *testing.T) {
 			async:     true,
 		}
 
-		assert.Panics(t, func() {
-			messageContext.PipeTo(pid2, nil)
-		})
-
+		messageContext.PipeTo(pid2, nil)
+		require.Error(t, messageContext.getError())
 		assert.NoError(t, pid1.Shutdown(ctx))
 		assert.NoError(t, pid2.Shutdown(ctx))
 	})
diff --git a/internal/eventstream/stream_test.go b/internal/eventstream/stream_test.go
index 1784707..4adf40f 100644
--- a/internal/eventstream/stream_test.go
+++ b/internal/eventstream/stream_test.go
@@ -60,6 +60,7 @@ func TestBroker(t *testing.T) {
 	t.Run("With Unsubscription", func(t *testing.T) {
 		broker := New()
 
+		require.Empty(t, broker.Subscribers())
 		// add consumer
 		cons := broker.AddSubscriber()
 		require.NotNil(t, cons)
diff --git a/internal/slices/chunk_test.go b/internal/slice/chunk_test.go
similarity index 99%
rename from internal/slices/chunk_test.go
rename to internal/slice/chunk_test.go
index 17b3003..3e39c15 100644
--- a/internal/slices/chunk_test.go
+++ b/internal/slice/chunk_test.go
@@ -23,7 +23,7 @@
  *
  */
 
-package slices
+package slice
 
 import (
 	"testing"
diff --git a/internal/slices/slices_test.go b/internal/slice/slice_test.go
similarity index 92%
rename from internal/slices/slices_test.go
rename to internal/slice/slice_test.go
index 9f547b4..37df905 100644
--- a/internal/slices/slices_test.go
+++ b/internal/slice/slice_test.go
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package slices
+package slice
 
 import (
 	"testing"
@@ -30,9 +30,9 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestNewThreadSafeSlice(t *testing.T) {
+func TestSlice(t *testing.T) {
 	// create a concurrent slice of integer
-	sl := NewSafe[int]()
+	sl := New[int]()
 
 	// add some items
 	sl.Append(2)
@@ -41,6 +41,8 @@ func TestNewThreadSafeSlice(t *testing.T) {
 
 	// assert the length
 	assert.EqualValues(t, 3, sl.Len())
+	assert.NotEmpty(t, sl.Items())
+	assert.Len(t, sl.Items(), 3)
 	// get the element at index 2
 	assert.EqualValues(t, 5, sl.Get(2))
 	// remove the element at index 1